fix(ce): recover stale active sessions

Recover persisted Compound Engineering active/launching sessions that outlived their live agent handles on plugin load and session reads.

.changeset/ce-recover-stale-sessions.md

@@ -0,0 +1,5 @@
+---
+"@runfusion/fusion": patch
+---
+
+Recover stale Compound Engineering sessions on plugin load and session reads so persisted active rows without live agent handles no longer leave the dashboard stuck waiting for work that is not running.

plugins/fusion-plugin-compound-engineering/src/__tests__/session-routes.test.ts

@@ -117,6 +117,53 @@ describe("session routes (polling transport)", () => {
     expect(sessions.map((s) => s.stage).sort()).toEqual(["brainstorm", "plan"]);
   });
 
+  it("GET /sessions recovers stale active rows that have no live route handle", async () => {
+    const { getCeSessionStore } = await import("../session/session-store.js");
+    const store = getCeSessionStore(h.ctx);
+    const zombie = store.create({ stage: "strategy", turnIntervalMs: 1 });
+    store.update(zombie.id, {
+      status: "active",
+      currentQuestion: null,
+      lastActivityAt: Date.now() - 10_000,
+    });
+
+    const res = await call("GET", "/sessions", { params: {}, query: {} }, h.ctx);
+
+    expect(res.status).toBe(200);
+    const sessions = (res.body as { sessions: Array<{ id: string; status: string; error: string | null }> }).sessions;
+    expect(sessions.find((s) => s.id === zombie.id)).toMatchObject({
+      status: "interrupted",
+      error: "Session interrupted — progress preserved, resume to continue",
+    });
+    expect(store.get(zombie.id)).toMatchObject({
+      status: "interrupted",
+      error: "Session interrupted — progress preserved, resume to continue",
+    });
+  });
+
+  it("GET /sessions/:id recovers a stale active row before returning it", async () => {
+    const { getCeSessionStore } = await import("../session/session-store.js");
+    const store = getCeSessionStore(h.ctx);
+    const zombie = store.create({ stage: "strategy", turnIntervalMs: 1 });
+    store.update(zombie.id, {
+      status: "active",
+      currentQuestion: null,
+      lastActivityAt: Date.now() - 10_000,
+    });
+
+    const res = await call("GET", "/sessions/:id", { params: { id: zombie.id } }, h.ctx);
+
+    expect(res.status).toBe(200);
+    expect((res.body as { session: { status: string; error: string | null } }).session).toMatchObject({
+      status: "interrupted",
+      error: "Session interrupted — progress preserved, resume to continue",
+    });
+    expect(store.get(zombie.id)).toMatchObject({
+      status: "interrupted",
+      error: "Session interrupted — progress preserved, resume to continue",
+    });
+  });
+
   it("POST /sessions requires a stage", async () => {
     const res = await call("POST", "/sessions", { body: {} }, h.ctx);
     expect(res.status).toBe(400);

plugins/fusion-plugin-compound-engineering/src/index.ts

@@ -4,6 +4,7 @@ import { installBundledCeSkills } from "./skill-installation.js";
 import { ensureCeSchema } from "./schema.js";
 import { createSessionRoutes } from "./routes/session-routes.js";
 import { createArtifactRoutes } from "./routes/artifact-routes.js";
+import { recoverStaleSessionsForContext } from "./session/session-recovery.js";
 import { getCePipelineStore } from "./sync/pipeline-store.js";
 import { reconcileCePipelines } from "./sync/reconciler.js";
 import { settingsSchema } from "./settings.js";
@@ -128,6 +129,8 @@ const plugin = definePlugin({
         const message = error instanceof Error ? error.message : String(error);
         ctx.logger.error(`Compound Engineering skill install failed: ${message}`);
       }
+
+      recoverStaleSessionsForContext(ctx, { reason: "load", force: true, emitEvent: true });
     },
   },
   routes: [...createSessionRoutes(), ...createArtifactRoutes()],

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

@@ -1,5 +1,6 @@
 import type { PluginContext, PluginRouteDefinition, PluginRouteResponse } from "@fusion/core";
 import { CeOrchestrator } from "../session/orchestrator.js";
+import { recoverStaleSessionsForContext } from "../session/session-recovery.js";
 import { asCeSessionStatus, getCeSessionStore } from "../session/session-store.js";
 import { getCePipelineStore } from "../sync/pipeline-store.js";
 import { asString } from "./route-helpers.js";
@@ -121,6 +122,7 @@ export function createSessionRoutes(): PluginRouteDefinition[] {
       description: "Get current session state, including in-flight working output (liveActivity).",
       handler: async (req: unknown, ctx: PluginContext): Promise<PluginRouteResponse> => {
         const id = (req as RouteRequest).params.id;
+        recoverStaleSessionsForContext(ctx, { reason: "route" });
         const session = getCeSessionStore(ctx).get(id);
         if (!session) return { status: 404, body: { error: `Session ${id} not found` } };
         // Attach the orchestrator's transient mid-turn buffer so a polling
@@ -137,6 +139,7 @@ export function createSessionRoutes(): PluginRouteDefinition[] {
       path: "/sessions",
       description: "List CE sessions (optionally filtered by status/stage).",
       handler: async (req: unknown, ctx: PluginContext): Promise<PluginRouteResponse> => {
+        recoverStaleSessionsForContext(ctx, { reason: "route" });
         const query = (req as RouteRequest).query ?? {};
         const status = asCeSessionStatus(typeof query.status === "string" ? query.status : undefined);
         const stage = typeof query.stage === "string" ? query.stage : undefined;

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

@@ -0,0 +1,49 @@
+import type { PluginContext } from "@fusion/core";
+import { getCeSessionStore } from "./session-store.js";
+
+const DEFAULT_RECOVERY_SCAN_TTL_MS = 120_000;
+
+const lastRecoveryScanAt = new WeakMap<object, number>();
+
+interface RecoverStaleSessionsOptions {
+  reason: "load" | "route";
+  force?: boolean;
+  emitEvent?: boolean;
+  now?: number;
+  ttlMs?: number;
+}
+
+/**
+ * Best-effort stale-session recovery for persisted CE sessions that outlived
+ * their in-memory agent handle. Route callers use a TTL because the individual
+ * session endpoint is also the dashboard polling fallback.
+ */
+export function recoverStaleSessionsForContext(
+  ctx: PluginContext,
+  options: RecoverStaleSessionsOptions,
+): string[] {
+  const key = ctx.taskStore as object;
+  const now = options.now ?? Date.now();
+  const ttlMs = options.ttlMs ?? DEFAULT_RECOVERY_SCAN_TTL_MS;
+  if (!options.force) {
+    const last = lastRecoveryScanAt.get(key) ?? 0;
+    if (now - last < ttlMs) return [];
+  }
+  lastRecoveryScanAt.set(key, now);
+
+  try {
+    const recovered = getCeSessionStore(ctx).recoverStaleSessions(now);
+    if (recovered.length > 0) {
+      ctx.logger.info(`Compound Engineering recovered stale session(s) during ${options.reason}: ${recovered.join(", ")}`);
+      if (options.emitEvent) {
+        ctx.emitEvent("compound-engineering:sessions-recovered", { sessionIds: recovered, reason: options.reason });
+      }
+    }
+    return recovered;
+  } catch (err) {
+    ctx.logger.warn(
+      `Compound Engineering stale-session recovery skipped during ${options.reason}: ${err instanceof Error ? err.message : String(err)}`,
+    );
+    return [];
+  }
+}