diff --git a/.changeset/ce-question-id-drift.md b/.changeset/ce-question-id-drift.md new file mode 100644 index 0000000000..ac2b2e26fa --- /dev/null +++ b/.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. diff --git a/packages/core/src/plugin-types.ts b/packages/core/src/plugin-types.ts index 6d438ad385..671dd7c8d7 100644 --- a/packages/core/src/plugin-types.ts +++ b/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; } /** diff --git a/packages/engine/src/__tests__/interactive-ai-session.test.ts b/packages/engine/src/__tests__/interactive-ai-session.test.ts index bc94a206bb..87ebd6930c 100644 --- a/packages/engine/src/__tests__/interactive-ai-session.test.ts +++ b/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), { diff --git a/packages/engine/src/interactive-ai-session.ts b/packages/engine/src/interactive-ai-session.ts index cc34dcf808..dda12de0f4 100644 --- a/packages/engine/src/interactive-ai-session.ts +++ b/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({ type: "error", data: { message: `answer() questionId "${questionId}" does not match current question "${currentQuestion.id}".` }, 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 55f002ebfb..d2e46d6917 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 @@ -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" }), ); diff --git a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts index 40b6b7527a..0e8b1f46e2 100644 --- a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts +++ b/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 = {}, ): Parameters[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 { - 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