feat(FN-3330): add wait_for_completion polling for research runs in extensi
Adds `wait_for_completion` polling support for research runs in the CLI extension, along with tests covering the research status schema and wait behavior, and ships a patch changeset for `@runfusion/fusion`. Fusion-Task-Id: FN-3330
This commit is contained in:
5
.changeset/fix-research-list-status-enum.md
Normal file
5
.changeset/fix-research-list-status-enum.md
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
---
|
||||||
|
"@runfusion/fusion": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
Fix fn_research_list status enum to include all valid ResearchRunStatus values and add wait_for_completion support to fn_research_run.
|
||||||
@@ -370,6 +370,8 @@ Start a bounded research run and optionally wait for findings.
|
|||||||
| Parameter | Type | Required | Description |
|
| Parameter | Type | Required | Description |
|
||||||
|-----------|------|----------|-------------|
|
|-----------|------|----------|-------------|
|
||||||
| `query` | string | ✓ | Research query or question |
|
| `query` | string | ✓ | Research query or question |
|
||||||
|
| `wait_for_completion` | boolean | — | Wait for the run to complete before returning (default: false) |
|
||||||
|
| `max_wait_ms` | number | — | Max wait time when wait_for_completion=true (default: 90000, capped by settings) |
|
||||||
|
|
||||||
### fn_research_list
|
### fn_research_list
|
||||||
|
|
||||||
@@ -377,7 +379,7 @@ List recent research runs.
|
|||||||
|
|
||||||
| Parameter | Type | Required | Description |
|
| Parameter | Type | Required | Description |
|
||||||
|-----------|------|----------|-------------|
|
|-----------|------|----------|-------------|
|
||||||
| `status` | string(enum) | — | |
|
| `status` | string(enum) | — | Filter by run status |
|
||||||
| `limit` | number | — | Max runs to return (default: 10) |
|
| `limit` | number | — | Max runs to return (default: 10) |
|
||||||
|
|
||||||
### fn_research_get
|
### fn_research_get
|
||||||
|
|||||||
@@ -27,7 +27,7 @@ vi.mock("../commands/task.js", () => ({
|
|||||||
}));
|
}));
|
||||||
|
|
||||||
import kbExtension from "../extension.js";
|
import kbExtension from "../extension.js";
|
||||||
import { TaskStore, AgentStore } from "@fusion/core";
|
import { TaskStore, AgentStore, RESEARCH_RUN_STATUSES } from "@fusion/core";
|
||||||
import { isGhAvailable, isGhAuthenticated, runGhJsonAsync } from "@fusion/core/gh-cli";
|
import { isGhAvailable, isGhAuthenticated, runGhJsonAsync } from "@fusion/core/gh-cli";
|
||||||
import { runTaskPlan } from "../commands/task.js";
|
import { runTaskPlan } from "../commands/task.js";
|
||||||
|
|
||||||
@@ -116,6 +116,23 @@ async function removeDirWithRetries(path: string) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async function enableResearch(cwd: string): Promise<TaskStore> {
|
||||||
|
const store = new TaskStore(cwd);
|
||||||
|
await store.init();
|
||||||
|
await store.updateGlobalSettings({
|
||||||
|
researchGlobalEnabled: true,
|
||||||
|
researchGlobalDefaults: { searchProvider: "searxng" },
|
||||||
|
researchSearxngUrl: "http://localhost:8888",
|
||||||
|
});
|
||||||
|
await store.updateSettings({
|
||||||
|
researchEnabled: true,
|
||||||
|
researchSettings: { enabled: true, searchProvider: "searxng" },
|
||||||
|
researchWebSearchProvider: "searxng",
|
||||||
|
researchSearxngUrl: "http://localhost:8888",
|
||||||
|
});
|
||||||
|
return store;
|
||||||
|
}
|
||||||
|
|
||||||
// ── Tests ──────────────────────────────────────────────────────────
|
// ── Tests ──────────────────────────────────────────────────────────
|
||||||
|
|
||||||
// Audited in FN-3189: this suite is expensive (~62s) and currently stale
|
// Audited in FN-3189: this suite is expensive (~62s) and currently stale
|
||||||
@@ -1327,6 +1344,61 @@ describe("fn pi extension (runnable structured-output regression slice)", () =>
|
|||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("research tools", () => {
|
||||||
|
it("fn_research_list status parameter matches RESEARCH_RUN_STATUSES", () => {
|
||||||
|
const tool = api.tools.get("fn_research_list") as any;
|
||||||
|
const statusSchema = tool.parameters.properties.status;
|
||||||
|
const enumValues = statusSchema.enum ?? statusSchema.anyOf?.[0]?.enum;
|
||||||
|
expect(enumValues).toEqual([...RESEARCH_RUN_STATUSES]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("fn_research_run preserves fire-and-forget behavior when wait_for_completion is false", async () => {
|
||||||
|
await enableResearch(tmpDir);
|
||||||
|
const tool = api.tools.get("fn_research_run")!;
|
||||||
|
|
||||||
|
const result = await tool.execute(
|
||||||
|
"research-run-ff",
|
||||||
|
{ query: "test query", wait_for_completion: false },
|
||||||
|
undefined,
|
||||||
|
undefined,
|
||||||
|
makeCtx(tmpDir),
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(result.content[0].text).toContain("Start the project engine to process pending runs");
|
||||||
|
expect(result.details.status).toBe("queued");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("fn_research_run waits and returns terminal run details when wait_for_completion is true", async () => {
|
||||||
|
const store = await enableResearch(tmpDir);
|
||||||
|
const tool = api.tools.get("fn_research_run")!;
|
||||||
|
const researchStore = store.getResearchStore();
|
||||||
|
|
||||||
|
setTimeout(() => {
|
||||||
|
const queuedRun = researchStore.listRuns({ limit: 1 })[0];
|
||||||
|
if (!queuedRun) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
researchStore.updateRun(queuedRun.id, { status: "running" });
|
||||||
|
researchStore.updateRun(queuedRun.id, {
|
||||||
|
status: "completed",
|
||||||
|
results: { summary: "done", findings: [{ heading: "h1", content: "f1", sources: [] }], citations: [] },
|
||||||
|
});
|
||||||
|
}, 25);
|
||||||
|
|
||||||
|
const result = await tool.execute(
|
||||||
|
"research-run-wait",
|
||||||
|
{ query: "terminal query", wait_for_completion: true, max_wait_ms: 4000 },
|
||||||
|
undefined,
|
||||||
|
undefined,
|
||||||
|
makeCtx(tmpDir),
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(result.details.status).toBe("completed");
|
||||||
|
expect(result.details.summary).toBe("done");
|
||||||
|
expect(result.content[0].text).toContain("is completed");
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe("fn_delegate_task", () => {
|
describe("fn_delegate_task", () => {
|
||||||
it("delegates task to agent", async () => {
|
it("delegates task to agent", async () => {
|
||||||
const agentId = await seedAgent(tmpDir, { name: "delegate-target" });
|
const agentId = await seedAgent(tmpDir, { name: "delegate-target" });
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ import {
|
|||||||
type InsightRunTrigger,
|
type InsightRunTrigger,
|
||||||
type ResearchRun,
|
type ResearchRun,
|
||||||
type ResearchRunStatus,
|
type ResearchRunStatus,
|
||||||
|
RESEARCH_RUN_STATUSES,
|
||||||
resolveResearchSettings,
|
resolveResearchSettings,
|
||||||
} from "@fusion/core";
|
} from "@fusion/core";
|
||||||
import {
|
import {
|
||||||
@@ -160,6 +161,14 @@ async function getResearchAvailability(store: TaskStore): Promise<{ ok: boolean;
|
|||||||
return { ok: true };
|
return { ok: true };
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const RESEARCH_RUN_TERMINAL_STATUSES = new Set<ResearchRunStatus>([
|
||||||
|
"completed",
|
||||||
|
"failed",
|
||||||
|
"cancelled",
|
||||||
|
"timed_out",
|
||||||
|
"retry_exhausted",
|
||||||
|
]);
|
||||||
|
|
||||||
function toResearchRunDetails(run: ResearchRun) {
|
function toResearchRunDetails(run: ResearchRun) {
|
||||||
return {
|
return {
|
||||||
runId: run.id,
|
runId: run.id,
|
||||||
@@ -174,6 +183,35 @@ function toResearchRunDetails(run: ResearchRun) {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function isResearchRunTerminal(status: ResearchRunStatus): boolean {
|
||||||
|
return RESEARCH_RUN_TERMINAL_STATUSES.has(status);
|
||||||
|
}
|
||||||
|
|
||||||
|
async function sleepWithSignal(ms: number, signal?: AbortSignal): Promise<void> {
|
||||||
|
if (ms <= 0) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
await new Promise<void>((resolve, reject) => {
|
||||||
|
const timer = setTimeout(() => {
|
||||||
|
signal?.removeEventListener("abort", onAbort);
|
||||||
|
resolve();
|
||||||
|
}, ms);
|
||||||
|
|
||||||
|
const onAbort = () => {
|
||||||
|
clearTimeout(timer);
|
||||||
|
reject(new Error("Operation aborted"));
|
||||||
|
};
|
||||||
|
|
||||||
|
if (signal?.aborted) {
|
||||||
|
onAbort();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
signal?.addEventListener("abort", onAbort, { once: true });
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
interface GitHubIssueApiResult {
|
interface GitHubIssueApiResult {
|
||||||
number: number;
|
number: number;
|
||||||
title: string;
|
title: string;
|
||||||
@@ -1247,8 +1285,10 @@ export default function kbExtension(pi: ExtensionAPI) {
|
|||||||
description: "Start a bounded research run and optionally wait for findings.",
|
description: "Start a bounded research run and optionally wait for findings.",
|
||||||
parameters: Type.Object({
|
parameters: Type.Object({
|
||||||
query: Type.String({ description: "Research query or question" }),
|
query: Type.String({ description: "Research query or question" }),
|
||||||
|
wait_for_completion: Type.Optional(Type.Boolean({ description: "Wait for the run to complete before returning (default: false)" })),
|
||||||
|
max_wait_ms: Type.Optional(Type.Number({ description: "Max wait time when wait_for_completion=true (default: 90000, capped by settings)" })),
|
||||||
}),
|
}),
|
||||||
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
|
async execute(_toolCallId, params, signal, _onUpdate, ctx) {
|
||||||
const store = await getStore(ctx.cwd);
|
const store = await getStore(ctx.cwd);
|
||||||
const availability = await getResearchAvailability(store);
|
const availability = await getResearchAvailability(store);
|
||||||
if (!availability.ok) {
|
if (!availability.ok) {
|
||||||
@@ -1258,15 +1298,47 @@ export default function kbExtension(pi: ExtensionAPI) {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
const run = store.getResearchStore().createRun({
|
const researchStore = store.getResearchStore();
|
||||||
|
const run = researchStore.createRun({
|
||||||
query: params.query,
|
query: params.query,
|
||||||
topic: params.query,
|
topic: params.query,
|
||||||
providerConfig: {},
|
providerConfig: {},
|
||||||
});
|
});
|
||||||
|
|
||||||
|
if (!params.wait_for_completion) {
|
||||||
|
return {
|
||||||
|
content: [{ type: "text", text: `Created research run ${run.id}. Start the project engine to process pending runs, then use fn_research_get.` }],
|
||||||
|
details: toResearchRunDetails(run),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
const maxWaitMs = Number.isFinite(params.max_wait_ms)
|
||||||
|
? Math.max(0, params.max_wait_ms ?? 90_000)
|
||||||
|
: 90_000;
|
||||||
|
const pollIntervalMs = 2_000;
|
||||||
|
const deadline = Date.now() + maxWaitMs;
|
||||||
|
let latestRun = run;
|
||||||
|
|
||||||
|
while (Date.now() <= deadline) {
|
||||||
|
const current = researchStore.getRun(run.id);
|
||||||
|
if (!current) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
latestRun = current;
|
||||||
|
if (isResearchRunTerminal(current.status)) {
|
||||||
|
return {
|
||||||
|
content: [{ type: "text", text: `Research run ${current.id} is ${current.status}.` }],
|
||||||
|
details: toResearchRunDetails(current),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
await sleepWithSignal(pollIntervalMs, signal);
|
||||||
|
}
|
||||||
|
|
||||||
return {
|
return {
|
||||||
content: [{ type: "text", text: `Created research run ${run.id}. Start the project engine to process pending runs, then use fn_research_get.` }],
|
content: [{ type: "text", text: `Research run ${latestRun.id} is ${latestRun.status}.` }],
|
||||||
details: toResearchRunDetails(run),
|
details: toResearchRunDetails(latestRun),
|
||||||
};
|
};
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
@@ -1276,7 +1348,7 @@ export default function kbExtension(pi: ExtensionAPI) {
|
|||||||
label: "fn: List Research Runs",
|
label: "fn: List Research Runs",
|
||||||
description: "List recent research runs.",
|
description: "List recent research runs.",
|
||||||
parameters: Type.Object({
|
parameters: Type.Object({
|
||||||
status: Type.Optional(StringEnum(["pending", "running", "completed", "failed", "cancelled"]) as unknown as TSchema),
|
status: Type.Optional(StringEnum([...RESEARCH_RUN_STATUSES], { description: "Filter by run status" }) as unknown as TSchema),
|
||||||
limit: Type.Optional(Type.Number({ description: "Max runs to return (default: 10)" })),
|
limit: Type.Optional(Type.Number({ description: "Max runs to return (default: 10)" })),
|
||||||
}),
|
}),
|
||||||
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
|
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
|
||||||
|
|||||||
Reference in New Issue
Block a user