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 index 6bafe7aba1..176c44cc5c 100644 --- a/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-flow.test.ts +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-flow.test.ts @@ -58,6 +58,7 @@ describe("orchestrator happy path", () => { // Adding a stage = data only. registerStage({ stageId: "compound", + order: 600, skillId: "ce-compound", artifactLocation: "docs/solutions/", icon: "BookOpen", 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 index 04cfbb693d..c1d9edafa4 100644 --- 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 @@ -3,7 +3,7 @@ import type { InteractiveAiSession, InteractiveAiSessionEvent, PlanningQuestion 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"; +import { makeHarness, makeScriptedSession, type TestHarness } from "./_harness.js"; /** * CHARACTERIZATION TEST — written first (U5 execution note: cover the @@ -82,10 +82,11 @@ describe("interrupt + resume (no 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). + it("leaves an awaiting_input session (waiting on a human) untouched even far past the stale band, and resume returns the same question with full history", async () => { + // A session legitimately paused on a human question: status awaiting_input + // with currentQuestion set, lastActivity well past the interval stale band. + // Human response time is unbounded, so this is NOT a crashed turn — the + // interval rubric must not misclassify it as stale. 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() }); @@ -93,26 +94,109 @@ describe("interrupt + resume (no silent loss)", () => { store.update(created.id, { status: "awaiting_input", currentQuestion: QUESTION, - // 10× interval old → unambiguously stale. + // 10× interval old → far past the band, yet legitimately awaiting a human. lastActivityAt: Date.now() - 10_000, }); const recovered = store.recoverStaleSessions(); - expect(recovered).toContain(created.id); + // Not flagged stale / not recovered — a human wait is not a crashed turn. + expect(recovered).not.toContain(created.id); const after = store.get(created.id)!; - // Awaiting-input session with a question stays resumable, not dropped. + // Awaiting-input session with a question stays resumable, unchanged. 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); + // Rehydration re-creates a live session and replays the opening message, + // draining the agent's response (the question) during replay. + const replaySession = makeScriptedSession([{ type: "question", data: QUESTION }]); + const orch = new CeOrchestrator({ + ctx: h.ctx, + createInteractiveAiSession: vi.fn(async () => ({ session: replaySession })), + projectRoot: h.projectRoot, + }); + const resumed = await 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("Bug 5: an interrupted/awaiting session with a currentQuestion + history can be resumed (rehydrated) and then ANSWERED to continue to completion", async () => { + // Simulate the post-interrupt / post-restart state: a session persisted + // mid-question (awaiting_input, currentQuestion set, full history) whose live + // handle was disposed and removed from this.live. This is exactly the state + // resume() must be able to back with a real live handle. + const store = getCeSessionStore(h.ctx); + const created = store.create({ stage: "brainstorm", turnIntervalMs: 5000 }); + 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 }); + const sessionId = created.id; + + // The rehydration factory: replays the opening prompt (yields the question, + // which replay discards), then on the real answer turn completes the stage. + const rehydrated = makeScriptedSession([ + { type: "question", data: QUESTION }, + { type: "complete", data: { artifact: "# Done\n" } }, + ]); + const factory = vi.fn(async () => ({ session: rehydrated })); + + const orch = new CeOrchestrator({ + ctx: h.ctx, + createInteractiveAiSession: factory, + projectRoot: h.projectRoot, + turnTimeoutMs: 5000, + }); + + // Pre-fix: resume() flips status to awaiting_input but never re-establishes a + // live handle, so the subsequent answer() throws "no live handle; call + // resume() first" — a dead-end loop. Post-fix: resume rehydrates a live one. + const resumed = await orch.resume(sessionId); + expect(resumed.session.status).toBe("awaiting_input"); + expect(resumed.session.currentQuestion?.id).toBe("q1"); + expect(factory).toHaveBeenCalledTimes(1); // rehydration created a live session. + + // The resumed session is genuinely answerable now — drive it to completion. + const done = await orch.answer(sessionId, "q1", "a"); + expect(done.event?.type).toBe("complete"); + expect(done.session.status).toBe("completed"); + }); + + it("Bug 4: answering with a wrong questionId throws and leaves the session awaiting_input with its currentQuestion preserved", 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.session.status).toBe("awaiting_input"); + expect(started.session.currentQuestion?.id).toBe("q1"); + + // Answer with the WRONG questionId → must reject without mutating state. + await expect(orch.answer(started.session.id, "WRONG-ID", "a")).rejects.toThrow(/q1|WRONG-ID/); + + // The recovery anchor is intact: still awaiting_input with currentQuestion. + const after = orch.getState(started.session.id)!; + expect(after.status).toBe("awaiting_input"); + expect(after.currentQuestion?.id).toBe("q1"); + // No spurious answer turn was appended to history. + expect(after.conversationHistory.some((t) => t.text.includes("WRONG-ID"))).toBe(false); + + // The correct questionId is still accepted (the live handle wasn't disturbed). + // The session hangs on the answer turn → it interrupts, but it DID accept the + // answer, proving the rejection above didn't break the seam. + const accepted = await orch.answer(started.session.id, "q1", "a"); + expect(accepted.session.status).toBe("interrupted"); + }); + 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 }); 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 index 6c875d787b..6c92f31aeb 100644 --- a/plugins/fusion-plugin-compound-engineering/src/__tests__/session-store.test.ts +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/session-store.test.ts @@ -74,4 +74,32 @@ describe("interval-relative staleness (FN-4172 rubric)", () => { const completed = store.update(s.id, { status: "completed", lastActivityAt: Date.now() - 1_000_000 })!; expect(store.isStale(completed)).toBe(false); }); + + it("Bug 2: a human-slow awaiting_input session past 3× is NOT recovered, while a stuck active one still is", () => { + const store = new CeSessionStore(h.db); + const now = Date.now(); + + // A session legitimately waiting on a human, far past 3× the interval. Human + // response time is unbounded — this is not a crashed turn. + const waiting = store.create({ stage: "brainstorm", turnIntervalMs: 1000 }); + store.update(waiting.id, { + status: "awaiting_input", + currentQuestion: { id: "q", type: "text", question: "?" }, + lastActivityAt: now - 100_000, // 100× interval + }); + + // A genuinely stuck in-flight agent turn past the threshold. + const stuck = store.create({ stage: "brainstorm", turnIntervalMs: 1000 }); + store.update(stuck.id, { status: "active", lastActivityAt: now - 100_000 }); + + const recovered = store.recoverStaleSessions(now); + + // The human-wait is excluded from the interval rubric entirely. + expect(recovered).not.toContain(waiting.id); + expect(store.get(waiting.id)!.status).toBe("awaiting_input"); + + // The stuck active turn is still recovered (here: no question → interrupted). + expect(recovered).toContain(stuck.id); + expect(store.get(stuck.id)!.status).toBe("interrupted"); + }); }); diff --git a/plugins/fusion-plugin-compound-engineering/src/__tests__/sync.test.ts b/plugins/fusion-plugin-compound-engineering/src/__tests__/sync.test.ts index 0fb0a0efa7..651ce2cbd4 100644 --- a/plugins/fusion-plugin-compound-engineering/src/__tests__/sync.test.ts +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/sync.test.ts @@ -12,6 +12,7 @@ import plugin, { } from "../index.js"; import { getCePipelineStore } from "../sync/pipeline-store.js"; import { CeReconciler, reconcileCePipelines } from "../sync/reconciler.js"; +import { registerStage, unregisterStage } from "../session/stage-registry.js"; import { makeScriptedSession } from "./_harness.js"; /** @@ -216,6 +217,61 @@ describe("U8 reconciler (convergence + outbound)", () => { await reconcileCePipelines(ctx); expect(store.getState(cePipelineId)!.currentStage).toBe("work"); // advanced. }); + + it("Bug 1: a deleted current-stage task does NOT wedge the pipeline — one terminal + one deleted still advances", async () => { + const { cePipelineId, task } = await landPipeline("plan"); + const store = getCePipelineStore(ctx); + + // Add a SECOND current-stage task linked to the same pipeline/stage, then + // DELETE it from the board (loadTasks will yield undefined for it). + const doomed = await taskStore.createTask({ description: "second plan task (to delete)" }); + store.createLink({ taskId: doomed.id, cePipelineId, ceStageId: "plan", ceArtifactPath: null }); + await taskStore.deleteTask(doomed.id); + + // The remaining task reaches terminal. Pre-fix: the deleted task made + // `every(... t && ...)` false, wedging the pipeline at "plan" forever. + await moveTo(task.id, "done"); + await reconcileCePipelines(ctx); + + // Post-fix: terminality is computed over EXISTING tasks only → it advances. + expect(store.getState(cePipelineId)!.currentStage).toBe("work"); + }); + + it("Bug 1: if ALL current-stage tasks were deleted, the pipeline is left unchanged (no wedge, no crash)", async () => { + const { cePipelineId, task } = await landPipeline("plan"); + const store = getCePipelineStore(ctx); + + await taskStore.deleteTask(task.id); // every current-stage task gone. + + // Safe non-wedging behavior: state unchanged, no advancement, no throw. + await expect(reconcileCePipelines(ctx)).resolves.toBeTruthy(); + expect(store.getState(cePipelineId)!.currentStage).toBe("plan"); + }); + + it("Bug 3: a stage registered with an `order` between two existing stages is the next stage (not append-at-end)", async () => { + // Insert a stage between plan(400) and work(500). Registry/Map insertion + // order would append it at the end; the explicit `order` slots it mid-pipeline. + registerStage({ + stageId: "refine", + order: 450, + skillId: "ce-refine", + artifactLocation: "docs/refine/", + icon: "Wand", + label: "Refine", + }); + try { + const { cePipelineId, task } = await landPipeline("plan"); + const store = getCePipelineStore(ctx); + + await moveTo(task.id, "done"); + await reconcileCePipelines(ctx); + + // Advances to the inserted stage, NOT to "work" (the old append-at-end). + expect(store.getState(cePipelineId)!.currentStage).toBe("refine"); + } finally { + unregisterStage("refine"); + } + }); }); describe("U8 conflict resolution (board vs CE authority)", () => { diff --git a/plugins/fusion-plugin-compound-engineering/src/artifacts/discovery.ts b/plugins/fusion-plugin-compound-engineering/src/artifacts/discovery.ts index cb551546d1..959a4b3b37 100644 --- a/plugins/fusion-plugin-compound-engineering/src/artifacts/discovery.ts +++ b/plugins/fusion-plugin-compound-engineering/src/artifacts/discovery.ts @@ -149,8 +149,10 @@ function readArtifactEntry( if (st.size > MAX_ARTIFACT_BYTES) { return makeError(stage, relPath, `Artifact too large to read (${st.size} bytes)`); } - // Probe readability (no bytes transferred) so a malformed/unreadable file is + // Probe READ PERMISSION only (no bytes transferred) so an unreadable file is // surfaced now as an error entry rather than crashing later at render time. + // NOTE: this is a permission probe, NOT a content check — malformed/corrupt + // file CONTENT is only detected at read time (readCeArtifact), not here. accessSync(abs, constants.R_OK); return { id: makeId(stage, relPath), 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 015691a6aa..86be65e303 100644 --- a/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts +++ b/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts @@ -98,7 +98,7 @@ export function createSessionRoutes(): PluginRouteDefinition[] { const id = (req as RouteRequest).params.id; const orch = getOrchestrator(ctx); try { - const result = orch.resume(id); + const result = await orch.resume(id); return { status: 200, body: { session: result.session } }; } catch (err) { return { status: 404, body: { error: err instanceof Error ? err.message : String(err) } }; diff --git a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts index 3a81db1728..7c18e70613 100644 --- a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts +++ b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts @@ -226,13 +226,23 @@ export class CeOrchestrator { /** Answer the awaiting question and continue the loop. */ async answer(sessionId: string, questionId: string, response: unknown): Promise { const session = this.requireSession(sessionId); + if (session.status !== "awaiting_input") { + throw new Error(`Session ${sessionId} is not awaiting input (status=${session.status}).`); + } + // Validate the questionId BEFORE mutating any persisted state. A stale/wrong + // questionId must NOT clear `currentQuestion` or flip status to active — + // doing so would destroy the recovery anchor while the seam rejects the + // mismatch, leaving the DB diverged from the live session. Reject cleanly and + // leave `currentQuestion`/status intact so the session stays answerable. + if (questionId !== session.currentQuestion?.id) { + throw new Error( + `Session ${sessionId} is awaiting question "${session.currentQuestion?.id ?? "(none)"}", not "${questionId}".`, + ); + } 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 }), @@ -243,25 +253,144 @@ export class CeOrchestrator { } /** - * 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 an `awaiting_input`, `interrupted`, or `error` session — and, crucially, + * RE-ESTABLISH a live interactive handle so the resumed session can actually be + * answered (Bug 5). After an interrupt/timeout the live handle was disposed and + * removed from `this.live`; flipping persisted status back to `awaiting_input` + * without a live handle was a dead end (resume → answer → "call resume() first"). + * + * When a question is still pending and no live handle exists, we rehydrate via + * the factory and REPLAY the persisted conversation history (opening message + + * prior answers) to prime the fresh agent back to the current awaiting question, + * repopulating `this.live`. Replay is side-effect-suppressed: it reconstructs + * the agent's context only — no artifact writes, no event emits, no history + * re-append (the DB already reflects the final state). + * + * If no factory is available (no live handle can ever be created in this + * process), we DO NOT advertise a misleading answerable status: the session is + * left `interrupted` with a clear error explaining it can't be continued here. */ - resume(sessionId: string): CeStepResult { + async resume(sessionId: string): Promise { 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; + + // Terminal / already-answerable-with-a-live-handle cases need no rehydration. + if (session.status === "completed") return { session }; + if (session.status === "awaiting_input" && this.live.has(sessionId)) { + return { session }; // already live + answerable. + } + + // No pending question → nothing to re-prime to. Mark active so the caller can + // re-run the turn with fresh input (retry for `error`, resume for others). + if (!session.currentQuestion) { + const next = this.store.update(sessionId, { status: "active", error: null }) ?? session; return { session: next }; } - // Already awaiting_input (or terminal completed) — return as-is (idempotent). - return { session }; + + // A live handle already exists (e.g. interrupted but not disposed) — just + // restore the answerable status. + if (this.live.has(sessionId)) { + const next = this.store.update(sessionId, { status: "awaiting_input", error: null }) ?? session; + return { session: next }; + } + + // Rehydration path: re-create the live session and replay history back to the + // current question. + if (!this.factory) { + // Honest status: we cannot back an answerable state in this process, so do + // not pretend the session is resumable here. Surface a clear error. + const next = + this.store.update(sessionId, { + status: "interrupted", + error: + "Session cannot be continued in this process: interactive AI sessions are unavailable (no factory on this context). Resume from a route context with the engine loaded.", + }) ?? session; + return { session: next }; + } + + try { + await this.rehydrate(session); + } catch (err) { + // Rehydration failed — keep progress, surface the failure, do not advertise + // an answerable status we can't back. + const next = this.interruptSession(sessionId, err); + return { session: next }; + } + + const next = this.store.update(sessionId, { status: "awaiting_input", error: null }) ?? session; + return { session: next }; + } + + /** + * Re-create a live interactive session and REPLAY the persisted conversation so + * the fresh agent is primed back to the current awaiting question. Side effects + * are suppressed: we drain the seam's events to advance the agent's context but + * do NOT persist/emit/write — the DB already holds the authoritative final + * state. Populates `this.live[session.id]` on success. + */ + private async rehydrate(session: CeSession): Promise { + const stage = getStage(session.stage); + if (!stage) throw new Error(`Unknown CE stage: ${session.stage}`); + + const cwd = resolveStageSkillCwd(); + const systemPrompt = buildStageSystemPrompt(stage); + const defaultProvider = getDefaultProvider(this.ctx.settings); + const defaultModelId = getDefaultModelId(this.ctx.settings); + + const interactive = await this.factory!({ + cwd, + systemPrompt, + tools: "coding", + ...(defaultProvider ? { defaultProvider } : {}), + ...(defaultModelId ? { defaultModelId } : {}), + }); + const live = interactive.session; + + // Walk the recorded user turns in order. The FIRST user turn is the opening + // message (raw text); each subsequent user turn is a serialized + // {answer, questionId} produced by answer(). Drive the seam with each, and + // drain exactly one event per drive to advance the agent's context — but + // suppress all side effects (no persist/emit/artifact-write). + const userTurns = session.conversationHistory.filter((t) => t.role === "user"); + try { + for (let i = 0; i < userTurns.length; i++) { + const turn = userTurns[i]; + if (i === 0) { + await live.prompt(turn.text); + } else { + const parsed = this.parseAnswerTurn(turn.text); + if (!parsed) continue; // tolerate non-answer user turns. + await live.answer(parsed.questionId, parsed.answer); + } + // Drain the agent's response for this turn to keep the seam in lockstep, + // but DISCARD it — replay reconstructs context, it does not re-run the + // turn loop's side effects. + await live.nextEvent(); + } + } catch (err) { + // Replay failed mid-way — dispose the half-primed handle so we don't leave + // a broken live session behind, then propagate. + try { + live.dispose(); + } catch { + // best-effort + } + throw err; + } + + this.live.set(session.id, live); + } + + /** Parse a serialized `{ answer, questionId }` user turn produced by answer(). */ + private parseAnswerTurn(text: string): { questionId: string; answer: unknown } | undefined { + try { + const obj = JSON.parse(text) as { answer?: unknown; questionId?: unknown }; + if (typeof obj?.questionId === "string") { + return { questionId: obj.questionId, answer: obj.answer }; + } + } catch { + // not a JSON answer turn + } + return undefined; } /** Read-through accessor for routes. */ diff --git a/plugins/fusion-plugin-compound-engineering/src/session/session-store.ts b/plugins/fusion-plugin-compound-engineering/src/session/session-store.ts index 261b2afa9c..547f6065e8 100644 --- a/plugins/fusion-plugin-compound-engineering/src/session/session-store.ts +++ b/plugins/fusion-plugin-compound-engineering/src/session/session-store.ts @@ -234,13 +234,22 @@ export class CeSessionStore { * silently dropped. Returns the ids transitioned. */ recoverStaleSessions(now = Date.now(), multiple = STALE_INTERVAL_MULTIPLE): string[] { + // Only IN-FLIGHT agent turns are subject to the interval-staleness rubric: + // `active`/`launching` mean an agent turn should be progressing, so exceeding + // the interval band signals a crashed/abandoned turn worth recovering. + // + // `awaiting_input` is DELIBERATELY excluded: a session waiting on human input + // is not a crashed turn — human response time is unbounded, and the interval + // rubric measures agent turns, not human waits. Flagging it stale would + // misclassify a legitimately-paused session. It is already in its resumable + // state, so no recovery action is needed. const candidates = this.list().filter( - (s) => (s.status === "active" || s.status === "launching" || s.status === "awaiting_input") && this.isStale(s, now, multiple), + (s) => (s.status === "active" || s.status === "launching") && 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" }); + this.update(s.id, { status: "awaiting_input" }); } else { this.update(s.id, { status: "interrupted", error: s.error ?? "Session interrupted — progress preserved, resume to continue" }); } 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 0150541fa7..b47ae94b7a 100644 --- a/plugins/fusion-plugin-compound-engineering/src/session/stage-registry.ts +++ b/plugins/fusion-plugin-compound-engineering/src/session/stage-registry.ts @@ -13,6 +13,13 @@ export interface CeStageDefinition { /** Stage id (stable, kebab-case). */ stageId: string; + /** + * Explicit pipeline ordinal. Pipeline progression (nextStageAfter) sorts by + * THIS value, NOT by registry/Map insertion order — so a stage registered out + * of order, or inserted mid-pipeline later, advances correctly. Lower runs + * earlier; values need not be contiguous (gaps leave room to insert between). + */ + order: number; /** Bundled skill the orchestrator loads for this stage. */ skillId: string; /** @@ -45,6 +52,7 @@ export interface CeStageDefinition { const STAGE_DEFINITIONS: CeStageDefinition[] = [ { stageId: "strategy", + order: 100, skillId: "ce-strategy", artifactLocation: "STRATEGY.md", icon: "Compass", @@ -53,6 +61,7 @@ const STAGE_DEFINITIONS: CeStageDefinition[] = [ }, { stageId: "ideate", + order: 200, skillId: "ce-ideate", artifactLocation: "docs/ideation/", icon: "Lightbulb", @@ -61,6 +70,7 @@ const STAGE_DEFINITIONS: CeStageDefinition[] = [ }, { stageId: "brainstorm", + order: 300, skillId: "ce-brainstorm", artifactLocation: "docs/brainstorms/", icon: "Sparkles", @@ -69,6 +79,7 @@ const STAGE_DEFINITIONS: CeStageDefinition[] = [ }, { stageId: "plan", + order: 400, skillId: "ce-plan", artifactLocation: "docs/plans/", icon: "ListChecks", @@ -81,6 +92,7 @@ const STAGE_DEFINITIONS: CeStageDefinition[] = [ // board (tagged CE-originated + recorded as pipeline links). The artifact is // the work log / summary for this stage. stageId: "work", + order: 500, skillId: "ce-work", artifactLocation: "docs/work/", icon: "Hammer", @@ -95,8 +107,12 @@ export function getStage(stageId: string): CeStageDefinition | undefined { return REGISTRY.get(stageId); } +/** + * All registered stages sorted by their explicit `order` ordinal (NOT Map + * insertion order). Ties break by `stageId` for a stable, deterministic order. + */ export function listStages(): CeStageDefinition[] { - return [...REGISTRY.values()]; + return [...REGISTRY.values()].sort((a, b) => a.order - b.order || a.stageId.localeCompare(b.stageId)); } /** @@ -106,3 +122,13 @@ export function listStages(): CeStageDefinition[] { export function registerStage(def: CeStageDefinition): void { REGISTRY.set(def.stageId, def); } + +/** + * Remove a runtime-registered stage. Production stages from STAGE_DEFINITIONS are + * protected (no-op) so tests can't accidentally drop a built-in stage. Used by + * tests to keep the shared global registry clean across cases. + */ +export function unregisterStage(stageId: string): void { + if (STAGE_DEFINITIONS.some((s) => s.stageId === stageId)) return; + REGISTRY.delete(stageId); +} diff --git a/plugins/fusion-plugin-compound-engineering/src/sync/reconciler.ts b/plugins/fusion-plugin-compound-engineering/src/sync/reconciler.ts index f6cdf4cef1..7668ef3712 100644 --- a/plugins/fusion-plugin-compound-engineering/src/sync/reconciler.ts +++ b/plugins/fusion-plugin-compound-engineering/src/sync/reconciler.ts @@ -55,7 +55,12 @@ export interface ReconcileResult { inspected: number; } -/** The linear CE stage order. The pipeline advances along this sequence. */ +/** + * The linear CE stage order. The pipeline advances along this sequence. + * `listStages()` is sorted by each stage's explicit `order` ordinal (NOT Map + * insertion order), so a stage registered out of order — or inserted mid- + * pipeline later — slots into the correct position here. + */ function stageOrder(): string[] { return listStages().map((s) => s.stageId); } @@ -130,9 +135,23 @@ export class CeReconciler { const tasks = await this.loadTasks(currentStageLinks); 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)); + // A deleted/missing task yields `undefined` (treated as ABSENT, not + // terminal AND not blocking). Compute terminality over the tasks that still + // EXIST so one deleted current-stage task cannot wedge the pipeline forever. + const existing = tasks.filter((t): t is Task => t != null); + if (existing.length === 0) { + // Every current-stage task was deleted — there is nothing left on the + // board to gate advancement, but also no completion signal to act on. + // Safest non-wedging behavior: leave the pipeline state unchanged (do not + // advance off a vanished stage, do not crash). A later sweep with a real + // board task re-derives the transition. + return undefined; + } + + // Advancement rule: every EXISTING current-stage board task has reached a + // terminal column (board-authoritative read). Partial completion keeps it + // running; deleted tasks are excluded above rather than counted as blocking. + const allTerminal = existing.every((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") {