root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(FN-3013): add durable insight lifecycle with bounded run executor

Browse Source 
Adds durable insight lifecycle management to the Fusion system, including a bounded insight run executor, storage contracts for insight and research data, and API integration through the insights routes. Legacy schema compatibility is stabilized, and lifecycle safeguards are documented in the archit

Fusion-Task-Id: FN-3013
This commit is contained in:
Fusion
2026-05-03 02:24:15 -07:00
committed by gsxdsm
parent 6984bcde07
commit f2c4b77b34
14 changed files with 1478 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

185
packages/core/src/__tests__/insight-run-executor.test.ts Normal file
View File

@@ -0,0 +1,185 @@
import { describe, expect, it } from "vitest";
import { mkdtempSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createDatabase } from "../db.js";
import { InsightLifecycleError, InsightStore } from "../insight-store.js";
import { classifyInsightRunError, executeInsightRunLifecycle, retryInsightRunLifecycle } from "../insight-run-executor.js";
function createStore(): InsightStore {
const fusionDir = mkdtempSync(join(tmpdir(), "fn-insight-executor-"));
const db = createDatabase(fusionDir, { inMemory: true });
db.init();
return new InsightStore(db);
}
describe("classifyInsightRunError", () => {
it("classifies cancellation", () => {
const result = classifyInsightRunError(new DOMException("Aborted", "AbortError"));
expect(result.failureClass).toBe("cancelled");
});
it("classifies timeout", () => {
const result = classifyInsightRunError(new Error("timed out while calling provider"));
expect(result.failureClass).toBe("timed_out");
expect(result.retryable).toBe(true);
});
it("classifies transient provider errors as retryable", () => {
const result = classifyInsightRunError(new Error("HTTP 503 from provider"));
expect(result.failureClass).toBe("retryable_transient");
expect(result.retryable).toBe(true);
});
it("classifies deterministic failures as non-retryable", () => {
const result = classifyInsightRunError(new Error("invalid JSON contract response"));
expect(result.failureClass).toBe("non_retryable");
expect(result.retryable).toBe(false);
});
});
describe("executeInsightRunLifecycle", () => {
it("completes and persists events", async () => {
const store = createStore();
const run = await executeInsightRunLifecycle({
store,
projectId: "proj",
input: { trigger: "manual" },
executeAttempt: async () => ({
summary: "done",
insightsCreated: 2,
insightsUpdated: 1,
}),
});
expect(run.status).toBe("completed");
const events = store.listRunEvents(run.id);
expect(events.map((event) => event.type)).toEqual(["status_changed", "status_changed", "info", "status_changed"]);
});
it("retries transient failures with bounded attempts", async () => {
const store = createStore();
let calls = 0;
const run = await executeInsightRunLifecycle({
store,
projectId: "proj",
input: { trigger: "manual" },
maxAttempts: 2,
retryDelayMs: 0,
executeAttempt: async () => {
calls += 1;
if (calls === 1) {
throw new Error("HTTP 503");
}
return {
summary: "recovered",
insightsCreated: 1,
insightsUpdated: 0,
};
},
});
expect(calls).toBe(2);
expect(run.status).toBe("completed");
const events = store.listRunEvents(run.id);
expect(events.some((event) => event.type === "retry_scheduled")).toBe(true);
});
it("fails non-retryable errors without retry", async () => {
const store = createStore();
const run = await executeInsightRunLifecycle({
store,
projectId: "proj",
input: { trigger: "manual" },
maxAttempts: 3,
executeAttempt: async () => {
throw new Error("validation failed");
},
});
expect(run.status).toBe("failed");
expect(run.lifecycle.failureClass).toBe("non_retryable");
expect(run.lifecycle.retryable).toBe(false);
});
it("blocks duplicate active runs for same project+trigger", async () => {
const store = createStore();
store.createRun("proj", { trigger: "manual" });
await expect(() => executeInsightRunLifecycle({
store,
projectId: "proj",
input: { trigger: "manual" },
executeAttempt: async () => ({ insightsCreated: 0, insightsUpdated: 0 }),
})).rejects.toMatchObject({ code: "active_run_conflict" } satisfies Partial<InsightLifecycleError>);
});
it("marks timeout as terminal failure classification", async () => {
const store = createStore();
const run = await executeInsightRunLifecycle({
store,
projectId: "proj",
input: { trigger: "manual" },
timeoutMs: 10,
maxAttempts: 1,
executeAttempt: async ({ signal }) => {
await new Promise((resolve, reject) => {
const timeout = setTimeout(resolve, 50);
signal.addEventListener("abort", () => {
clearTimeout(timeout);
reject(signal.reason ?? new Error("aborted"));
});
});
return { insightsCreated: 0, insightsUpdated: 0 };
},
});
expect(run.status).toBe("failed");
expect(run.lifecycle.failureClass).toBe("timed_out");
});
});
describe("retryInsightRunLifecycle", () => {
it("creates a new run from retryable failed run", async () => {
const store = createStore();
const failed = await executeInsightRunLifecycle({
store,
projectId: "proj",
input: { trigger: "manual" },
maxAttempts: 1,
executeAttempt: async () => {
throw new Error("HTTP 503");
},
});
const retried = await retryInsightRunLifecycle({
store,
runId: failed.id,
executeAttempt: async () => ({ insightsCreated: 1, insightsUpdated: 0 }),
});
expect(retried.run.id).not.toBe(failed.id);
expect(retried.run.lifecycle.retryOfRunId).toBe(failed.id);
expect(retried.run.status).toBe("completed");
});
it("rejects retry for non-retryable failures", async () => {
const store = createStore();
const failed = await executeInsightRunLifecycle({
store,
projectId: "proj",
input: { trigger: "manual" },
maxAttempts: 1,
executeAttempt: async () => {
throw new Error("invalid input");
},
});
await expect(retryInsightRunLifecycle({
store,
runId: failed.id,
executeAttempt: async () => ({ insightsCreated: 1, insightsUpdated: 0 }),
})).rejects.toMatchObject({ code: "not_retryable" } satisfies Partial<InsightLifecycleError>);
});
});

12
packages/core/src/__tests__/insight-store.test.ts
View File

@@ -749,16 +749,14 @@ describe("InsightStore Run CRUD", () => {
expect(fromDb).toEqual(updated);
});
it("preserves existing completedAt on later updates", () => {
it("rejects updates after terminal completion", () => {
const run = store.createRun("proj", { trigger: "manual" });
const completed = store.updateRun(run.id, { status: "failed", error: "boom" });
const firstCompletedAt = completed?.completedAt;
expect(completed?.completedAt).toBeTruthy();
const patched = store.updateRun(run.id, { summary: "postmortem" });
expect(firstCompletedAt).toBeTruthy();
expect(patched?.completedAt).toBe(firstCompletedAt);
expect(patched?.summary).toBe("postmortem");
expect(() => store.updateRun(run.id, { summary: "postmortem" })).toThrow(
/terminal and immutable/i,
);
});
it("does not override completedAt if already provided", () => {

34
packages/core/src/__tests__/research-store.test.ts
View File

@@ -3,7 +3,7 @@ import { mkdtempSync } from "node:fs";
import { join } from "node:path";
import { tmpdir } from "node:os";
import { createDatabase, type Database } from "../db.js";
import { ResearchStore } from "../research-store.js";
import { ResearchLifecycleError, ResearchStore } from "../research-store.js";
describe("ResearchStore", () => {
let db: Database;
@@ -51,6 +51,36 @@ describe("ResearchStore", () => {
expect(store.getRun(cancelled.id)?.cancelledAt).toBeTruthy();
});
it("enforces terminal immutability and valid transitions", () => {
const run = store.createRun({ query: "guarded" });
store.updateStatus(run.id, "running");
store.updateStatus(run.id, "completed");
expect(() => store.updateRun(run.id, { topic: "changed" })).toThrow(ResearchLifecycleError);
const pending = store.createRun({ query: "pending" });
expect(() => store.updateStatus(pending.id, "completed")).toThrow(/Invalid run status transition/i);
});
it("persists lifecycle events to research_run_events", () => {
const run = store.createRun({ query: "events" });
store.updateStatus(run.id, "running");
store.appendLifecycleEvent(run.id, { type: "info", message: "custom event" });
const events = store.listRunEvents(run.id);
expect(events.length).toBeGreaterThanOrEqual(2);
expect(events.at(-1)?.message).toBe("custom event");
});
it("guards against duplicate active runs per project and trigger", () => {
const run = store.createRun({ query: "r1", projectId: "p1", trigger: "manual" });
expect(store.getActiveRun("p1", "manual")?.id).toBe(run.id);
expect(() => store.assertNoActiveRun("p1", "manual")).toThrow(ResearchLifecycleError);
store.updateStatus(run.id, "cancelled");
expect(() => store.assertNoActiveRun("p1", "manual")).not.toThrow();
});
it("appends events, manages sources, and sets results", () => {
const run = store.createRun({ query: "events" });
const event = store.appendEvent(run.id, { type: "info", message: "started" });
@@ -91,6 +121,7 @@ describe("ResearchStore", () => {
expect(store.getExport("REXP-missing")).toBeUndefined();
store.updateStatus(r1.id, "running");
store.updateStatus(r2.id, "running");
store.updateStatus(r2.id, "completed");
const stats = store.getStats();
expect(stats.total).toBeGreaterThanOrEqual(2);
@@ -107,6 +138,7 @@ describe("ResearchStore", () => {
store.on("run:completed", onCompleted);
const run = store.createRun({ query: "events" });
store.updateStatus(run.id, "running");
store.updateStatus(run.id, "completed");
expect(onStatus).toHaveBeenCalled();
expect(onCompleted).toHaveBeenCalled();

97
packages/core/src/db.ts
View File

@@ -419,6 +419,8 @@ CREATE TABLE IF NOT EXISTS research_runs (
query TEXT NOT NULL,
topic TEXT,
status TEXT NOT NULL,
projectId TEXT,
trigger TEXT,
providerConfig TEXT,
sources TEXT NOT NULL DEFAULT '[]',
events TEXT NOT NULL DEFAULT '[]',
@@ -427,6 +429,7 @@ CREATE TABLE IF NOT EXISTS research_runs (
tokenUsage TEXT,
tags TEXT NOT NULL DEFAULT '[]',
metadata TEXT,
lifecycle TEXT,
createdAt TEXT NOT NULL,
updatedAt TEXT NOT NULL,
startedAt TEXT,
@@ -448,6 +451,20 @@ CREATE TABLE IF NOT EXISTS research_exports (
);
CREATE INDEX IF NOT EXISTS idxResearchExportsRunId ON research_exports(runId);
CREATE TABLE IF NOT EXISTS research_run_events (
id TEXT PRIMARY KEY,
runId TEXT NOT NULL,
seq INTEGER NOT NULL,
type TEXT NOT NULL,
message TEXT NOT NULL,
status TEXT,
classification TEXT,
metadata TEXT,
createdAt TEXT NOT NULL,
FOREIGN KEY (runId) REFERENCES research_runs(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idxResearchRunEventsRunIdSeq ON research_run_events(runId, seq);
-- Schema version tracking
CREATE TABLE IF NOT EXISTS __meta (
key TEXT PRIMARY KEY,
@@ -643,9 +660,11 @@ CREATE TABLE IF NOT EXISTS project_insight_runs (
insightsUpdated INTEGER NOT NULL DEFAULT 0,
inputMetadata TEXT,
outputMetadata TEXT,
lifecycle TEXT,
createdAt TEXT NOT NULL,
startedAt TEXT,
completedAt TEXT
completedAt TEXT,
cancelledAt TEXT
);
-- Index for filtering insights by projectId
@@ -663,6 +682,23 @@ CREATE INDEX IF NOT EXISTS idxProjectInsightsCategory
-- Index for filtering runs by projectId
CREATE INDEX IF NOT EXISTS idxInsightRunsProjectId
ON project_insight_runs(projectId);
CREATE INDEX IF NOT EXISTS idxInsightRunsProjectTriggerStatus
ON project_insight_runs(projectId, trigger, status);
CREATE TABLE IF NOT EXISTS project_insight_run_events (
id TEXT PRIMARY KEY,
runId TEXT NOT NULL,
seq INTEGER NOT NULL,
type TEXT NOT NULL,
message TEXT NOT NULL,
status TEXT,
classification TEXT,
metadata TEXT,
createdAt TEXT NOT NULL,
FOREIGN KEY (runId) REFERENCES project_insight_runs(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idxInsightRunEventsRunIdSeq
ON project_insight_run_events(runId, seq);
-- Todo list persistence tables (FN-2575)
-- Project-scoped todo lists and ordered checklist items
@@ -1789,9 +1825,11 @@ export class Database {
insightsUpdated INTEGER NOT NULL DEFAULT 0,
inputMetadata TEXT,
outputMetadata TEXT,
lifecycle TEXT,
createdAt TEXT NOT NULL,
startedAt TEXT,
completedAt TEXT
completedAt TEXT,
cancelledAt TEXT
)
`);
@@ -2155,6 +2193,8 @@ export class Database {
query TEXT NOT NULL,
topic TEXT,
status TEXT NOT NULL,
projectId TEXT,
trigger TEXT,
providerConfig TEXT,
sources TEXT NOT NULL DEFAULT '[]',
events TEXT NOT NULL DEFAULT '[]',
@@ -2163,6 +2203,7 @@ export class Database {
tokenUsage TEXT,
tags TEXT NOT NULL DEFAULT '[]',
metadata TEXT,
lifecycle TEXT,
createdAt TEXT NOT NULL,
updatedAt TEXT NOT NULL,
startedAt TEXT,
@@ -2174,6 +2215,7 @@ export class Database {
this.db.exec(`CREATE INDEX IF NOT EXISTS idxResearchRunsStatus ON research_runs(status)`);
this.db.exec(`CREATE INDEX IF NOT EXISTS idxResearchRunsCreatedAt ON research_runs(createdAt)`);
this.db.exec(`CREATE INDEX IF NOT EXISTS idxResearchRunsUpdatedAt ON research_runs(updatedAt)`);
this.db.exec(`CREATE INDEX IF NOT EXISTS idxResearchRunsProjectTriggerStatus ON research_runs(projectId, trigger, status)`);
this.db.exec(`
CREATE TABLE IF NOT EXISTS research_exports (
@@ -2254,6 +2296,57 @@ export class Database {
this.applyMigration(59, () => {
this.db.exec(`CREATE INDEX IF NOT EXISTS idxTasksColumn ON tasks("column")`);
this.db.exec(`CREATE INDEX IF NOT EXISTS idxTasksUpdatedAt ON tasks(updatedAt DESC)`);
if (this.hasTable("research_runs")) {
this.addColumnIfMissing("research_runs", "projectId", "TEXT");
this.addColumnIfMissing("research_runs", "trigger", "TEXT");
this.addColumnIfMissing("research_runs", "lifecycle", "TEXT");
this.db.exec(`CREATE INDEX IF NOT EXISTS idxResearchRunsProjectTriggerStatus ON research_runs(projectId, trigger, status)`);
}
this.db.exec(`
CREATE TABLE IF NOT EXISTS research_run_events (
id TEXT PRIMARY KEY,
runId TEXT NOT NULL,
seq INTEGER NOT NULL,
type TEXT NOT NULL,
message TEXT NOT NULL,
status TEXT,
classification TEXT,
metadata TEXT,
createdAt TEXT NOT NULL,
FOREIGN KEY (runId) REFERENCES research_runs(id) ON DELETE CASCADE
)
`);
if (this.hasTable("research_run_events")) {
this.addColumnIfMissing("research_run_events", "seq", "INTEGER NOT NULL DEFAULT 0");
}
this.db.exec(`CREATE INDEX IF NOT EXISTS idxResearchRunEventsRunIdSeq ON research_run_events(runId, seq)`);
if (this.hasTable("project_insight_runs")) {
this.addColumnIfMissing("project_insight_runs", "lifecycle", "TEXT");
this.addColumnIfMissing("project_insight_runs", "cancelledAt", "TEXT");
this.db.exec(`CREATE INDEX IF NOT EXISTS idxInsightRunsProjectTriggerStatus ON project_insight_runs(projectId, trigger, status)`);
}
this.db.exec(`
CREATE TABLE IF NOT EXISTS project_insight_run_events (
id TEXT PRIMARY KEY,
runId TEXT NOT NULL,
seq INTEGER NOT NULL,
type TEXT NOT NULL,
message TEXT NOT NULL,
status TEXT,
classification TEXT,
metadata TEXT,
createdAt TEXT NOT NULL,
FOREIGN KEY (runId) REFERENCES project_insight_runs(id) ON DELETE CASCADE
)
`);
if (this.hasTable("project_insight_run_events")) {
this.addColumnIfMissing("project_insight_run_events", "seq", "INTEGER NOT NULL DEFAULT 0");
}
this.db.exec(`CREATE INDEX IF NOT EXISTS idxInsightRunEventsRunIdSeq ON project_insight_run_events(runId, seq)`);
});
}

23
packages/core/src/index.ts
View File

@@ -586,7 +586,12 @@ export type { AgentDreamProcessorResult, DreamProcessorResult, DreamPromptExecut
// ── Project Insights ──────────────────────────────────────────────────────
export { InsightStore, computeInsightFingerprint } from "./insight-store.js";
export { InsightLifecycleError, InsightStore, computeInsightFingerprint } from "./insight-store.js";
export {
classifyInsightRunError,
executeInsightRunLifecycle,
retryInsightRunLifecycle,
} from "./insight-run-executor.js";
export type {
InsightCategory,
InsightStatus,
@@ -599,6 +604,10 @@ export type {
InsightRun,
InsightRunStatus,
InsightRunTrigger,
InsightRunFailureClass,
InsightRunLifecycle,
InsightRunEventType,
InsightRunEvent,
InsightRunInputMetadata,
InsightRunOutputMetadata,
InsightRunCreateInput,
@@ -606,10 +615,16 @@ export type {
InsightRunListOptions,
InsightStoreEvents,
} from "./insight-types.js";
export type {
InsightRunAttemptContext,
InsightRunAttemptResult,
InsightRunExecutorErrorClassification,
InsightRunExecutorOptions,
} from "./insight-run-executor.js";
// ── Research System ───────────────────────────────────────────────────────
export { ResearchStore } from "./research-store.js";
export { ResearchLifecycleError, ResearchStore } from "./research-store.js";
export {
RESEARCH_RUN_STATUSES,
RESEARCH_SOURCE_STATUSES,
@@ -618,6 +633,7 @@ export {
RESEARCH_EVENT_TYPES,
RESEARCH_ORCHESTRATION_PHASES,
RESEARCH_ORCHESTRATION_STEP_STATUSES,
RESEARCH_RUN_FAILURE_CLASSES,
} from "./research-types.js";
export type {
ResearchRunStatus,
@@ -631,6 +647,9 @@ export type {
ResearchResult,
ResearchTokenUsage,
ResearchRun,
ResearchRunLifecycle,
ResearchRunFailureClass,
ResearchRunEvent,
ResearchExport,
ResearchRunCreateInput,
ResearchRunUpdateInput,

303
packages/core/src/insight-run-executor.ts Normal file
View File

@@ -0,0 +1,303 @@
import { setTimeout as delay } from "node:timers/promises";
import type {
InsightRun,
InsightRunCreateInput,
InsightRunFailureClass,
InsightRunOutputMetadata,
InsightRunTrigger,
InsightRunUpdateInput,
} from "./insight-types.js";
import { InsightLifecycleError, InsightStore } from "./insight-store.js";
export interface InsightRunAttemptResult {
summary?: string | null;
insightsCreated: number;
insightsUpdated: number;
outputMetadata?: InsightRunOutputMetadata;
}
export interface InsightRunAttemptContext {
run: InsightRun;
attempt: number;
maxAttempts: number;
signal: AbortSignal;
}
export interface InsightRunExecutorOptions {
store: InsightStore;
projectId: string;
input: InsightRunCreateInput;
executeAttempt: (ctx: InsightRunAttemptContext) => Promise<InsightRunAttemptResult>;
timeoutMs?: number;
maxAttempts?: number;
retryDelayMs?: number;
signal?: AbortSignal;
}
export interface InsightRunExecutorErrorClassification {
failureClass: InsightRunFailureClass;
retryable: boolean;
terminalReason: "cancelled" | "failed" | "timed_out";
terminalCause: string;
}
function isAbortLike(error: unknown): boolean {
return error instanceof DOMException && error.name === "AbortError";
}
function asErrorMessage(error: unknown): string {
return error instanceof Error ? error.message : String(error);
}
export function classifyInsightRunError(error: unknown): InsightRunExecutorErrorClassification {
if (isAbortLike(error)) {
return {
failureClass: "cancelled",
retryable: false,
terminalReason: "cancelled",
terminalCause: asErrorMessage(error),
};
}
const message = asErrorMessage(error);
if (/timeout|timed out|deadline/i.test(message)) {
return {
failureClass: "timed_out",
retryable: true,
terminalReason: "timed_out",
terminalCause: message,
};
}
if (/ECONNRESET|ENOTFOUND|EAI_AGAIN|ETIMEDOUT|429|5\d\d/i.test(message)) {
return {
failureClass: "retryable_transient",
retryable: true,
terminalReason: "failed",
terminalCause: message,
};
}
return {
failureClass: "non_retryable",
retryable: false,
terminalReason: "failed",
terminalCause: message,
};
}
function composeSignal(timeoutMs: number | undefined, parent?: AbortSignal): { signal: AbortSignal; clear: () => void } {
const controller = new AbortController();
const timeoutId = timeoutMs && timeoutMs > 0
? setTimeout(() => controller.abort(new Error(`Insight run timed out after ${timeoutMs}ms`)), timeoutMs)
: undefined;
const onAbort = () => {
controller.abort(parent?.reason ?? new DOMException("Aborted", "AbortError"));
};
if (parent) {
if (parent.aborted) onAbort();
else parent.addEventListener("abort", onAbort, { once: true });
}
return {
signal: controller.signal,
clear: () => {
if (timeoutId) clearTimeout(timeoutId);
if (parent) parent.removeEventListener("abort", onAbort);
},
};
}
function patchForStatus(status: "completed" | "failed" | "cancelled", patch: InsightRunUpdateInput): InsightRunUpdateInput {
if (status === "cancelled") {
return {
...patch,
cancelledAt: patch.cancelledAt ?? new Date().toISOString(),
};
}
return patch;
}
async function executeExistingRun(
store: InsightStore,
run: InsightRun,
options: Omit<InsightRunExecutorOptions, "input" | "projectId"> & { maxAttempts: number; retryDelayMs: number },
): Promise<InsightRun> {
const started = store.updateRun(run.id, {
status: "running",
startedAt: run.startedAt ?? new Date().toISOString(),
lifecycle: {
...run.lifecycle,
maxAttempts: options.maxAttempts,
attempt: run.lifecycle.attempt ?? 1,
},
});
let active = started ?? run;
store.appendRunEvent(active.id, { type: "status_changed", status: "running", message: "Run started" });
for (let attempt = active.lifecycle.attempt ?? 1; attempt <= options.maxAttempts; attempt += 1) {
const { signal, clear } = composeSignal(options.timeoutMs, options.signal);
try {
if (signal.aborted) {
throw signal.reason instanceof Error ? signal.reason : new DOMException("Aborted", "AbortError");
}
store.appendRunEvent(active.id, {
type: "info",
message: `Attempt ${attempt}/${options.maxAttempts}`,
metadata: { attempt, maxAttempts: options.maxAttempts },
});
const result = await options.executeAttempt({ run: active, attempt, maxAttempts: options.maxAttempts, signal });
const completed = store.updateRun(active.id, {
status: "completed",
summary: result.summary ?? null,
insightsCreated: result.insightsCreated,
insightsUpdated: result.insightsUpdated,
outputMetadata: result.outputMetadata,
lifecycle: {
...active.lifecycle,
attempt,
maxAttempts: options.maxAttempts,
terminalReason: "completed",
retryable: false,
},
});
if (!completed) throw new Error(`Run disappeared while completing: ${active.id}`);
store.appendRunEvent(completed.id, { type: "status_changed", status: "completed", message: "Run completed" });
return completed;
} catch (error) {
const classification = classifyInsightRunError(error);
const canRetry = classification.retryable && attempt < options.maxAttempts;
store.appendRunEvent(active.id, {
type: canRetry ? "retry_scheduled" : "error",
status: canRetry ? "running" : classification.terminalReason === "cancelled" ? "cancelled" : "failed",
classification: classification.failureClass,
message: canRetry
? `Attempt ${attempt} failed (${classification.failureClass}); retrying`
: `Run failed (${classification.failureClass})`,
metadata: { attempt, maxAttempts: options.maxAttempts, error: asErrorMessage(error) },
});
if (canRetry) {
active = store.updateRun(active.id, {
lifecycle: {
...active.lifecycle,
attempt: attempt + 1,
maxAttempts: options.maxAttempts,
failureClass: classification.failureClass,
retryable: true,
},
}) ?? active;
if (options.retryDelayMs > 0) {
await delay(options.retryDelayMs, undefined, { signal: options.signal });
}
continue;
}
const terminalStatus = classification.terminalReason === "cancelled" ? "cancelled" : "failed";
const terminal = store.updateRun(active.id, patchForStatus(terminalStatus, {
status: terminalStatus,
error: asErrorMessage(error),
lifecycle: {
...active.lifecycle,
attempt,
maxAttempts: options.maxAttempts,
terminalReason: classification.terminalReason,
terminalCause: classification.terminalCause,
failureClass: classification.failureClass,
retryable: classification.failureClass === "retryable_transient",
timeoutAt: classification.failureClass === "timed_out" ? new Date().toISOString() : active.lifecycle.timeoutAt,
},
}));
if (!terminal) throw new Error(`Run disappeared while failing: ${active.id}`);
return terminal;
} finally {
clear();
}
}
const failed = store.updateRun(active.id, {
status: "failed",
error: "Run exhausted attempts",
lifecycle: {
...active.lifecycle,
terminalReason: "failed",
terminalCause: "Run exhausted attempts",
failureClass: "non_retryable",
retryable: false,
attempt: options.maxAttempts,
maxAttempts: options.maxAttempts,
},
});
if (!failed) throw new Error(`Run disappeared after attempts exhausted: ${active.id}`);
return failed;
}
export async function executeInsightRunLifecycle(options: InsightRunExecutorOptions): Promise<InsightRun> {
const maxAttempts = Math.max(1, options.maxAttempts ?? 2);
const retryDelayMs = Math.max(0, options.retryDelayMs ?? 250);
let run: InsightRun;
try {
run = options.store.createRunOrThrowConflict(options.projectId, {
...options.input,
lifecycle: {
...options.input.lifecycle,
attempt: options.input.lifecycle?.attempt ?? 1,
maxAttempts,
rootRunId: options.input.lifecycle?.rootRunId,
},
});
} catch (error) {
if (error instanceof InsightLifecycleError && error.code === "active_run_conflict") {
throw error;
}
throw error;
}
options.store.appendRunEvent(run.id, {
type: "status_changed",
status: "pending",
message: "Run created",
});
return executeExistingRun(options.store, run, {
...options,
maxAttempts,
retryDelayMs,
});
}
export async function retryInsightRunLifecycle(
options: Omit<InsightRunExecutorOptions, "input" | "projectId"> & { runId: string; trigger?: InsightRunTrigger; inputMetadata?: InsightRunCreateInput["inputMetadata"] },
): Promise<{ run: InsightRun; retryOf: InsightRun }> {
const original = options.store.getRun(options.runId);
if (!original) {
throw new Error(`Insight run not found: ${options.runId}`);
}
if (original.status !== "failed") {
throw new InsightLifecycleError(`Run ${original.id} must be failed to retry`, "not_retryable");
}
if (!original.lifecycle.retryable || original.lifecycle.failureClass !== "retryable_transient") {
throw new InsightLifecycleError(`Run ${original.id} is non-retryable`, "not_retryable");
}
const run = await executeInsightRunLifecycle({
...options,
projectId: original.projectId,
input: {
trigger: options.trigger ?? original.trigger,
inputMetadata: options.inputMetadata ?? original.inputMetadata,
lifecycle: {
retryOfRunId: original.id,
rootRunId: original.lifecycle.rootRunId ?? original.id,
attempt: 1,
},
},
});
return { run, retryOf: original };
}

190
packages/core/src/insight-store.ts
View File

@@ -27,6 +27,7 @@
*/
import { EventEmitter } from "node:events";
import { randomUUID } from "node:crypto";
import type { Database } from "./db.js";
import { toJsonNullable, fromJson } from "./db.js";
import type {
@@ -46,6 +47,10 @@ import type {
InsightRunTrigger,
InsightRunInputMetadata,
InsightRunOutputMetadata,
InsightRunLifecycle,
InsightRunFailureClass,
InsightRunEvent,
InsightRunEventType,
} from "./insight-types.js";
import type { InsightStoreEvents } from "./insight-types.js";
@@ -63,6 +68,29 @@ function generateRunId(): string {
return `INSR-${timestamp}-${random}`;
}
function generateRunEventId(): string {
return `INSEVT-${randomUUID()}`;
}
const TERMINAL_RUN_STATUSES = new Set<InsightRunStatus>(["completed", "failed", "cancelled"]);
const VALID_RUN_STATUS_TRANSITIONS: Record<InsightRunStatus, InsightRunStatus[]> = {
pending: ["running", "completed", "failed", "cancelled"],
running: ["completed", "failed", "cancelled"],
completed: [],
failed: [],
cancelled: [],
};
export class InsightLifecycleError extends Error {
constructor(
message: string,
readonly code: "invalid_transition" | "terminal_immutable" | "active_run_conflict" | "not_retryable",
) {
super(message);
this.name = "InsightLifecycleError";
}
}
// ── Fingerprint Helper ────────────────────────────────────────────────
/**
@@ -403,14 +431,21 @@ export class InsightStore extends EventEmitter<InsightStoreEvents> {
const now = new Date().toISOString();
const id = generateRunId();
const inputMetadata = input.inputMetadata ?? {};
const lifecycle: InsightRunLifecycle = {
attempt: input.lifecycle?.attempt ?? 1,
maxAttempts: input.lifecycle?.maxAttempts ?? 1,
rootRunId: input.lifecycle?.rootRunId,
retryOfRunId: input.lifecycle?.retryOfRunId,
...input.lifecycle,
};
this.db.prepare(`
INSERT INTO project_insight_runs (
id, projectId, trigger, status, summary, error,
insightsCreated, insightsUpdated,
inputMetadata, outputMetadata,
createdAt, startedAt, completedAt
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
inputMetadata, outputMetadata, lifecycle,
createdAt, startedAt, completedAt, cancelledAt
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`).run(
id,
projectId,
@@ -422,9 +457,11 @@ export class InsightStore extends EventEmitter<InsightStoreEvents> {
0,
toJsonNullable(inputMetadata) ?? null,
null,
toJsonNullable(lifecycle),
now,
null,
null,
null,
);
this.db.bumpLastModified();
@@ -443,6 +480,8 @@ export class InsightStore extends EventEmitter<InsightStoreEvents> {
createdAt: now,
startedAt: null,
completedAt: null,
cancelledAt: null,
lifecycle,
};
this.emit("run:created", run);
@@ -499,9 +538,27 @@ export class InsightStore extends EventEmitter<InsightStoreEvents> {
const existing = this.getRun(id);
if (!existing) return undefined;
const mutatingKeys = Object.keys(input);
if (TERMINAL_RUN_STATUSES.has(existing.status) && mutatingKeys.length > 0) {
throw new InsightLifecycleError(`Run ${id} is terminal and immutable`, "terminal_immutable");
}
if (input.status && input.status !== existing.status) {
const allowed = VALID_RUN_STATUS_TRANSITIONS[existing.status];
if (!allowed.includes(input.status)) {
throw new InsightLifecycleError(
`Invalid run status transition: ${existing.status} -> ${input.status}`,
"invalid_transition",
);
}
}
const now = new Date().toISOString();
const isTerminal = input.status !== undefined && ["completed", "failed", "cancelled"].includes(input.status);
const autoComplete = isTerminal && input.completedAt === undefined && existing.completedAt === null;
const nextStatus = input.status ?? existing.status;
const isTerminal = TERMINAL_RUN_STATUSES.has(nextStatus);
const lifecycle = { ...existing.lifecycle, ...(input.lifecycle ?? {}) };
const autoCompleteAt = isTerminal && input.completedAt === undefined && existing.completedAt === null ? now : undefined;
const autoCancelledAt = nextStatus === "cancelled" && input.cancelledAt === undefined && existing.cancelledAt === null ? now : undefined;
const sets: string[] = [];
const params: (string | number | null)[] = [];
@@ -530,6 +587,10 @@ export class InsightStore extends EventEmitter<InsightStoreEvents> {
sets.push("outputMetadata = ?");
params.push(toJsonNullable(input.outputMetadata));
}
if (input.lifecycle !== undefined) {
sets.push("lifecycle = ?");
params.push(toJsonNullable(lifecycle));
}
if (input.startedAt !== undefined) {
sets.push("startedAt = ?");
params.push(input.startedAt);
@@ -538,22 +599,29 @@ export class InsightStore extends EventEmitter<InsightStoreEvents> {
sets.push("completedAt = ?");
params.push(input.completedAt);
}
if (input.cancelledAt !== undefined) {
sets.push("cancelledAt = ?");
params.push(input.cancelledAt);
}
if (autoCompleteAt !== undefined) {
sets.push("completedAt = ?");
params.push(autoCompleteAt);
}
if (autoCancelledAt !== undefined) {
sets.push("cancelledAt = ?");
params.push(autoCancelledAt);
}
if (sets.length === 0) return existing;
// Auto-set completedAt for terminal transitions
if (autoComplete) {
sets.push("completedAt = ?");
params.push(now);
}
params.push(id);
this.db.prepare(`UPDATE project_insight_runs SET ${sets.join(", ")} WHERE id = ?`).run(...params);
this.db.bumpLastModified();
const updated = this.getRun(id)!;
if (isTerminal) {
if (isTerminal && updated.status !== existing.status) {
this.emit("run:completed", updated);
}
this.emit("run:updated", updated);
@@ -573,21 +641,102 @@ export class InsightStore extends EventEmitter<InsightStoreEvents> {
* @returns The created or existing run
*/
upsertRun(projectId: string, trigger: InsightRunTrigger, input: InsightRunCreateInput): InsightRun {
// Find most recent pending/running run for this project + trigger
const existing = this.findActiveRun(projectId, trigger);
if (existing) {
return existing;
}
return this.createRun(projectId, input);
}
findActiveRun(projectId: string, trigger: InsightRunTrigger): InsightRun | undefined {
const existingRow = this.db.prepare(`
SELECT * FROM project_insight_runs
SELECT id FROM project_insight_runs
WHERE projectId = ? AND trigger = ? AND status IN ('pending', 'running')
ORDER BY createdAt DESC, id DESC
LIMIT 1
`).get(projectId, trigger) as Record<string, unknown> | undefined;
`).get(projectId, trigger) as { id: string } | undefined;
return existingRow ? this.getRun(existingRow.id) : undefined;
}
if (existingRow) {
return this.getRun(existingRow.id as string)!;
createRunOrThrowConflict(projectId: string, input: InsightRunCreateInput): InsightRun {
const existing = this.findActiveRun(projectId, input.trigger);
if (existing) {
throw new InsightLifecycleError(
`Active run already exists for project ${projectId} trigger ${input.trigger}: ${existing.id}`,
"active_run_conflict",
);
}
return this.createRun(projectId, input);
}
appendRunEvent(
runId: string,
event: {
type: InsightRunEventType;
message: string;
status?: InsightRunStatus;
classification?: InsightRunFailureClass;
metadata?: Record<string, unknown>;
},
): InsightRunEvent {
const run = this.getRun(runId);
if (!run) {
throw new Error(`Insight run not found: ${runId}`);
}
const createdAt = new Date().toISOString();
const row = this.db.prepare("SELECT COALESCE(MAX(seq), 0) + 1 as nextSeq FROM project_insight_run_events WHERE runId = ?").get(runId) as { nextSeq: number };
const runEvent: InsightRunEvent = {
id: generateRunEventId(),
runId,
seq: Number(row?.nextSeq ?? 1),
type: event.type,
message: event.message,
status: event.status,
classification: event.classification,
metadata: event.metadata,
createdAt,
};
this.db.prepare(`
INSERT INTO project_insight_run_events (id, runId, seq, type, message, status, classification, metadata, createdAt)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
`).run(
runEvent.id,
runEvent.runId,
runEvent.seq,
runEvent.type,
runEvent.message,
runEvent.status ?? null,
runEvent.classification ?? null,
toJsonNullable(runEvent.metadata),
runEvent.createdAt,
);
this.db.bumpLastModified();
this.emit("run:event", { runId, event: runEvent });
return runEvent;
}
listRunEvents(runId: string): InsightRunEvent[] {
const rows = this.db.prepare(`
SELECT * FROM project_insight_run_events
WHERE runId = ?
ORDER BY seq ASC
`).all(runId) as Record<string, unknown>[];
return rows.map((row) => ({
id: row.id as string,
runId: row.runId as string,
seq: Number(row.seq),
type: row.type as InsightRunEventType,
message: row.message as string,
status: (row.status as InsightRunStatus | null) ?? undefined,
classification: (row.classification as InsightRunFailureClass | null) ?? undefined,
metadata: fromJson<Record<string, unknown>>(row.metadata as string | null),
createdAt: row.createdAt as string,
}));
}
/**
* Get the count of runs matching the given filter.
*/
@@ -642,6 +791,11 @@ export class InsightStore extends EventEmitter<InsightStoreEvents> {
createdAt: row.createdAt as string,
startedAt: row.startedAt as string | null,
completedAt: row.completedAt as string | null,
cancelledAt: row.cancelledAt as string | null,
lifecycle: (() => {
const m = fromJson<InsightRunLifecycle>(row.lifecycle as string | null);
return m ?? {};
})(),
};
}
}

48
packages/core/src/insight-types.ts
View File

@@ -275,6 +275,42 @@ export type InsightRunTrigger = "schedule" | "manual" | "task_completion" | "mer
* Runs track the full lifecycle of an analysis pass — from scheduling
* through input processing to output persistence.
*/
export type InsightRunFailureClass = "cancelled" | "timed_out" | "retryable_transient" | "non_retryable";
export interface InsightRunLifecycle {
terminalReason?: "completed" | "cancelled" | "failed" | "timed_out";
terminalCause?: string;
failureClass?: InsightRunFailureClass;
retryable?: boolean;
cancellationRequestedAt?: string;
timeoutAt?: string;
retryOfRunId?: string;
rootRunId?: string;
attempt?: number;
maxAttempts?: number;
}
export type InsightRunEventType =
| "status_changed"
| "retry_scheduled"
| "cancel_requested"
| "timeout"
| "info"
| "warning"
| "error";
export interface InsightRunEvent {
id: string;
runId: string;
seq: number;
type: InsightRunEventType;
message: string;
status?: InsightRunStatus;
classification?: InsightRunFailureClass;
metadata?: Record<string, unknown>;
createdAt: string;
}
export interface InsightRun {
/**
* Unique identifier for this run (e.g., "INSR-xxx").
@@ -344,6 +380,13 @@ export interface InsightRun {
* When the run reached a terminal state.
*/
completedAt: string | null;
/**
* When cancellation was applied.
*/
cancelledAt: string | null;
lifecycle: InsightRunLifecycle;
}
// ── Run Input / Output Metadata ──────────────────────────────────────
@@ -422,6 +465,7 @@ export interface InsightRunOutputMetadata {
export interface InsightRunCreateInput {
trigger: InsightRunTrigger;
inputMetadata?: InsightRunInputMetadata;
lifecycle?: InsightRunLifecycle;
}
// ── Run Update Input ─────────────────────────────────────────────────
@@ -437,8 +481,10 @@ export interface InsightRunUpdateInput {
insightsCreated?: number;
insightsUpdated?: number;
outputMetadata?: InsightRunOutputMetadata;
lifecycle?: InsightRunLifecycle;
startedAt?: string | null;
completedAt?: string | null;
cancelledAt?: string | null;
}
// ── Run List Options ─────────────────────────────────────────────────
@@ -478,4 +524,6 @@ export interface InsightStoreEvents {
"run:updated": [InsightRun];
/** Emitted when a run reaches a terminal state */
"run:completed": [InsightRun];
/** Emitted when a durable run event is appended */
"run:event": [{ runId: string; event: InsightRunEvent }];
}

243
packages/core/src/research-store.ts
View File

@@ -9,6 +9,8 @@ import type {
ResearchResult,
ResearchRun,
ResearchRunCreateInput,
ResearchRunEvent,
ResearchRunFailureClass,
ResearchRunListOptions,
ResearchRunStatus,
ResearchRunUpdateInput,
@@ -26,6 +28,16 @@ function generateId(prefix: string): string {
return `${prefix}-${randomUUID()}`;
}
export class ResearchLifecycleError extends Error {
constructor(
message: string,
readonly code: "invalid_transition" | "terminal_immutable" | "active_run_conflict" | "not_retryable",
) {
super(message);
this.name = "ResearchLifecycleError";
}
}
function mergeRecord(
currentValue: Record<string, unknown> | undefined,
patchValue: Record<string, unknown> | undefined,
@@ -35,6 +47,15 @@ function mergeRecord(
return Object.keys(merged).length > 0 ? merged : undefined;
}
const TERMINAL_STATUSES = new Set<ResearchRunStatus>(["completed", "failed", "cancelled"]);
const VALID_STATUS_TRANSITIONS: Record<ResearchRunStatus, ResearchRunStatus[]> = {
pending: ["running", "cancelled", "failed"],
running: ["completed", "failed", "cancelled"],
completed: [],
failed: [],
cancelled: [],
};
export class ResearchStore extends EventEmitter<ResearchStoreEvents> {
constructor(private readonly db: Database) {
super();
@@ -48,26 +69,37 @@ export class ResearchStore extends EventEmitter<ResearchStoreEvents> {
query: input.query,
topic: input.topic,
status: "pending",
projectId: input.projectId,
trigger: input.trigger,
providerConfig: input.providerConfig,
sources: input.sources ?? [],
events: input.events ?? [],
results: input.results,
tags: input.tags ?? [],
metadata: input.metadata,
lifecycle: {
attempt: input.lifecycle?.attempt ?? 1,
maxAttempts: input.lifecycle?.maxAttempts ?? 1,
rootRunId: input.lifecycle?.rootRunId,
retryOfRunId: input.lifecycle?.retryOfRunId,
...input.lifecycle,
},
createdAt: now,
updatedAt: now,
};
this.db.prepare(`
INSERT INTO research_runs (
id, query, topic, status, providerConfig, sources, events, results, error,
tokenUsage, tags, metadata, createdAt, updatedAt, startedAt, completedAt, cancelledAt
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
id, query, topic, status, projectId, trigger, providerConfig, sources, events, results, error,
tokenUsage, tags, metadata, lifecycle, createdAt, updatedAt, startedAt, completedAt, cancelledAt
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`).run(
run.id,
run.query,
run.topic ?? null,
run.status,
run.projectId ?? null,
run.trigger ?? null,
toJsonNullable(run.providerConfig),
toJson(run.sources),
toJson(run.events),
@@ -76,6 +108,7 @@ export class ResearchStore extends EventEmitter<ResearchStoreEvents> {
null,
toJson(run.tags),
toJsonNullable(run.metadata),
toJsonNullable(run.lifecycle),
run.createdAt,
run.updatedAt,
null,
@@ -97,15 +130,31 @@ export class ResearchStore extends EventEmitter<ResearchStoreEvents> {
const existing = this.getRun(id);
if (!existing) return undefined;
if (TERMINAL_STATUSES.has(existing.status) && Object.keys(input).some((key) => key !== "events" && key !== "metadata")) {
throw new ResearchLifecycleError(`Run ${id} is terminal and immutable`, "terminal_immutable");
}
if (input.status && input.status !== existing.status) {
const allowed = VALID_STATUS_TRANSITIONS[existing.status];
if (!allowed.includes(input.status)) {
throw new ResearchLifecycleError(
`Invalid run status transition: ${existing.status} -> ${input.status}`,
"invalid_transition",
);
}
}
const now = new Date().toISOString();
const mergedProviderConfig = mergeRecord(existing.providerConfig, input.providerConfig);
const mergedMetadata = mergeRecord(existing.metadata, input.metadata);
const mergedLifecycle = { ...(existing.lifecycle ?? {}), ...(input.lifecycle ?? {}) };
const updated: ResearchRun = {
...existing,
...input,
providerConfig: mergedProviderConfig,
metadata: mergedMetadata,
lifecycle: Object.keys(mergedLifecycle).length > 0 ? mergedLifecycle : undefined,
error: input.error === null ? undefined : (input.error ?? existing.error),
updatedAt: now,
startedAt: input.startedAt === null ? undefined : (input.startedAt ?? existing.startedAt),
@@ -180,7 +229,24 @@ export class ResearchStore extends EventEmitter<ResearchStoreEvents> {
metadata: event.metadata,
};
this.updateRun(runId, { events: [...run.events, created] });
const seq = this.getNextEventSeq(runId);
this.db.prepare(`
INSERT INTO research_run_events (id, runId, seq, type, message, status, classification, metadata, createdAt)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
`).run(
created.id,
runId,
seq,
created.type,
created.message,
run.status,
null,
toJsonNullable(created.metadata),
created.timestamp,
);
this.persistRun({ ...run, events: [...run.events, created], updatedAt: new Date().toISOString() });
this.db.bumpLastModified();
this.emit("event:added", { runId, event: created });
return created;
}
@@ -189,6 +255,62 @@ export class ResearchStore extends EventEmitter<ResearchStoreEvents> {
return this.addEvent(runId, event);
}
appendLifecycleEvent(
runId: string,
event: { type: ResearchEvent["type"]; message: string; status?: ResearchRunStatus; classification?: ResearchRunFailureClass; metadata?: Record<string, unknown> },
): ResearchRunEvent {
const run = this.getRun(runId);
if (!run) throw new Error(`Research run not found: ${runId}`);
const createdAt = new Date().toISOString();
const lifecycleEvent: ResearchRunEvent = {
id: generateId("REVT"),
runId,
seq: this.getNextEventSeq(runId),
type: event.type,
message: event.message,
status: event.status,
classification: event.classification,
metadata: event.metadata,
createdAt,
};
this.db.prepare(`
INSERT INTO research_run_events (id, runId, seq, type, message, status, classification, metadata, createdAt)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
`).run(
lifecycleEvent.id,
lifecycleEvent.runId,
lifecycleEvent.seq,
lifecycleEvent.type,
lifecycleEvent.message,
lifecycleEvent.status ?? null,
lifecycleEvent.classification ?? null,
toJsonNullable(lifecycleEvent.metadata),
lifecycleEvent.createdAt,
);
this.db.bumpLastModified();
return lifecycleEvent;
}
listRunEvents(runId: string): ResearchRunEvent[] {
const rows = this.db.prepare(`
SELECT * FROM research_run_events
WHERE runId = ?
ORDER BY seq ASC
`).all(runId) as Record<string, unknown>[];
return rows.map((row) => ({
id: row.id as string,
runId: row.runId as string,
seq: Number(row.seq),
type: row.type as ResearchEvent["type"],
message: row.message as string,
status: (row.status as ResearchRunStatus | null) ?? undefined,
classification: (row.classification as ResearchRunFailureClass | null) ?? undefined,
metadata: fromJson<Record<string, unknown>>(row.metadata as string | null),
createdAt: row.createdAt as string,
}));
}
addSource(runId: string, source: Omit<ResearchSource, "id">): ResearchSource {
const run = this.getRun(runId);
if (!run) throw new Error(`Research run not found: ${runId}`);
@@ -228,6 +350,9 @@ export class ResearchStore extends EventEmitter<ResearchStoreEvents> {
const patch: ResearchRunUpdateInput = {
...(extra ?? {}),
status,
lifecycle: {
...(run.lifecycle ?? {}),
},
};
if (status === "running" && !run.startedAt) {
@@ -240,9 +365,24 @@ export class ResearchStore extends EventEmitter<ResearchStoreEvents> {
patch.cancelledAt = now;
}
if (status === "completed") {
patch.lifecycle = { ...(patch.lifecycle ?? {}), terminalReason: "completed", retryable: false };
} else if (status === "failed") {
patch.lifecycle = { ...(patch.lifecycle ?? {}), terminalReason: "failed", retryable: patch.lifecycle?.failureClass === "retryable_transient" };
} else if (status === "cancelled") {
patch.lifecycle = { ...(patch.lifecycle ?? {}), terminalReason: "cancelled", retryable: false, failureClass: "cancelled" };
}
const updated = this.updateRun(runId, patch);
if (!updated) return;
this.appendLifecycleEvent(runId, {
type: "status_changed",
message: `Status changed to ${status}`,
status,
classification: updated.lifecycle?.failureClass,
});
this.emit("run:status_changed", updated);
if (status === "completed") this.emit("run:completed", updated);
if (status === "failed") this.emit("run:failed", updated);
@@ -319,17 +459,104 @@ export class ResearchStore extends EventEmitter<ResearchStoreEvents> {
return { total, byStatus };
}
getActiveRun(projectId: string, trigger: string): ResearchRun | undefined {
const row = this.db.prepare(`
SELECT * FROM research_runs
WHERE projectId = ? AND trigger = ? AND status IN ('pending', 'running')
ORDER BY createdAt DESC
LIMIT 1
`).get(projectId, trigger) as Record<string, unknown> | undefined;
return row ? this.rowToRun(row) : undefined;
}
assertNoActiveRun(projectId: string, trigger: string): void {
const active = this.getActiveRun(projectId, trigger);
if (active) {
throw new ResearchLifecycleError(
`Active run already exists for projectId=${projectId} trigger=${trigger}: ${active.id}`,
"active_run_conflict",
);
}
}
requestCancellation(runId: string, reason = "Cancelled by user"): ResearchRun {
const run = this.getRun(runId);
if (!run) throw new Error(`Research run not found: ${runId}`);
if (TERMINAL_STATUSES.has(run.status)) {
throw new ResearchLifecycleError(`Run ${runId} is already terminal`, "invalid_transition");
}
const now = new Date().toISOString();
const updated = this.updateRun(runId, {
status: "cancelled",
cancelledAt: run.cancelledAt ?? now,
lifecycle: {
...(run.lifecycle ?? {}),
cancellationRequestedAt: now,
terminalReason: "cancelled",
terminalCause: reason,
failureClass: "cancelled",
retryable: false,
},
});
if (!updated) throw new Error(`Research run not found: ${runId}`);
this.appendLifecycleEvent(runId, { type: "cancel_requested", message: reason, status: "cancelled", classification: "cancelled" });
return updated;
}
createRetryRun(runId: string, maxAttempts?: number): ResearchRun {
const run = this.getRun(runId);
if (!run) throw new Error(`Research run not found: ${runId}`);
if (run.status !== "failed") {
throw new ResearchLifecycleError(`Run ${runId} is not failed`, "invalid_transition");
}
if (!run.lifecycle?.retryable) {
throw new ResearchLifecycleError(`Run ${runId} is non-retryable`, "not_retryable");
}
const nextAttempt = (run.lifecycle?.attempt ?? 1) + 1;
const rootRunId = run.lifecycle?.rootRunId ?? run.id;
const retryRun = this.createRun({
query: run.query,
topic: run.topic,
projectId: run.projectId,
trigger: run.trigger,
providerConfig: run.providerConfig,
tags: run.tags,
metadata: run.metadata,
lifecycle: {
attempt: nextAttempt,
maxAttempts: maxAttempts ?? run.lifecycle?.maxAttempts ?? nextAttempt,
retryOfRunId: run.id,
rootRunId,
},
});
this.appendLifecycleEvent(retryRun.id, {
type: "retry_scheduled",
message: `Retry scheduled from ${run.id}`,
metadata: { retryOfRunId: run.id, rootRunId, attempt: nextAttempt },
});
return retryRun;
}
private getNextEventSeq(runId: string): number {
const row = this.db.prepare("SELECT COALESCE(MAX(seq), 0) AS seq FROM research_run_events WHERE runId = ?").get(runId) as { seq?: number };
return Number(row?.seq ?? 0) + 1;
}
private persistRun(run: ResearchRun): void {
this.db.prepare(`
UPDATE research_runs
SET query = ?, topic = ?, status = ?, providerConfig = ?, sources = ?, events = ?,
results = ?, error = ?, tokenUsage = ?, tags = ?, metadata = ?, updatedAt = ?,
SET query = ?, topic = ?, status = ?, projectId = ?, trigger = ?, providerConfig = ?, sources = ?, events = ?,
results = ?, error = ?, tokenUsage = ?, tags = ?, metadata = ?, lifecycle = ?, updatedAt = ?,
startedAt = ?, completedAt = ?, cancelledAt = ?
WHERE id = ?
`).run(
run.query,
run.topic ?? null,
run.status,
run.projectId ?? null,
run.trigger ?? null,
toJsonNullable(run.providerConfig),
toJson(run.sources),
toJson(run.events),
@@ -338,6 +565,7 @@ export class ResearchStore extends EventEmitter<ResearchStoreEvents> {
toJsonNullable(run.tokenUsage),
toJson(run.tags),
toJsonNullable(run.metadata),
toJsonNullable(run.lifecycle),
run.updatedAt,
run.startedAt ?? null,
run.completedAt ?? null,
@@ -354,6 +582,8 @@ export class ResearchStore extends EventEmitter<ResearchStoreEvents> {
query: row.query as string,
topic: (row.topic as string | null) ?? undefined,
status: row.status as ResearchRunStatus,
projectId: (row.projectId as string | null) ?? undefined,
trigger: (row.trigger as string | null) ?? undefined,
providerConfig: fromJson<Record<string, unknown>>(row.providerConfig as string | null),
sources: fromJson<ResearchSource[]>(row.sources as string | null) ?? [],
events: fromJson<ResearchEvent[]>(row.events as string | null) ?? [],
@@ -362,6 +592,7 @@ export class ResearchStore extends EventEmitter<ResearchStoreEvents> {
tokenUsage: fromJson<ResearchRun["tokenUsage"]>(row.tokenUsage as string | null),
tags: fromJson<string[]>(row.tags as string | null) ?? [],
metadata: fromJson<Record<string, unknown>>(row.metadata as string | null),
lifecycle: fromJson<ResearchRun["lifecycle"]>(row.lifecycle as string | null),
createdAt: row.createdAt as string,
updatedAt: row.updatedAt as string,
startedAt: (row.startedAt as string | null) ?? undefined,

47
packages/core/src/research-types.ts
View File

@@ -38,10 +38,48 @@ export const RESEARCH_EVENT_TYPES = [
"source_added",
"result_updated",
"progress",
"status_changed",
"retry_scheduled",
"cancel_requested",
"timeout",
] as const;
export type ResearchEventType = typeof RESEARCH_EVENT_TYPES[number];
export const RESEARCH_RUN_FAILURE_CLASSES = [
"cancelled",
"timed_out",
"retryable_transient",
"non_retryable",
] as const;
export type ResearchRunFailureClass = typeof RESEARCH_RUN_FAILURE_CLASSES[number];
export interface ResearchRunLifecycle {
terminalReason?: "completed" | "cancelled" | "failed" | "timed_out";
terminalCause?: string;
failureClass?: ResearchRunFailureClass;
retryable?: boolean;
cancellationRequestedAt?: string;
timeoutAt?: string;
retryOfRunId?: string;
rootRunId?: string;
attempt?: number;
maxAttempts?: number;
}
export interface ResearchRunEvent {
id: string;
runId: string;
seq: number;
type: ResearchEventType;
message: string;
status?: ResearchRunStatus;
classification?: ResearchRunFailureClass;
metadata?: Record<string, unknown>;
createdAt: string;
}
export interface ResearchSource {
id: string;
type: ResearchSourceType;
@@ -89,6 +127,8 @@ export interface ResearchRun {
query: string;
topic?: string;
status: ResearchRunStatus;
projectId?: string;
trigger?: string;
providerConfig?: Record<string, unknown>;
sources: ResearchSource[];
events: ResearchEvent[];
@@ -97,6 +137,7 @@ export interface ResearchRun {
tokenUsage?: ResearchTokenUsage;
tags: string[];
metadata?: Record<string, unknown>;
lifecycle?: ResearchRunLifecycle;
createdAt: string;
updatedAt: string;
startedAt?: string;
@@ -116,18 +157,23 @@ export interface ResearchExport {
export interface ResearchRunCreateInput {
query: string;
topic?: string;
projectId?: string;
trigger?: string;
providerConfig?: Record<string, unknown>;
sources?: ResearchSource[];
events?: ResearchEvent[];
results?: ResearchResult;
tags?: string[];
metadata?: Record<string, unknown>;
lifecycle?: ResearchRunLifecycle;
}
export interface ResearchRunUpdateInput {
query?: string;
topic?: string;
status?: ResearchRunStatus;
projectId?: string;
trigger?: string;
providerConfig?: Record<string, unknown>;
sources?: ResearchSource[];
events?: ResearchEvent[];
@@ -136,6 +182,7 @@ export interface ResearchRunUpdateInput {
tokenUsage?: ResearchTokenUsage;
tags?: string[];
metadata?: Record<string, unknown>;
lifecycle?: ResearchRunLifecycle;
startedAt?: string | null;
completedAt?: string | null;
cancelledAt?: string | null;