fix(compound-engineering): correctness fixes from code review

- reconciler: a deleted current-stage board task no longer wedges the pipeline
  in 'running' forever (terminality computed over existing tasks only; all-deleted
  is a no-op, not a wedge)
- session-store: a human-slow awaiting_input session is no longer misclassified
  stale (interval rubric applies only to in-flight active/launching turns)
- stage-registry: pipeline progression uses an explicit order ordinal instead of
  registry insertion order, so out-of-order registration can't corrupt advancement
- orchestrator.answer(): validate questionId before mutating state, so a stale id
  can't destroy the persisted currentQuestion recovery anchor
- orchestrator.resume(): rehydrate a live interactive session by replaying
  persisted history (side effects suppressed) so a resumed session is actually
  answerable instead of dead-ending; honest interrupted+error fallback when no
  factory is available

95 tests (6 new regression tests, each confirmed failing pre-fix).

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",

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 });

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");
+  });
 });

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)", () => {

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),

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) } };

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<CeStepResult> {
     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<CeStepResult> {
     const session = this.requireSession(sessionId);
-    if (session.status === "interrupted" || session.status === "error") {
-      // Resumable transition (retry for `error`, resume for `interrupted`): if a
-      // question is still pending, return to awaiting_input; otherwise mark
-      // active so the caller can re-run the turn with fresh input. The persisted
-      // question/history is the no-loss recovery anchor either way.
-      const next = session.currentQuestion
-        ? this.store.update(sessionId, { status: "awaiting_input", error: null }) ?? session
-        : this.store.update(sessionId, { status: "active", error: null }) ?? session;
+
+    // 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<void> {
+    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. */

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" });
       }

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);
+}

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") {