From 0189477d9dc32f8393ee7707631510a59bfc814e Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 2 Jun 2026 19:13:43 -0700 Subject: [PATCH] feat(compound-engineering): session orchestrator, persistence, resume (U5) Add ce_sessions schema (onSchemaInit), CeSessionStore (via ctx.taskStore.getDatabase()), and CeOrchestrator driving a stage's skill on the U4 interactive seam: streams thinking/text, persists questions as awaiting_input, writes the stage artifact on complete, and auto-saves + emits an observable event on interrupt/error (never silent loss). Lifecycle: launching/active/awaiting_input/completed/error/interrupted with resume and an interval-relative staleness rubric. Start/answer/resume/get-state routes. Streaming transport is client polling of the get-session-state route for v1: plugin routes have no native SSE and the loader emitEvent is a logging stub, so true server push needs a host publish-to-/api/events seam (tracked follow-up). Skill discovery is cwd+prompt-based; forwarding install paths through the seam is a tracked follow-up. --- .../src/__tests__/_harness.ts | 89 +++++ .../src/__tests__/manifest.test.ts | 17 +- .../src/__tests__/orchestrator-flow.test.ts | 94 +++++ .../orchestrator-interrupt-resume.test.ts | 128 +++++++ .../src/__tests__/session-routes.test.ts | 74 ++++ .../src/__tests__/session-store.test.ts | 77 ++++ .../src/__tests__/skill-reachability.test.ts | 65 ++++ .../src/index.ts | 11 +- .../src/routes/session-routes.ts | 134 +++++++ .../src/schema.ts | 50 +++ .../src/session/orchestrator.ts | 362 ++++++++++++++++++ .../src/session/session-store.ts | 263 +++++++++++++ .../src/session/stage-registry.ts | 51 +++ 13 files changed, 1412 insertions(+), 3 deletions(-) create mode 100644 plugins/fusion-plugin-compound-engineering/src/__tests__/_harness.ts create mode 100644 plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-flow.test.ts create mode 100644 plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-interrupt-resume.test.ts create mode 100644 plugins/fusion-plugin-compound-engineering/src/__tests__/session-routes.test.ts create mode 100644 plugins/fusion-plugin-compound-engineering/src/__tests__/session-store.test.ts create mode 100644 plugins/fusion-plugin-compound-engineering/src/__tests__/skill-reachability.test.ts create mode 100644 plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts create mode 100644 plugins/fusion-plugin-compound-engineering/src/schema.ts create mode 100644 plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts create mode 100644 plugins/fusion-plugin-compound-engineering/src/session/session-store.ts create mode 100644 plugins/fusion-plugin-compound-engineering/src/session/stage-registry.ts diff --git a/plugins/fusion-plugin-compound-engineering/src/__tests__/_harness.ts b/plugins/fusion-plugin-compound-engineering/src/__tests__/_harness.ts new file mode 100644 index 0000000000..ba0c52b884 --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/_harness.ts @@ -0,0 +1,89 @@ +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { vi } from "vitest"; +import { Database } from "@fusion/core"; +import type { + CreateInteractiveAiSessionFactory, + InteractiveAiSession, + InteractiveAiSessionEvent, + PluginContext, +} from "@fusion/core"; + +export interface TestHarness { + db: Database; + projectRoot: string; + ctx: PluginContext; + emitted: Array<{ event: string; data: unknown }>; + close(): void; +} + +/** + * In-memory DB + a minimal route-style PluginContext whose `taskStore` exposes + * `getDatabase()` / `getRootDir()` (the only surfaces the orchestrator uses) and + * a recording `emitEvent` so tests can assert observable events. + */ +export function makeHarness(): TestHarness { + const projectRoot = mkdtempSync(join(tmpdir(), "ce-session-test-")); + const db = new Database(join(projectRoot, ".fusion"), { inMemory: true }); + db.init(); + + const emitted: Array<{ event: string; data: unknown }> = []; + + const taskStore = { + getDatabase: () => db, + getRootDir: () => projectRoot, + } as unknown as PluginContext["taskStore"]; + + const ctx: PluginContext = { + pluginId: "fusion-plugin-compound-engineering", + taskStore, + settings: {}, + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, + emitEvent: (event: string, data: unknown) => { + emitted.push({ event, data }); + }, + }; + + return { + db, + projectRoot, + ctx, + emitted, + close: () => db.close(), + }; +} + +/** + * A scripted fake interactive session: each prompt/answer advances a cursor and + * the next `nextEvent()` yields the scripted event for that turn. Mirrors the + * U4 seam tests' scripted fake. + */ +export function makeScriptedSession(script: InteractiveAiSessionEvent[]): InteractiveAiSession { + let cursor = -1; + return { + prompt: vi.fn(async () => { + cursor++; + }), + answer: vi.fn(async () => { + cursor++; + }), + nextEvent: vi.fn(async () => script[Math.min(cursor, script.length - 1)]), + dispose: vi.fn(), + }; +} + +/** A factory that returns the given scripted session. */ +export function scriptedFactory(session: InteractiveAiSession): CreateInteractiveAiSessionFactory { + return vi.fn(async () => ({ session, sessionFile: "/tmp/ce.json" })); +} + +/** A session whose first turn never produces an event (forces a turn timeout). */ +export function hangingSession(): InteractiveAiSession { + return { + prompt: vi.fn(async () => undefined), + answer: vi.fn(async () => undefined), + nextEvent: vi.fn(() => new Promise(() => undefined)), + dispose: vi.fn(), + }; +} diff --git a/plugins/fusion-plugin-compound-engineering/src/__tests__/manifest.test.ts b/plugins/fusion-plugin-compound-engineering/src/__tests__/manifest.test.ts index ef9e9b66e3..bca6b5765b 100644 --- a/plugins/fusion-plugin-compound-engineering/src/__tests__/manifest.test.ts +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/manifest.test.ts @@ -31,8 +31,21 @@ describe("compound engineering plugin manifest", () => { expect(manifest.dashboardViews).toEqual(plugin.dashboardViews); }); - it("ships an empty routes scaffold (U1)", () => { - expect(plugin.routes).toEqual([]); + it("registers the session orchestration routes (U5)", () => { + const paths = (plugin.routes ?? []).map((r) => `${r.method} ${r.path}`); + expect(paths).toEqual( + expect.arrayContaining([ + "POST /sessions", + "POST /sessions/:id/answer", + "POST /sessions/:id/resume", + "GET /sessions/:id", + "GET /sessions", + ]), + ); + }); + + it("wires onSchemaInit for the plugin-local CE tables (U5)", () => { + expect(typeof plugin.hooks.onSchemaInit).toBe("function"); }); it("registers the bundled CE pipeline-stage skills on plugin and manifest (U2)", () => { diff --git a/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-flow.test.ts b/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-flow.test.ts new file mode 100644 index 0000000000..65423f822b --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-flow.test.ts @@ -0,0 +1,94 @@ +import { existsSync, readFileSync } from "node:fs"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { InteractiveAiSessionEvent, PlanningQuestion } from "@fusion/core"; +import { CeOrchestrator, CE_EVENTS } from "../session/orchestrator.js"; +import { registerStage, getStage } from "../session/stage-registry.js"; +import { makeHarness, makeScriptedSession, type TestHarness } from "./_harness.js"; + +const QUESTION: PlanningQuestion = { + id: "q1", + type: "text", + question: "What is the topic?", +}; + +let h: TestHarness; +beforeEach(() => { + h = makeHarness(); +}); +afterEach(() => { + h.close(); +}); + +function makeOrch(script: InteractiveAiSessionEvent[]) { + const session = makeScriptedSession(script); + return new CeOrchestrator({ + ctx: h.ctx, + createInteractiveAiSession: vi.fn(async () => ({ session })), + projectRoot: h.projectRoot, + turnTimeoutMs: 5000, + }); +} + +describe("orchestrator happy path", () => { + it("start → question → answer → complete writes the artifact to the conventional location", async () => { + const orch = makeOrch([ + { type: "question", data: QUESTION }, + { type: "complete", data: { artifact: "# Brainstorm\n\nThe plan.\n" } }, + ]); + + const started = await orch.start("brainstorm", { openingMessage: "let's brainstorm widgets" }); + expect(started.session.status).toBe("awaiting_input"); + expect(started.session.currentQuestion?.id).toBe("q1"); + + const done = await orch.answer(started.session.id, "q1", "widgets"); + expect(done.event?.type).toBe("complete"); + expect(done.session.status).toBe("completed"); + + // Artifact written to docs/brainstorms/ (the stage's conventional location). + const artifactPath = done.session.artifactPath!; + expect(artifactPath).toContain("docs/brainstorms/"); + expect(existsSync(artifactPath)).toBe(true); + expect(readFileSync(artifactPath, "utf-8")).toContain("# Brainstorm"); + + // Observable completion event emitted. + expect(h.emitted.map((e) => e.event)).toContain(CE_EVENTS.completed); + }); + + it("runs a SECOND stage through the SAME orchestrator with only a registry-data entry (no new route/store code)", async () => { + // Adding a stage = data only. + registerStage({ stageId: "compound", skillId: "ce-compound", artifactLocation: "docs/solutions/" }); + expect(getStage("compound")?.skillId).toBe("ce-compound"); + + const orch = makeOrch([{ type: "complete", data: { artifact: "# Learning\n" } }]); + const started = await orch.start("compound", { openingMessage: "document this" }); + expect(started.event?.type).toBe("complete"); + expect(started.session.stage).toBe("compound"); + expect(started.session.status).toBe("completed"); + expect(started.session.artifactPath).toContain("docs/solutions/"); + expect(readFileSync(started.session.artifactPath!, "utf-8")).toContain("# Learning"); + }); +}); + +describe("orchestrator error + retry", () => { + it("agent error → status error, progress preserved, observable event; retry resumes to the question", async () => { + const orch = makeOrch([ + { type: "question", data: QUESTION }, + { type: "error", data: { message: "model overloaded" } }, + ]); + + const started = await orch.start("brainstorm", { openingMessage: "topic" }); + expect(started.session.currentQuestion?.id).toBe("q1"); + + const errored = await orch.answer(started.session.id, "q1", "answer-text"); + expect(errored.session.status).toBe("error"); + expect(errored.session.error).toContain("model overloaded"); + // Progress preserved: history retained. + expect(errored.session.conversationHistory.length).toBeGreaterThan(0); + expect(h.emitted.map((e) => e.event)).toContain(CE_EVENTS.error); + + // Retry: resume() moves an errored session forward. (Error keeps it + // resumable; resume reads persisted state — the no-loss anchor.) + const state = orch.getState(errored.session.id)!; + expect(state.conversationHistory.length).toBeGreaterThan(0); + }); +}); diff --git a/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-interrupt-resume.test.ts b/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-interrupt-resume.test.ts new file mode 100644 index 0000000000..04cfbb693d --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-interrupt-resume.test.ts @@ -0,0 +1,128 @@ +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import type { InteractiveAiSession, InteractiveAiSessionEvent, PlanningQuestion } from "@fusion/core"; +import { vi } from "vitest"; +import { CeOrchestrator, CE_EVENTS } from "../session/orchestrator.js"; +import { CeSessionStore, getCeSessionStore } from "../session/session-store.js"; +import { makeHarness, type TestHarness } from "./_harness.js"; + +/** + * CHARACTERIZATION TEST — written first (U5 execution note: cover the + * no-silent-loss invariant before the happy path). Asserts that an interrupted + * mid-question session auto-saves progress, lands in `interrupted`, emits an + * observable event, and resumes to the SAME question with full history. + */ + +const QUESTION: PlanningQuestion = { + id: "q1", + type: "single_select", + question: "Which direction?", + options: [ + { id: "a", label: "A" }, + { id: "b", label: "B" }, + ], +}; + +let h: TestHarness; + +beforeEach(() => { + h = makeHarness(); +}); + +afterEach(() => { + h.close(); +}); + +/** + * Session that yields a question on turn 1, then HANGS on the next turn + * (the answer turn never produces an event) — forcing a turn timeout. + */ +function questionThenHangSession(): InteractiveAiSession { + let cursor = -1; + return { + prompt: vi.fn(async () => { + cursor++; + }), + answer: vi.fn(async () => { + cursor++; + }), + nextEvent: vi.fn(async (): Promise => { + if (cursor === 0) return { type: "question", data: QUESTION }; + // turn 2+ hangs forever + return new Promise(() => undefined); + }), + dispose: vi.fn(), + }; +} + +describe("interrupt + resume (no silent loss)", () => { + it("auto-saves progress on a turn timeout, marks interrupted, emits an event", async () => { + const session = questionThenHangSession(); + const orch = new CeOrchestrator({ + ctx: h.ctx, + createInteractiveAiSession: vi.fn(async () => ({ session })), + projectRoot: h.projectRoot, + turnTimeoutMs: 20, + }); + + const started = await orch.start("brainstorm", { openingMessage: "kick off" }); + expect(started.event?.type).toBe("question"); + expect(started.session.status).toBe("awaiting_input"); + expect(started.session.currentQuestion?.id).toBe("q1"); + + // Answering triggers the next turn, which hangs → timeout → interrupted. + const interrupted = await orch.answer(started.session.id, "q1", "a"); + expect(interrupted.session.status).toBe("interrupted"); + // Progress preserved: full history including the question and the answer. + const history = interrupted.session.conversationHistory; + expect(history.some((t) => t.text.includes("kick off"))).toBe(true); + expect(history.some((t) => t.text.includes("question"))).toBe(true); + expect(history.some((t) => t.text.includes("\"answer\""))).toBe(true); + + // Observable event emitted — never silent loss. + expect(h.emitted.map((e) => e.event)).toContain(CE_EVENTS.interrupted); + }); + + it("recoverStaleSessions restores an awaiting_input session left by a crash, and resume returns the same question with full history", () => { + // Simulate a session persisted mid-question whose process died: status is + // awaiting_input with currentQuestion set, lastActivity well past the stale + // band (interval-relative). + const store = new CeSessionStore(h.db); + const created = store.create({ stage: "brainstorm", turnIntervalMs: 1000 }); + store.appendHistory(created.id, { role: "user", text: "kick off", at: new Date().toISOString() }); + store.appendHistory(created.id, { role: "agent", text: JSON.stringify({ question: QUESTION }), at: new Date().toISOString() }); + store.update(created.id, { + status: "awaiting_input", + currentQuestion: QUESTION, + // 10× interval old → unambiguously stale. + lastActivityAt: Date.now() - 10_000, + }); + + const recovered = store.recoverStaleSessions(); + expect(recovered).toContain(created.id); + + const after = store.get(created.id)!; + // Awaiting-input session with a question stays resumable, not dropped. + expect(after.status).toBe("awaiting_input"); + expect(after.currentQuestion?.id).toBe("q1"); + + // Resume via the orchestrator returns to the same question + full history. + const orch = new CeOrchestrator({ ctx: h.ctx, createInteractiveAiSession: vi.fn(), projectRoot: h.projectRoot }); + const resumed = orch.resume(created.id); + expect(resumed.session.status).toBe("awaiting_input"); + expect(resumed.session.currentQuestion?.id).toBe("q1"); + expect(resumed.session.conversationHistory).toHaveLength(2); + }); + + it("a crash with no pending question is marked interrupted (progress preserved), not silently dropped", () => { + const store = getCeSessionStore(h.ctx); + const created = store.create({ stage: "brainstorm", turnIntervalMs: 1000 }); + store.appendHistory(created.id, { role: "user", text: "kick off", at: new Date().toISOString() }); + store.update(created.id, { status: "active", lastActivityAt: Date.now() - 10_000 }); + + store.recoverStaleSessions(); + const after = store.get(created.id)!; + expect(after.status).toBe("interrupted"); + expect(after.error).toMatch(/progress preserved/i); + expect(after.conversationHistory).toHaveLength(1); + }); +}); diff --git a/plugins/fusion-plugin-compound-engineering/src/__tests__/session-routes.test.ts b/plugins/fusion-plugin-compound-engineering/src/__tests__/session-routes.test.ts new file mode 100644 index 0000000000..346085acf9 --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/session-routes.test.ts @@ -0,0 +1,74 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { PluginContext, PluginRouteResponse } from "@fusion/core"; +import { createSessionRoutes } from "../routes/session-routes.js"; +import { makeHarness, type TestHarness } from "./_harness.js"; + +/** + * Routes-level smoke test for the POLLING transport. Exercises validation and + * the get-session-state read path that clients poll. The orchestrator's live + * interactive flow is covered by orchestrator-flow.test.ts; here createInter- + * activeAiSession is absent (non-engine context), so `start` returns a 400 — + * which is the correct, non-hanging behavior. + */ + +let h: TestHarness; +beforeEach(() => { + h = makeHarness(); +}); +afterEach(() => { + h.close(); + vi.restoreAllMocks(); +}); + +function route(method: string, path: string) { + const r = createSessionRoutes().find((x) => x.method === method && x.path === path); + if (!r) throw new Error(`route ${method} ${path} not found`); + return r; +} + +async function call(method: string, path: string, req: unknown, ctx: PluginContext): Promise { + return (await route(method, path).handler(req, ctx)) as PluginRouteResponse; +} + +describe("session routes (polling transport)", () => { + it("exposes start / answer / resume / get-session-state / list", () => { + const paths = createSessionRoutes().map((r) => `${r.method} ${r.path}`); + expect(paths).toEqual( + expect.arrayContaining([ + "POST /sessions", + "POST /sessions/:id/answer", + "POST /sessions/:id/resume", + "GET /sessions/:id", + "GET /sessions", + ]), + ); + }); + + it("POST /sessions requires a stage", async () => { + const res = await call("POST", "/sessions", { body: {} }, h.ctx); + expect(res.status).toBe(400); + }); + + it("POST /sessions without engine interactive factory returns a clean 400 (no hang)", async () => { + const res = await call("POST", "/sessions", { body: { stage: "brainstorm", message: "go" } }, h.ctx); + expect(res.status).toBe(400); + expect((res.body as { error: string }).error).toMatch(/not available/i); + }); + + it("GET /sessions/:id returns 404 for an unknown id and 200 for a known one", async () => { + const missing = await call("GET", "/sessions/:id", { params: { id: "nope" } }, h.ctx); + expect(missing.status).toBe(404); + + // Seed a session directly so the poll route has something to return. + const { getCeSessionStore } = await import("../session/session-store.js"); + const seeded = getCeSessionStore(h.ctx).create({ stage: "brainstorm" }); + const found = await call("GET", "/sessions/:id", { params: { id: seeded.id } }, h.ctx); + expect(found.status).toBe(200); + expect((found.body as { session: { id: string } }).session.id).toBe(seeded.id); + }); + + it("POST /sessions/:id/answer validates questionId and response", async () => { + const res = await call("POST", "/sessions/:id/answer", { params: { id: "x" }, body: {} }, h.ctx); + expect(res.status).toBe(400); + }); +}); diff --git a/plugins/fusion-plugin-compound-engineering/src/__tests__/session-store.test.ts b/plugins/fusion-plugin-compound-engineering/src/__tests__/session-store.test.ts new file mode 100644 index 0000000000..6c875d787b --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/session-store.test.ts @@ -0,0 +1,77 @@ +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { CeSessionStore, STALE_INTERVAL_MULTIPLE } from "../session/session-store.js"; +import { ensureCeSchema } from "../schema.js"; +import { makeHarness, type TestHarness } from "./_harness.js"; + +let h: TestHarness; +beforeEach(() => { + h = makeHarness(); +}); +afterEach(() => { + h.close(); +}); + +describe("ensureCeSchema", () => { + it("is idempotent (safe to run repeatedly)", () => { + ensureCeSchema(h.db); + ensureCeSchema(h.db); + const cols = h.db.prepare("PRAGMA table_info(ce_sessions)").all() as Array<{ name: string }>; + const names = cols.map((c) => c.name); + expect(names).toEqual( + expect.arrayContaining([ + "id", + "stage", + "status", + "currentQuestion", + "conversationHistory", + "projectId", + "lastActivityAt", + ]), + ); + }); +}); + +describe("CeSessionStore CRUD + JSON round-trip", () => { + it("creates, reads back, and round-trips JSON fields", () => { + const store = new CeSessionStore(h.db); + const created = store.create({ stage: "brainstorm", projectId: "p1" }); + expect(created.status).toBe("launching"); + + store.update(created.id, { + currentQuestion: { id: "q", type: "confirm", question: "ok?" }, + status: "awaiting_input", + }); + store.appendHistory(created.id, { role: "user", text: "hi", at: "2026-06-02T00:00:00Z" }); + + const read = store.get(created.id)!; + expect(read.currentQuestion?.id).toBe("q"); + expect(read.conversationHistory).toHaveLength(1); + expect(read.status).toBe("awaiting_input"); + expect(read.projectId).toBe("p1"); + }); +}); + +describe("interval-relative staleness (FN-4172 rubric)", () => { + it("does NOT misclassify a healthy-but-slow session as stale", () => { + const store = new CeSessionStore(h.db); + const s = store.create({ stage: "brainstorm", turnIntervalMs: 1000 }); + store.update(s.id, { status: "active" }); + + const now = Date.now(); + // 2.5× the interval old: slow, but within the 3× band → NOT stale. + const slow = store.update(s.id, { status: "active", lastActivityAt: now - 2_500 })!; + expect(STALE_INTERVAL_MULTIPLE).toBe(3); + expect(store.isStale(slow, now)).toBe(false); + + // 4× the interval old → stale. + const stalled = store.update(s.id, { status: "active", lastActivityAt: now - 4_000 })!; + expect(store.isStale(stalled, now)).toBe(true); + }); + + it("never flags terminal sessions as stale regardless of age", () => { + const store = new CeSessionStore(h.db); + const s = store.create({ stage: "brainstorm", turnIntervalMs: 1000 }); + const completed = store.update(s.id, { status: "completed", lastActivityAt: Date.now() - 1_000_000 })!; + expect(store.isStale(completed)).toBe(false); + }); +}); diff --git a/plugins/fusion-plugin-compound-engineering/src/__tests__/skill-reachability.test.ts b/plugins/fusion-plugin-compound-engineering/src/__tests__/skill-reachability.test.ts new file mode 100644 index 0000000000..7ed2116696 --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/skill-reachability.test.ts @@ -0,0 +1,65 @@ +import { existsSync, mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { installBundledCeSkills } from "../skill-installation.js"; +import { resolveStageSkillCwd, buildStageSystemPrompt } from "../session/orchestrator.js"; +import { getStage } from "../session/stage-registry.js"; + +/** + * CARRY-FORWARD (U2 → U5): prove the launched stage's ce-* skill is REACHABLE + * for the session. + * + * HONEST SCOPE — proved at the installer/resolver layer (exactly as U2 did), + * NOT at the live session layer. The U4 `CreateInteractiveAiSessionOptions` + * surface carries only `cwd` (no `requestedSkillNames`/`additionalSkillPaths`/ + * `skillSelection`), so the orchestrator cannot hand the session an explicit + * skill-discovery path. The closest honest wiring is: + * 1. point the session `cwd` at the install-target root (where pi's + * DefaultResourceLoader can discover `/SKILL.md`), and + * 2. name the skill id in the system prompt. + * This test asserts BOTH: the resolved cwd contains the stage's installed + * SKILL.md, and the system prompt names the stage's skill id. + * + * A complete fix needs U4's options to gain a forwarded + * `requestedSkillNames`/`additionalSkillPaths` field — flagged as a carry- + * forward for U6/follow-up. + */ + +describe("stage skill reachability (carry-forward, resolver-layer proof)", () => { + let targets: string[] = []; + + afterEach(() => { + targets = []; + vi.restoreAllMocks(); + }); + + it("the resolved session cwd contains the stage's installed SKILL.md", () => { + const target = mkdtempSync(join(tmpdir(), "ce-skill-target-")); + targets.push(target); + + // Install bundled skills into a plugin-local target. + const { results } = installBundledCeSkills({ targetRoot: target }); + expect(results.every((r) => r.outcome === "installed" || r.outcome === "skipped")).toBe(true); + + const stage = getStage("brainstorm")!; + // The orchestrator resolves the discovery cwd to the default install-target + // root. For this isolation test we assert the SAME structure exists at the + // explicit target we installed into (resolveStageSkillCwd returns the + // default root, which the production onLoad install populates identically). + const installedSkillMd = join(target, stage.skillId, "SKILL.md"); + expect(existsSync(installedSkillMd)).toBe(true); + + // resolveStageSkillCwd returns a plugin-local directory (never a global one). + const cwd = resolveStageSkillCwd(stage); + expect(cwd).toMatch(/\.fusion-ce-skills$/); + }); + + it("the stage system prompt names the stage's ce-* skill id", () => { + const stage = getStage("brainstorm")!; + const prompt = buildStageSystemPrompt(stage); + expect(prompt).toContain(stage.skillId); // "ce-brainstorm" + expect(prompt).toContain("question"); + expect(prompt).toContain("complete"); + }); +}); diff --git a/plugins/fusion-plugin-compound-engineering/src/index.ts b/plugins/fusion-plugin-compound-engineering/src/index.ts index ef2e33cd8f..11b7178310 100644 --- a/plugins/fusion-plugin-compound-engineering/src/index.ts +++ b/plugins/fusion-plugin-compound-engineering/src/index.ts @@ -1,6 +1,8 @@ import { definePlugin } from "@fusion/plugin-sdk"; import { COMPOUND_ENGINEERING_SKILLS } from "./skills.js"; import { installBundledCeSkills } from "./skill-installation.js"; +import { ensureCeSchema } from "./schema.js"; +import { createSessionRoutes } from "./routes/session-routes.js"; export { CompoundEngineeringDashboardView } from "./dashboard-view.js"; export { COMPOUND_ENGINEERING_SKILLS } from "./skills.js"; @@ -10,6 +12,10 @@ export { resolveDefaultInstallTargetRoot, isPluginLocalPath, } from "./skill-installation.js"; +export { ensureCeSchema } from "./schema.js"; +export { CeSessionStore, getCeSessionStore } from "./session/session-store.js"; +export { CeOrchestrator } from "./session/orchestrator.js"; +export { getStage, listStages, registerStage } from "./session/stage-registry.js"; const plugin = definePlugin({ manifest: { @@ -24,6 +30,9 @@ const plugin = definePlugin({ state: "installed", skills: COMPOUND_ENGINEERING_SKILLS, hooks: { + // Idempotent DDL for the plugin-local CE tables (ce_sessions). Runs against + // the same DB route handlers reach via ctx.taskStore.getDatabase() (U5). + onSchemaInit: ensureCeSchema, // Install the bundled, pinned ce-* SKILL.md files into a plugin-local, // discoverable directory on load. The engine ingests // PluginSkillContribution only as a name; physical discovery requires the @@ -51,7 +60,7 @@ const plugin = definePlugin({ } }, }, - routes: [], + routes: [...createSessionRoutes()], dashboardViews: [ { viewId: "compound-engineering", diff --git a/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts b/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts new file mode 100644 index 0000000000..1e7a62bbc6 --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts @@ -0,0 +1,134 @@ +import type { PluginContext, PluginRouteDefinition, PluginRouteResponse } from "@fusion/core"; +import { CeOrchestrator } from "../session/orchestrator.js"; +import { getCeSessionStore } from "../session/session-store.js"; + +/** + * Session routes (U5): start / answer / resume / get-session-state. + * + * STREAMING TRANSPORT — HONEST STATEMENT. + * Plugin routes return `{status, body}` with no native server-push; the loader + * `emitEvent` is a logging stub, so there is no real plugin→client push path + * today. v1 therefore uses POLLING: clients poll `GET /sessions/:id` for the + * current persisted state (status, currentQuestion, conversationHistory). This + * keeps U5 plugin-local and shippable and uses NO raw EventSource. The + * orchestrator still emits observable events via `ctx.emitEvent` (a no-silent- + * loss requirement); turning those into true client push needs a host + * event-publish seam (publish-to-`/api/events`) — that is a carry-forward for + * U6/follow-up, not faked here as push. + */ + +interface RouteRequest { + params: Record; + query?: Record; + body?: unknown; +} + +/** + * Cache the orchestrator per TaskStore so live in-process interactive-session + * handles survive across requests within a process (a fresh orchestrator per + * request would lose the live handle needed to answer a question). + */ +const orchestratorCache = new WeakMap(); + +function getOrchestrator(ctx: PluginContext): CeOrchestrator { + const key = ctx.taskStore as object; + const cached = orchestratorCache.get(key); + if (cached) return cached; + const orch = new CeOrchestrator({ ctx }); + orchestratorCache.set(key, orch); + return orch; +} + +function badRequest(message: string): PluginRouteResponse { + return { status: 400, body: { error: message } }; +} + +function asString(v: unknown): string | undefined { + return typeof v === "string" && v.length > 0 ? v : undefined; +} + +export function createSessionRoutes(): PluginRouteDefinition[] { + return [ + { + method: "POST", + path: "/sessions", + description: "Start an interactive CE stage session.", + handler: async (req: unknown, ctx: PluginContext): Promise => { + const body = (req as RouteRequest).body as Record | undefined; + const stageId = asString(body?.stage); + const openingMessage = asString(body?.message) ?? ""; + if (!stageId) return badRequest("`stage` is required"); + + const orch = getOrchestrator(ctx); + try { + const result = await orch.start(stageId, { + openingMessage, + projectId: asString(body?.projectId) ?? null, + }); + return { status: 201, body: { session: result.session, event: result.event } }; + } catch (err) { + return { status: 400, body: { error: err instanceof Error ? err.message : String(err) } }; + } + }, + }, + { + method: "POST", + path: "/sessions/:id/answer", + description: "Answer the awaiting question and continue the session.", + handler: async (req: unknown, ctx: PluginContext): Promise => { + const request = req as RouteRequest; + const id = request.params.id; + const body = request.body as Record | undefined; + const questionId = asString(body?.questionId); + if (!questionId) return badRequest("`questionId` is required"); + if (!("response" in (body ?? {}))) return badRequest("`response` is required"); + + const orch = getOrchestrator(ctx); + try { + const result = await orch.answer(id, questionId, (body as Record).response); + return { status: 200, body: { session: result.session, event: result.event } }; + } catch (err) { + return { status: 409, body: { error: err instanceof Error ? err.message : String(err) } }; + } + }, + }, + { + method: "POST", + path: "/sessions/:id/resume", + description: "Resume an awaiting_input or interrupted session to its current question.", + handler: async (req: unknown, ctx: PluginContext): Promise => { + const id = (req as RouteRequest).params.id; + const orch = getOrchestrator(ctx); + try { + const result = orch.resume(id); + return { status: 200, body: { session: result.session } }; + } catch (err) { + return { status: 404, body: { error: err instanceof Error ? err.message : String(err) } }; + } + }, + }, + { + method: "GET", + path: "/sessions/:id", + description: "Get current persisted session state (polling transport).", + handler: async (req: unknown, ctx: PluginContext): Promise => { + const id = (req as RouteRequest).params.id; + const session = getCeSessionStore(ctx).get(id); + if (!session) return { status: 404, body: { error: `Session ${id} not found` } }; + return { status: 200, body: { session } }; + }, + }, + { + method: "GET", + path: "/sessions", + description: "List CE sessions (optionally filtered by status/stage).", + handler: async (req: unknown, ctx: PluginContext): Promise => { + const query = (req as RouteRequest).query ?? {}; + const status = typeof query.status === "string" ? query.status : undefined; + const stage = typeof query.stage === "string" ? query.stage : undefined; + const sessions = getCeSessionStore(ctx).list({ status: status as never, stage }); + return { status: 200, body: { sessions } }; + }, + }, + ]; +} diff --git a/plugins/fusion-plugin-compound-engineering/src/schema.ts b/plugins/fusion-plugin-compound-engineering/src/schema.ts new file mode 100644 index 0000000000..8c31236bf0 --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/schema.ts @@ -0,0 +1,50 @@ +import type { Database } from "@fusion/core"; + +/** + * Idempotent DDL for the Compound Engineering plugin-local tables (U5). + * + * Wired via `hooks.onSchemaInit` and run against the same DB that route + * handlers reach through `ctx.taskStore.getDatabase()` (the sanctioned + * plugin-table access path; `PluginContext` exposes no `db` handle and the + * loader `emitEvent` is a logging stub — see the U5 storage/event seam note). + * + * `ce_sessions` is the no-silent-loss core: every interactive stage session is + * persisted here so an interrupt/error never destroys progress (lesson: + * docs/incidents/2026-05-23-lost-work-tasks.md). The `currentQuestion` and + * `conversationHistory` columns are JSON; resume reconstructs the awaiting + * question and full history from them. + * + * `lastActivityAt` is an interval-relative liveness field (epoch millis of the + * last produced event). Staleness is judged relative to the session's + * configured turn interval, NOT by raw last-event age, so a healthy-but-slow + * agent turn is not misclassified stale (docs/fn-4172-heartbeat-investigation.md). + */ +export function ensureCeSchema(db: Database): void { + db.exec(` + CREATE TABLE IF NOT EXISTS ce_sessions ( + id TEXT PRIMARY KEY, + stage TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ( + 'launching','active','awaiting_input','completed','error','interrupted' + )), + currentQuestion TEXT, + conversationHistory TEXT NOT NULL DEFAULT '[]', + projectId TEXT, + artifactPath TEXT, + error TEXT, + turnIntervalMs INTEGER NOT NULL DEFAULT 120000, + lastActivityAt INTEGER NOT NULL, + createdAt TEXT NOT NULL, + updatedAt TEXT NOT NULL + ); + + CREATE INDEX IF NOT EXISTS idxCeSessionsStatusUpdated + ON ce_sessions(status, updatedAt DESC, id); + + CREATE INDEX IF NOT EXISTS idxCeSessionsStageCreated + ON ce_sessions(stage, createdAt DESC, id); + + CREATE INDEX IF NOT EXISTS idxCeSessionsProject + ON ce_sessions(projectId, updatedAt DESC, id); + `); +} diff --git a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts new file mode 100644 index 0000000000..1e8d2ffb3d --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts @@ -0,0 +1,362 @@ +import { mkdirSync, writeFileSync } from "node:fs"; +import { dirname, isAbsolute, join } from "node:path"; +import type { + CreateInteractiveAiSessionFactory, + InteractiveAiSession, + InteractiveAiSessionEvent, + PlanningQuestion, + PluginContext, +} from "@fusion/core"; +import { resolveDefaultInstallTargetRoot } from "../skill-installation.js"; +import type { CeSession, CeSessionStore } from "./session-store.js"; +import { getCeSessionStore } from "./session-store.js"; +import { getStage, type CeStageDefinition } from "./stage-registry.js"; + +/** Default per-turn timeout. A turn that exceeds this is treated as a stall. */ +const DEFAULT_TURN_TIMEOUT_MS = 120000; + +/** + * Observable event names emitted via `ctx.emitEvent`. The no-silent-loss + * invariant requires that interrupt/error ALWAYS emit one of these AND persist + * progress first. + */ +export const CE_EVENTS = { + turn: "compound-engineering:session-turn", + question: "compound-engineering:session-question", + completed: "compound-engineering:session-completed", + error: "compound-engineering:session-error", + interrupted: "compound-engineering:session-interrupted", +} as const; + +export class CeTurnTimeoutError extends Error { + constructor(ms: number) { + super(`CE session turn timed out after ${ms}ms`); + this.name = "CeTurnTimeoutError"; + } +} + +function timeoutAfter(ms: number): Promise { + return new Promise((_, reject) => { + globalThis.setTimeout(() => reject(new CeTurnTimeoutError(ms)), ms).unref?.(); + }); +} + +export interface OrchestratorDeps { + ctx: PluginContext; + /** + * Interactive-session factory. Defaults to `ctx.createInteractiveAiSession` + * (route contexts only); injectable for deterministic, scripted-fake tests. + */ + createInteractiveAiSession?: CreateInteractiveAiSessionFactory; + /** Project root used for the session cwd and artifact writes. */ + projectRoot?: string; + /** Override the per-turn timeout (ms). */ + turnTimeoutMs?: number; +} + +/** + * CARRY-FORWARD (U2 → U5) — skill discovery wiring. + * + * U2 proved a `PluginSkillContribution` is NOT auto-ingested by the engine + * skill-resolver; a physical install onto a cwd-discoverable path is required. + * The U4 `CreateInteractiveAiSessionOptions` surface carries ONLY `cwd` (plus + * systemPrompt/tools/provider/model) — it has NO `requestedSkillNames` / + * `additionalSkillPaths` / `skillSelection` field. So we cannot point the + * session at the install target the way `createFnAgent`'s `skills`/ + * `skillSelection` options would. + * + * The closest honest thing we CAN do today: + * 1. Set the session `cwd` to a root under which the installed ce-* skills are + * discoverable by pi's DefaultResourceLoader (the install-target root). + * 2. Name the required skill id in the systemPrompt so the agent is told which + * ce-* skill to apply (protocol-level instruction). + * + * This function computes that cwd. We PROVE reachability at the installer/ + * resolver layer (like U2 did) in the tests — the U4 options surface cannot yet + * carry an explicit skill-path, so a complete fix needs U4's options to gain a + * `requestedSkillNames`/`additionalSkillPaths` field forwarded into + * `createFnAgent`. That gap is documented as a carry-forward for U6/follow-up. + */ +export function resolveStageSkillCwd(stage: CeStageDefinition, projectRoot?: string): string { + // The install target root holds `/SKILL.md` for each installed + // skill; using it as the discovery root makes the stage's skill loadable. + void stage; + void projectRoot; + return resolveDefaultInstallTargetRoot(); +} + +/** + * Build the system prompt: instruct the agent to (a) apply the named ce-* skill + * and (b) emit the JSON question/complete protocol the U4 seam parses. + */ +export function buildStageSystemPrompt(stage: CeStageDefinition): string { + return [ + `You are running the Compound Engineering "${stage.stageId}" stage.`, + `Apply the bundled skill "${stage.skillId}" (its SKILL.md is discoverable in your working directory).`, + "", + "Drive the stage as an interactive question/answer flow. On every turn respond with ONLY a JSON object:", + ' - To ask the user something: {"type":"question","data":{"id":"","type":"single_select|multi_select|text|confirm","question":"...","options":[{"id":"..","label":".."}]}}', + ' - When the stage is finished: {"type":"complete","data":{"artifact":"", ...}}', + "No markdown fences, no prose outside the JSON object.", + ].join("\n"); +} + +export interface StartStageOptions { + /** Opening user message (the stage prompt / topic). */ + openingMessage: string; + projectId?: string | null; +} + +/** Result of a single orchestrator step (start / answer / resume). */ +export interface CeStepResult { + session: CeSession; + /** The event the seam produced for this step, if a turn ran. */ + event?: InteractiveAiSessionEvent; +} + +/** + * Drives a stage's interactive skill session: streams thinking/text, surfaces + * questions (persisted as `awaiting_input`), accepts answers, and on `complete` + * writes the artifact to the stage's conventional location. On interrupt/error + * it AUTO-SAVES progress and emits an observable event — never silent loss. + * + * Liveness uses the interval-relative rubric (CeSessionStore.isStale): a slow + * turn is not misclassified stale. + */ +export class CeOrchestrator { + private readonly ctx: PluginContext; + private readonly store: CeSessionStore; + private readonly factory: CreateInteractiveAiSessionFactory | undefined; + private readonly projectRoot: string; + private readonly turnTimeoutMs: number; + /** Live in-memory session handles keyed by ce_session id. */ + private readonly live = new Map(); + + constructor(deps: OrchestratorDeps) { + this.ctx = deps.ctx; + this.store = getCeSessionStore(deps.ctx); + this.factory = deps.createInteractiveAiSession ?? deps.ctx.createInteractiveAiSession; + this.projectRoot = deps.projectRoot ?? deps.ctx.taskStore.getRootDir(); + this.turnTimeoutMs = deps.turnTimeoutMs ?? DEFAULT_TURN_TIMEOUT_MS; + } + + /** Start a fresh session for a registered stage and run the opening turn. */ + async start(stageId: string, opts: StartStageOptions): Promise { + const stage = getStage(stageId); + if (!stage) throw new Error(`Unknown CE stage: ${stageId}`); + if (!this.factory) { + throw new Error( + "Interactive AI sessions are not available (createInteractiveAiSession is only injected on route contexts with the engine loaded).", + ); + } + + let session = this.store.create({ + stage: stageId, + projectId: opts.projectId ?? null, + turnIntervalMs: this.turnTimeoutMs, + }); + this.store.appendHistory(session.id, { role: "user", text: opts.openingMessage, at: new Date().toISOString() }); + + const cwd = resolveStageSkillCwd(stage, this.projectRoot); + const systemPrompt = buildStageSystemPrompt(stage); + + let interactive; + try { + interactive = await this.factory({ cwd, systemPrompt, tools: "coding" }); + } catch (err) { + return { session: this.failSession(session.id, err), event: undefined }; + } + this.live.set(session.id, interactive.session); + session = this.store.update(session.id, { status: "active" }) ?? session; + + return this.runTurn(session.id, () => interactive.session.prompt(opts.openingMessage), interactive.session); + } + + /** Answer the awaiting question and continue the loop. */ + async answer(sessionId: string, questionId: string, response: unknown): Promise { + const session = this.requireSession(sessionId); + const live = this.live.get(sessionId); + if (!live) { + throw new Error(`Session ${sessionId} has no live handle in this process; call resume() first.`); + } + if (session.status !== "awaiting_input") { + throw new Error(`Session ${sessionId} is not awaiting input (status=${session.status}).`); + } + this.store.appendHistory(sessionId, { + role: "user", + text: JSON.stringify({ answer: response, questionId }), + at: new Date().toISOString(), + }); + this.store.update(sessionId, { status: "active", currentQuestion: null }); + return this.runTurn(sessionId, () => live.answer(questionId, response), live); + } + + /** + * Resume an `awaiting_input` or `interrupted` session. Returns its persisted + * state pointed back at the current question. A resumed session with no live + * in-process handle requires a fresh interactive session to continue past the + * current question (the persisted question/history is the recovery anchor). + */ + resume(sessionId: string): CeStepResult { + const session = this.requireSession(sessionId); + if (session.status === "interrupted" || session.status === "error") { + // Resumable transition (retry for `error`, resume for `interrupted`): if a + // question is still pending, return to awaiting_input; otherwise mark + // active so the caller can re-run the turn with fresh input. The persisted + // question/history is the no-loss recovery anchor either way. + const next = session.currentQuestion + ? this.store.update(sessionId, { status: "awaiting_input", error: null }) ?? session + : this.store.update(sessionId, { status: "active", error: null }) ?? session; + return { session: next }; + } + // Already awaiting_input (or terminal completed) — return as-is (idempotent). + return { session }; + } + + /** Read-through accessor for routes. */ + getState(sessionId: string): CeSession | undefined { + return this.store.get(sessionId); + } + + /** + * Run one turn behind a timeout race, persist the resulting event, and on a + * turn-level failure auto-save + emit. The `driver` performs the prompt/answer + * against the live session; we then pull exactly one event. + */ + private async runTurn( + sessionId: string, + driver: () => Promise, + live: InteractiveAiSession, + ): Promise { + let event: InteractiveAiSessionEvent; + try { + event = await Promise.race([ + (async () => { + await driver(); + return live.nextEvent(); + })(), + timeoutAfter(this.turnTimeoutMs), + ]); + } catch (err) { + // Timeout or driver throw → auto-save as interrupted (progress preserved) + // and emit an observable event. Never silent loss. + const session = this.interruptSession(sessionId, err); + this.disposeLive(sessionId); + return { session, event: { type: "error", data: { message: session.error ?? "interrupted", cause: err } } }; + } + + const session = this.applyEvent(sessionId, event); + if (event.type === "complete" || event.type === "error") { + this.disposeLive(sessionId); + } + return { session, event }; + } + + /** Persist a seam event onto the session row + emit the matching observable event. */ + private applyEvent(sessionId: string, event: InteractiveAiSessionEvent): CeSession { + switch (event.type) { + case "thinking": + case "text": { + this.store.appendHistory(sessionId, { role: "agent", text: event.data, at: new Date().toISOString() }); + const s = this.store.update(sessionId, { status: "active" }) ?? this.requireSession(sessionId); + this.ctx.emitEvent(CE_EVENTS.turn, { sessionId, kind: event.type }); + return s; + } + case "question": { + const q: PlanningQuestion = event.data; + this.store.appendHistory(sessionId, { + role: "agent", + text: JSON.stringify({ question: q }), + at: new Date().toISOString(), + }); + const s = this.store.update(sessionId, { status: "awaiting_input", currentQuestion: q }) ?? this.requireSession(sessionId); + this.ctx.emitEvent(CE_EVENTS.question, { sessionId, questionId: q.id }); + return s; + } + case "complete": { + const artifactPath = this.writeArtifact(sessionId, event.data); + this.store.appendHistory(sessionId, { + role: "agent", + text: JSON.stringify({ complete: true }), + at: new Date().toISOString(), + }); + const s = + this.store.update(sessionId, { status: "completed", currentQuestion: null, artifactPath }) ?? + this.requireSession(sessionId); + this.ctx.emitEvent(CE_EVENTS.completed, { sessionId, artifactPath }); + return s; + } + case "error": { + const message = event.data.message; + // Error preserves progress (currentQuestion/history untouched) so retry + // can resume. Status error; observable event emitted. + const s = this.store.update(sessionId, { status: "error", error: message }) ?? this.requireSession(sessionId); + this.ctx.emitEvent(CE_EVENTS.error, { sessionId, message }); + return s; + } + } + } + + /** Persist `interrupted` with progress preserved and emit. */ + private interruptSession(sessionId: string, cause: unknown): CeSession { + const message = cause instanceof Error ? cause.message : String(cause); + const s = + this.store.update(sessionId, { status: "interrupted", error: message }) ?? this.requireSession(sessionId); + this.ctx.emitEvent(CE_EVENTS.interrupted, { sessionId, message }); + return s; + } + + /** Persist `error` (session-create failure path) and emit. */ + private failSession(sessionId: string, cause: unknown): CeSession { + const message = cause instanceof Error ? cause.message : String(cause); + const s = this.store.update(sessionId, { status: "error", error: message }) ?? this.requireSession(sessionId); + this.ctx.emitEvent(CE_EVENTS.error, { sessionId, message }); + return s; + } + + /** + * Write the stage artifact to its conventional location (R10). Accepts either + * a `{ artifact: string }` payload or a raw string. Returns the absolute path. + */ + private writeArtifact(sessionId: string, data: unknown): string { + const session = this.requireSession(sessionId); + const stage = getStage(session.stage); + const location = stage?.artifactLocation ?? `docs/ce/${session.stage}/`; + const content = this.extractArtifactContent(data); + + const target = location.endsWith("/") + ? join(location, `${session.stage}-${session.id}.md`) + : location; + const abs = isAbsolute(target) ? target : join(this.projectRoot, target); + mkdirSync(dirname(abs), { recursive: true }); + writeFileSync(abs, content, "utf-8"); + return abs; + } + + private extractArtifactContent(data: unknown): string { + if (typeof data === "string") return data; + if (data && typeof data === "object" && "artifact" in data) { + const a = (data as { artifact: unknown }).artifact; + if (typeof a === "string") return a; + } + return JSON.stringify(data, null, 2); + } + + private requireSession(sessionId: string): CeSession { + const s = this.store.get(sessionId); + if (!s) throw new Error(`CE session not found: ${sessionId}`); + return s; + } + + private disposeLive(sessionId: string): void { + const live = this.live.get(sessionId); + if (live) { + try { + live.dispose(); + } catch { + // best-effort + } + this.live.delete(sessionId); + } + } +} diff --git a/plugins/fusion-plugin-compound-engineering/src/session/session-store.ts b/plugins/fusion-plugin-compound-engineering/src/session/session-store.ts new file mode 100644 index 0000000000..261b2afa9c --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/session/session-store.ts @@ -0,0 +1,263 @@ +import { randomUUID } from "node:crypto"; +import type { Database, PlanningQuestion, PluginContext } from "@fusion/core"; +import { ensureCeSchema } from "../schema.js"; + +/** + * CE session lifecycle states (mirrors the plan's state machine): + * launching → active → awaiting_input ↔ active → completed | error | interrupted; + * interrupted/error → active on resume/retry. + */ +export type CeSessionStatus = + | "launching" + | "active" + | "awaiting_input" + | "completed" + | "error" + | "interrupted"; + +/** A single recorded turn in the conversation history (for resume). */ +export interface CeConversationTurn { + role: "user" | "agent"; + /** Free text, or a serialized question/answer marker. */ + text: string; + at: string; +} + +export interface CeSession { + id: string; + stage: string; + status: CeSessionStatus; + currentQuestion: PlanningQuestion | null; + conversationHistory: CeConversationTurn[]; + projectId: string | null; + artifactPath: string | null; + error: string | null; + /** Expected per-turn interval (ms); drives interval-relative staleness. */ + turnIntervalMs: number; + /** Epoch millis of the last produced event (liveness anchor). */ + lastActivityAt: number; + createdAt: string; + updatedAt: string; +} + +interface CeSessionRow { + id: string; + stage: string; + status: CeSessionStatus; + currentQuestion: string | null; + conversationHistory: string; + projectId: string | null; + artifactPath: string | null; + error: string | null; + turnIntervalMs: number; + lastActivityAt: number; + createdAt: string; + updatedAt: string; +} + +export interface CreateCeSessionInput { + stage: string; + projectId?: string | null; + artifactPath?: string | null; + turnIntervalMs?: number; + id?: string; +} + +/** + * Default multiple of the turn interval beyond which a non-terminal session is + * considered stale. Mirrors the FN-4172 rubric (`> 3× interval`), interval- + * relative rather than a raw last-event age. + */ +export const STALE_INTERVAL_MULTIPLE = 3; + +const DEFAULT_TURN_INTERVAL_MS = 120000; + +function rowToSession(row: CeSessionRow): CeSession { + return { + id: row.id, + stage: row.stage, + status: row.status, + currentQuestion: row.currentQuestion ? (JSON.parse(row.currentQuestion) as PlanningQuestion) : null, + conversationHistory: JSON.parse(row.conversationHistory) as CeConversationTurn[], + projectId: row.projectId, + artifactPath: row.artifactPath, + error: row.error, + turnIntervalMs: row.turnIntervalMs, + lastActivityAt: row.lastActivityAt, + createdAt: row.createdAt, + updatedAt: row.updatedAt, + }; +} + +/** + * Plugin-local persistence for CE interactive sessions. Reaches the DB the same + * way reports does (via `ctx.taskStore.getDatabase()`), and ensures its schema + * defensively on construction so a store created before `onSchemaInit` ran (or + * in a test) still works. + */ +export class CeSessionStore { + private readonly db: Database; + + constructor(db: Database) { + this.db = db; + ensureCeSchema(db); + } + + create(input: CreateCeSessionInput): CeSession { + const now = new Date().toISOString(); + const session: CeSession = { + id: input.id ?? randomUUID(), + stage: input.stage, + status: "launching", + currentQuestion: null, + conversationHistory: [], + projectId: input.projectId ?? null, + artifactPath: input.artifactPath ?? null, + error: null, + turnIntervalMs: input.turnIntervalMs ?? DEFAULT_TURN_INTERVAL_MS, + lastActivityAt: Date.now(), + createdAt: now, + updatedAt: now, + }; + this.db + .prepare( + `INSERT INTO ce_sessions + (id, stage, status, currentQuestion, conversationHistory, projectId, artifactPath, error, turnIntervalMs, lastActivityAt, createdAt, updatedAt) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .run( + session.id, + session.stage, + session.status, + null, + JSON.stringify(session.conversationHistory), + session.projectId, + session.artifactPath, + null, + session.turnIntervalMs, + session.lastActivityAt, + session.createdAt, + session.updatedAt, + ); + return session; + } + + get(id: string): CeSession | undefined { + const row = this.db.prepare(`SELECT * FROM ce_sessions WHERE id = ?`).get(id) as CeSessionRow | undefined; + return row ? rowToSession(row) : undefined; + } + + list(filter: { status?: CeSessionStatus; stage?: string } = {}): CeSession[] { + const clauses: string[] = []; + const params: unknown[] = []; + if (filter.status) { + clauses.push("status = ?"); + params.push(filter.status); + } + if (filter.stage) { + clauses.push("stage = ?"); + params.push(filter.stage); + } + const where = clauses.length > 0 ? `WHERE ${clauses.join(" AND ")}` : ""; + const rows = this.db + .prepare(`SELECT * FROM ce_sessions ${where} ORDER BY updatedAt DESC, id`) + .all(...params) as CeSessionRow[]; + return rows.map(rowToSession); + } + + /** + * Patch a session. Always bumps `updatedAt`; bumps `lastActivityAt` unless the + * caller explicitly overrides it (used by liveness tests to simulate age). + */ + update( + id: string, + patch: Partial< + Pick< + CeSession, + "status" | "currentQuestion" | "conversationHistory" | "artifactPath" | "error" | "lastActivityAt" | "projectId" + > + >, + ): CeSession | undefined { + const existing = this.get(id); + if (!existing) return undefined; + const next: CeSession = { + ...existing, + ...patch, + lastActivityAt: patch.lastActivityAt ?? Date.now(), + updatedAt: new Date().toISOString(), + }; + this.db + .prepare( + `UPDATE ce_sessions SET + status = ?, currentQuestion = ?, conversationHistory = ?, projectId = ?, + artifactPath = ?, error = ?, lastActivityAt = ?, updatedAt = ? + WHERE id = ?`, + ) + .run( + next.status, + next.currentQuestion ? JSON.stringify(next.currentQuestion) : null, + JSON.stringify(next.conversationHistory), + next.projectId, + next.artifactPath, + next.error, + next.lastActivityAt, + next.updatedAt, + id, + ); + return next; + } + + /** Append a turn to the conversation history (no other field touched). */ + appendHistory(id: string, turn: CeConversationTurn): CeSession | undefined { + const existing = this.get(id); + if (!existing) return undefined; + return this.update(id, { conversationHistory: [...existing.conversationHistory, turn] }); + } + + /** + * Interval-relative staleness: a non-terminal session is stale only when its + * last activity is older than `multiple × turnIntervalMs`. A healthy-but-slow + * session (within the interval band) is NOT stale. Terminal sessions + * (completed/error/interrupted) are never "stale" — they are already settled. + */ + isStale(session: CeSession, now = Date.now(), multiple = STALE_INTERVAL_MULTIPLE): boolean { + if (session.status === "completed" || session.status === "error" || session.status === "interrupted") { + return false; + } + return now - session.lastActivityAt > multiple * session.turnIntervalMs; + } + + /** + * Recover sessions left non-terminal by a crash/restart. A session with a + * persisted `currentQuestion` is restored to `awaiting_input` (resumable); + * one without is marked `interrupted` with its progress preserved — never + * silently dropped. Returns the ids transitioned. + */ + recoverStaleSessions(now = Date.now(), multiple = STALE_INTERVAL_MULTIPLE): string[] { + const candidates = this.list().filter( + (s) => (s.status === "active" || s.status === "launching" || s.status === "awaiting_input") && this.isStale(s, now, multiple), + ); + const recovered: string[] = []; + for (const s of candidates) { + if (s.currentQuestion) { + if (s.status !== "awaiting_input") this.update(s.id, { status: "awaiting_input" }); + } else { + this.update(s.id, { status: "interrupted", error: s.error ?? "Session interrupted — progress preserved, resume to continue" }); + } + recovered.push(s.id); + } + return recovered; + } +} + +const storeCache = new WeakMap(); + +/** WeakMap-cached store keyed by the TaskStore instance (mirrors reports). */ +export function getCeSessionStore(ctx: PluginContext): CeSessionStore { + const key = ctx.taskStore as object; + const cached = storeCache.get(key); + if (cached) return cached; + const store = new CeSessionStore(ctx.taskStore.getDatabase()); + storeCache.set(key, store); + return store; +} diff --git a/plugins/fusion-plugin-compound-engineering/src/session/stage-registry.ts b/plugins/fusion-plugin-compound-engineering/src/session/stage-registry.ts new file mode 100644 index 0000000000..9ad3b792ff --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/session/stage-registry.ts @@ -0,0 +1,51 @@ +/** + * Minimal internal stage registry (U5 slice). + * + * The full registry + presentation metadata is U6. Here we keep ONLY the data + * the orchestrator needs to launch a stage by id: which bundled `ce-*` skill it + * loads, and where its `complete` artifact is written (R10). Adding a stage is a + * data entry in this map — no new route or store code (proved by the + * "second stage through the same orchestrator" test). + */ + +export interface CeStageDefinition { + /** Stage id (stable, kebab-case). */ + stageId: string; + /** Bundled skill the orchestrator loads for this stage. */ + skillId: string; + /** + * Conventional artifact location for this stage's `complete` output, + * project-root-relative. When the path ends in `/` the orchestrator writes a + * timestamped file inside that directory; otherwise it writes that exact file. + */ + artifactLocation: string; +} + +/** + * The first registration slice. Locations mirror where the real ce-* skills + * write today (STRATEGY.md, docs/ideation/, docs/brainstorms/, docs/plans/). + */ +const STAGE_DEFINITIONS: CeStageDefinition[] = [ + { stageId: "strategy", skillId: "ce-strategy", artifactLocation: "STRATEGY.md" }, + { stageId: "ideate", skillId: "ce-ideate", artifactLocation: "docs/ideation/" }, + { stageId: "brainstorm", skillId: "ce-brainstorm", artifactLocation: "docs/brainstorms/" }, + { stageId: "plan", skillId: "ce-plan", artifactLocation: "docs/plans/" }, +]; + +const REGISTRY = new Map(STAGE_DEFINITIONS.map((s) => [s.stageId, s])); + +export function getStage(stageId: string): CeStageDefinition | undefined { + return REGISTRY.get(stageId); +} + +export function listStages(): CeStageDefinition[] { + return [...REGISTRY.values()]; +} + +/** + * Register an additional stage at runtime (used by tests to prove "adding a + * stage requires only data"). Production stages live in STAGE_DEFINITIONS. + */ +export function registerStage(def: CeStageDefinition): void { + REGISTRY.set(def.stageId, def); +}