fix: tolerate recovered CE question ids (#1855)

## Summary
- add an opt-in `allowAnswerQuestionIdDrift` flag for interactive AI
sessions
- keep strict question-id validation by default
- enable the tolerance only for Compound Engineering recovered sessions
so persisted session rows can answer after dashboard
restarts/non-deterministic rehydration

## Test Plan
- `corepack pnpm --filter @fusion/engine exec vitest run
src/__tests__/interactive-ai-session.test.ts --silent=passed-only
--reporter=dot`
- `corepack pnpm --filter @fusion/engine typecheck`
- `corepack pnpm --filter @fusion-plugin-examples/compound-engineering
build`


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Bug Fixes**
* Improved interactive session recovery so persisted answers can
continue after dashboard restarts, even if the question ID changes
during rehydration.
* Keeps strict question-ID validation by default; mismatches still fail
unless drift is explicitly allowed.
* **New Features**
* Added `allowAnswerQuestionIdDrift` option to permit accepting the
persisted question ID during recovered session answering.
* **Tests**
* Added/updated coverage for strict mismatch error behavior and the
successful completion path when drift is enabled.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

.changeset/ce-question-id-drift.md

@@ -0,0 +1,7 @@
+---
+"@runfusion/fusion": patch
+---
+
+summary: Allow Compound Engineering recovered sessions to answer persisted questions after dashboard restarts.
+category: fix
+dev: Keeps strict question-id validation by default while letting CE trust its persisted session row as the recovery anchor.

packages/core/src/plugin-types.ts

@@ -185,6 +185,13 @@ export interface CreateInteractiveAiSessionOptions {
    * stream. Must not throw — implementations should swallow callback errors.
    */
   onProgress?: (event: InteractiveAiSessionProgressEvent) => void;
+  /**
+   * Trust the caller's persisted/current question id when answering, even if a
+   * rehydrated live handle generated a different question id while replaying.
+   * Default remains strict for fresh planning/CE sessions; recovery paths may
+   * enable this when the persisted session row is the authoritative anchor.
+   */
+  allowAnswerQuestionIdDrift?: boolean;
 }
 
 /**

packages/engine/src/__tests__/interactive-ai-session.test.ts

@@ -330,6 +330,47 @@ describe("interactive-ai-session seam", () => {
     expect(ev.type === "error" && ev.data.message).toMatch(/transport exploded/);
   });
 
+  it("rejects mismatched question ids by default", async () => {
+    const question: PlanningQuestion = { id: "current", type: "text", question: "Current?" };
+    const scripted = makeScriptedAgent([q(question), complete({ ok: true })]);
+    const { session } = await createInteractiveAiSessionWith(factoryFor(scripted.session), {
+      cwd: "/tmp",
+      systemPrompt: "protocol",
+    });
+
+    await session.prompt("start");
+    expect((await session.nextEvent()).type).toBe("question");
+
+    await session.answer("persisted", "answer");
+    const ev = await session.nextEvent();
+    expect(ev.type).toBe("error");
+    expect(ev.type === "error" && ev.data.message).toContain('questionId "persisted" does not match current question "current"');
+    expect(scripted.promptCalls()).toHaveLength(1);
+  });
+
+  it("can trust the caller's persisted question id after non-deterministic rehydration", async () => {
+    const question: PlanningQuestion = { id: "rehydrated-different", type: "text", question: "Rehydrated?" };
+    const scripted = makeScriptedAgent([q(question), complete({ ok: true })]);
+    const { session } = await createInteractiveAiSessionWith(factoryFor(scripted.session), {
+      cwd: "/tmp",
+      systemPrompt: "protocol",
+      allowAnswerQuestionIdDrift: true,
+    });
+
+    await session.prompt("start");
+    expect((await session.nextEvent()).type).toBe("question");
+
+    await session.answer("persisted-original", "answer");
+    const done = await session.nextEvent();
+    expect(done.type).toBe("complete");
+    const lastPrompt = scripted.promptCalls().at(-1)!;
+    expect(JSON.parse(lastPrompt)).toMatchObject({
+      type: "answer",
+      questionId: "persisted-original",
+      response: "answer",
+    });
+  });
+
   it("ignores answer() when not awaiting input", async () => {
     const scripted = makeScriptedAgent([complete({ ok: true })]);
     const { session } = await createInteractiveAiSessionWith(factoryFor(scripted.session), {

packages/engine/src/interactive-ai-session.ts

@@ -444,7 +444,7 @@ export async function createInteractiveAiSessionWith(
         });
         return;
       }
-      if (currentQuestion && questionId !== currentQuestion.id) {
+      if (currentQuestion && questionId !== currentQuestion.id && !options.allowAnswerQuestionIdDrift) {
         pendingEvent = Promise.resolve<InteractiveAiSessionEvent>({
           type: "error",
           data: { message: `answer() questionId "${questionId}" does not match current question "${currentQuestion.id}".` },

plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-interrupt-resume.test.ts

@@ -155,6 +155,7 @@ describe("interrupt + resume (no silent loss)", () => {
     expect(done.event?.type).toBe("complete");
     expect(done.session.status).toBe("completed");
     expect(factory).toHaveBeenCalledTimes(1);
+    expect(factory).toHaveBeenCalledWith(expect.objectContaining({ allowAnswerQuestionIdDrift: true }));
     expect(rehydrated.prompt).toHaveBeenCalledTimes(1);
     expect(rehydrated.answer).toHaveBeenCalledTimes(1);
     const hasAnswerTurn = done.session.conversationHistory.some(
@@ -183,6 +184,7 @@ describe("interrupt + resume (no silent loss)", () => {
     const done = await orch.answer(started.session.id, "q1", "a");
     expect(done.session.status).toBe("completed");
     expect(factory).toHaveBeenCalledTimes(1);
+    expect(factory).toHaveBeenCalledWith(expect.not.objectContaining({ allowAnswerQuestionIdDrift: true }));
     expect(live.prompt).toHaveBeenCalledTimes(1);
     expect(live.answer).toHaveBeenCalledTimes(1);
   });
@@ -276,6 +278,7 @@ describe("interrupt + resume (no silent loss)", () => {
     await new Promise((resolve) => setImmediate(resolve));
     const after = store.get(created.id)!;
     expect(after.status).toBe("completed");
+    expect(rehydrated.answer).toHaveBeenCalledTimes(1);
     const hasAnswerTurn = after.conversationHistory.some(
       (t) => t.text === JSON.stringify({ answer: "a", questionId: "q1" }),
     );

plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts

@@ -2,6 +2,7 @@ import { existsSync, mkdirSync, writeFileSync } from "node:fs";
 import { dirname, isAbsolute, join } from "node:path";
 import type {
   CreateInteractiveAiSessionFactory,
+  CreateInteractiveAiSessionOptions,
   InteractiveAiSession,
   InteractiveAiSessionEvent,
   InteractiveAiSessionProgressEvent,
@@ -291,6 +292,7 @@ export class CeOrchestrator {
   private buildSessionOptions(
     stage: CeStageDefinition,
     sessionId: string,
+    opts: Pick<CreateInteractiveAiSessionOptions, "allowAnswerQuestionIdDrift"> = {},
   ): Parameters<CreateInteractiveAiSessionFactory>[0] {
     const defaultProvider = getDefaultProvider(this.ctx.settings);
     const defaultModelId = getDefaultModelId(this.ctx.settings);
@@ -302,6 +304,11 @@ export class CeOrchestrator {
       tools: "coding",
       requestedSkillNames: [stage.skillId],
       additionalSkillPaths,
+      /*
+       * FNXC:CompoundEngineering 2026-07-01-17:31:
+       * Question-id drift tolerance is recovery-only. Fresh CE sessions keep the strict interactive seam guard so live/DB question divergence is surfaced immediately; rehydration enables the tolerance because the persisted session row is the recovery anchor after a restart.
+       */
+      ...(opts.allowAnswerQuestionIdDrift ? { allowAnswerQuestionIdDrift: true } : {}),
       onProgress: (event) => this.handleProgress(sessionId, event),
       ...(defaultProvider ? { defaultProvider } : {}),
       ...(defaultModelId ? { defaultModelId } : {}),
@@ -642,7 +649,9 @@ export class CeOrchestrator {
   }
 
   private async rehydrateReplay(session: CeSession, stage: CeStageDefinition): Promise<void> {
-    const interactive = await this.factory!(this.buildSessionOptions(stage, session.id));
+    const interactive = await this.factory!(
+      this.buildSessionOptions(stage, session.id, { allowAnswerQuestionIdDrift: true }),
+    );
     const live = interactive.session;
 
     // Walk the recorded user turns in order. The FIRST user turn is the opening