FN-6251: rehydrate compound answers before resuming

Allow interrupted compound engineering sessions to answer pending questions after the live handle is lost.

- Rehydrate missing live session handles before sending answer turns.
- Keep detached answer resumes active while background rehydration completes.
- Cover route and orchestrator resume paths for old interrupted sessions.
- Document the resume behavior and add a patch changeset.

Files changed:
 .changeset/fn-6251-ce-answer-rehydrate.md          |   5 +
 .../fusion-plugin-compound-engineering/README.md   |   5 +-
 .../orchestrator-interrupt-resume.test.ts          | 156 ++++++++++++++++++++-
 .../src/__tests__/session-routes.test.ts           |  72 +++++++++-
 .../src/session/orchestrator.ts                    |  52 +++++--
 5 files changed, 276 insertions(+), 14 deletions(-)

Fusion-Task-Id: FN-6251
Fusion-Task-Lineage: b6930c72-6106-4da9-a0c3-4bcf882f20da

.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.

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
 

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

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

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<CeStepResult> {
+    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 };
     }