feat(compound-engineering): CE-pipeline state model + bidirectional sync (U8)
Add a ce_pipeline_state machine (currentStage/status) kept distinct from board-task ownership (task column) — separate tables, no shared column, per FN-5719. onTaskMoved/onTaskCompleted hooks do only an indexed lookup + enqueue and return well under the 5s budget; advancement happens in an on-demand reconciler sweep that re-derives correct pipeline state from board truth, so a dropped hook event still converges (no tight poll loop). Outbound CE-flow changes create the next-stage board task. Conflict policy: board authoritative for task state, CE flow authoritative for artifact/pipeline content (inbound read and outbound write target different rows, so they cannot contend). Host-scheduler note: no host timer wired; sweeps run on hook-drain and on the dashboard/route refresh surface (tracked for the host event-publish follow-up).
This commit is contained in:
gsxdsm
@@ -0,0 +1,250 @@
|
||||
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, Task } from "@fusion/core";
|
||||
import { TaskStore } from "@fusion/core";
|
||||
import plugin, {
|
||||
CeOrchestrator,
|
||||
CE_PLUGIN_ID,
|
||||
WORK_STAGE_ID,
|
||||
} from "../index.js";
|
||||
import { getCePipelineStore } from "../sync/pipeline-store.js";
|
||||
import { CeReconciler, reconcileCePipelines } from "../sync/reconciler.js";
|
||||
import { makeScriptedSession } from "./_harness.js";
|
||||
|
||||
/**
|
||||
* U8 bidirectional-sync tests. REAL in-memory TaskStore (genuine board tasks) +
|
||||
* the actual lifecycle-hook handlers and reconciler. We exercise the two
|
||||
* separate state machines (board columns vs ce_pipeline_state) and prove the
|
||||
* dropped-event convergence path independently of the hooks.
|
||||
*/
|
||||
|
||||
let rootDir: string;
|
||||
let taskStore: TaskStore;
|
||||
let ctx: PluginContext;
|
||||
let emitted: Array<{ event: string; data: unknown }>;
|
||||
|
||||
beforeEach(async () => {
|
||||
rootDir = mkdtempSync(join(tmpdir(), "ce-sync-"));
|
||||
taskStore = new TaskStore(rootDir, join(rootDir, ".fusion-global"), { 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 });
|
||||
});
|
||||
|
||||
/** The board enforces ordered transitions; walk a task forward to a target column. */
|
||||
const COLUMN_PATH = ["triage", "todo", "in-progress", "in-review", "done"];
|
||||
async function moveTo(taskId: string, target: string): Promise<void> {
|
||||
const current = (await taskStore.getTask(taskId))!.column;
|
||||
const from = COLUMN_PATH.indexOf(current);
|
||||
const to = COLUMN_PATH.indexOf(target);
|
||||
for (let i = from + 1; i <= to; i++) {
|
||||
await taskStore.moveTask(taskId, COLUMN_PATH[i] as never);
|
||||
}
|
||||
}
|
||||
|
||||
/** Run the work stage so a CE pipeline + its first board task + state record exist. */
|
||||
async function landPipeline(stage = "plan"): Promise<{ cePipelineId: string; task: Task }> {
|
||||
// Register-free: drive the WORK stage (which seeds state) but point the link at
|
||||
// `stage` so we can advance through the real stage order. Simplest: use the
|
||||
// work bridge directly via the orchestrator at the work stage, then rewrite the
|
||||
// pipeline state's currentStage to `stage` for ordering tests.
|
||||
const script: InteractiveAiSessionEvent[] = [
|
||||
{ type: "complete", data: { artifact: "# log\n", tasks: [{ description: "do stage work" }] } },
|
||||
];
|
||||
const orch = new CeOrchestrator({
|
||||
ctx,
|
||||
createInteractiveAiSession: vi.fn(async () => ({ session: makeScriptedSession(script) })),
|
||||
projectRoot: rootDir,
|
||||
turnTimeoutMs: 5000,
|
||||
});
|
||||
const started = await orch.start(WORK_STAGE_ID, { openingMessage: "go" });
|
||||
const cePipelineId = started.session.id;
|
||||
const store = getCePipelineStore(ctx);
|
||||
|
||||
if (stage !== WORK_STAGE_ID) {
|
||||
// Reposition both the link stage and the state stage to `stage` so the
|
||||
// pipeline has a non-terminal stage to advance FROM.
|
||||
const links = store.listByPipeline(cePipelineId);
|
||||
const db = taskStore.getDatabase();
|
||||
for (const l of links) {
|
||||
db.prepare(`UPDATE ce_pipeline_links SET ceStageId = ? WHERE id = ?`).run(stage, l.id);
|
||||
}
|
||||
store.upsertState({ cePipelineId, currentStage: stage, status: "running" });
|
||||
}
|
||||
const tasks = await taskStore.listTasks();
|
||||
return { cePipelineId, task: tasks[0] };
|
||||
}
|
||||
|
||||
describe("U8 inbound hooks (board → pipeline)", () => {
|
||||
it("onTaskMoved only enqueues when the task is CE-linked; ignores unrelated tasks fast", async () => {
|
||||
const { task } = await landPipeline("plan");
|
||||
const store = getCePipelineStore(ctx);
|
||||
|
||||
// Unrelated (non-CE) board task → hook is a no-op (no queue row).
|
||||
const other = await taskStore.createTask({ description: "unrelated work" });
|
||||
await plugin.hooks.onTaskMoved!(other, "triage", "todo", ctx);
|
||||
expect(store.listPendingSync()).toHaveLength(0);
|
||||
|
||||
// CE-linked task move → a queue row is appended synchronously. We do NOT
|
||||
// await the hook (its body is synchronous; awaiting would let the
|
||||
// fired-and-forgotten reconcile drain the row), so we observe the pending
|
||||
// entry the fast path wrote before any deferred work runs.
|
||||
void plugin.hooks.onTaskMoved!(task, "todo", "in-progress", ctx);
|
||||
const pending = store.listPendingSync();
|
||||
expect(pending.length).toBeGreaterThanOrEqual(1);
|
||||
expect(pending.some((p) => p.taskId === task.id && p.reason === "task_moved")).toBe(true);
|
||||
});
|
||||
|
||||
it("the hook handler does NOT advance the pipeline inline (heavy work is deferred)", async () => {
|
||||
const { cePipelineId, task } = await landPipeline("plan");
|
||||
const store = getCePipelineStore(ctx);
|
||||
|
||||
// Move the task to a terminal column then fire ONLY the synchronous part of
|
||||
// the hook. We assert that synchronously the pipeline stage is unchanged —
|
||||
// advancement happens in the (deferred) reconcile, not inline.
|
||||
await moveTo(task.id, "done");
|
||||
const stageBefore = store.getState(cePipelineId)!.currentStage;
|
||||
// Drive the hook but capture state immediately after the synchronous body.
|
||||
const p = plugin.hooks.onTaskMoved!(task, "in-progress", "done", ctx);
|
||||
// The synchronous body has already run (enqueue) but the fired-and-forgotten
|
||||
// reconcile has not been awaited. Inline, the stage must be unchanged.
|
||||
expect(store.getState(cePipelineId)!.currentStage).toBe(stageBefore);
|
||||
// A queue row exists (the fast path did its job).
|
||||
expect(store.listPendingSync().some((q) => q.taskId === task.id)).toBe(true);
|
||||
await p; // let the fire-and-forget settle for clean teardown.
|
||||
});
|
||||
|
||||
it("the hook handler completes well under the 5s budget even with a slow reconciler", async () => {
|
||||
const { task } = await landPipeline("plan");
|
||||
await moveTo(task.id, "done");
|
||||
const start = Date.now();
|
||||
await plugin.hooks.onTaskMoved!(task, "in-progress", "done", ctx);
|
||||
// The hook awaits NOTHING heavy; it returns synchronously-ish.
|
||||
expect(Date.now() - start).toBeLessThan(1000);
|
||||
});
|
||||
});
|
||||
|
||||
describe("U8 reconciler (convergence + outbound)", () => {
|
||||
it("AE3: a CE task reaching a terminal column advances the pipeline to the next stage with NO manual step", async () => {
|
||||
const { cePipelineId, task } = await landPipeline("plan");
|
||||
const store = getCePipelineStore(ctx);
|
||||
expect(store.getState(cePipelineId)!.currentStage).toBe("plan");
|
||||
|
||||
// Board moves the task to done (the only manual-equivalent action: a normal
|
||||
// board transition). The hook enqueues; reconcile advances.
|
||||
await moveTo(task.id, "done");
|
||||
await plugin.hooks.onTaskCompleted!({ ...task, column: "done" }, ctx);
|
||||
await reconcileCePipelines(ctx);
|
||||
|
||||
// Pipeline advanced plan → work (next in stage order) with no manual step.
|
||||
const state = store.getState(cePipelineId)!;
|
||||
expect(state.currentStage).toBe("work");
|
||||
});
|
||||
|
||||
it("outbound: advancing the pipeline propagates a NEW next-stage board task", async () => {
|
||||
const { cePipelineId, task } = await landPipeline("plan");
|
||||
const before = (await taskStore.listTasks()).length;
|
||||
|
||||
await moveTo(task.id, "in-review");
|
||||
await reconcileCePipelines(ctx); // no hook fired — pure re-derivation.
|
||||
|
||||
const after = await taskStore.listTasks();
|
||||
expect(after.length).toBe(before + 1);
|
||||
const newTask = after.find((t) => t.id !== task.id)!;
|
||||
const meta = newTask.sourceMetadata as Record<string, unknown>;
|
||||
expect(meta.pluginId).toBe(CE_PLUGIN_ID);
|
||||
expect(meta.cePipelineId).toBe(cePipelineId);
|
||||
expect(meta.ceStageId).toBe("work");
|
||||
// The pipeline is now awaiting the new board task.
|
||||
expect(getCePipelineStore(ctx).getState(cePipelineId)!.status).toBe("awaiting_board");
|
||||
});
|
||||
|
||||
it("MISSED HOOK EVENT → the reconcile sweep still converges (no queue row needed)", async () => {
|
||||
const { cePipelineId, task } = await landPipeline("plan");
|
||||
const store = getCePipelineStore(ctx);
|
||||
|
||||
// Simulate a DROPPED hook: move the board task to a terminal column but do
|
||||
// NOT call any hook and do NOT enqueue anything.
|
||||
await moveTo(task.id, "done");
|
||||
expect(store.listPendingSync()).toHaveLength(0); // nothing was enqueued.
|
||||
expect(store.getState(cePipelineId)!.currentStage).toBe("plan"); // not advanced yet.
|
||||
|
||||
// The on-demand sweep re-derives the transition from board truth alone.
|
||||
const result = await new CeReconciler(ctx).reconcile();
|
||||
expect(result.advanced).toBe(1);
|
||||
expect(store.getState(cePipelineId)!.currentStage).toBe("work");
|
||||
});
|
||||
|
||||
it("reconcile is idempotent: a second sweep does not double-advance or duplicate tasks", async () => {
|
||||
const { cePipelineId, task } = await landPipeline("plan");
|
||||
await moveTo(task.id, "done");
|
||||
await reconcileCePipelines(ctx);
|
||||
const afterFirst = (await taskStore.listTasks()).length;
|
||||
const stageFirst = getCePipelineStore(ctx).getState(cePipelineId)!.currentStage;
|
||||
|
||||
await reconcileCePipelines(ctx);
|
||||
expect((await taskStore.listTasks()).length).toBe(afterFirst);
|
||||
expect(getCePipelineStore(ctx).getState(cePipelineId)!.currentStage).toBe(stageFirst);
|
||||
});
|
||||
|
||||
it("partial completion does not advance: pipeline stays running until ALL current-stage tasks are terminal", async () => {
|
||||
const { cePipelineId, task } = await landPipeline("plan");
|
||||
const store = getCePipelineStore(ctx);
|
||||
// Add a second current-stage task to the SAME pipeline/stage.
|
||||
const t2 = await taskStore.createTask({ description: "second plan task" });
|
||||
store.createLink({ taskId: t2.id, cePipelineId, ceStageId: "plan", ceArtifactPath: null });
|
||||
|
||||
await moveTo(task.id, "done"); // only one terminal.
|
||||
await reconcileCePipelines(ctx);
|
||||
expect(store.getState(cePipelineId)!.currentStage).toBe("plan"); // not advanced.
|
||||
|
||||
await moveTo(t2.id, "done"); // now both terminal.
|
||||
await reconcileCePipelines(ctx);
|
||||
expect(store.getState(cePipelineId)!.currentStage).toBe("work"); // advanced.
|
||||
});
|
||||
});
|
||||
|
||||
describe("U8 conflict resolution (board vs CE authority)", () => {
|
||||
it("simultaneous board move + CE advance: board keeps the task column, CE keeps the pipeline content", async () => {
|
||||
const { cePipelineId, task } = await landPipeline("plan");
|
||||
const store = getCePipelineStore(ctx);
|
||||
|
||||
// CE-flow side: the pipeline owns its content; record an artifact (CE-authoritative).
|
||||
store.transitionState(cePipelineId, { lastArtifactPath: "/docs/plans/p.md" });
|
||||
|
||||
// Board side: move the task to done (board-authoritative for the column).
|
||||
await moveTo(task.id, "done");
|
||||
|
||||
// Reconcile resolves the collision: it READS the board column (never rewrites
|
||||
// the terminal task) and WRITES only CE-owned fields + a NEW task.
|
||||
await reconcileCePipelines(ctx);
|
||||
|
||||
// Board authority: the original task's column is exactly what the board set.
|
||||
const reread = await taskStore.getTask(task.id);
|
||||
expect(reread!.column).toBe("done");
|
||||
|
||||
// CE authority: the pipeline content (stage + artifact) is what CE wrote.
|
||||
const state = store.getState(cePipelineId)!;
|
||||
expect(state.currentStage).toBe("work");
|
||||
expect(state.lastArtifactPath).toBe("/docs/plans/p.md");
|
||||
|
||||
// The new outbound task is a fresh row — the writers never contended on one cell.
|
||||
const tasks = await taskStore.listTasks();
|
||||
const next = tasks.find((t) => t.id !== task.id)!;
|
||||
expect((next.sourceMetadata as Record<string, unknown>).ceStageId).toBe("work");
|
||||
});
|
||||
});
|
||||
@@ -4,6 +4,8 @@ import { installBundledCeSkills } from "./skill-installation.js";
|
||||
import { ensureCeSchema } from "./schema.js";
|
||||
import { createSessionRoutes } from "./routes/session-routes.js";
|
||||
import { createArtifactRoutes } from "./routes/artifact-routes.js";
|
||||
import { getCePipelineStore } from "./sync/pipeline-store.js";
|
||||
import { reconcileCePipelines } from "./sync/reconciler.js";
|
||||
|
||||
export { CompoundEngineeringDashboardView } from "./dashboard-view.js";
|
||||
export { COMPOUND_ENGINEERING_SKILLS } from "./skills.js";
|
||||
@@ -16,7 +18,16 @@ export {
|
||||
export { ensureCeSchema } from "./schema.js";
|
||||
export { CeSessionStore, getCeSessionStore } from "./session/session-store.js";
|
||||
export { CePipelineStore, getCePipelineStore } from "./sync/pipeline-store.js";
|
||||
export type { CePipelineLink, CreateCePipelineLinkInput } from "./sync/pipeline-store.js";
|
||||
export type {
|
||||
CePipelineLink,
|
||||
CreateCePipelineLinkInput,
|
||||
CePipelineState,
|
||||
CePipelineStatus,
|
||||
CeSyncQueueEntry,
|
||||
CeSyncReason,
|
||||
} from "./sync/pipeline-store.js";
|
||||
export { CeReconciler, reconcileCePipelines } from "./sync/reconciler.js";
|
||||
export type { ReconcileResult } from "./sync/reconciler.js";
|
||||
export {
|
||||
CeOrchestrator,
|
||||
WORK_STAGE_ID,
|
||||
@@ -41,6 +52,42 @@ const plugin = definePlugin({
|
||||
// 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,
|
||||
// INBOUND board→pipeline sync (U8 / FN-5719). The 5s hook budget
|
||||
// (plugin-runner invokeHookSafe) means these MUST be fast: resolve the link,
|
||||
// ENQUEUE a sync signal, and return. Heavy advancement (board reads, outbound
|
||||
// task creation) happens in the reconciler, NOT inline here. A reconcile
|
||||
// drain is fired-and-forgotten (never awaited) so a slow sweep cannot blow
|
||||
// the hook budget; correctness does not depend on it firing because the next
|
||||
// reconcile() sweep re-derives the transition from board truth.
|
||||
onTaskMoved: (task, fromColumn, toColumn, ctx) => {
|
||||
const store = getCePipelineStore(ctx);
|
||||
const link = store.findByTaskId(task.id);
|
||||
if (!link) return; // not a CE-linked task → ignore fast.
|
||||
store.enqueueSync({
|
||||
cePipelineId: link.cePipelineId,
|
||||
taskId: task.id,
|
||||
reason: "task_moved",
|
||||
fromColumn,
|
||||
toColumn,
|
||||
});
|
||||
void Promise.resolve()
|
||||
.then(() => reconcileCePipelines(ctx))
|
||||
.catch((err) => ctx.logger.warn(`CE reconcile (onTaskMoved) failed: ${String(err)}`));
|
||||
},
|
||||
onTaskCompleted: (task, ctx) => {
|
||||
const store = getCePipelineStore(ctx);
|
||||
const link = store.findByTaskId(task.id);
|
||||
if (!link) return;
|
||||
store.enqueueSync({
|
||||
cePipelineId: link.cePipelineId,
|
||||
taskId: task.id,
|
||||
reason: "task_completed",
|
||||
toColumn: "done",
|
||||
});
|
||||
void Promise.resolve()
|
||||
.then(() => reconcileCePipelines(ctx))
|
||||
.catch((err) => ctx.logger.warn(`CE reconcile (onTaskCompleted) failed: ${String(err)}`));
|
||||
},
|
||||
// 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
|
||||
|
||||
@@ -25,6 +25,20 @@ import type { Database } from "@fusion/core";
|
||||
* 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.
|
||||
*
|
||||
* `ce_pipeline_state` (U8) is the CE-pipeline's OWN state machine — DISTINCT from
|
||||
* board-task column state (KTD4 / FN-5719: two separate ownership state machines,
|
||||
* never one shared column encoding two concerns). It tracks where the pipeline
|
||||
* itself is: `currentStage` (the CE stage the pipeline has reached) and `status`
|
||||
* (`running` | `advancing` | `awaiting_board` | `completed`). The board owns task
|
||||
* columns; this table owns pipeline progress. They are reconciled — never merged.
|
||||
*
|
||||
* `ce_pipeline_sync_queue` (U8) is the event-enqueue seam (FN-5719). Lifecycle
|
||||
* hooks write a row here FAST (5s hook budget) and return; the reconciler drains
|
||||
* it. A dropped/never-enqueued event is still recovered because the reconciler
|
||||
* ALSO re-derives transitions from board state — the queue is an optimization,
|
||||
* board+state comparison is the convergence guarantee. `processedAt NULL` =
|
||||
* pending; non-null = drained (kept for audit, swept idempotently).
|
||||
*/
|
||||
export function ensureCeSchema(db: Database): void {
|
||||
db.exec(`
|
||||
@@ -68,5 +82,36 @@ export function ensureCeSchema(db: Database): void {
|
||||
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idxCePipelineLinksTask
|
||||
ON ce_pipeline_links(taskId);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS ce_pipeline_state (
|
||||
cePipelineId TEXT PRIMARY KEY,
|
||||
currentStage TEXT NOT NULL,
|
||||
status TEXT NOT NULL CHECK (status IN (
|
||||
'running','advancing','awaiting_board','completed'
|
||||
)),
|
||||
lastArtifactPath TEXT,
|
||||
createdAt TEXT NOT NULL,
|
||||
updatedAt TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idxCePipelineStateStatus
|
||||
ON ce_pipeline_state(status, updatedAt DESC, cePipelineId);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS ce_pipeline_sync_queue (
|
||||
id TEXT PRIMARY KEY,
|
||||
cePipelineId TEXT NOT NULL,
|
||||
taskId TEXT NOT NULL,
|
||||
reason TEXT NOT NULL,
|
||||
fromColumn TEXT,
|
||||
toColumn TEXT,
|
||||
enqueuedAt TEXT NOT NULL,
|
||||
processedAt TEXT
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idxCePipelineSyncQueuePending
|
||||
ON ce_pipeline_sync_queue(processedAt, enqueuedAt, id);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idxCePipelineSyncQueuePipeline
|
||||
ON ce_pipeline_sync_queue(cePipelineId, enqueuedAt, id);
|
||||
`);
|
||||
}
|
||||
|
||||
@@ -361,6 +361,16 @@ export class CeOrchestrator {
|
||||
const ceStageId = session.stage;
|
||||
const ceArtifactPath = session.artifactPath ?? null;
|
||||
|
||||
// Seed the CE-pipeline STATE record (U8). This is the pipeline's OWN state
|
||||
// machine, distinct from the board task columns it will spawn. The pipeline
|
||||
// is "running" at this stage until a board signal advances it.
|
||||
this.pipelineStore.upsertState({
|
||||
cePipelineId,
|
||||
currentStage: ceStageId,
|
||||
status: "running",
|
||||
lastArtifactPath: ceArtifactPath,
|
||||
});
|
||||
|
||||
for (const spec of specs) {
|
||||
const description = spec.description.trim();
|
||||
if (!description) continue; // createTask rejects blank descriptions.
|
||||
|
||||
@@ -49,6 +49,106 @@ export interface CreateCePipelineLinkInput {
|
||||
id?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* CE-pipeline STATUS — the pipeline's own lifecycle, DISTINCT from board-task
|
||||
* columns (KTD4 / FN-5719). The board owns task columns; this owns pipeline
|
||||
* progress. The two are never encoded in one shared column.
|
||||
*
|
||||
* running — pipeline is at `currentStage`, work in flight on the board.
|
||||
* advancing — a board signal arrived; reconciler is moving it on / feeding
|
||||
* the next stage (transient, set inside the reconciler sweep).
|
||||
* awaiting_board — pipeline advanced and is waiting on board task(s) again.
|
||||
* completed — pipeline reached its terminal stage and finished.
|
||||
*/
|
||||
export type CePipelineStatus = "running" | "advancing" | "awaiting_board" | "completed";
|
||||
|
||||
/** The CE-pipeline's own state record (separate state machine from the board). */
|
||||
export interface CePipelineState {
|
||||
cePipelineId: string;
|
||||
/** The CE stage the pipeline has reached (a stage id, e.g. "work"). */
|
||||
currentStage: string;
|
||||
status: CePipelineStatus;
|
||||
/** Last artifact the pipeline produced/propagated (CE-authoritative content). */
|
||||
lastArtifactPath: string | null;
|
||||
createdAt: string;
|
||||
updatedAt: string;
|
||||
}
|
||||
|
||||
interface CePipelineStateRow {
|
||||
cePipelineId: string;
|
||||
currentStage: string;
|
||||
status: CePipelineStatus;
|
||||
lastArtifactPath: string | null;
|
||||
createdAt: string;
|
||||
updatedAt: string;
|
||||
}
|
||||
|
||||
export interface UpsertCePipelineStateInput {
|
||||
cePipelineId: string;
|
||||
currentStage: string;
|
||||
status?: CePipelineStatus;
|
||||
lastArtifactPath?: string | null;
|
||||
}
|
||||
|
||||
/** Why a board change was enqueued for the pipeline (audit + reconcile routing). */
|
||||
export type CeSyncReason = "task_moved" | "task_completed" | "reconcile";
|
||||
|
||||
/** A pending (or drained) board→pipeline sync signal. */
|
||||
export interface CeSyncQueueEntry {
|
||||
id: string;
|
||||
cePipelineId: string;
|
||||
taskId: string;
|
||||
reason: CeSyncReason;
|
||||
fromColumn: string | null;
|
||||
toColumn: string | null;
|
||||
enqueuedAt: string;
|
||||
processedAt: string | null;
|
||||
}
|
||||
|
||||
interface CeSyncQueueRow {
|
||||
id: string;
|
||||
cePipelineId: string;
|
||||
taskId: string;
|
||||
reason: CeSyncReason;
|
||||
fromColumn: string | null;
|
||||
toColumn: string | null;
|
||||
enqueuedAt: string;
|
||||
processedAt: string | null;
|
||||
}
|
||||
|
||||
export interface EnqueueSyncInput {
|
||||
cePipelineId: string;
|
||||
taskId: string;
|
||||
reason: CeSyncReason;
|
||||
fromColumn?: string | null;
|
||||
toColumn?: string | null;
|
||||
id?: string;
|
||||
}
|
||||
|
||||
function rowToState(row: CePipelineStateRow): CePipelineState {
|
||||
return {
|
||||
cePipelineId: row.cePipelineId,
|
||||
currentStage: row.currentStage,
|
||||
status: row.status,
|
||||
lastArtifactPath: row.lastArtifactPath,
|
||||
createdAt: row.createdAt,
|
||||
updatedAt: row.updatedAt,
|
||||
};
|
||||
}
|
||||
|
||||
function rowToQueueEntry(row: CeSyncQueueRow): CeSyncQueueEntry {
|
||||
return {
|
||||
id: row.id,
|
||||
cePipelineId: row.cePipelineId,
|
||||
taskId: row.taskId,
|
||||
reason: row.reason,
|
||||
fromColumn: row.fromColumn,
|
||||
toColumn: row.toColumn,
|
||||
enqueuedAt: row.enqueuedAt,
|
||||
processedAt: row.processedAt,
|
||||
};
|
||||
}
|
||||
|
||||
function rowToLink(row: CePipelineLinkRow): CePipelineLink {
|
||||
return {
|
||||
id: row.id,
|
||||
@@ -109,6 +209,115 @@ export class CePipelineStore {
|
||||
.get(taskId) as CePipelineLinkRow | undefined;
|
||||
return row ? rowToLink(row) : undefined;
|
||||
}
|
||||
|
||||
// ── CE-pipeline STATE machine (U8) ───────────────────────────────────
|
||||
// Separate from board-task columns: this table is the pipeline's OWN state.
|
||||
|
||||
/** Read a pipeline's own state record. */
|
||||
getState(cePipelineId: string): CePipelineState | undefined {
|
||||
const row = this.db
|
||||
.prepare(`SELECT * FROM ce_pipeline_state WHERE cePipelineId = ?`)
|
||||
.get(cePipelineId) as CePipelineStateRow | undefined;
|
||||
return row ? rowToState(row) : undefined;
|
||||
}
|
||||
|
||||
/** All pipeline state records (the reconciler sweeps every one). */
|
||||
listAllState(): CePipelineState[] {
|
||||
const rows = this.db
|
||||
.prepare(`SELECT * FROM ce_pipeline_state ORDER BY updatedAt DESC, cePipelineId`)
|
||||
.all() as CePipelineStateRow[];
|
||||
return rows.map(rowToState);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create or update a pipeline's state. Idempotent on `cePipelineId`. `status`
|
||||
* defaults to `running` on first write and is preserved on update unless given.
|
||||
*/
|
||||
upsertState(input: UpsertCePipelineStateInput): CePipelineState {
|
||||
const now = new Date().toISOString();
|
||||
const existing = this.getState(input.cePipelineId);
|
||||
const status = input.status ?? existing?.status ?? "running";
|
||||
const lastArtifactPath =
|
||||
input.lastArtifactPath !== undefined ? input.lastArtifactPath : existing?.lastArtifactPath ?? null;
|
||||
if (existing) {
|
||||
this.db
|
||||
.prepare(
|
||||
`UPDATE ce_pipeline_state
|
||||
SET currentStage = ?, status = ?, lastArtifactPath = ?, updatedAt = ?
|
||||
WHERE cePipelineId = ?`,
|
||||
)
|
||||
.run(input.currentStage, status, lastArtifactPath, now, input.cePipelineId);
|
||||
} else {
|
||||
this.db
|
||||
.prepare(
|
||||
`INSERT INTO ce_pipeline_state
|
||||
(cePipelineId, currentStage, status, lastArtifactPath, createdAt, updatedAt)
|
||||
VALUES (?, ?, ?, ?, ?, ?)`,
|
||||
)
|
||||
.run(input.cePipelineId, input.currentStage, status, lastArtifactPath, now, now);
|
||||
}
|
||||
return this.getState(input.cePipelineId)!;
|
||||
}
|
||||
|
||||
/**
|
||||
* Transition a pipeline to a new stage/status. Returns the updated state, or
|
||||
* `undefined` if the pipeline has no state row yet (caller should seed first).
|
||||
*/
|
||||
transitionState(
|
||||
cePipelineId: string,
|
||||
next: { currentStage?: string; status?: CePipelineStatus; lastArtifactPath?: string | null },
|
||||
): CePipelineState | undefined {
|
||||
const existing = this.getState(cePipelineId);
|
||||
if (!existing) return undefined;
|
||||
return this.upsertState({
|
||||
cePipelineId,
|
||||
currentStage: next.currentStage ?? existing.currentStage,
|
||||
status: next.status ?? existing.status,
|
||||
lastArtifactPath:
|
||||
next.lastArtifactPath !== undefined ? next.lastArtifactPath : existing.lastArtifactPath,
|
||||
});
|
||||
}
|
||||
|
||||
// ── Event-enqueue seam (U8 / FN-5719) ────────────────────────────────
|
||||
// Hooks write here FAST and return; the reconciler drains. A missed enqueue is
|
||||
// still recovered because reconcile() re-derives from board state too.
|
||||
|
||||
/** Append a pending board→pipeline sync signal. Fast, append-only. */
|
||||
enqueueSync(input: EnqueueSyncInput): CeSyncQueueEntry {
|
||||
const entry: CeSyncQueueEntry = {
|
||||
id: input.id ?? randomUUID(),
|
||||
cePipelineId: input.cePipelineId,
|
||||
taskId: input.taskId,
|
||||
reason: input.reason,
|
||||
fromColumn: input.fromColumn ?? null,
|
||||
toColumn: input.toColumn ?? null,
|
||||
enqueuedAt: new Date().toISOString(),
|
||||
processedAt: null,
|
||||
};
|
||||
this.db
|
||||
.prepare(
|
||||
`INSERT INTO ce_pipeline_sync_queue
|
||||
(id, cePipelineId, taskId, reason, fromColumn, toColumn, enqueuedAt, processedAt)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, NULL)`,
|
||||
)
|
||||
.run(entry.id, entry.cePipelineId, entry.taskId, entry.reason, entry.fromColumn, entry.toColumn, entry.enqueuedAt);
|
||||
return entry;
|
||||
}
|
||||
|
||||
/** All pending (un-drained) queue entries, oldest first. */
|
||||
listPendingSync(): CeSyncQueueEntry[] {
|
||||
const rows = this.db
|
||||
.prepare(`SELECT * FROM ce_pipeline_sync_queue WHERE processedAt IS NULL ORDER BY enqueuedAt, id`)
|
||||
.all() as CeSyncQueueRow[];
|
||||
return rows.map(rowToQueueEntry);
|
||||
}
|
||||
|
||||
/** Mark a queue entry drained (idempotent). */
|
||||
markSyncProcessed(id: string): void {
|
||||
this.db
|
||||
.prepare(`UPDATE ce_pipeline_sync_queue SET processedAt = ? WHERE id = ? AND processedAt IS NULL`)
|
||||
.run(new Date().toISOString(), id);
|
||||
}
|
||||
}
|
||||
|
||||
const storeCache = new WeakMap<object, CePipelineStore>();
|
||||
|
||||
@@ -0,0 +1,246 @@
|
||||
import type { PluginContext, Task } from "@fusion/core";
|
||||
import { listStages } from "../session/stage-registry.js";
|
||||
import {
|
||||
CE_PLUGIN_ID,
|
||||
CE_WORK_SOURCE_TYPE,
|
||||
} from "../session/orchestrator.js";
|
||||
import {
|
||||
getCePipelineStore,
|
||||
type CePipelineLink,
|
||||
type CePipelineState,
|
||||
type CePipelineStore,
|
||||
} from "./pipeline-store.js";
|
||||
|
||||
/**
|
||||
* BIDIRECTIONAL SYNC RECONCILER (U8 / FN-5719 pattern).
|
||||
*
|
||||
* Two SEPARATE state machines are kept in sync, never merged (KTD4):
|
||||
* - Board-task ownership → the task's `column` (board is authoritative).
|
||||
* - CE-pipeline ownership → `ce_pipeline_state.{currentStage,status}` (CE flow
|
||||
* is authoritative for artifact/pipeline content).
|
||||
*
|
||||
* INBOUND (board → pipeline): the lifecycle hooks (`onTaskMoved`/`onTaskCompleted`
|
||||
* in index.ts) do the MINIMUM under the 5s hook budget — resolve the link and
|
||||
* `enqueueSync(...)`, then return. They do NOT advance the pipeline inline.
|
||||
*
|
||||
* RECONCILE (the convergence guarantee): `reconcile()` is a single on-demand
|
||||
* sweep — NOT a tight interval poll (per docs/performance/dashboard-load.md).
|
||||
* It (1) drains the queue and (2) INDEPENDENTLY re-derives transitions by
|
||||
* comparing live board state (`ctx.taskStore`) against pipeline state. Step (2)
|
||||
* is why a DROPPED or never-enqueued hook event still converges: the queue is an
|
||||
* optimization; the board↔state comparison is the source of truth.
|
||||
*
|
||||
* OUTBOUND (pipeline → board): when a pipeline advances to a stage that produces
|
||||
* board work, the reconciler creates the next-stage board task via
|
||||
* `ctx.taskStore.createTask` and links it — propagating the CE-flow change onto
|
||||
* the board.
|
||||
*
|
||||
* TRIGGER MODEL (honest about the host seam): there is NO host scheduler wired to
|
||||
* call this on a timer. In production the sweep is invoked (a) right after the
|
||||
* hooks enqueue (a cheap drain on the same board mutation that triggered the
|
||||
* hook), and (b) on demand from a route (U9 settings/refresh surface) or on a
|
||||
* dashboard session-change. Because step (2) re-derives from board truth, any
|
||||
* single missed trigger is recovered on the NEXT sweep — no continuous poll loop
|
||||
* is needed for correctness.
|
||||
*/
|
||||
|
||||
/** Columns that mean "this stage's board work is finished" → advance the pipeline. */
|
||||
const TERMINAL_COLUMNS = new Set(["in-review", "done"]);
|
||||
|
||||
export interface ReconcileResult {
|
||||
/** Queue entries drained this sweep. */
|
||||
drained: number;
|
||||
/** Pipelines whose state advanced this sweep. */
|
||||
advanced: number;
|
||||
/** Board tasks created outbound this sweep (next-stage propagation). */
|
||||
tasksCreated: number;
|
||||
/** Pipelines inspected. */
|
||||
inspected: number;
|
||||
}
|
||||
|
||||
/** The linear CE stage order. The pipeline advances along this sequence. */
|
||||
function stageOrder(): string[] {
|
||||
return listStages().map((s) => s.stageId);
|
||||
}
|
||||
|
||||
/** The stage AFTER `stageId` in the pipeline, or `undefined` if it's terminal. */
|
||||
function nextStageAfter(stageId: string): string | undefined {
|
||||
const order = stageOrder();
|
||||
const idx = order.indexOf(stageId);
|
||||
if (idx < 0 || idx >= order.length - 1) return undefined;
|
||||
return order[idx + 1];
|
||||
}
|
||||
|
||||
export class CeReconciler {
|
||||
private readonly ctx: PluginContext;
|
||||
private readonly store: CePipelineStore;
|
||||
|
||||
constructor(ctx: PluginContext) {
|
||||
this.ctx = ctx;
|
||||
this.store = getCePipelineStore(ctx);
|
||||
}
|
||||
|
||||
/**
|
||||
* Drain the queue AND re-derive missed transitions from live board state, then
|
||||
* apply any pipeline advancement (with outbound board propagation). Idempotent:
|
||||
* running it twice is a no-op once everything has converged.
|
||||
*/
|
||||
async reconcile(): Promise<ReconcileResult> {
|
||||
const result: ReconcileResult = { drained: 0, advanced: 0, tasksCreated: 0, inspected: 0 };
|
||||
|
||||
// (1) Drain the queue. Draining is just an audit/ack — the actual decision is
|
||||
// re-derived from board truth below, so a queue entry for an already-handled
|
||||
// transition is harmless.
|
||||
const pending = this.store.listPendingSync();
|
||||
const pipelineIds = new Set<string>();
|
||||
for (const entry of pending) {
|
||||
pipelineIds.add(entry.cePipelineId);
|
||||
this.store.markSyncProcessed(entry.id);
|
||||
result.drained++;
|
||||
}
|
||||
|
||||
// (2) Convergence sweep: inspect EVERY pipeline that has state, not only the
|
||||
// ones with queued entries. This is what recovers a dropped/never-enqueued
|
||||
// hook event — board truth is compared against pipeline state regardless of
|
||||
// whether a queue row exists.
|
||||
void pipelineIds;
|
||||
const states = this.store.listAllState();
|
||||
for (const state of states) {
|
||||
result.inspected++;
|
||||
const advanced = await this.reconcileOne(state);
|
||||
if (advanced) {
|
||||
result.advanced++;
|
||||
if (advanced.created) result.tasksCreated++;
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-derive whether ONE pipeline should advance by reading the live board
|
||||
* column of its current-stage task(s). Board is authoritative for task state;
|
||||
* we never write the task column from here for the current stage.
|
||||
*/
|
||||
private async reconcileOne(
|
||||
state: CePipelineState,
|
||||
): Promise<{ created: boolean } | undefined> {
|
||||
if (state.status === "completed") return undefined;
|
||||
|
||||
// Links produced by this pipeline AT its current stage are the board tasks
|
||||
// whose completion gates advancement.
|
||||
const links = this.store
|
||||
.listByPipeline(state.cePipelineId)
|
||||
.filter((l) => l.ceStageId === state.currentStage);
|
||||
if (links.length === 0) return undefined;
|
||||
|
||||
const tasks = await this.loadTasks(links);
|
||||
if (tasks.length === 0) return undefined;
|
||||
|
||||
// Advancement rule: every current-stage board task has reached a terminal
|
||||
// column (board-authoritative read). Partial completion keeps it running.
|
||||
const allTerminal = tasks.every((t) => t && TERMINAL_COLUMNS.has(t.column));
|
||||
if (!allTerminal) {
|
||||
// Still running on the board — make sure our status reflects that and stop.
|
||||
if (state.status !== "running") {
|
||||
this.store.transitionState(state.cePipelineId, { status: "running" });
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const next = nextStageAfter(state.currentStage);
|
||||
if (!next) {
|
||||
// Terminal stage finished → pipeline completed. No outbound task.
|
||||
this.store.transitionState(state.cePipelineId, { status: "completed" });
|
||||
this.ctx.emitEvent("compound-engineering:pipeline-completed", {
|
||||
cePipelineId: state.cePipelineId,
|
||||
stage: state.currentStage,
|
||||
});
|
||||
return { created: false };
|
||||
}
|
||||
|
||||
// CONFLICT POLICY (explicit): board is authoritative for the task columns we
|
||||
// just READ (we never rewrote them); CE flow is authoritative for the
|
||||
// pipeline content we WRITE (currentStage, artifact, the next-stage task).
|
||||
// Advancing only moves the CE-owned fields + creates a NEW board task; it
|
||||
// never mutates the already-terminal board tasks, so the two writers never
|
||||
// contend over the same cell.
|
||||
const created = await this.advance(state, next);
|
||||
return { created };
|
||||
}
|
||||
|
||||
/**
|
||||
* Advance the pipeline to `nextStage` (CE-owned write) and propagate OUTBOUND
|
||||
* by creating the next-stage board task (board-owned write on a NEW row).
|
||||
* Idempotent: if a link for the next stage already exists, we don't duplicate.
|
||||
*/
|
||||
private async advance(state: CePipelineState, nextStage: string): Promise<boolean> {
|
||||
// Idempotency guard: if we already advanced (a next-stage link exists), just
|
||||
// ensure state is consistent and skip the outbound create.
|
||||
const already = this.store
|
||||
.listByPipeline(state.cePipelineId)
|
||||
.some((l) => l.ceStageId === nextStage);
|
||||
|
||||
this.store.transitionState(state.cePipelineId, {
|
||||
currentStage: nextStage,
|
||||
status: already ? "running" : "awaiting_board",
|
||||
});
|
||||
|
||||
if (already) return false;
|
||||
|
||||
const task = await this.ctx.taskStore.createTask({
|
||||
title: `CE ${nextStage}: continue pipeline`,
|
||||
description: `Continue the compound-engineering pipeline at the "${nextStage}" stage.`,
|
||||
source: {
|
||||
sourceType: CE_WORK_SOURCE_TYPE,
|
||||
sourceSessionId: state.cePipelineId,
|
||||
sourceMetadata: {
|
||||
pluginId: CE_PLUGIN_ID,
|
||||
cePipelineId: state.cePipelineId,
|
||||
ceStageId: nextStage,
|
||||
ceArtifactPath: state.lastArtifactPath,
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
this.store.createLink({
|
||||
taskId: task.id,
|
||||
cePipelineId: state.cePipelineId,
|
||||
ceStageId: nextStage,
|
||||
ceArtifactPath: state.lastArtifactPath,
|
||||
});
|
||||
|
||||
// Pipeline is now waiting on the freshly-created board task.
|
||||
this.store.transitionState(state.cePipelineId, { status: "awaiting_board" });
|
||||
|
||||
this.ctx.emitEvent("compound-engineering:pipeline-advanced", {
|
||||
cePipelineId: state.cePipelineId,
|
||||
fromStage: state.currentStage,
|
||||
toStage: nextStage,
|
||||
taskId: task.id,
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Load the live board tasks for a set of links (board-authoritative read). */
|
||||
private async loadTasks(links: CePipelineLink[]): Promise<Array<Task | undefined>> {
|
||||
const out: Array<Task | undefined> = [];
|
||||
for (const link of links) {
|
||||
try {
|
||||
const task = await this.ctx.taskStore.getTask(link.taskId);
|
||||
out.push(task ?? undefined);
|
||||
} catch {
|
||||
// A deleted/missing task is treated as absent, not terminal.
|
||||
out.push(undefined);
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Convenience: build a reconciler and run one sweep. This is the entry point a
|
||||
* route handler or post-hook drain calls.
|
||||
*/
|
||||
export async function reconcileCePipelines(ctx: PluginContext): Promise<ReconcileResult> {
|
||||
return new CeReconciler(ctx).reconcile();
|
||||
}
|
||||
Reference in New Issue
Block a user