feat(FN-2992): merge fusion/fn-2992
Commits merged: - fix(FN-2992): align ResearchView styles with design tokens - fix(FN-2992): align Research view with standalone layout conventions - test(FN-2992): complete Step 6 — expand orchestration verification - feat(FN-2992): complete Step 5 — wire research settings and engine exports - test(FN-2992): cover provider error classification in step runner - feat(FN-2992): complete Step 4 — implement research step runner - test(FN-2992): add orchestrator lifecycle regression coverage - feat(FN-2992): complete Step 3 — add research orchestrator lifecycle - feat(FN-2992): complete Step 2 — align research store API events - feat(FN-2992): complete Step 1 — define orchestration domain types - feat(FN-2992): complete Step 1 — define orchestration domain types Files changed: packages/core/src/index.ts | 15 + packages/core/src/research-store.ts | 8 +- packages/core/src/research-types.ts | 109 +++++ packages/core/src/settings-schema.ts | 10 + packages/core/src/types.ts | 31 ++ packages/dashboard/app/components/ResearchView.css | 87 +++- packages/dashboard/app/components/ResearchView.tsx | 144 +++--- .../app/components/__tests__/ResearchView.test.tsx | 3 +- .../src/__tests__/research-orchestrator.test.ts | 256 ++++++++++ .../src/__tests__/research-step-runner.test.ts | 102 ++++ packages/engine/src/index.ts | 11 + packages/engine/src/project-engine.ts | 17 + packages/engine/src/research-orchestrator.ts | 517 +++++++++++++++++++++ packages/engine/src/research-step-runner.ts | 235 ++++++++++ 14 files changed, 1464 insertions(+), 81 deletions(-) Fusion-Task-Id: FN-2992
This commit is contained in:
Fusion
256
packages/engine/src/__tests__/research-orchestrator.test.ts
Normal file
256
packages/engine/src/__tests__/research-orchestrator.test.ts
Normal file
@@ -0,0 +1,256 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import type { ResearchRun, ResearchSource } from "@fusion/core";
|
||||
import { ResearchOrchestrator } from "../research-orchestrator.js";
|
||||
|
||||
function createHarness() {
|
||||
const runs = new Map<string, ResearchRun>();
|
||||
const counter = { value: 0 };
|
||||
|
||||
const store = {
|
||||
createRun: vi.fn((input: { query: string; providerConfig?: Record<string, unknown>; metadata?: Record<string, unknown> }) => {
|
||||
const id = `RR-test-${++counter.value}`;
|
||||
const now = new Date().toISOString();
|
||||
const run: ResearchRun = {
|
||||
id,
|
||||
query: input.query,
|
||||
status: "pending",
|
||||
providerConfig: input.providerConfig,
|
||||
sources: [],
|
||||
events: [],
|
||||
tags: [],
|
||||
metadata: input.metadata,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
};
|
||||
runs.set(id, run);
|
||||
return run;
|
||||
}),
|
||||
getRun: vi.fn((id: string) => runs.get(id)),
|
||||
updateRun: vi.fn((id: string, patch: Partial<ResearchRun>) => {
|
||||
const run = runs.get(id);
|
||||
if (!run) return undefined;
|
||||
const next = { ...run, ...patch, updatedAt: new Date().toISOString() };
|
||||
runs.set(id, next);
|
||||
return next;
|
||||
}),
|
||||
addEvent: vi.fn((id: string, event: { type: string; message: string; metadata?: Record<string, unknown> }) => {
|
||||
const run = runs.get(id);
|
||||
if (!run) throw new Error("missing run");
|
||||
run.events.push({
|
||||
id: `evt-${run.events.length + 1}`,
|
||||
timestamp: new Date().toISOString(),
|
||||
type: event.type as never,
|
||||
message: event.message,
|
||||
metadata: event.metadata,
|
||||
});
|
||||
return run.events.at(-1)!;
|
||||
}),
|
||||
addSource: vi.fn((id: string, source: Omit<ResearchSource, "id">) => {
|
||||
const run = runs.get(id);
|
||||
if (!run) throw new Error("missing run");
|
||||
const created: ResearchSource = { ...source, id: `src-${run.sources.length + 1}` };
|
||||
run.sources.push(created);
|
||||
return created;
|
||||
}),
|
||||
updateSource: vi.fn((id: string, sourceId: string, patch: Partial<ResearchSource>) => {
|
||||
const run = runs.get(id);
|
||||
if (!run) throw new Error("missing run");
|
||||
run.sources = run.sources.map((s) => (s.id === sourceId ? { ...s, ...patch } : s));
|
||||
}),
|
||||
setResults: vi.fn((id: string, results: ResearchRun["results"]) => {
|
||||
const run = runs.get(id);
|
||||
if (!run) throw new Error("missing run");
|
||||
run.results = results;
|
||||
}),
|
||||
updateStatus: vi.fn((id: string, status: ResearchRun["status"], extra?: Partial<ResearchRun>) => {
|
||||
const run = runs.get(id);
|
||||
if (!run) throw new Error("missing run");
|
||||
runs.set(id, { ...run, ...extra, status });
|
||||
}),
|
||||
};
|
||||
|
||||
const stepRunner = {
|
||||
runSourceQuery: vi.fn(async () => ({ ok: true, data: [{ type: "web", reference: "https://example.com", status: "pending" }] })),
|
||||
runContentFetch: vi.fn(async () => ({ ok: true, data: { content: "body", metadata: { lang: "en" } } })),
|
||||
runSynthesis: vi.fn(async () => ({ ok: true, data: { output: "summary", citations: ["src-1"], confidence: 0.9 } })),
|
||||
};
|
||||
|
||||
return { store, stepRunner, runs };
|
||||
}
|
||||
|
||||
describe("ResearchOrchestrator", () => {
|
||||
it("runs full lifecycle and completes", async () => {
|
||||
const { store, stepRunner } = createHarness();
|
||||
const orchestrator = new ResearchOrchestrator({
|
||||
store: store as never,
|
||||
stepRunner: stepRunner as never,
|
||||
maxConcurrentRuns: 2,
|
||||
});
|
||||
|
||||
const runId = orchestrator.createRun({
|
||||
providers: [{ type: "web" }],
|
||||
maxSources: 2,
|
||||
maxSynthesisRounds: 1,
|
||||
});
|
||||
|
||||
const run = await orchestrator.startRun(runId, "fusion research");
|
||||
expect(run.status).toBe("completed");
|
||||
expect(stepRunner.runSourceQuery).toHaveBeenCalledTimes(1);
|
||||
expect(stepRunner.runContentFetch).toHaveBeenCalledTimes(1);
|
||||
expect(stepRunner.runSynthesis).toHaveBeenCalledTimes(1);
|
||||
|
||||
const status = orchestrator.getRunStatus(runId);
|
||||
expect(status.phase).toBe("completed");
|
||||
});
|
||||
|
||||
it("cancels a running run", async () => {
|
||||
const { store, stepRunner } = createHarness();
|
||||
stepRunner.runSourceQuery.mockImplementation(
|
||||
(async (_query: string, _provider: string, _config: unknown, signal?: AbortSignal) => {
|
||||
await new Promise((resolve, reject) => {
|
||||
const timer = setTimeout(resolve, 100);
|
||||
signal?.addEventListener("abort", () => {
|
||||
clearTimeout(timer);
|
||||
reject(new Error("aborted"));
|
||||
});
|
||||
});
|
||||
return { ok: true, data: [] };
|
||||
}) as never,
|
||||
);
|
||||
|
||||
const orchestrator = new ResearchOrchestrator({
|
||||
store: store as never,
|
||||
stepRunner: stepRunner as never,
|
||||
maxConcurrentRuns: 1,
|
||||
});
|
||||
|
||||
const runId = orchestrator.createRun({
|
||||
providers: [{ type: "web" }],
|
||||
maxSources: 2,
|
||||
maxSynthesisRounds: 1,
|
||||
});
|
||||
|
||||
const runPromise = orchestrator.startRun(runId, "cancel me");
|
||||
await Promise.resolve();
|
||||
expect(orchestrator.cancelRun(runId)).toBe(true);
|
||||
|
||||
const run = await runPromise;
|
||||
expect(run.status).toBe("cancelled");
|
||||
});
|
||||
|
||||
it("records step failures and continues when later providers succeed", async () => {
|
||||
const { store, stepRunner } = createHarness();
|
||||
stepRunner.runSourceQuery
|
||||
.mockResolvedValueOnce({ ok: false, error: { code: "provider_error", message: "provider down" } } as never)
|
||||
.mockResolvedValueOnce({ ok: true, data: [{ type: "web", reference: "https://backup.com", status: "pending" }] } as never);
|
||||
|
||||
const orchestrator = new ResearchOrchestrator({
|
||||
store: store as never,
|
||||
stepRunner: stepRunner as never,
|
||||
maxConcurrentRuns: 2,
|
||||
});
|
||||
|
||||
const runId = orchestrator.createRun({
|
||||
providers: [{ type: "primary" }, { type: "backup" }],
|
||||
maxSources: 2,
|
||||
maxSynthesisRounds: 1,
|
||||
});
|
||||
|
||||
const run = await orchestrator.startRun(runId, "fallback query");
|
||||
expect(run.status).toBe("completed");
|
||||
expect(store.addEvent).toHaveBeenCalledWith(
|
||||
runId,
|
||||
expect.objectContaining({
|
||||
type: "error",
|
||||
metadata: expect.objectContaining({ orchestrationEventType: "step-failed" }),
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("emits step-failed for timeout-classified step errors", async () => {
|
||||
const { store, stepRunner } = createHarness();
|
||||
stepRunner.runSourceQuery
|
||||
.mockResolvedValueOnce({ ok: false, error: { code: "timeout", message: "search timed out" } } as never)
|
||||
.mockResolvedValueOnce({ ok: true, data: [{ type: "web", reference: "https://backup.com", status: "pending" }] } as never);
|
||||
|
||||
const orchestrator = new ResearchOrchestrator({
|
||||
store: store as never,
|
||||
stepRunner: stepRunner as never,
|
||||
maxConcurrentRuns: 1,
|
||||
});
|
||||
|
||||
const runId = orchestrator.createRun({
|
||||
providers: [{ type: "slow" }, { type: "backup" }],
|
||||
maxSources: 1,
|
||||
maxSynthesisRounds: 1,
|
||||
});
|
||||
|
||||
await orchestrator.startRun(runId, "timeout query");
|
||||
expect(store.addEvent).toHaveBeenCalledWith(
|
||||
runId,
|
||||
expect.objectContaining({
|
||||
type: "error",
|
||||
message: expect.stringContaining("failed"),
|
||||
metadata: expect.objectContaining({ orchestrationEventType: "step-failed" }),
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("respects max concurrent run limit", async () => {
|
||||
const { store, stepRunner } = createHarness();
|
||||
let releaseFirst: (() => void) | undefined;
|
||||
const firstBlocked = new Promise<void>((resolve) => {
|
||||
releaseFirst = resolve;
|
||||
});
|
||||
|
||||
stepRunner.runSourceQuery.mockImplementationOnce(async () => {
|
||||
await firstBlocked;
|
||||
return { ok: true, data: [{ type: "web", reference: "https://example.com/a", status: "pending" }] };
|
||||
});
|
||||
|
||||
const orchestrator = new ResearchOrchestrator({
|
||||
store: store as never,
|
||||
stepRunner: stepRunner as never,
|
||||
maxConcurrentRuns: 1,
|
||||
});
|
||||
|
||||
const runA = orchestrator.createRun({ providers: [{ type: "web" }], maxSources: 1, maxSynthesisRounds: 1 });
|
||||
const runB = orchestrator.createRun({ providers: [{ type: "web" }], maxSources: 1, maxSynthesisRounds: 1 });
|
||||
|
||||
const p1 = orchestrator.startRun(runA, "A");
|
||||
await Promise.resolve();
|
||||
const p2 = orchestrator.startRun(runB, "B");
|
||||
|
||||
await Promise.resolve();
|
||||
expect(stepRunner.runSourceQuery).toHaveBeenCalledTimes(1);
|
||||
|
||||
releaseFirst?.();
|
||||
await p1;
|
||||
await p2;
|
||||
expect(stepRunner.runSourceQuery).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("retries failed run with inherited config", () => {
|
||||
const { store } = createHarness();
|
||||
const orchestrator = new ResearchOrchestrator({
|
||||
store: store as never,
|
||||
stepRunner: {
|
||||
runSourceQuery: vi.fn(),
|
||||
runContentFetch: vi.fn(),
|
||||
runSynthesis: vi.fn(),
|
||||
},
|
||||
});
|
||||
|
||||
const baseId = orchestrator.createRun({
|
||||
providers: [{ type: "web", config: { timeoutMs: 1000 } }],
|
||||
maxSources: 1,
|
||||
maxSynthesisRounds: 1,
|
||||
});
|
||||
store.updateStatus(baseId, "failed", { error: "boom" });
|
||||
|
||||
const retryId = orchestrator.retryRun(baseId);
|
||||
expect(retryId).not.toBe(baseId);
|
||||
const retried = store.getRun(retryId)!;
|
||||
expect(retried.metadata?.retryOfRunId).toBe(baseId);
|
||||
});
|
||||
});
|
||||
102
packages/engine/src/__tests__/research-step-runner.test.ts
Normal file
102
packages/engine/src/__tests__/research-step-runner.test.ts
Normal file
@@ -0,0 +1,102 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { ResearchStepRunner } from "../research-step-runner.js";
|
||||
|
||||
describe("ResearchStepRunner", () => {
|
||||
it("returns provider_not_configured when provider missing", async () => {
|
||||
const runner = new ResearchStepRunner();
|
||||
const result = await runner.runSourceQuery("hello", "web");
|
||||
expect(result.ok).toBe(false);
|
||||
expect(result.error?.code).toBe("provider_not_configured");
|
||||
});
|
||||
|
||||
it("classifies timeout errors", async () => {
|
||||
const provider = {
|
||||
type: "web",
|
||||
isConfigured: () => true,
|
||||
search: async () => {
|
||||
await new Promise((resolve) => setTimeout(resolve, 25));
|
||||
return [];
|
||||
},
|
||||
fetchContent: async () => ({ content: "", metadata: {} }),
|
||||
};
|
||||
|
||||
const runner = new ResearchStepRunner({ providers: [provider] });
|
||||
const result = await runner.runSourceQuery("q", "web", { timeoutMs: 1 });
|
||||
expect(result.ok).toBe(false);
|
||||
expect(result.error?.code).toBe("timeout");
|
||||
});
|
||||
|
||||
it("classifies provider errors", async () => {
|
||||
const provider = {
|
||||
type: "web",
|
||||
isConfigured: () => true,
|
||||
search: async () => {
|
||||
throw new Error("rate limit exceeded");
|
||||
},
|
||||
fetchContent: async () => ({ content: "", metadata: {} }),
|
||||
};
|
||||
|
||||
const runner = new ResearchStepRunner({ providers: [provider] });
|
||||
const result = await runner.runSourceQuery("q", "web");
|
||||
expect(result.ok).toBe(false);
|
||||
expect(result.error?.code).toBe("provider_error");
|
||||
expect(result.error?.message).toContain("rate limit exceeded");
|
||||
});
|
||||
|
||||
it("propagates abort signals", async () => {
|
||||
const provider = {
|
||||
type: "web",
|
||||
isConfigured: () => true,
|
||||
search: async (_query: string, _options: unknown, signal?: AbortSignal) => {
|
||||
await new Promise((resolve, reject) => {
|
||||
const timer = setTimeout(resolve, 50);
|
||||
signal?.addEventListener("abort", () => {
|
||||
clearTimeout(timer);
|
||||
reject(new Error("aborted by user"));
|
||||
});
|
||||
});
|
||||
return [];
|
||||
},
|
||||
fetchContent: async () => ({ content: "", metadata: {} }),
|
||||
};
|
||||
|
||||
const runner = new ResearchStepRunner({ providers: [provider] });
|
||||
const ac = new AbortController();
|
||||
const promise = runner.runSourceQuery("q", "web", { timeoutMs: 3000 }, ac.signal);
|
||||
ac.abort();
|
||||
|
||||
const result = await promise;
|
||||
expect(result.ok).toBe(false);
|
||||
expect(result.error?.code).toBe("aborted");
|
||||
});
|
||||
|
||||
it("returns provider_not_configured for content fetch without configured providers", async () => {
|
||||
const runner = new ResearchStepRunner();
|
||||
const result = await runner.runContentFetch("https://example.com");
|
||||
expect(result.ok).toBe(false);
|
||||
expect(result.error?.code).toBe("provider_not_configured");
|
||||
});
|
||||
|
||||
it("returns provider_not_configured for synthesis when no runner configured", async () => {
|
||||
const runner = new ResearchStepRunner();
|
||||
const result = await runner.runSynthesis({ query: "q", sources: [], round: 1 });
|
||||
expect(result.ok).toBe(false);
|
||||
expect(result.error?.code).toBe("provider_not_configured");
|
||||
});
|
||||
|
||||
it("classifies synthesis timeout", async () => {
|
||||
const runner = new ResearchStepRunner({
|
||||
synthesisRunner: async () => {
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
return { output: "done", citations: [] };
|
||||
},
|
||||
});
|
||||
|
||||
const result = await runner.runSynthesis(
|
||||
{ query: "q", sources: [], round: 1 },
|
||||
{ timeoutMs: 1 },
|
||||
);
|
||||
expect(result.ok).toBe(false);
|
||||
expect(result.error?.code).toBe("timeout");
|
||||
});
|
||||
});
|
||||
@@ -53,6 +53,17 @@ export { WorktreePool, scanIdleWorktrees, cleanupOrphanedWorktrees, reapOrphanWo
|
||||
export { createLogger, type Logger } from "./logger.js";
|
||||
export { isUsageLimitError, UsageLimitPauser } from "./usage-limit-detector.js";
|
||||
export { withRateLimitRetry } from "./rate-limit-retry.js";
|
||||
export { ResearchOrchestrator, type ResearchOrchestratorOptions, type ResearchOrchestratorStatus, type ResearchOrchestratorStartOptions } from "./research-orchestrator.js";
|
||||
export {
|
||||
ResearchStepRunner,
|
||||
ResearchStepTimeoutError,
|
||||
ResearchStepAbortError,
|
||||
ResearchStepProviderError,
|
||||
type ResearchProvider,
|
||||
type ResearchStepRunnerApi,
|
||||
type ResearchStepRunnerOptions,
|
||||
type ResearchStepResult,
|
||||
} from "./research-step-runner.js";
|
||||
export { PrMonitor, type PrComment, type TrackedPr, type OnNewCommentsCallback } from "./pr-monitor.js";
|
||||
export { PrCommentHandler } from "./pr-comment-handler.js";
|
||||
export {
|
||||
|
||||
@@ -24,6 +24,8 @@ import { aiMergeTask } from "./merger.js";
|
||||
import { PRIORITY_MERGE } from "./concurrency.js";
|
||||
import { runtimeLog } from "./logger.js";
|
||||
import type { HeartbeatTriggerScheduler } from "./agent-heartbeat.js";
|
||||
import { ResearchOrchestrator } from "./research-orchestrator.js";
|
||||
import { ResearchStepRunner } from "./research-step-runner.js";
|
||||
import { TunnelProcessManager } from "./remote-access/tunnel-process-manager.js";
|
||||
import type {
|
||||
TunnelProvider,
|
||||
@@ -142,6 +144,7 @@ export class ProjectEngine {
|
||||
private gridlockDetector?: GridlockDetector;
|
||||
private cronRunner?: CronRunner;
|
||||
private automationStore?: AutomationStoreType;
|
||||
private researchOrchestrator?: ResearchOrchestrator;
|
||||
private remoteTunnelManager?: TunnelProcessManager;
|
||||
private remoteTunnelRestoreDiagnostics: TunnelRestoreDiagnostics = {
|
||||
outcome: "skipped",
|
||||
@@ -243,6 +246,15 @@ export class ProjectEngine {
|
||||
|
||||
const store = this.runtime.getTaskStore();
|
||||
const cwd = this.config.workingDirectory;
|
||||
const settings = await store.getSettings();
|
||||
|
||||
if (typeof (store as { getResearchStore?: () => unknown }).getResearchStore === "function") {
|
||||
this.researchOrchestrator = new ResearchOrchestrator({
|
||||
store: store.getResearchStore(),
|
||||
stepRunner: new ResearchStepRunner(),
|
||||
maxConcurrentRuns: settings.researchMaxConcurrentRuns ?? 3,
|
||||
});
|
||||
}
|
||||
|
||||
this.remoteTunnelManager = new TunnelProcessManager();
|
||||
try {
|
||||
@@ -546,6 +558,11 @@ export class ProjectEngine {
|
||||
return this.runtime.getRoutineStore();
|
||||
}
|
||||
|
||||
/** Get the ResearchOrchestrator (if initialized). Returns undefined before start(). */
|
||||
getResearchOrchestrator(): ResearchOrchestrator | undefined {
|
||||
return this.researchOrchestrator;
|
||||
}
|
||||
|
||||
/** Get the remote tunnel manager (available after start()). */
|
||||
getRemoteTunnelManager(): TunnelProcessManager | undefined {
|
||||
return this.remoteTunnelManager;
|
||||
|
||||
517
packages/engine/src/research-orchestrator.ts
Normal file
517
packages/engine/src/research-orchestrator.ts
Normal file
@@ -0,0 +1,517 @@
|
||||
import type { ResearchStore } from "@fusion/core";
|
||||
import type {
|
||||
ResearchCancellationState,
|
||||
ResearchOrchestrationConfig,
|
||||
ResearchOrchestrationPhase,
|
||||
ResearchOrchestrationStep,
|
||||
ResearchRun,
|
||||
ResearchSource,
|
||||
ResearchSynthesisRequest,
|
||||
} from "@fusion/core";
|
||||
import { AgentSemaphore } from "./concurrency.js";
|
||||
import { createLogger, formatError } from "./logger.js";
|
||||
import type { ResearchStepRunnerApi } from "./research-step-runner.js";
|
||||
|
||||
const log = createLogger("research-orchestrator");
|
||||
|
||||
export interface ResearchOrchestratorStatus {
|
||||
runId: string;
|
||||
status: ResearchRun["status"];
|
||||
phase: ResearchOrchestrationPhase;
|
||||
stepIndex: number;
|
||||
totalSteps: number;
|
||||
progress: number;
|
||||
active: boolean;
|
||||
}
|
||||
|
||||
export interface ResearchOrchestratorStartOptions {
|
||||
abortSignal?: AbortSignal;
|
||||
}
|
||||
|
||||
export interface ResearchOrchestratorOptions {
|
||||
store: ResearchStore;
|
||||
stepRunner: ResearchStepRunnerApi;
|
||||
maxConcurrentRuns?: number;
|
||||
}
|
||||
|
||||
interface ActiveRunState {
|
||||
controller: AbortController;
|
||||
phase: ResearchOrchestrationPhase;
|
||||
stepIndex: number;
|
||||
totalSteps: number;
|
||||
config: ResearchOrchestrationConfig;
|
||||
}
|
||||
|
||||
export class ResearchOrchestrator {
|
||||
private readonly store: ResearchStore;
|
||||
private readonly stepRunner: ResearchStepRunnerApi;
|
||||
private readonly semaphore: AgentSemaphore;
|
||||
private readonly activeRuns = new Map<string, ActiveRunState>();
|
||||
private readonly cancellation = new Map<string, ResearchCancellationState>();
|
||||
|
||||
constructor(options: ResearchOrchestratorOptions) {
|
||||
this.store = options.store;
|
||||
this.stepRunner = options.stepRunner;
|
||||
this.semaphore = new AgentSemaphore(options.maxConcurrentRuns ?? 3);
|
||||
}
|
||||
|
||||
createRun(config: ResearchOrchestrationConfig): string {
|
||||
const run = this.store.createRun({
|
||||
query: "",
|
||||
providerConfig: config as unknown as Record<string, unknown>,
|
||||
metadata: {
|
||||
orchestration: {
|
||||
phase: "planning",
|
||||
stepIndex: 0,
|
||||
totalSteps: this.computeTotalSteps(config),
|
||||
},
|
||||
},
|
||||
});
|
||||
return run.id;
|
||||
}
|
||||
|
||||
async startRun(runId: string, query: string, options: ResearchOrchestratorStartOptions = {}): Promise<ResearchRun> {
|
||||
const run = this.store.getRun(runId);
|
||||
if (!run) throw new Error(`Research run not found: ${runId}`);
|
||||
|
||||
const config = (run.providerConfig ?? {}) as unknown as ResearchOrchestrationConfig;
|
||||
const controller = new AbortController();
|
||||
if (options.abortSignal) {
|
||||
options.abortSignal.addEventListener("abort", () => controller.abort(options.abortSignal?.reason), { once: true });
|
||||
}
|
||||
|
||||
const totalSteps = this.computeTotalSteps(config);
|
||||
this.activeRuns.set(runId, {
|
||||
controller,
|
||||
phase: "planning",
|
||||
stepIndex: 0,
|
||||
totalSteps,
|
||||
config,
|
||||
});
|
||||
|
||||
await this.semaphore.run(async () => {
|
||||
this.store.updateRun(runId, { query, status: "running", startedAt: new Date().toISOString(), error: null });
|
||||
await this.runPhases(runId, query, config, controller.signal);
|
||||
});
|
||||
|
||||
const updated = this.store.getRun(runId);
|
||||
if (!updated) throw new Error(`Research run not found after start: ${runId}`);
|
||||
return updated;
|
||||
}
|
||||
|
||||
cancelRun(runId: string): boolean {
|
||||
const active = this.activeRuns.get(runId);
|
||||
if (!active) return false;
|
||||
|
||||
const state: ResearchCancellationState = {
|
||||
runId,
|
||||
controller: active.controller,
|
||||
requestedAt: new Date().toISOString(),
|
||||
gracefulShutdown: true,
|
||||
reason: "Cancelled by user",
|
||||
};
|
||||
this.cancellation.set(runId, state);
|
||||
active.controller.abort(new Error("Research run cancelled"));
|
||||
return true;
|
||||
}
|
||||
|
||||
retryRun(runId: string): string {
|
||||
const run = this.store.getRun(runId);
|
||||
if (!run) throw new Error(`Research run not found: ${runId}`);
|
||||
if (run.status !== "failed" && run.status !== "cancelled") {
|
||||
throw new Error(`Research run ${runId} is not retryable (status=${run.status})`);
|
||||
}
|
||||
|
||||
const next = this.store.createRun({
|
||||
query: run.query,
|
||||
topic: run.topic,
|
||||
providerConfig: run.providerConfig,
|
||||
tags: [...run.tags],
|
||||
metadata: {
|
||||
...(run.metadata ?? {}),
|
||||
retryOfRunId: run.id,
|
||||
},
|
||||
});
|
||||
this.store.addEvent(next.id, {
|
||||
type: "info",
|
||||
message: `Retry run created from ${run.id}`,
|
||||
metadata: { retryOfRunId: run.id },
|
||||
});
|
||||
return next.id;
|
||||
}
|
||||
|
||||
getRunStatus(runId: string): ResearchOrchestratorStatus {
|
||||
const run = this.store.getRun(runId);
|
||||
if (!run) throw new Error(`Research run not found: ${runId}`);
|
||||
|
||||
const active = this.activeRuns.get(runId);
|
||||
const metadata = (run.metadata?.orchestration as Record<string, unknown> | undefined) ?? {};
|
||||
const phase = (active?.phase ?? metadata.phase ?? this.statusToPhase(run.status)) as ResearchOrchestrationPhase;
|
||||
const stepIndex = active?.stepIndex ?? Number(metadata.stepIndex ?? 0);
|
||||
const totalSteps = active?.totalSteps ?? Number(metadata.totalSteps ?? 0);
|
||||
|
||||
return {
|
||||
runId,
|
||||
status: run.status,
|
||||
phase,
|
||||
stepIndex,
|
||||
totalSteps,
|
||||
progress: totalSteps > 0 ? Math.min(1, stepIndex / totalSteps) : 0,
|
||||
active: this.activeRuns.has(runId),
|
||||
};
|
||||
}
|
||||
|
||||
private async runPhases(
|
||||
runId: string,
|
||||
query: string,
|
||||
config: ResearchOrchestrationConfig,
|
||||
signal: AbortSignal,
|
||||
): Promise<void> {
|
||||
try {
|
||||
await this.runPlanning(runId, query, config, signal);
|
||||
const sources = await this.runSearching(runId, query, config, signal);
|
||||
const fetchedSources = await this.runFetching(runId, sources, config, signal);
|
||||
const synthesis = await this.runSynthesis(runId, query, fetchedSources, config, signal);
|
||||
await this.runFinalizing(runId, synthesis.output, synthesis.citations, synthesis.confidence, signal);
|
||||
|
||||
this.store.updateStatus(runId, "completed");
|
||||
this.transitionPhase(runId, "completed", "Research run completed");
|
||||
} catch (err) {
|
||||
if (signal.aborted) {
|
||||
this.onCancelled(runId);
|
||||
} else {
|
||||
const { message, detail } = formatError(err);
|
||||
this.store.addEvent(runId, {
|
||||
type: "error",
|
||||
message: `Research run failed: ${message}`,
|
||||
metadata: { detail },
|
||||
});
|
||||
this.store.updateStatus(runId, "failed", { error: message });
|
||||
this.transitionPhase(runId, "failed", "Research run failed", { error: message });
|
||||
}
|
||||
} finally {
|
||||
this.activeRuns.delete(runId);
|
||||
this.cancellation.delete(runId);
|
||||
}
|
||||
}
|
||||
|
||||
private async runPlanning(runId: string, query: string, config: ResearchOrchestrationConfig, _signal: AbortSignal): Promise<void> {
|
||||
this.transitionPhase(runId, "planning", "Planning research execution");
|
||||
this.stepStarted(runId, {
|
||||
id: `${runId}-planning`,
|
||||
type: "synthesis-pass",
|
||||
phase: "planning",
|
||||
status: "running",
|
||||
order: 0,
|
||||
name: "Create plan",
|
||||
input: { query, providerCount: config.providers.length },
|
||||
startedAt: new Date().toISOString(),
|
||||
});
|
||||
this.stepCompleted(runId, `${runId}-planning`, { query });
|
||||
}
|
||||
|
||||
private async runSearching(
|
||||
runId: string,
|
||||
query: string,
|
||||
config: ResearchOrchestrationConfig,
|
||||
signal: AbortSignal,
|
||||
): Promise<ResearchSource[]> {
|
||||
this.throwIfAborted(signal);
|
||||
this.transitionPhase(runId, "searching", "Searching sources");
|
||||
|
||||
const allSources: ResearchSource[] = [];
|
||||
for (const provider of config.providers) {
|
||||
this.throwIfAborted(signal);
|
||||
const step = this.createStep(runId, "source-query", "searching", `Search with ${provider.type}`, {
|
||||
query,
|
||||
provider: provider.type,
|
||||
});
|
||||
this.stepStarted(runId, step);
|
||||
|
||||
const result = await this.stepRunner.runSourceQuery(query, provider.type, provider.config, signal);
|
||||
if (!result.ok || !result.data) {
|
||||
this.stepFailed(runId, step.id, result.error?.message ?? `Provider ${provider.type} returned no data`, result.error);
|
||||
continue;
|
||||
}
|
||||
|
||||
for (const source of result.data.slice(0, Math.max(0, config.maxSources - allSources.length))) {
|
||||
const saved = this.store.addSource(runId, source);
|
||||
allSources.push(saved);
|
||||
this.store.addEvent(runId, {
|
||||
type: "source_added",
|
||||
message: `Source found: ${saved.reference}`,
|
||||
metadata: { sourceId: saved.id, provider: provider.type },
|
||||
});
|
||||
}
|
||||
|
||||
this.stepCompleted(runId, step.id, { sourceCount: result.data.length });
|
||||
if (allSources.length >= config.maxSources) break;
|
||||
}
|
||||
|
||||
if (allSources.length === 0) {
|
||||
throw new Error("No sources discovered during search phase");
|
||||
}
|
||||
|
||||
return allSources;
|
||||
}
|
||||
|
||||
private async runFetching(
|
||||
runId: string,
|
||||
sources: ResearchSource[],
|
||||
config: ResearchOrchestrationConfig,
|
||||
signal: AbortSignal,
|
||||
): Promise<ResearchSource[]> {
|
||||
this.throwIfAborted(signal);
|
||||
this.transitionPhase(runId, "fetching", "Fetching source content");
|
||||
|
||||
const fetched: ResearchSource[] = [];
|
||||
const provider = config.providers[0];
|
||||
for (const source of sources.slice(0, config.maxSources)) {
|
||||
this.throwIfAborted(signal);
|
||||
const step = this.createStep(runId, "content-fetch", "fetching", `Fetch ${source.reference}`, {
|
||||
sourceId: source.id,
|
||||
});
|
||||
this.stepStarted(runId, step);
|
||||
|
||||
const result = await this.stepRunner.runContentFetch(source.reference, provider?.config, signal);
|
||||
if (!result.ok || !result.data) {
|
||||
this.stepFailed(runId, step.id, result.error?.message ?? "Failed to fetch source content", result.error);
|
||||
continue;
|
||||
}
|
||||
|
||||
const updated: ResearchSource = {
|
||||
...source,
|
||||
content: result.data.content,
|
||||
metadata: {
|
||||
...(source.metadata ?? {}),
|
||||
...(result.data.metadata ?? {}),
|
||||
},
|
||||
status: "completed",
|
||||
fetchedAt: new Date().toISOString(),
|
||||
};
|
||||
this.store.updateSource(runId, source.id, updated);
|
||||
fetched.push(updated);
|
||||
this.stepCompleted(runId, step.id, { fetched: true });
|
||||
}
|
||||
|
||||
if (fetched.length === 0) {
|
||||
throw new Error("No source content fetched");
|
||||
}
|
||||
|
||||
return fetched;
|
||||
}
|
||||
|
||||
private async runSynthesis(
|
||||
runId: string,
|
||||
query: string,
|
||||
sources: ResearchSource[],
|
||||
config: ResearchOrchestrationConfig,
|
||||
signal: AbortSignal,
|
||||
): Promise<{ output: string; citations: string[]; confidence?: number }> {
|
||||
this.throwIfAborted(signal);
|
||||
this.transitionPhase(runId, "synthesizing", "Synthesizing findings");
|
||||
|
||||
let final: { output: string; citations: string[]; confidence?: number } | undefined;
|
||||
|
||||
for (let round = 1; round <= Math.max(1, config.maxSynthesisRounds); round++) {
|
||||
this.throwIfAborted(signal);
|
||||
const step = this.createStep(runId, "synthesis-pass", "synthesizing", `Synthesis round ${round}`, {
|
||||
round,
|
||||
});
|
||||
this.stepStarted(runId, step);
|
||||
|
||||
const request: ResearchSynthesisRequest = {
|
||||
query,
|
||||
sources,
|
||||
round,
|
||||
desiredFormat: "markdown",
|
||||
};
|
||||
const result = await this.stepRunner.runSynthesis(request, config.synthesisModel, signal);
|
||||
if (!result.ok || !result.data) {
|
||||
this.stepFailed(runId, step.id, result.error?.message ?? "Synthesis failed", result.error);
|
||||
continue;
|
||||
}
|
||||
|
||||
final = result.data;
|
||||
this.store.addEvent(runId, {
|
||||
type: "progress",
|
||||
message: `Synthesis round ${round} completed`,
|
||||
metadata: { round, confidence: result.data.confidence },
|
||||
});
|
||||
this.stepCompleted(runId, step.id, { round, citations: result.data.citations.length });
|
||||
}
|
||||
|
||||
if (!final) {
|
||||
throw new Error("All synthesis rounds failed");
|
||||
}
|
||||
|
||||
return final;
|
||||
}
|
||||
|
||||
private async runFinalizing(
|
||||
runId: string,
|
||||
output: string,
|
||||
citations: string[],
|
||||
confidence: number | undefined,
|
||||
signal: AbortSignal,
|
||||
): Promise<void> {
|
||||
this.throwIfAborted(signal);
|
||||
this.transitionPhase(runId, "finalizing", "Finalizing research results");
|
||||
this.store.setResults(runId, {
|
||||
summary: output,
|
||||
findings: [
|
||||
{
|
||||
heading: "Synthesis",
|
||||
content: output,
|
||||
sources: citations,
|
||||
confidence,
|
||||
},
|
||||
],
|
||||
citations,
|
||||
synthesizedOutput: output,
|
||||
});
|
||||
}
|
||||
|
||||
private onCancelled(runId: string): void {
|
||||
const cancellation = this.cancellation.get(runId);
|
||||
this.store.addEvent(runId, {
|
||||
type: "warning",
|
||||
message: "Research run cancelled",
|
||||
metadata: {
|
||||
requestedAt: cancellation?.requestedAt,
|
||||
reason: cancellation?.reason,
|
||||
},
|
||||
});
|
||||
this.store.updateStatus(runId, "cancelled", {
|
||||
cancelledAt: new Date().toISOString(),
|
||||
error: cancellation?.reason,
|
||||
});
|
||||
this.transitionPhase(runId, "cancelled", "Research run cancelled");
|
||||
}
|
||||
|
||||
private transitionPhase(
|
||||
runId: string,
|
||||
phase: ResearchOrchestrationPhase,
|
||||
message: string,
|
||||
metadata?: Record<string, unknown>,
|
||||
): void {
|
||||
const active = this.activeRuns.get(runId);
|
||||
if (active) {
|
||||
active.phase = phase;
|
||||
}
|
||||
this.store.updateRun(runId, {
|
||||
metadata: {
|
||||
orchestration: {
|
||||
phase,
|
||||
stepIndex: active?.stepIndex ?? 0,
|
||||
totalSteps: active?.totalSteps ?? 0,
|
||||
},
|
||||
},
|
||||
});
|
||||
this.store.addEvent(runId, {
|
||||
type: "progress",
|
||||
message,
|
||||
metadata: {
|
||||
orchestrationEventType: "phase-changed",
|
||||
phase,
|
||||
...(metadata ?? {}),
|
||||
},
|
||||
});
|
||||
log.log(`${runId}: phase changed -> ${phase}`);
|
||||
}
|
||||
|
||||
private stepStarted(runId: string, step: ResearchOrchestrationStep): void {
|
||||
this.bumpStep(runId, step.order);
|
||||
this.store.addEvent(runId, {
|
||||
type: "progress",
|
||||
message: `${step.name} started`,
|
||||
metadata: {
|
||||
orchestrationEventType: "step-started",
|
||||
step,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
private stepCompleted(runId: string, stepId: string, output?: Record<string, unknown>): void {
|
||||
this.store.addEvent(runId, {
|
||||
type: "progress",
|
||||
message: `${stepId} completed`,
|
||||
metadata: {
|
||||
orchestrationEventType: "step-completed",
|
||||
stepId,
|
||||
output,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
private stepFailed(
|
||||
runId: string,
|
||||
stepId: string,
|
||||
errorMessage: string,
|
||||
errorMeta?: Record<string, unknown>,
|
||||
): void {
|
||||
this.store.addEvent(runId, {
|
||||
type: "error",
|
||||
message: `${stepId} failed: ${errorMessage}`,
|
||||
metadata: {
|
||||
orchestrationEventType: "step-failed",
|
||||
stepId,
|
||||
...(errorMeta ?? {}),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
private bumpStep(runId: string, stepIndex: number): void {
|
||||
const active = this.activeRuns.get(runId);
|
||||
if (!active) return;
|
||||
active.stepIndex = stepIndex;
|
||||
this.store.updateRun(runId, {
|
||||
metadata: {
|
||||
orchestration: {
|
||||
phase: active.phase,
|
||||
stepIndex: active.stepIndex,
|
||||
totalSteps: active.totalSteps,
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
private createStep(
|
||||
runId: string,
|
||||
type: ResearchOrchestrationStep["type"],
|
||||
phase: ResearchOrchestrationPhase,
|
||||
name: string,
|
||||
input?: Record<string, unknown>,
|
||||
): ResearchOrchestrationStep {
|
||||
const active = this.activeRuns.get(runId);
|
||||
const order = (active?.stepIndex ?? 0) + 1;
|
||||
return {
|
||||
id: `${runId}-${phase}-${order}`,
|
||||
type,
|
||||
phase,
|
||||
status: "running",
|
||||
order,
|
||||
name,
|
||||
input,
|
||||
startedAt: new Date().toISOString(),
|
||||
};
|
||||
}
|
||||
|
||||
private computeTotalSteps(config: ResearchOrchestrationConfig): number {
|
||||
const providers = Math.max(1, config.providers.length);
|
||||
return 1 + providers + Math.max(1, config.maxSources) + Math.max(1, config.maxSynthesisRounds) + 1;
|
||||
}
|
||||
|
||||
private statusToPhase(status: ResearchRun["status"]): ResearchOrchestrationPhase {
|
||||
if (status === "completed") return "completed";
|
||||
if (status === "failed") return "failed";
|
||||
if (status === "cancelled") return "cancelled";
|
||||
return "planning";
|
||||
}
|
||||
|
||||
private throwIfAborted(signal: AbortSignal): void {
|
||||
if (signal.aborted) {
|
||||
throw signal.reason ?? new Error("Research run aborted");
|
||||
}
|
||||
}
|
||||
}
|
||||
235
packages/engine/src/research-step-runner.ts
Normal file
235
packages/engine/src/research-step-runner.ts
Normal file
@@ -0,0 +1,235 @@
|
||||
import type {
|
||||
ResearchModelSettings,
|
||||
ResearchProviderConfig,
|
||||
ResearchSource,
|
||||
ResearchSynthesisRequest,
|
||||
ResearchSynthesisResult,
|
||||
} from "@fusion/core";
|
||||
import { createLogger, formatError } from "./logger.js";
|
||||
|
||||
const log = createLogger("research-step-runner");
|
||||
|
||||
const DEFAULT_QUERY_TIMEOUT_MS = 30_000;
|
||||
const DEFAULT_FETCH_TIMEOUT_MS = 60_000;
|
||||
const DEFAULT_SYNTHESIS_TIMEOUT_MS = 120_000;
|
||||
|
||||
export class ResearchStepTimeoutError extends Error {
|
||||
constructor(step: string, timeoutMs: number) {
|
||||
super(`${step} timed out after ${timeoutMs}ms`);
|
||||
this.name = "ResearchStepTimeoutError";
|
||||
}
|
||||
}
|
||||
|
||||
export class ResearchStepAbortError extends Error {
|
||||
constructor(step: string) {
|
||||
super(`${step} aborted`);
|
||||
this.name = "ResearchStepAbortError";
|
||||
}
|
||||
}
|
||||
|
||||
export class ResearchStepProviderError extends Error {
|
||||
constructor(step: string, message: string) {
|
||||
super(`${step} provider error: ${message}`);
|
||||
this.name = "ResearchStepProviderError";
|
||||
}
|
||||
}
|
||||
|
||||
export interface ResearchProvider {
|
||||
readonly type: string;
|
||||
search(query: string, options: ResearchProviderConfig, signal?: AbortSignal): Promise<ResearchSource[]>;
|
||||
fetchContent(
|
||||
url: string,
|
||||
options: ResearchProviderConfig,
|
||||
signal?: AbortSignal,
|
||||
): Promise<{ content: string; metadata: Record<string, unknown> }>;
|
||||
isConfigured(): boolean;
|
||||
}
|
||||
|
||||
export interface ResearchStepResult<T> {
|
||||
ok: boolean;
|
||||
data?: T;
|
||||
error?: {
|
||||
code: "provider_not_configured" | "timeout" | "aborted" | "provider_error";
|
||||
message: string;
|
||||
retryable: boolean;
|
||||
};
|
||||
}
|
||||
|
||||
export interface ResearchStepRunnerApi {
|
||||
runSourceQuery(
|
||||
query: string,
|
||||
providerType: string,
|
||||
config?: ResearchProviderConfig,
|
||||
signal?: AbortSignal,
|
||||
): Promise<ResearchStepResult<ResearchSource[]>>;
|
||||
runContentFetch(
|
||||
url: string,
|
||||
config?: ResearchProviderConfig,
|
||||
signal?: AbortSignal,
|
||||
): Promise<ResearchStepResult<{ content: string; metadata: Record<string, unknown> }>>;
|
||||
runSynthesis(
|
||||
request: ResearchSynthesisRequest,
|
||||
modelSettings?: ResearchModelSettings,
|
||||
signal?: AbortSignal,
|
||||
): Promise<ResearchStepResult<ResearchSynthesisResult>>;
|
||||
}
|
||||
|
||||
export interface ResearchStepRunnerOptions {
|
||||
providers?: ResearchProvider[];
|
||||
synthesisRunner?: (
|
||||
request: ResearchSynthesisRequest,
|
||||
modelSettings: ResearchModelSettings,
|
||||
signal?: AbortSignal,
|
||||
) => Promise<ResearchSynthesisResult>;
|
||||
}
|
||||
|
||||
export class ResearchStepRunner implements ResearchStepRunnerApi {
|
||||
private readonly providers: Map<string, ResearchProvider>;
|
||||
private readonly synthesisRunner?: ResearchStepRunnerOptions["synthesisRunner"];
|
||||
|
||||
constructor(options: ResearchStepRunnerOptions = {}) {
|
||||
this.providers = new Map((options.providers ?? []).map((provider) => [provider.type, provider]));
|
||||
this.synthesisRunner = options.synthesisRunner;
|
||||
}
|
||||
|
||||
async runSourceQuery(
|
||||
query: string,
|
||||
providerType: string,
|
||||
config: ResearchProviderConfig = {},
|
||||
signal?: AbortSignal,
|
||||
): Promise<ResearchStepResult<ResearchSource[]>> {
|
||||
const provider = this.providers.get(providerType);
|
||||
if (!provider || !provider.isConfigured()) {
|
||||
return this.unconfigured(`provider ${providerType} is not configured`);
|
||||
}
|
||||
|
||||
try {
|
||||
const data = await this.withTimeout(
|
||||
`source-query:${providerType}`,
|
||||
provider.search(query, config, signal),
|
||||
config.timeoutMs ?? DEFAULT_QUERY_TIMEOUT_MS,
|
||||
signal,
|
||||
);
|
||||
return { ok: true, data };
|
||||
} catch (error) {
|
||||
return this.classifyError("source-query", error);
|
||||
}
|
||||
}
|
||||
|
||||
async runContentFetch(
|
||||
url: string,
|
||||
config: ResearchProviderConfig = {},
|
||||
signal?: AbortSignal,
|
||||
): Promise<ResearchStepResult<{ content: string; metadata: Record<string, unknown> }>> {
|
||||
const provider = this.findFirstConfiguredProvider();
|
||||
if (!provider) {
|
||||
return this.unconfigured("no configured provider available for content fetch");
|
||||
}
|
||||
|
||||
try {
|
||||
const data = await this.withTimeout(
|
||||
`content-fetch:${provider.type}`,
|
||||
provider.fetchContent(url, config, signal),
|
||||
config.timeoutMs ?? DEFAULT_FETCH_TIMEOUT_MS,
|
||||
signal,
|
||||
);
|
||||
return { ok: true, data };
|
||||
} catch (error) {
|
||||
return this.classifyError("content-fetch", error);
|
||||
}
|
||||
}
|
||||
|
||||
async runSynthesis(
|
||||
request: ResearchSynthesisRequest,
|
||||
modelSettings: ResearchModelSettings = {},
|
||||
signal?: AbortSignal,
|
||||
): Promise<ResearchStepResult<ResearchSynthesisResult>> {
|
||||
if (!this.synthesisRunner) {
|
||||
return this.unconfigured("synthesis provider is not configured");
|
||||
}
|
||||
|
||||
try {
|
||||
const timeoutMs = modelSettings.timeoutMs ?? DEFAULT_SYNTHESIS_TIMEOUT_MS;
|
||||
const data = await this.withTimeout(
|
||||
"synthesis",
|
||||
this.synthesisRunner(request, modelSettings, signal),
|
||||
timeoutMs,
|
||||
signal,
|
||||
);
|
||||
return { ok: true, data };
|
||||
} catch (error) {
|
||||
return this.classifyError("synthesis", error);
|
||||
}
|
||||
}
|
||||
|
||||
private findFirstConfiguredProvider(): ResearchProvider | undefined {
|
||||
for (const provider of this.providers.values()) {
|
||||
if (provider.isConfigured()) return provider;
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
private classifyError<T>(step: string, error: unknown): ResearchStepResult<T> {
|
||||
if (error instanceof ResearchStepTimeoutError) {
|
||||
return { ok: false, error: { code: "timeout", message: error.message, retryable: true } };
|
||||
}
|
||||
if (error instanceof ResearchStepAbortError) {
|
||||
return { ok: false, error: { code: "aborted", message: error.message, retryable: false } };
|
||||
}
|
||||
|
||||
const { message, detail } = formatError(error);
|
||||
log.warn(`${step} failed`, detail);
|
||||
return {
|
||||
ok: false,
|
||||
error: {
|
||||
code: "provider_error",
|
||||
message,
|
||||
retryable: true,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
private unconfigured<T>(message: string): ResearchStepResult<T> {
|
||||
return {
|
||||
ok: false,
|
||||
error: {
|
||||
code: "provider_not_configured",
|
||||
message,
|
||||
retryable: false,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
private async withTimeout<T>(
|
||||
step: string,
|
||||
promise: Promise<T>,
|
||||
timeoutMs: number,
|
||||
signal?: AbortSignal,
|
||||
): Promise<T> {
|
||||
if (signal?.aborted) {
|
||||
throw new ResearchStepAbortError(step);
|
||||
}
|
||||
|
||||
let timeoutId: NodeJS.Timeout | undefined;
|
||||
let abortListener: (() => void) | undefined;
|
||||
|
||||
const abortPromise = new Promise<never>((_, reject) => {
|
||||
if (!signal) return;
|
||||
abortListener = () => reject(new ResearchStepAbortError(step));
|
||||
signal.addEventListener("abort", abortListener, { once: true });
|
||||
});
|
||||
|
||||
const timeoutPromise = new Promise<never>((_, reject) => {
|
||||
timeoutId = setTimeout(() => reject(new ResearchStepTimeoutError(step, timeoutMs)), timeoutMs);
|
||||
});
|
||||
|
||||
try {
|
||||
return await Promise.race([promise, timeoutPromise, abortPromise]);
|
||||
} finally {
|
||||
if (timeoutId) clearTimeout(timeoutId);
|
||||
if (signal && abortListener) {
|
||||
signal.removeEventListener("abort", abortListener);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user