diff --git a/.changeset/fn-6251-ce-answer-rehydrate.md b/.changeset/fn-6251-ce-answer-rehydrate.md new file mode 100644 index 0000000000..18182e8ef5 --- /dev/null +++ b/.changeset/fn-6251-ce-answer-rehydrate.md @@ -0,0 +1,5 @@ +--- +"@runfusion/fusion": patch +--- + +Self-heal compound-engineering answer submission for restarted awaiting-input sessions by rehydrating the interactive session before sending the answer. diff --git a/plugins/fusion-plugin-compound-engineering/README.md b/plugins/fusion-plugin-compound-engineering/README.md index 9881c2fa66..b354ecc037 100644 --- a/plugins/fusion-plugin-compound-engineering/README.md +++ b/plugins/fusion-plugin-compound-engineering/README.md @@ -57,7 +57,10 @@ Lifecycle states are `launching → active → awaiting_input → completed`, pl `error` and `interrupted`. On interrupt or error the orchestrator **auto-saves progress and emits an observable event — never silent loss** — and an `interrupted`/`error` session can be resumed/retried back to its current -question. +question. If the server restarts while a session is already `awaiting_input`, +submitting the pending answer rehydrates the live interactive handle from the +persisted conversation history before continuing, so old answerable sessions do +not require a separate resume action. ### Multiple sessions 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 c1d9edafa4..3ec210ad9d 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, makeScriptedSession, type TestHarness } from "./_harness.js"; +import { makeHarness, makeScriptedSession, scriptedFactory, type TestHarness } from "./_harness.js"; /** * CHARACTERIZATION TEST — written first (U5 execution note: cover the @@ -122,6 +122,160 @@ describe("interrupt + resume (no silent loss)", () => { expect(resumed.session.conversationHistory).toHaveLength(2); }); + it("answer() rehydrates an old awaiting_input session with no live handle and drives the answer to completion", async () => { + 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 rehydrated = makeScriptedSession([ + { type: "question", data: QUESTION }, + { type: "complete", data: { artifact: "# Done\n" } }, + ]); + const factory = scriptedFactory(rehydrated); + const orch = new CeOrchestrator({ + ctx: h.ctx, + createInteractiveAiSession: factory, + projectRoot: h.projectRoot, + turnTimeoutMs: 5000, + }); + + const done = await orch.answer(created.id, "q1", "a"); + expect(done.event?.type).toBe("complete"); + expect(done.session.status).toBe("completed"); + expect(factory).toHaveBeenCalledTimes(1); + expect(rehydrated.prompt).toHaveBeenCalledTimes(1); + expect(rehydrated.answer).toHaveBeenCalledTimes(1); + const hasAnswerTurn = done.session.conversationHistory.some( + (t) => t.text === JSON.stringify({ answer: "a", questionId: "q1" }), + ); + expect(hasAnswerTurn).toBe(true); + }); + + it("answer() uses an existing live handle directly without rehydrating", async () => { + const live = makeScriptedSession([ + { type: "question", data: QUESTION }, + { type: "complete", data: { artifact: "# Done\n" } }, + ]); + const factory = scriptedFactory(live); + const orch = new CeOrchestrator({ + ctx: h.ctx, + createInteractiveAiSession: factory, + projectRoot: h.projectRoot, + turnTimeoutMs: 5000, + }); + + const started = await orch.start("brainstorm", { openingMessage: "kick off" }); + expect(started.session.status).toBe("awaiting_input"); + expect(factory).toHaveBeenCalledTimes(1); + + const done = await orch.answer(started.session.id, "q1", "a"); + expect(done.session.status).toBe("completed"); + expect(factory).toHaveBeenCalledTimes(1); + expect(live.prompt).toHaveBeenCalledTimes(1); + expect(live.answer).toHaveBeenCalledTimes(1); + }); + + it("answer() without a live handle and without a factory reports an honest error without corrupting the question", async () => { + 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 orch = new CeOrchestrator({ ctx: h.ctx, projectRoot: h.projectRoot, turnTimeoutMs: 5000 }); + + await expect(orch.answer(created.id, "q1", "a")).rejects.toThrow(/cannot be continued in this process/i); + const after = store.get(created.id)!; + expect(after.status).toBe("awaiting_input"); + expect(after.currentQuestion?.id).toBe("q1"); + expect(after.conversationHistory.some((t) => t.text.includes('"answer"'))).toBe(false); + }); + + it("answer() rejects a stale questionId before rehydration and leaves state untouched", async () => { + 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 factory = scriptedFactory(makeScriptedSession([{ type: "question", data: QUESTION }])); + const orch = new CeOrchestrator({ + ctx: h.ctx, + createInteractiveAiSession: factory, + projectRoot: h.projectRoot, + turnTimeoutMs: 5000, + }); + + await expect(orch.answer(created.id, "stale-q", "a")).rejects.toThrow(/q1|stale-q/); + expect(factory).not.toHaveBeenCalled(); + const after = store.get(created.id)!; + expect(after.status).toBe("awaiting_input"); + expect(after.currentQuestion?.id).toBe("q1"); + expect(after.conversationHistory.some((t) => t.text.includes("stale-q"))).toBe(false); + }); + + it("answer() preserves the existing not-awaiting guard before rehydration", async () => { + const store = getCeSessionStore(h.ctx); + const created = store.create({ stage: "brainstorm", turnIntervalMs: 5000 }); + store.update(created.id, { status: "active", currentQuestion: QUESTION }); + const factory = scriptedFactory(makeScriptedSession([{ type: "question", data: QUESTION }])); + const orch = new CeOrchestrator({ + ctx: h.ctx, + createInteractiveAiSession: factory, + projectRoot: h.projectRoot, + turnTimeoutMs: 5000, + }); + + await expect(orch.answer(created.id, "q1", "a")).rejects.toThrow(/not awaiting input/); + expect(factory).not.toHaveBeenCalled(); + expect(store.get(created.id)!.status).toBe("active"); + }); + + it("detached answer() rehydrates an old awaiting_input session in the background", async () => { + 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 rehydrated = makeScriptedSession([ + { type: "question", data: QUESTION }, + { type: "complete", data: { artifact: "# Done\n" } }, + ]); + const orch = new CeOrchestrator({ + ctx: h.ctx, + createInteractiveAiSession: scriptedFactory(rehydrated), + projectRoot: h.projectRoot, + turnTimeoutMs: 5000, + }); + + const returned = await orch.answer(created.id, "q1", "a", { detach: true }); + expect(returned.session.status).toBe("active"); + + await new Promise((resolve) => setImmediate(resolve)); + const after = store.get(created.id)!; + expect(after.status).toBe("completed"); + const hasAnswerTurn = after.conversationHistory.some( + (t) => t.text === JSON.stringify({ answer: "a", questionId: "q1" }), + ); + expect(hasAnswerTurn).toBe(true); + }); + 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 diff --git a/plugins/fusion-plugin-compound-engineering/src/__tests__/session-routes.test.ts b/plugins/fusion-plugin-compound-engineering/src/__tests__/session-routes.test.ts index ca064e135f..2d06285cca 100644 --- a/plugins/fusion-plugin-compound-engineering/src/__tests__/session-routes.test.ts +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/session-routes.test.ts @@ -1,7 +1,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import type { PluginContext, PluginRouteResponse } from "@fusion/core"; +import type { PlanningQuestion, PluginContext, PluginRouteResponse } from "@fusion/core"; import { createSessionRoutes } from "../routes/session-routes.js"; -import { makeHarness, type TestHarness } from "./_harness.js"; +import { makeHarness, makeScriptedSession, scriptedFactory, type TestHarness } from "./_harness.js"; /** * Routes-level smoke test for the POLLING transport. Exercises validation and @@ -11,6 +11,16 @@ import { makeHarness, type TestHarness } from "./_harness.js"; * which is the correct, non-hanging behavior. */ +const QUESTION: PlanningQuestion = { + id: "q1", + type: "single_select", + question: "Which direction?", + options: [ + { id: "a", label: "A" }, + { id: "b", label: "B" }, + ], +}; + let h: TestHarness; beforeEach(() => { h = makeHarness(); @@ -99,4 +109,62 @@ describe("session routes (polling transport)", () => { const res = await call("POST", "/sessions/:id/answer", { params: { id: "x" }, body: {} }, h.ctx); expect(res.status).toBe(400); }); + + it("POST /sessions/:id/answer rehydrates an old awaiting_input session instead of returning call-resume-first", async () => { + const { getCeSessionStore } = await import("../session/session-store.js"); + const store = getCeSessionStore(h.ctx); + const created = store.create({ stage: "brainstorm" }); + 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 }); + + h.ctx.createInteractiveAiSession = scriptedFactory( + makeScriptedSession([ + { type: "question", data: QUESTION }, + { type: "complete", data: { artifact: "# Done\n" } }, + ]), + ); + + const res = await call( + "POST", + "/sessions/:id/answer", + { params: { id: created.id }, body: { questionId: "q1", response: "a" } }, + h.ctx, + ); + expect(res.status).toBe(200); + expect((res.body as { session: { status: string } }).session.status).toBe("active"); + + await new Promise((resolve) => setImmediate(resolve)); + expect(store.get(created.id)!.status).toBe("completed"); + }); + + it("POST /sessions/:id/answer returns an honest no-factory error without corrupting an old awaiting_input session", async () => { + const { getCeSessionStore } = await import("../session/session-store.js"); + const store = getCeSessionStore(h.ctx); + const created = store.create({ stage: "brainstorm" }); + 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 res = await call( + "POST", + "/sessions/:id/answer", + { params: { id: created.id }, body: { questionId: "q1", response: "a" } }, + h.ctx, + ); + expect(res.status).toBe(409); + expect((res.body as { error: string }).error).toMatch(/cannot be continued in this process/i); + expect((res.body as { error: string }).error).not.toMatch(/call resume\(\) first/i); + const after = store.get(created.id)!; + expect(after.status).toBe("awaiting_input"); + expect(after.currentQuestion?.id).toBe("q1"); + }); }); diff --git a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts index 05040c17ee..1f51d00092 100644 --- a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts +++ b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts @@ -65,6 +65,9 @@ const MAX_ACTIVITY_TURN_CHARS = 16000; const MAX_PERSISTED_ACTIVITY_TURNS = 50; const MAX_PERSISTED_ACTIVITY_TURN_CHARS = 4000; +const INTERACTIVE_AI_UNAVAILABLE_MESSAGE = + "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."; + /** * Observable event names emitted via `ctx.emitEvent`. The no-silent-loss * invariant requires that interrupt/error ALWAYS emit one of these AND persist @@ -445,22 +448,52 @@ export class CeOrchestrator { ); } const live = this.live.get(sessionId); - if (!live) { - throw new Error(`Session ${sessionId} has no live handle in this process; call resume() first.`); + if (!live && !this.factory) { + throw new Error(INTERACTIVE_AI_UNAVAILABLE_MESSAGE); } + + const turn = this.runAnswerTurn(session, questionId, response); + if (opts.detach) { + // If the process lost its live handle, rehydration can take time. Mirror + // resume(detach): mark the row active immediately while the background + // turn re-creates the handle and converges through persisted state. + if (!live) { + this.store.update(sessionId, { status: "active", error: null }); + } + // runAnswerTurn never rejects after the preflight guards above (failures + // persist into session state). + void turn; + return { session: this.requireSession(sessionId) }; + } + return turn; + } + + private async runAnswerTurn(session: CeSession, questionId: string, response: unknown): Promise { + const sessionId = session.id; + let live = this.live.get(sessionId); + if (!live) { + try { + await this.rehydrate(session); + live = this.live.get(sessionId); + if (!live) { + throw new Error(`Session ${sessionId} could not be rehydrated with a live handle.`); + } + } catch (err) { + const interrupted = this.interruptSession(sessionId, err); + return { + session: interrupted, + event: { type: "error", data: { message: interrupted.error ?? "interrupted", cause: err } }, + }; + } + } + this.store.appendHistory(sessionId, { role: "user", text: JSON.stringify({ answer: response, questionId }), at: new Date().toISOString(), }); this.store.update(sessionId, { status: "active", currentQuestion: null }); - const turn = this.runTurn(sessionId, () => live.answer(questionId, response), live); - if (opts.detach) { - // runTurn never rejects (all failures persist into session state). - void turn; - return { session: this.requireSession(sessionId) }; - } - return turn; + return this.runTurn(sessionId, () => live.answer(questionId, response), live); } /** @@ -512,8 +545,7 @@ export class CeOrchestrator { 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.", + error: INTERACTIVE_AI_UNAVAILABLE_MESSAGE, }) ?? session; return { session: next }; }