diff --git a/.changeset/ce-recover-stale-sessions.md b/.changeset/ce-recover-stale-sessions.md new file mode 100644 index 0000000000..a76727d30e --- /dev/null +++ b/.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. 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 f7906b1ea6..f76ea7b8d8 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 @@ -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); diff --git a/plugins/fusion-plugin-compound-engineering/src/index.ts b/plugins/fusion-plugin-compound-engineering/src/index.ts index 7aaf230632..e3cc1f2e13 100644 --- a/plugins/fusion-plugin-compound-engineering/src/index.ts +++ b/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()], diff --git a/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts b/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts index 4f597f1f63..072403ee8a 100644 --- a/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts +++ b/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 => { 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 => { + 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; diff --git a/plugins/fusion-plugin-compound-engineering/src/session/session-recovery.ts b/plugins/fusion-plugin-compound-engineering/src/session/session-recovery.ts new file mode 100644 index 0000000000..63386bf1aa --- /dev/null +++ b/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(); + +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 []; + } +}