From 7f8db9f892737b63be95a8d0def4f844cf9bbb8b Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 3 Jun 2026 13:47:02 -0700 Subject: [PATCH] feat(compound-engineering): manage and work across multiple CE sessions The store/orchestrator were already multi-session (independent rows + live handles per session); this surfaces it end to end: - Sessions panel in the dashboard view: lists every session with stage, status badge ("needs your input" for awaiting_input), and last activity; stays visible while a flow is open so switching is one click. Closing a flow returns to the overview without stopping the session. - useCeSession.open(): adopt an existing session (pins its projectId for answer/resume/poll); useCeSessions list hook with push-event refresh and poll fallback while any session is mid-turn. - DELETE /sessions/:id + orchestrator.discard(): dispose the live handle before deleting the row (pipeline-link rows kept for task provenance); Discard affordance on settled sessions. - Tests: cross-session independence through one orchestrator, store delete, route list/delete, hook open/list/remove/push/poll, view panel open/switch/discard. 116 tests green; plugin + dashboard tsc clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../compound-engineering-plugin-scaffold.md | 2 +- .../README.md | 19 +++ .../src/__tests__/orchestrator-flow.test.ts | 55 +++++++ .../src/__tests__/session-routes.test.ts | 28 ++++ .../src/__tests__/session-store.test.ts | 17 ++ .../src/dashboard/CompoundEngineeringView.css | 65 ++++++++ .../src/dashboard/CompoundEngineeringView.tsx | 139 +++++++++++++++- .../CompoundEngineeringView.test.tsx | 107 +++++++++++++ .../hooks/__tests__/useCeSession.test.tsx | 23 +++ .../hooks/__tests__/useCeSessions.test.tsx | 149 ++++++++++++++++++ .../src/dashboard/hooks/api.ts | 18 +++ .../src/dashboard/hooks/useCeSession.ts | 16 +- .../src/dashboard/hooks/useCeSessions.ts | 127 +++++++++++++++ .../src/routes/session-routes.ts | 14 ++ .../src/session/orchestrator.ts | 11 ++ .../src/session/session-store.ts | 6 + 16 files changed, 790 insertions(+), 6 deletions(-) create mode 100644 plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/__tests__/useCeSessions.test.tsx create mode 100644 plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/useCeSessions.ts diff --git a/.changeset/compound-engineering-plugin-scaffold.md b/.changeset/compound-engineering-plugin-scaffold.md index 6a08635b4d..d11f71787d 100644 --- a/.changeset/compound-engineering-plugin-scaffold.md +++ b/.changeset/compound-engineering-plugin-scaffold.md @@ -2,7 +2,7 @@ "@runfusion/fusion": minor --- -Add the Compound Engineering bundled plugin: a dedicated dashboard surface for compound-engineering artifacts and interactive `ce-*` sessions, a work→board bridge, and bidirectional board↔pipeline sync. +Add the Compound Engineering bundled plugin: a dedicated dashboard surface for compound-engineering artifacts and interactive `ce-*` sessions, a work→board bridge, and bidirectional board↔pipeline sync. Sessions are fully multi-session: a Sessions panel lists every run with stage/status/last-activity, lets you open and switch between concurrent sessions (each keeps running server-side), resume interrupted ones, and discard settled ones (`DELETE /sessions/:id` disposes the live handle before deleting the row). This also adds two reusable host capabilities that any plugin benefits from: diff --git a/plugins/fusion-plugin-compound-engineering/README.md b/plugins/fusion-plugin-compound-engineering/README.md index 09f730f2ac..5dc7457ccd 100644 --- a/plugins/fusion-plugin-compound-engineering/README.md +++ b/plugins/fusion-plugin-compound-engineering/README.md @@ -59,6 +59,25 @@ progress and emits an observable event — never silent loss** — and an `interrupted`/`error` session can be resumed/retried back to its current question. +### Multiple sessions + +Sessions are independent pipeline runs — the store, routes, and orchestrator +all hold many at once (each with its own live agent handle). The dashboard's +**Sessions panel** lists every session with its stage, status, and last +activity; from there you can: + +- **open** any session and keep working on it (an `awaiting_input` session is + flagged "needs your input"), +- **switch** between sessions — the panel stays visible while a flow is open, + and a session you switch away from keeps running server-side, +- **resume** an `interrupted`/`error` session from where it stopped, +- **discard** a settled (completed/error/interrupted) session via + `DELETE /sessions/:id`, which disposes any live handle before deleting the + row (pipeline-link rows are kept — board-task provenance survives). + +The list refreshes on any CE push event and falls back to polling +`GET /sessions` while any session has a turn in flight. + ### Transport Session updates are **pushed** over the shared `/api/events` SSE stream. The diff --git a/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-flow.test.ts b/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-flow.test.ts index 176c44cc5c..4ca7cdaf7a 100644 --- a/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-flow.test.ts +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-flow.test.ts @@ -76,6 +76,61 @@ describe("orchestrator happy path", () => { }); }); +describe("multiple concurrent sessions", () => { + it("drives two independent sessions through the SAME orchestrator without cross-talk", async () => { + // Two scripted live sessions; the factory hands them out in creation order. + const liveA = makeScriptedSession([ + { type: "question", data: QUESTION }, + { type: "complete", data: { artifact: "# A\n" } }, + ]); + const liveB = makeScriptedSession([ + { type: "question", data: { ...QUESTION, id: "q-b" } }, + { type: "complete", data: { artifact: "# B\n" } }, + ]); + const handles = [liveA, liveB]; + const orch = new CeOrchestrator({ + ctx: h.ctx, + createInteractiveAiSession: vi.fn(async () => ({ session: handles.shift()! })), + projectRoot: h.projectRoot, + turnTimeoutMs: 5000, + }); + + const a = await orch.start("brainstorm", { openingMessage: "topic A" }); + const b = await orch.start("brainstorm", { openingMessage: "topic B" }); + expect(a.session.id).not.toBe(b.session.id); + expect(a.session.status).toBe("awaiting_input"); + expect(b.session.status).toBe("awaiting_input"); + + // Answer B first — A must stay awaiting, untouched. + const doneB = await orch.answer(b.session.id, "q-b", "bee"); + expect(doneB.session.status).toBe("completed"); + expect(orch.getState(a.session.id)?.status).toBe("awaiting_input"); + + // A is still answerable on ITS live handle (not B's). + const doneA = await orch.answer(a.session.id, "q1", "ay"); + expect(doneA.session.status).toBe("completed"); + expect(liveA.answer).toHaveBeenCalledTimes(1); + expect(liveB.answer).toHaveBeenCalledTimes(1); + }); + + it("discard disposes the live handle and deletes only that session", async () => { + const live = makeScriptedSession([{ type: "question", data: QUESTION }]); + const orch = new CeOrchestrator({ + ctx: h.ctx, + createInteractiveAiSession: vi.fn(async () => ({ session: live })), + projectRoot: h.projectRoot, + turnTimeoutMs: 5000, + }); + const started = await orch.start("brainstorm", { openingMessage: "topic" }); + + expect(orch.discard(started.session.id)).toBe(true); + expect(live.dispose).toHaveBeenCalled(); + expect(orch.getState(started.session.id)).toBeUndefined(); + // Idempotent-ish: a second discard reports false, no throw. + expect(orch.discard(started.session.id)).toBe(false); + }); +}); + describe("orchestrator error + retry", () => { it("agent error → status error, progress preserved, observable event; retry resumes to the question", async () => { const orch = makeOrch([ 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 346085acf9..ca064e135f 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 @@ -40,10 +40,38 @@ describe("session routes (polling transport)", () => { "POST /sessions/:id/resume", "GET /sessions/:id", "GET /sessions", + "DELETE /sessions/:id", ]), ); }); + it("DELETE /sessions/:id discards a session (404 for unknown, gone afterwards, others kept)", async () => { + const { getCeSessionStore } = await import("../session/session-store.js"); + const store = getCeSessionStore(h.ctx); + const keep = store.create({ stage: "brainstorm" }); + const drop = store.create({ stage: "plan" }); + + const missing = await call("DELETE", "/sessions/:id", { params: { id: "nope" } }, h.ctx); + expect(missing.status).toBe(404); + + const deleted = await call("DELETE", "/sessions/:id", { params: { id: drop.id } }, h.ctx); + expect(deleted.status).toBe(200); + expect(store.get(drop.id)).toBeUndefined(); + expect(store.get(keep.id)).toBeDefined(); + }); + + it("GET /sessions lists every session so a client can manage multiple concurrently", async () => { + const { getCeSessionStore } = await import("../session/session-store.js"); + const store = getCeSessionStore(h.ctx); + store.create({ stage: "brainstorm" }); + store.create({ stage: "plan" }); + + const res = await call("GET", "/sessions", { params: {}, query: {} }, h.ctx); + expect(res.status).toBe(200); + const sessions = (res.body as { sessions: Array<{ stage: string }> }).sessions; + expect(sessions.map((s) => s.stage).sort()).toEqual(["brainstorm", "plan"]); + }); + 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/__tests__/session-store.test.ts b/plugins/fusion-plugin-compound-engineering/src/__tests__/session-store.test.ts index d0340cab4a..6528a3c94b 100644 --- a/plugins/fusion-plugin-compound-engineering/src/__tests__/session-store.test.ts +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/session-store.test.ts @@ -51,6 +51,23 @@ describe("CeSessionStore CRUD + JSON round-trip", () => { }); }); +describe("multi-session independence + delete", () => { + it("holds many independent sessions; deleting one leaves the others untouched", () => { + const store = new CeSessionStore(h.db); + const a = store.create({ stage: "brainstorm", projectId: "p1" }); + const b = store.create({ stage: "plan", projectId: "p1" }); + const c = store.create({ stage: "work" }); + expect(store.list()).toHaveLength(3); + + expect(store.delete(b.id)).toBe(true); + expect(store.get(b.id)).toBeUndefined(); + expect(store.get(a.id)).toBeDefined(); + expect(store.get(c.id)).toBeDefined(); + // Deleting a missing row reports false, no throw. + expect(store.delete(b.id)).toBe(false); + }); +}); + describe("interval-relative staleness (FN-4172 rubric)", () => { it("does NOT misclassify a healthy-but-slow session as stale", () => { const store = new CeSessionStore(h.db); diff --git a/plugins/fusion-plugin-compound-engineering/src/dashboard/CompoundEngineeringView.css b/plugins/fusion-plugin-compound-engineering/src/dashboard/CompoundEngineeringView.css index 587c1383bf..35c7edda25 100644 --- a/plugins/fusion-plugin-compound-engineering/src/dashboard/CompoundEngineeringView.css +++ b/plugins/fusion-plugin-compound-engineering/src/dashboard/CompoundEngineeringView.css @@ -303,3 +303,68 @@ margin: 0 0 0.4rem; padding-left: 1.1rem; } + +/* Sessions panel — manage/switch across multiple concurrent CE sessions. */ +.ce-sessions { + margin-bottom: 0.8rem; +} +.ce-sessions-list { + list-style: none; + margin: 0; + padding: 0; + display: flex; + flex-direction: column; + gap: 0.3rem; +} +.ce-session-row { + display: flex; + align-items: center; + gap: 0.5rem; +} +.ce-session-row.is-active .ce-session-open { + border-color: var(--color-accent, #36c); + background: color-mix(in srgb, var(--color-accent, #36c) 8%, transparent); +} +.ce-session-open { + flex: 1; + display: flex; + align-items: baseline; + gap: 0.6rem; + text-align: left; + padding: 0.4rem 0.6rem; + border: 1px solid var(--color-border, #ddd); + border-radius: 6px; + background: transparent; + cursor: pointer; +} +.ce-session-open:disabled { + cursor: default; + opacity: 0.6; +} +.ce-session-stage { + font-weight: 600; +} +.ce-session-status { + font-size: 0.74rem; + text-transform: capitalize; + opacity: 0.8; +} +.ce-session-status-awaiting_input { + color: var(--color-warning, #a60); + font-weight: 600; + opacity: 1; +} +.ce-session-status-error, +.ce-session-status-interrupted { + color: var(--color-danger, #d23); + opacity: 1; +} +.ce-session-status-completed { + color: var(--color-success, #2a7); + opacity: 1; +} +.ce-session-updated { + margin-left: auto; + font-size: 0.72rem; + opacity: 0.6; +} diff --git a/plugins/fusion-plugin-compound-engineering/src/dashboard/CompoundEngineeringView.tsx b/plugins/fusion-plugin-compound-engineering/src/dashboard/CompoundEngineeringView.tsx index 8cc2302113..a7d7917e14 100644 --- a/plugins/fusion-plugin-compound-engineering/src/dashboard/CompoundEngineeringView.tsx +++ b/plugins/fusion-plugin-compound-engineering/src/dashboard/CompoundEngineeringView.tsx @@ -6,10 +6,12 @@ import type { PluginDashboardViewContext } from "@fusion/dashboard/app/plugins/t import { useArtifacts } from "./hooks/useArtifacts.js"; import { useViewportMode } from "./hooks/useViewportMode.js"; import { useCeSession, type CeSessionSubscribe } from "./hooks/useCeSession.js"; +import { useCeSessions, type CeSessionsSubscribe } from "./hooks/useCeSessions.js"; import { getArtifactPreviewUrl } from "./hooks/api.js"; import { CeFlow } from "./CeFlow.js"; -import { listStages, type CeStageDefinition } from "../session/stage-registry.js"; +import { getStage, listStages, type CeStageDefinition } from "../session/stage-registry.js"; import type { CeArtifactEntry, CeArtifactGroup } from "../artifacts/discovery.js"; +import type { CeSession, CeSessionStatus } from "../session/session-store.js"; const CE_PLUGIN_ID = "fusion-plugin-compound-engineering"; @@ -56,6 +58,82 @@ function StageLauncher({ ); } +/** Statuses that are settled (no agent turn in flight). */ +const TERMINAL: ReadonlySet = new Set(["completed", "error", "interrupted"]); + +function statusLabel(status: CeSessionStatus): string { + return status.replace("_", " "); +} + +/** + * Sessions panel: every CE session (each an independent pipeline run) with its + * stage, status, and last activity — open any to keep working on it, discard + * settled ones. Sessions keep running server-side while not open here. + */ +function SessionsPanel({ + sessions, + activeSessionId, + disabled, + onOpen, + onDiscard, +}: { + sessions: CeSession[]; + activeSessionId?: string; + disabled: boolean; + onOpen: (session: CeSession) => void; + onDiscard: (session: CeSession) => void; +}) { + if (sessions.length === 0) return null; + return ( +
+
+

Sessions

+ {sessions.length} +
+
    + {sessions.map((s) => { + const stageLabel = getStage(s.stage)?.label ?? s.stage; + const awaiting = s.status === "awaiting_input"; + return ( +
  • + + {TERMINAL.has(s.status) ? ( + + ) : null} +
    • + ); + })} +
+
+ ); +} + interface CompoundEngineeringViewProps { context?: PluginDashboardViewContext; /** Test seam: override the active project id without a host context. */ @@ -193,6 +271,16 @@ export function CompoundEngineeringView(props: CompoundEngineeringViewProps) { }); }, [subscribePluginEvents]); const ceSession = useCeSession(subscribe ? { subscribe } : {}); + // Session list refresh: ANY CE push event means some session changed. + const subscribeList = useMemo(() => { + if (!subscribePluginEvents) return undefined; + return (onAnyEvent) => subscribePluginEvents(CE_PLUGIN_ID, () => onAnyEvent()); + }, [subscribePluginEvents]); + const ceSessions = useCeSessions({ + projectId, + enabled, + ...(subscribeList ? { subscribe: subscribeList } : {}), + }); const [launcherOpen, setLauncherOpen] = useState(false); const totalArtifacts = result?.totalArtifacts ?? 0; @@ -208,20 +296,50 @@ export function CompoundEngineeringView(props: CompoundEngineeringViewProps) { const onLaunch = useCallback( (stage: CeStageDefinition) => { setLauncherOpen(false); - void ceSession.start(stage.stageId, { message: `Start the ${stage.label} stage.`, projectId }); + void ceSession + .start(stage.stageId, { message: `Start the ${stage.label} stage.`, projectId }) + .then(() => ceSessions.refresh()); + }, + [ceSession, ceSessions, projectId], + ); + + const onOpenSession = useCallback( + (s: CeSession) => { + void ceSession.open(s.id, { projectId }); }, [ceSession, projectId], ); - const onCloseFlow = useCallback(() => ceSession.reset(), [ceSession]); + const onDiscardSession = useCallback( + (s: CeSession) => { + void ceSessions.remove(s.id); + }, + [ceSessions], + ); - // Once a session exists, the flow renderer owns the surface until closed. + // Closing the flow returns to the overview WITHOUT stopping the session — + // it keeps running server-side and stays reachable from the sessions panel. + const onCloseFlow = useCallback(() => { + ceSession.reset(); + void ceSessions.refresh(); + }, [ceSession, ceSessions]); + + // Once a session is active here, the flow renderer owns the surface until + // closed — but the sessions panel stays visible so other sessions remain + // one click away (switching does not stop the open one). if (ceSession.session) { return (

Compound Engineering

+ ) : null} + + + {ceSessions.error ? ( +
+ Failed to load sessions: {ceSessions.error} +
+ ) : null} + {ceSession.error && !ceSession.session ? (
Failed to start session: {ceSession.error} diff --git a/plugins/fusion-plugin-compound-engineering/src/dashboard/__tests__/CompoundEngineeringView.test.tsx b/plugins/fusion-plugin-compound-engineering/src/dashboard/__tests__/CompoundEngineeringView.test.tsx index 24bfc2ca6c..6a91782667 100644 --- a/plugins/fusion-plugin-compound-engineering/src/dashboard/__tests__/CompoundEngineeringView.test.tsx +++ b/plugins/fusion-plugin-compound-engineering/src/dashboard/__tests__/CompoundEngineeringView.test.tsx @@ -6,13 +6,43 @@ import type { DiscoveryResult } from "../../artifacts/discovery.js"; const listArtifacts = vi.fn(async (): Promise => { throw new Error("listArtifacts mock not configured"); }); +const listSessions = vi.fn(async (): Promise => []); +const deleteSession = vi.fn(async (_id: string, _projectId?: string): Promise => undefined); +const getSession = vi.fn(async (_id: string, _projectId?: string): Promise => { + throw new Error("getSession mock not configured"); +}); vi.mock("../hooks/api.js", () => ({ listArtifacts: () => listArtifacts(), getArtifactPreviewUrl: (id: string) => `/preview/${id}`, + listSessions: () => listSessions(), + deleteSession: (id: string, projectId?: string) => deleteSession(id, projectId), + getSession: (id: string, projectId?: string) => getSession(id, projectId), + startSession: vi.fn(), + answerSession: vi.fn(), + resumeSession: vi.fn(), })); import { CompoundEngineeringView } from "../CompoundEngineeringView.js"; import { __test_clearArtifactsCache } from "../hooks/useArtifacts.js"; +import type { CeSession } from "../../session/session-store.js"; + +function mkCeSession(over: Partial): CeSession { + return { + id: "sess-1", + stage: "brainstorm", + status: "awaiting_input", + currentQuestion: null, + conversationHistory: [], + projectId: "p1", + artifactPath: null, + error: null, + turnIntervalMs: 1000, + lastActivityAt: Date.now(), + createdAt: "2026-06-03T00:00:00Z", + updatedAt: "2026-06-03T00:00:00Z", + ...over, + }; +} const ALL_STAGES: Array<{ stage: DiscoveryResult["groups"][number]["stage"]; label: string }> = [ { stage: "strategy", label: "Strategy" }, @@ -45,6 +75,11 @@ describe("CompoundEngineeringView", () => { beforeEach(() => { __test_clearArtifactsCache(); listArtifacts.mockReset(); + listSessions.mockReset(); + listSessions.mockResolvedValue([]); + deleteSession.mockReset(); + deleteSession.mockResolvedValue(undefined); + getSession.mockReset(); }); afterEach(() => vi.clearAllMocks()); @@ -101,6 +136,78 @@ describe("CompoundEngineeringView", () => { expect(screen.getByTestId("ce-summary").textContent).toMatch(/unreadable/i); }); + it("lists multiple sessions with status badges; terminal sessions get a discard affordance", async () => { + listArtifacts.mockResolvedValue(makeResult({})); + listSessions.mockResolvedValue([ + mkCeSession({ id: "a", stage: "brainstorm", status: "awaiting_input" }), + mkCeSession({ id: "b", stage: "plan", status: "active" }), + mkCeSession({ id: "c", stage: "work", status: "completed" }), + ]); + render(); + + await screen.findByTestId("ce-sessions"); + const rows = screen.getAllByTestId("ce-session-row"); + expect(rows).toHaveLength(3); + expect(rows.map((r) => r.getAttribute("data-status"))).toEqual([ + "awaiting_input", + "active", + "completed", + ]); + // Awaiting sessions advertise that they need the user. + expect(rows[0].textContent).toMatch(/needs your input/i); + // Only the terminal session can be discarded. + expect(screen.getAllByTestId("ce-session-discard")).toHaveLength(1); + }); + + it("opens an existing session from the list into the flow (and back without losing it)", async () => { + listArtifacts.mockResolvedValue(makeResult({})); + listSessions.mockResolvedValue([ + mkCeSession({ id: "a", stage: "brainstorm", status: "awaiting_input" }), + mkCeSession({ id: "b", stage: "plan", status: "active" }), + ]); + getSession.mockResolvedValue( + mkCeSession({ + id: "a", + status: "awaiting_input", + currentQuestion: { id: "q1", type: "text", question: "Topic?" }, + }), + ); + render(); + + await screen.findByTestId("ce-sessions"); + fireEvent.click(screen.getAllByTestId("ce-session-open")[0]); + + // The flow surface opens on the adopted session… + const flow = await screen.findByTestId("ce-flow"); + expect(flow.getAttribute("data-stage")).toBe("brainstorm"); + expect(getSession).toHaveBeenCalledWith("a", "p1"); + // …while the sessions panel stays visible for switching, with the open + // session marked active. + expect(screen.getByTestId("ce-sessions")).toBeInTheDocument(); + const rows = screen.getAllByTestId("ce-session-row"); + expect(rows[0].className).toMatch(/is-active/); + + // Closing returns to the overview; the session list survives (the session + // itself keeps running server-side — close does not delete anything). + fireEvent.click(screen.getByText("Close")); + await screen.findByTestId("ce-empty-state"); + expect(screen.getByTestId("ce-sessions")).toBeInTheDocument(); + expect(deleteSession).not.toHaveBeenCalled(); + }); + + it("discards a terminal session via the list", async () => { + listArtifacts.mockResolvedValue(makeResult({})); + listSessions.mockResolvedValue([mkCeSession({ id: "done", stage: "plan", status: "completed" })]); + render(); + + await screen.findByTestId("ce-sessions"); + listSessions.mockResolvedValue([]); + fireEvent.click(screen.getByTestId("ce-session-discard")); + + await waitFor(() => expect(deleteSession).toHaveBeenCalledWith("done", "p1")); + await waitFor(() => expect(screen.queryByTestId("ce-sessions")).not.toBeInTheDocument()); + }); + it("does not fetch when the viewport-gated flag is disabled", async () => { listArtifacts.mockResolvedValue(makeResult({})); render(); diff --git a/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/__tests__/useCeSession.test.tsx b/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/__tests__/useCeSession.test.tsx index 32b3bb9cc0..5175d5f535 100644 --- a/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/__tests__/useCeSession.test.tsx +++ b/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/__tests__/useCeSession.test.tsx @@ -32,6 +32,7 @@ function Harness({ transport }: { transport: CeSessionTransport }) { {s.busy ? "busy" : "idle"} {s.error ?? ""} + @@ -120,6 +121,28 @@ describe("useCeSession lifecycle", () => { expect(screen.getByTestId("status")).toHaveTextContent("awaiting_input"); }); + it("open() adopts an existing session and threads ITS projectId to later calls", async () => { + const transport: CeSessionTransport = { + start: vi.fn(), + answer: vi.fn(async () => mkSession({ id: "s2", status: "completed" })), + resume: vi.fn(), + get: vi.fn(async () => mkSession({ id: "s2", status: "awaiting_input", currentQuestion: Q })), + }; + render(); + + await act(async () => { + screen.getByText("open").click(); + }); + expect(transport.get).toHaveBeenCalledWith("s2", "p2"); + expect(screen.getByTestId("status")).toHaveTextContent("awaiting_input"); + + // Subsequent answer goes to the opened session with the opened projectId. + await act(async () => { + screen.getByText("answer").click(); + }); + expect(transport.answer).toHaveBeenCalledWith("s2", "q1", "yes", "p2"); + }); + it("surfaces a start error", async () => { const transport: CeSessionTransport = { start: vi.fn(async () => { diff --git a/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/__tests__/useCeSessions.test.tsx b/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/__tests__/useCeSessions.test.tsx new file mode 100644 index 0000000000..da0c740f85 --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/__tests__/useCeSessions.test.tsx @@ -0,0 +1,149 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { act, render, screen } from "@testing-library/react"; +import { useCeSessions, type CeSessionsTransport, type CeSessionsSubscribe } from "../useCeSessions.js"; +import type { CeSession } from "../../../session/session-store.js"; + +function mkSession(over: Partial): CeSession { + return { + id: "s1", + stage: "brainstorm", + status: "awaiting_input", + currentQuestion: null, + conversationHistory: [], + projectId: null, + artifactPath: null, + error: null, + turnIntervalMs: 1000, + lastActivityAt: Date.now(), + createdAt: "t", + updatedAt: "t", + ...over, + }; +} + +function Harness({ + transport, + subscribe, +}: { + transport: CeSessionsTransport; + subscribe?: CeSessionsSubscribe; +}) { + const s = useCeSessions({ + projectId: "p1", + transport, + pollIntervalMs: 5, + ...(subscribe ? { subscribe } : {}), + }); + return ( +
+ {s.sessions.length} + {s.sessions.map((x) => x.id).join(",")} + {s.error ?? ""} + + +
+ ); +} + +describe("useCeSessions (multi-session list)", () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it("lists all sessions on mount with the projectId", async () => { + const list = vi.fn(async () => [mkSession({ id: "s1" }), mkSession({ id: "s2", stage: "plan" })]); + const transport: CeSessionsTransport = { list, remove: vi.fn() }; + render(); + + await act(async () => {}); + expect(list).toHaveBeenCalledWith("p1"); + expect(screen.getByTestId("count")).toHaveTextContent("2"); + expect(screen.getByTestId("ids")).toHaveTextContent("s1,s2"); + }); + + it("remove() deletes via the transport then refreshes the list", async () => { + let removed = false; + const transport: CeSessionsTransport = { + list: vi.fn(async () => (removed ? [mkSession({ id: "s2" })] : [mkSession({ id: "s1" }), mkSession({ id: "s2" })])), + remove: vi.fn(async () => { + removed = true; + }), + }; + render(); + await act(async () => {}); + expect(screen.getByTestId("count")).toHaveTextContent("2"); + + await act(async () => { + screen.getByText("remove").click(); + }); + expect(transport.remove).toHaveBeenCalledWith("s1", "p1"); + expect(screen.getByTestId("ids")).toHaveTextContent("s2"); + }); + + it("refreshes when a push event fires", async () => { + let fire: (() => void) | undefined; + const subscribe: CeSessionsSubscribe = (onAnyEvent) => { + fire = onAnyEvent; + return () => { + fire = undefined; + }; + }; + let n = 1; + const transport: CeSessionsTransport = { + list: vi.fn(async () => Array.from({ length: n }, (_, i) => mkSession({ id: `s${i + 1}` }))), + remove: vi.fn(), + }; + render(); + await act(async () => {}); + expect(screen.getByTestId("count")).toHaveTextContent("1"); + + n = 2; + await act(async () => { + fire?.(); + await Promise.resolve(); + }); + expect(screen.getByTestId("count")).toHaveTextContent("2"); + }); + + it("polls while any session is mid-turn and stops when all settle", async () => { + vi.useFakeTimers(); + let calls = 0; + const transport: CeSessionsTransport = { + list: vi.fn(async () => { + calls += 1; + return [mkSession({ id: "s1", status: calls >= 3 ? "completed" : "active" })]; + }), + remove: vi.fn(), + }; + render(); + await act(async () => { + await vi.advanceTimersByTimeAsync(1); + }); + expect(screen.getByTestId("count")).toHaveTextContent("1"); + + await act(async () => { + await vi.advanceTimersByTimeAsync(50); + }); + const settledCalls = calls; + expect(calls).toBeGreaterThanOrEqual(3); + + // All settled → polling stops (no further list calls as time advances). + await act(async () => { + await vi.advanceTimersByTimeAsync(50); + }); + expect(calls).toBe(settledCalls); + }); + + it("surfaces a list error without crashing", async () => { + const transport: CeSessionsTransport = { + list: vi.fn(async () => { + throw new Error("kaput"); + }), + remove: vi.fn(), + }; + render(); + await act(async () => {}); + expect(screen.getByTestId("err")).toHaveTextContent("kaput"); + expect(screen.getByTestId("count")).toHaveTextContent("0"); + }); +}); diff --git a/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/api.ts b/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/api.ts index 2a031ceda9..b2a0b0433b 100644 --- a/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/api.ts +++ b/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/api.ts @@ -92,6 +92,24 @@ export async function resumeSession(sessionId: string, projectId?: string): Prom return data.session; } +/** List CE sessions, newest-activity first (optionally filtered by status/stage). */ +export async function listSessions( + opts: { projectId?: string; status?: string; stage?: string } = {}, +): Promise { + const data = await request<{ sessions: CeSession[] }>( + `/sessions${qp({ projectId: opts.projectId, status: opts.status, stage: opts.stage })}`, + ); + return data.sessions; +} + +/** Discard a session (disposes any live handle, deletes the row). `projectId` must match start. */ +export async function deleteSession(sessionId: string, projectId?: string): Promise { + await request<{ deleted: boolean }>( + `/sessions/${encodeURIComponent(sessionId)}${qp({ projectId })}`, + { method: "DELETE" }, + ); +} + /** Poll the current persisted session state. `projectId` must match start (see answerSession). */ export async function getSession(sessionId: string, projectId?: string): Promise { const data = await request<{ session: CeSession }>(`/sessions/${encodeURIComponent(sessionId)}${qp({ projectId })}`); diff --git a/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/useCeSession.ts b/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/useCeSession.ts index 42d0d26476..4219375de0 100644 --- a/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/useCeSession.ts +++ b/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/useCeSession.ts @@ -62,6 +62,8 @@ export interface UseCeSessionResult { busy: boolean; error?: string; start(stage: string, opts?: { message?: string; projectId?: string }): Promise; + /** Adopt an EXISTING session (e.g. from the session list) as the active one. */ + open(sessionId: string, opts?: { projectId?: string }): Promise; answer(questionId: string, response: unknown): Promise; resume(): Promise; reset(): void; @@ -130,6 +132,18 @@ export function useCeSession(options: UseCeSessionOptions = {}): UseCeSessionRes [run, transport], ); + // Adopt an existing session (started earlier, possibly in another view visit) + // as this hook's active session. Like start(), it pins the projectId used for + // every subsequent call — the session row lives in that project's store. + const open = useCallback( + (sessionId: string, opts: { projectId?: string } = {}) => { + projectIdRef.current = opts.projectId; + sessionIdRef.current = sessionId; + return run(() => transport.get(sessionId, opts.projectId)); + }, + [run, transport], + ); + const answer = useCallback( (questionId: string, response: unknown) => { const id = sessionIdRef.current; @@ -198,5 +212,5 @@ export function useCeSession(options: UseCeSessionOptions = {}): UseCeSessionRes }; }, [status, busy, transport, apply, pollIntervalMs]); - return { session, busy, error, start, answer, resume, reset }; + return { session, busy, error, start, open, answer, resume, reset }; } diff --git a/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/useCeSessions.ts b/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/useCeSessions.ts new file mode 100644 index 0000000000..663b7a5f65 --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/useCeSessions.ts @@ -0,0 +1,127 @@ +import { useCallback, useEffect, useRef, useState } from "react"; +import type { CeSession } from "../../session/session-store.js"; +import { deleteSession as deleteSessionApi, listSessions as listSessionsApi } from "./api.js"; + +/** + * Injectable list transport so component tests can drive the session list + * without a network. Defaults to the real routes. + */ +export interface CeSessionsTransport { + list(projectId?: string): Promise; + remove(sessionId: string, projectId?: string): Promise; +} + +const defaultTransport: CeSessionsTransport = { + list: (projectId) => listSessionsApi({ projectId }), + remove: (id, projectId) => deleteSessionApi(id, projectId), +}; + +/** + * Subscribe to ANY CE plugin push event (no per-session filter — any session + * turn/question/complete should refresh the list). Returns an unsubscribe fn. + * Default no-op = polling only, same posture as useCeSession's subscribe. + */ +export type CeSessionsSubscribe = (onAnyEvent: () => void) => () => void; + +export interface UseCeSessionsOptions { + projectId?: string; + /** Gate fetching (mirrors useArtifacts' viewport gating). Default true. */ + enabled?: boolean; + /** Poll interval (ms) while any session has a turn in flight. */ + pollIntervalMs?: number; + transport?: CeSessionsTransport; + subscribe?: CeSessionsSubscribe; +} + +export interface UseCeSessionsResult { + sessions: CeSession[]; + loading: boolean; + error?: string; + /** Re-fetch the list now (e.g. after launching or closing a session). */ + refresh(): Promise; + /** Discard a session and refresh the list. */ + remove(sessionId: string): Promise; +} + +/** Statuses with an agent turn in flight — the list keeps polling while any exist. */ +const IN_FLIGHT = new Set(["active", "launching"]); + +/** + * Multi-session management list (server state is already multi-session: each + * row is an independent pipeline run with its own live handle). Refreshes on + * any plugin push event, and polls as a fallback while any session is + * mid-turn so progress made in another tab/process still shows up. + */ +export function useCeSessions(options: UseCeSessionsOptions = {}): UseCeSessionsResult { + const { projectId } = options; + const enabled = options.enabled ?? true; + const pollIntervalMs = options.pollIntervalMs ?? 5000; + const transport = options.transport ?? defaultTransport; + const subscribe = options.subscribe; + + const [sessions, setSessions] = useState([]); + const [loading, setLoading] = useState(false); + const [error, setError] = useState(); + + const mounted = useRef(true); + useEffect(() => { + mounted.current = true; + return () => { + mounted.current = false; + }; + }, []); + + const refresh = useCallback(async () => { + try { + const next = await transport.list(projectId); + if (mounted.current) { + setSessions(next); + setError(undefined); + } + } catch (err) { + if (mounted.current) setError(err instanceof Error ? err.message : String(err)); + } finally { + if (mounted.current) setLoading(false); + } + }, [transport, projectId]); + + // Initial fetch (and on project switch). + useEffect(() => { + if (!enabled) return; + setLoading(true); + void refresh(); + }, [enabled, refresh]); + + // Live push: any CE event means some session changed — refresh the list. + useEffect(() => { + if (!enabled || !subscribe) return; + return subscribe(() => { + void refresh(); + }); + }, [enabled, subscribe, refresh]); + + // Poll fallback only while a turn is actually in flight somewhere. + const anyInFlight = sessions.some((s) => IN_FLIGHT.has(s.status)); + useEffect(() => { + if (!enabled || !anyInFlight) return; + const timer = setInterval(() => { + void refresh(); + }, pollIntervalMs); + return () => clearInterval(timer); + }, [enabled, anyInFlight, pollIntervalMs, refresh]); + + const remove = useCallback( + async (sessionId: string) => { + try { + await transport.remove(sessionId, projectId); + } catch (err) { + if (mounted.current) setError(err instanceof Error ? err.message : String(err)); + return; + } + await refresh(); + }, + [transport, projectId, refresh], + ); + + return { sessions, loading, error, refresh, remove }; +} 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 4015b6fbd6..69d44aeace 100644 --- a/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts +++ b/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts @@ -128,6 +128,20 @@ export function createSessionRoutes(): PluginRouteDefinition[] { return { status: 200, body: { sessions } }; }, }, + { + method: "DELETE", + path: "/sessions/:id", + description: "Discard a CE session (disposes any live handle, deletes the row).", + handler: async (req: unknown, ctx: PluginContext): Promise => { + const id = (req as RouteRequest).params.id; + // Go through the orchestrator so an in-flight live handle is disposed, + // not just the row removed (a bare store.delete would leave the agent + // running unobserved in this process). + const removed = getOrchestrator(ctx).discard(id); + if (!removed) return { status: 404, body: { error: `Session ${id} not found` } }; + return { status: 200, body: { deleted: true } }; + }, + }, { // U7 work bridge: observe the board tasks a CE pipeline (session) landed, // via their link records (the addressable back-reference, FN-5719). The diff --git a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts index fa9c48e29b..a4c3fc3bef 100644 --- a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts +++ b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts @@ -385,6 +385,17 @@ export class CeOrchestrator { return this.store.get(sessionId); } + /** + * Discard a session: dispose any live in-process handle (so an in-flight + * agent doesn't keep running unobserved) and delete the persisted row. + * Returns false when the session doesn't exist. Pipeline-link rows are NOT + * touched — board tasks the session landed keep their provenance records. + */ + discard(sessionId: string): boolean { + this.disposeLive(sessionId); + return this.store.delete(sessionId); + } + /** * Run one turn behind a timeout race, persist the resulting event, and on a * turn-level failure auto-save + emit. The `driver` performs the prompt/answer diff --git a/plugins/fusion-plugin-compound-engineering/src/session/session-store.ts b/plugins/fusion-plugin-compound-engineering/src/session/session-store.ts index db73f73c1a..75259ede3b 100644 --- a/plugins/fusion-plugin-compound-engineering/src/session/session-store.ts +++ b/plugins/fusion-plugin-compound-engineering/src/session/session-store.ts @@ -255,6 +255,12 @@ export class CeSessionStore { return next; } + /** Delete a session row. Returns true when a row was removed. */ + delete(id: string): boolean { + const result = this.db.prepare(`DELETE FROM ce_sessions WHERE id = ?`).run(id); + return Number(result.changes ?? 0) > 0; + } + /** Append a turn to the conversation history (no other field touched). */ appendHistory(id: string, turn: CeConversationTurn): CeSession | undefined { const existing = this.get(id);