diff --git a/plugins/fusion-plugin-compound-engineering/src/__tests__/work-bridge.test.ts b/plugins/fusion-plugin-compound-engineering/src/__tests__/work-bridge.test.ts new file mode 100644 index 0000000000..1ad88086ec --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/work-bridge.test.ts @@ -0,0 +1,159 @@ +import { rm } from "node:fs/promises"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { InteractiveAiSessionEvent, PluginContext } from "@fusion/core"; +import { TaskStore } from "@fusion/core"; +import { + CeOrchestrator, + CE_PLUGIN_ID, + CE_WORK_SOURCE_TYPE, + WORK_STAGE_ID, +} from "../session/orchestrator.js"; +import { getCePipelineStore } from "../sync/pipeline-store.js"; +import { makeScriptedSession } from "./_harness.js"; + +/** + * U7 work bridge tests. These use the REAL in-memory TaskStore (so created tasks + * are genuine board tasks under the normal lifecycle) and a scripted fake + * interactive session (the same deterministic driver U5/U6 use). + */ + +let rootDir: string; +let globalDir: string; +let taskStore: TaskStore; +let ctx: PluginContext; +let emitted: Array<{ event: string; data: unknown }>; + +beforeEach(async () => { + rootDir = mkdtempSync(join(tmpdir(), "ce-work-bridge-")); + globalDir = join(rootDir, ".fusion-global"); + taskStore = new TaskStore(rootDir, globalDir, { inMemoryDb: true }); + await taskStore.init(); + + emitted = []; + ctx = { + pluginId: CE_PLUGIN_ID, + taskStore, + settings: {}, + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, + emitEvent: (event: string, data: unknown) => { + emitted.push({ event, data }); + }, + } as unknown as PluginContext; +}); + +afterEach(async () => { + taskStore?.close(); + await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); +}); + +function makeOrch(script: InteractiveAiSessionEvent[]) { + const session = makeScriptedSession(script); + return new CeOrchestrator({ + ctx, + createInteractiveAiSession: vi.fn(async () => ({ session })), + projectRoot: rootDir, + turnTimeoutMs: 5000, + }); +} + +describe("work bridge (U7)", () => { + it("lands derived tasks on the board, tagged CE-originated with a resolvable back-reference", async () => { + const orch = makeOrch([ + { + type: "complete", + data: { + artifact: "# Work log\n", + tasks: [ + { title: "Wire the thing", description: "Implement the thing in module X." }, + { description: "Add tests for the thing.", column: "todo" }, + ], + }, + }, + ]); + + const started = await orch.start(WORK_STAGE_ID, { openingMessage: "do the work" }); + expect(started.session.status).toBe("completed"); + const cePipelineId = started.session.id; + + // Two board tasks created. + const tasks = await taskStore.listTasks(); + expect(tasks).toHaveLength(2); + + const pipelineStore = getCePipelineStore(ctx); + + for (const task of tasks) { + // CE-originated provenance: valid SourceType + CE marker + back-ref copy. + // (TaskStore exposes provenance as flat top-level fields on the Task.) + expect(task.sourceType).toBe(CE_WORK_SOURCE_TYPE); + const meta = task.sourceMetadata as Record | undefined; + expect(meta?.pluginId).toBe(CE_PLUGIN_ID); + expect(meta?.cePipelineId).toBe(cePipelineId); + expect(meta?.ceStageId).toBe(WORK_STAGE_ID); + + // Authoritative back-reference: the link row resolves task→pipeline/artifact. + const link = pipelineStore.findByTaskId(task.id); + expect(link).toBeDefined(); + expect(link?.cePipelineId).toBe(cePipelineId); + expect(link?.ceStageId).toBe(WORK_STAGE_ID); + expect(link?.ceArtifactPath).toBe(started.session.artifactPath); + } + + // Pipeline lists exactly its two links. + expect(pipelineStore.listByPipeline(cePipelineId)).toHaveLength(2); + + // Optional column honored. + const todoTask = tasks.find((t) => t.description.includes("Add tests")); + expect(todoTask?.column).toBe("todo"); + }); + + it("created tasks run the NORMAL lifecycle with no plugin interference", async () => { + const orch = makeOrch([ + { type: "complete", data: { tasks: [{ description: "A normal task." }] } }, + ]); + const started = await orch.start(WORK_STAGE_ID, { openingMessage: "go" }); + + const tasks = await taskStore.listTasks(); + expect(tasks).toHaveLength(1); + const task = tasks[0]; + + // It is an ordinary board task: default column, normal mutation works, and the + // plugin attached no extra status/hook state beyond provenance metadata. + expect(task.column).toBe("triage"); + const moved = await taskStore.moveTask(task.id, "todo"); + expect(moved.column).toBe("todo"); + + // Re-read is a clean, normal task (provenance is the only CE footprint). + const reread = await taskStore.getTask(task.id); + expect(reread?.column).toBe("todo"); + expect((reread?.sourceMetadata as Record)?.pluginId).toBe(CE_PLUGIN_ID); + void started; + }); + + it("zero derived tasks is a clean no-op (no board tasks, no orphan link rows)", async () => { + const orch = makeOrch([{ type: "complete", data: { artifact: "# Nothing to do\n", tasks: [] } }]); + const started = await orch.start(WORK_STAGE_ID, { openingMessage: "nothing here" }); + expect(started.session.status).toBe("completed"); + + expect(await taskStore.listTasks()).toHaveLength(0); + expect(getCePipelineStore(ctx).listByPipeline(started.session.id)).toHaveLength(0); + }); + + it("a completion payload with NO tasks field is also a no-op", async () => { + const orch = makeOrch([{ type: "complete", data: { artifact: "# Just an artifact\n" } }]); + const started = await orch.start(WORK_STAGE_ID, { openingMessage: "x" }); + expect(started.session.status).toBe("completed"); + expect(await taskStore.listTasks()).toHaveLength(0); + expect(getCePipelineStore(ctx).listByPipeline(started.session.id)).toHaveLength(0); + }); + + it("a non-work stage with a tasks payload does NOT land board tasks (bridge is work-only)", async () => { + const orch = makeOrch([ + { type: "complete", data: { artifact: "# Brainstorm\n", tasks: [{ description: "should be ignored" }] } }, + ]); + await orch.start("brainstorm", { openingMessage: "ideas" }); + expect(await taskStore.listTasks()).toHaveLength(0); + }); +}); diff --git a/plugins/fusion-plugin-compound-engineering/src/index.ts b/plugins/fusion-plugin-compound-engineering/src/index.ts index dc01d5bc91..87900c06b4 100644 --- a/plugins/fusion-plugin-compound-engineering/src/index.ts +++ b/plugins/fusion-plugin-compound-engineering/src/index.ts @@ -15,7 +15,14 @@ export { } from "./skill-installation.js"; export { ensureCeSchema } from "./schema.js"; export { CeSessionStore, getCeSessionStore } from "./session/session-store.js"; -export { CeOrchestrator } from "./session/orchestrator.js"; +export { CePipelineStore, getCePipelineStore } from "./sync/pipeline-store.js"; +export type { CePipelineLink, CreateCePipelineLinkInput } from "./sync/pipeline-store.js"; +export { + CeOrchestrator, + WORK_STAGE_ID, + CE_PLUGIN_ID, + CE_WORK_SOURCE_TYPE, +} from "./session/orchestrator.js"; export { getStage, listStages, registerStage } from "./session/stage-registry.js"; const plugin = definePlugin({ diff --git a/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts b/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts index 1e7a62bbc6..c7123e6a27 100644 --- a/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts +++ b/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts @@ -1,6 +1,7 @@ import type { PluginContext, PluginRouteDefinition, PluginRouteResponse } from "@fusion/core"; import { CeOrchestrator } from "../session/orchestrator.js"; import { getCeSessionStore } from "../session/session-store.js"; +import { getCePipelineStore } from "../sync/pipeline-store.js"; /** * Session routes (U5): start / answer / resume / get-session-state. @@ -130,5 +131,18 @@ export function createSessionRoutes(): PluginRouteDefinition[] { return { status: 200, body: { sessions } }; }, }, + { + // U7 work bridge: observe the board tasks a CE pipeline (session) landed, + // via their link records (the addressable back-reference, FN-5719). The + // session id IS the pipeline id. Outbound-only in U7; U8 layers state. + method: "GET", + path: "/sessions/:id/links", + description: "List the CE pipeline-link records (work→board) for a session/pipeline.", + handler: async (req: unknown, ctx: PluginContext): Promise => { + const id = (req as RouteRequest).params.id; + const links = getCePipelineStore(ctx).listByPipeline(id); + return { status: 200, body: { links } }; + }, + }, ]; } diff --git a/plugins/fusion-plugin-compound-engineering/src/schema.ts b/plugins/fusion-plugin-compound-engineering/src/schema.ts index 8c31236bf0..b2bf15102c 100644 --- a/plugins/fusion-plugin-compound-engineering/src/schema.ts +++ b/plugins/fusion-plugin-compound-engineering/src/schema.ts @@ -18,6 +18,13 @@ import type { Database } from "@fusion/core"; * 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). + * + * `ce_pipeline_links` (U7) is the addressable back-reference table: it links a + * board task to the CE pipeline/stage/artifact that produced it. Per FN-5719 the + * back-reference lives in this plugin-local table (NOT in task-row JSON) so + * board-task ownership and CE-pipeline ownership stay separate state machines. + * U7 keeps it minimal (link records only); U8 extends it with the bidirectional + * pipeline-state machine. */ export function ensureCeSchema(db: Database): void { db.exec(` @@ -46,5 +53,20 @@ export function ensureCeSchema(db: Database): void { CREATE INDEX IF NOT EXISTS idxCeSessionsProject ON ce_sessions(projectId, updatedAt DESC, id); + + CREATE TABLE IF NOT EXISTS ce_pipeline_links ( + id TEXT PRIMARY KEY, + taskId TEXT NOT NULL, + cePipelineId TEXT NOT NULL, + ceStageId TEXT NOT NULL, + ceArtifactPath TEXT, + createdAt TEXT NOT NULL + ); + + CREATE INDEX IF NOT EXISTS idxCePipelineLinksPipeline + ON ce_pipeline_links(cePipelineId, createdAt DESC, id); + + CREATE UNIQUE INDEX IF NOT EXISTS idxCePipelineLinksTask + ON ce_pipeline_links(taskId); `); } diff --git a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts index 1e8d2ffb3d..c02eb2a7c1 100644 --- a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts +++ b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts @@ -8,10 +8,49 @@ import type { PluginContext, } from "@fusion/core"; import { resolveDefaultInstallTargetRoot } from "../skill-installation.js"; +import { getCePipelineStore, type CePipelineStore } from "../sync/pipeline-store.js"; import type { CeSession, CeSessionStore } from "./session-store.js"; import { getCeSessionStore } from "./session-store.js"; import { getStage, type CeStageDefinition } from "./stage-registry.js"; +/** + * The stage id whose `complete` payload carries a derived task list to land on + * the board (U7). Its skill is `ce-work` (see the stage registry). + */ +export const WORK_STAGE_ID = "work"; + +/** + * The CE marker plugin id recorded on every CE-originated board task and link. + * Kept as a constant so U8's sync code reuses the same identity. + */ +export const CE_PLUGIN_ID = "fusion-plugin-compound-engineering"; + +/** + * SourceType chosen for CE-originated automated work. The work stage is a step in + * the CE pipeline, so `workflow_step` is the closest existing provenance value + * (vs. inventing a new SourceType). The CE marker + back-reference convenience + * copy ride in `sourceMetadata`; the authoritative link is the pipeline-link row. + */ +export const CE_WORK_SOURCE_TYPE = "workflow_step" as const; + +/** + * COMPLETION-PAYLOAD → TASKS CONTRACT (U7). + * + * The `work` stage's `complete` event `data` MAY carry a `tasks` array describing + * the board tasks to create. Each entry needs at least a `description` (the only + * required TaskCreateInput field); `title` and `column` are optional. + * + * { artifact?: string, tasks?: Array<{ title?: string, description: string, column?: Column }> } + * + * A missing/empty `tasks` array is a clean no-op (no board tasks, no link rows). + * Entries with a blank description are skipped (createTask would reject them). + */ +export interface CeDerivedTaskSpec { + title?: string; + description: string; + column?: string; +} + /** Default per-turn timeout. A turn that exceeds this is treated as a stall. */ const DEFAULT_TURN_TIMEOUT_MS = 120000; @@ -126,6 +165,7 @@ export interface CeStepResult { export class CeOrchestrator { private readonly ctx: PluginContext; private readonly store: CeSessionStore; + private readonly pipelineStore: CePipelineStore; private readonly factory: CreateInteractiveAiSessionFactory | undefined; private readonly projectRoot: string; private readonly turnTimeoutMs: number; @@ -135,6 +175,7 @@ export class CeOrchestrator { constructor(deps: OrchestratorDeps) { this.ctx = deps.ctx; this.store = getCeSessionStore(deps.ctx); + this.pipelineStore = getCePipelineStore(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; @@ -249,6 +290,12 @@ export class CeOrchestrator { if (event.type === "complete" || event.type === "error") { this.disposeLive(sessionId); } + // Work bridge (U7): the work stage's completion payload lands derived tasks + // on the board, tagged CE-originated + recorded as pipeline links. Outbound + // only — created tasks then run the NORMAL lifecycle with no plugin hooks. + if (event.type === "complete" && session.stage === WORK_STAGE_ID) { + await this.landWorkTasks(session, event.data); + } return { session, event }; } @@ -297,6 +344,73 @@ export class CeOrchestrator { } } + /** + * Work bridge (U7). Read the derived task list from the work stage's + * completion payload, create each as a board task tagged CE-originated, and + * record a pipeline-link row resolving task→pipeline/stage/artifact. Zero + * derived tasks is a clean no-op (no board tasks, no orphan link rows). The + * created tasks then run the normal lifecycle — no hooks attached here (U8). + */ + private async landWorkTasks(session: CeSession, data: unknown): Promise { + const specs = this.extractTaskSpecs(data); + if (specs.length === 0) return; + + // The session is the CE pipeline run; its id is the stable pipeline id the + // link rows (and U8's state machine) address. + const cePipelineId = session.id; + const ceStageId = session.stage; + const ceArtifactPath = session.artifactPath ?? null; + + for (const spec of specs) { + const description = spec.description.trim(); + if (!description) continue; // createTask rejects blank descriptions. + + const task = await this.ctx.taskStore.createTask({ + title: spec.title, + description, + column: spec.column as never, + source: { + sourceType: CE_WORK_SOURCE_TYPE, + sourceSessionId: cePipelineId, + sourceMetadata: { + pluginId: CE_PLUGIN_ID, + cePipelineId, + ceStageId, + ceArtifactPath, + }, + }, + }); + + // Authoritative back-reference (FN-5719): the link row, not task-row JSON. + this.pipelineStore.createLink({ + taskId: task.id, + cePipelineId, + ceStageId, + ceArtifactPath, + }); + } + } + + /** Parse the `{ tasks: [...] }` completion-payload contract; tolerant of shape. */ + private extractTaskSpecs(data: unknown): CeDerivedTaskSpec[] { + if (!data || typeof data !== "object") return []; + const raw = (data as { tasks?: unknown }).tasks; + if (!Array.isArray(raw)) return []; + const specs: CeDerivedTaskSpec[] = []; + for (const entry of raw) { + if (!entry || typeof entry !== "object") continue; + const e = entry as Record; + const description = typeof e.description === "string" ? e.description : ""; + if (!description.trim()) continue; + specs.push({ + description, + title: typeof e.title === "string" ? e.title : undefined, + column: typeof e.column === "string" ? e.column : undefined, + }); + } + return specs; + } + /** Persist `interrupted` with progress preserved and emit. */ private interruptSession(sessionId: string, cause: unknown): CeSession { const message = cause instanceof Error ? cause.message : String(cause); diff --git a/plugins/fusion-plugin-compound-engineering/src/session/stage-registry.ts b/plugins/fusion-plugin-compound-engineering/src/session/stage-registry.ts index 2703a4175b..0150541fa7 100644 --- a/plugins/fusion-plugin-compound-engineering/src/session/stage-registry.ts +++ b/plugins/fusion-plugin-compound-engineering/src/session/stage-registry.ts @@ -75,6 +75,18 @@ const STAGE_DEFINITIONS: CeStageDefinition[] = [ label: "Plan", artifactGlob: "docs/plans/**/*.md", }, + { + // The work stage (U7). Its `ce-work` skill drives execution and, on + // `complete`, carries a derived task list that the orchestrator lands on the + // board (tagged CE-originated + recorded as pipeline links). The artifact is + // the work log / summary for this stage. + stageId: "work", + skillId: "ce-work", + artifactLocation: "docs/work/", + icon: "Hammer", + label: "Work", + artifactGlob: "docs/work/**/*.md", + }, ]; const REGISTRY = new Map(STAGE_DEFINITIONS.map((s) => [s.stageId, s])); diff --git a/plugins/fusion-plugin-compound-engineering/src/sync/pipeline-store.ts b/plugins/fusion-plugin-compound-engineering/src/sync/pipeline-store.ts new file mode 100644 index 0000000000..3e2f57dca3 --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/sync/pipeline-store.ts @@ -0,0 +1,124 @@ +import { randomUUID } from "node:crypto"; +import type { Database, PluginContext } from "@fusion/core"; +import { ensureCeSchema } from "../schema.js"; + +/** + * Plugin-local store for CE pipeline LINK records (U7). + * + * A link record is the addressable, durable association between a board task and + * the CE pipeline/stage/artifact that produced it. Per FN-5719 the back-reference + * lives HERE (a plugin-local table) — NOT in task-row JSON — so board-task + * ownership and CE-pipeline ownership remain separate state machines and cannot + * oscillate. A convenience copy of the ids may also ride along in the task's + * `source.sourceMetadata`, but THIS ROW is the authoritative link. + * + * U7 SURFACE (intentionally minimal): create a link, list links by pipeline, and + * find the link by taskId. U8 will EXTEND this store with the full bidirectional + * pipeline-STATE machine (state column, status transitions, enqueue/reconcile). + * U7 deliberately does not add any state/status field or sync behaviour so U8 can + * layer it on without reworking the link surface. + */ +export interface CePipelineLink { + /** Stable link-record id. */ + id: string; + /** The board task this link points at (1:1 for U7). */ + taskId: string; + /** The CE pipeline this task was derived under (the originating run). */ + cePipelineId: string; + /** The CE stage id within that pipeline (e.g. "work"). */ + ceStageId: string; + /** Absolute path to the stage artifact that drove this task, if any. */ + ceArtifactPath: string | null; + createdAt: string; +} + +interface CePipelineLinkRow { + id: string; + taskId: string; + cePipelineId: string; + ceStageId: string; + ceArtifactPath: string | null; + createdAt: string; +} + +export interface CreateCePipelineLinkInput { + taskId: string; + cePipelineId: string; + ceStageId: string; + ceArtifactPath?: string | null; + id?: string; +} + +function rowToLink(row: CePipelineLinkRow): CePipelineLink { + return { + id: row.id, + taskId: row.taskId, + cePipelineId: row.cePipelineId, + ceStageId: row.ceStageId, + ceArtifactPath: row.ceArtifactPath, + createdAt: row.createdAt, + }; +} + +/** + * CRUD for CE pipeline link records. Reaches the DB the same way the session + * store / reports do (via `ctx.taskStore.getDatabase()`) and ensures its schema + * defensively on construction so a store built before `onSchemaInit` ran (or in a + * test) still works. + */ +export class CePipelineStore { + private readonly db: Database; + + constructor(db: Database) { + this.db = db; + ensureCeSchema(db); + } + + /** Record a task→pipeline/artifact link. */ + createLink(input: CreateCePipelineLinkInput): CePipelineLink { + const link: CePipelineLink = { + id: input.id ?? randomUUID(), + taskId: input.taskId, + cePipelineId: input.cePipelineId, + ceStageId: input.ceStageId, + ceArtifactPath: input.ceArtifactPath ?? null, + createdAt: new Date().toISOString(), + }; + this.db + .prepare( + `INSERT INTO ce_pipeline_links + (id, taskId, cePipelineId, ceStageId, ceArtifactPath, createdAt) + VALUES (?, ?, ?, ?, ?, ?)`, + ) + .run(link.id, link.taskId, link.cePipelineId, link.ceStageId, link.ceArtifactPath, link.createdAt); + return link; + } + + /** All links produced by a given CE pipeline, newest first. */ + listByPipeline(cePipelineId: string): CePipelineLink[] { + const rows = this.db + .prepare(`SELECT * FROM ce_pipeline_links WHERE cePipelineId = ? ORDER BY createdAt DESC, id`) + .all(cePipelineId) as CePipelineLinkRow[]; + return rows.map(rowToLink); + } + + /** Resolve a board task back to its CE link (the back-reference). */ + findByTaskId(taskId: string): CePipelineLink | undefined { + const row = this.db + .prepare(`SELECT * FROM ce_pipeline_links WHERE taskId = ?`) + .get(taskId) as CePipelineLinkRow | undefined; + return row ? rowToLink(row) : undefined; + } +} + +const storeCache = new WeakMap(); + +/** WeakMap-cached store keyed by the TaskStore instance (mirrors the session store). */ +export function getCePipelineStore(ctx: PluginContext): CePipelineStore { + const key = ctx.taskStore as object; + const cached = storeCache.get(key); + if (cached) return cached; + const store = new CePipelineStore(ctx.taskStore.getDatabase()); + storeCache.set(key, store); + return store; +}