diff --git a/docs/plugins/compound-engineering.md b/docs/plugins/compound-engineering.md index c74a391e49..4d7194a06b 100644 --- a/docs/plugins/compound-engineering.md +++ b/docs/plugins/compound-engineering.md @@ -49,8 +49,15 @@ sessions resume/retry back to their current question. Turn execution is **detached**: start/answer/resume return as soon as the session row reflects the request, with the agent turn running in the background -(failures persist into session state — never an unhandled rejection). While a -turn runs, the engine streams mid-turn progress (thinking/text deltas + tool +(failures persist into session state — never an unhandled rejection). **Close** +only leaves the flow UI; it does not stop the detached agent. **Cancel** is the +explicit stop action for `launching`/`active`/`awaiting_input` sessions: it +aborts any live in-process handle, flushes live working output into history, and +keeps the session row as terminal `interrupted` with `Cancelled by user` so the +conversation can be inspected or resumed. **Discard** is different: it removes a +settled session row entirely after disposing any live handle. + +While a turn runs, the engine streams mid-turn progress (thinking/text deltas + tool markers) through the seam's `onProgress` option; the orchestrator buffers it and `GET /sessions/:id` attaches it as transient `liveActivity`. The per-turn timeout is **inactivity-based** (progress re-arms it), so long actively-working @@ -69,9 +76,11 @@ HTTP endpoints (under `/api/plugins/fusion-plugin-compound-engineering/`): - `POST /sessions` → start a stage session - `POST /sessions/:id/answer` → answer the awaiting question (send `projectId`) - `POST /sessions/:id/resume` → resume an awaiting/interrupted session (send `projectId`) +- `POST /sessions/:id/cancel` → cancel an in-flight session; stops the agent and keeps the row as `interrupted` - `GET /sessions/:id` → current persisted session state (push + poll fallback) - `GET /sessions` → list sessions (filter by status/stage) - `GET /sessions/:id/links` → the work→board pipeline-link records for a session +- `DELETE /sessions/:id` → discard a session; stops any live handle and deletes the row ## Sync model diff --git a/plugins/fusion-plugin-compound-engineering/README.md b/plugins/fusion-plugin-compound-engineering/README.md index b354ecc037..643eb66132 100644 --- a/plugins/fusion-plugin-compound-engineering/README.md +++ b/plugins/fusion-plugin-compound-engineering/README.md @@ -74,10 +74,17 @@ activity; from there you can: - **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, +- **cancel** an in-flight (`launching`/`active`/`awaiting_input`) session via + `POST /sessions/:id/cancel`, which stops any live in-process handle, flushes + live progress into history, and keeps the row as `interrupted` with a + `Cancelled by user` marker for inspection/resume, - **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). +Cancel and discard are intentionally different: cancel stops work but preserves +conversation/progress; discard removes the row entirely. + The list refreshes on any CE push event and falls back to polling `GET /sessions` while any session has a turn in flight. @@ -85,7 +92,9 @@ The list refreshes on any CE push event and falls back to polling Turn execution is **detached**: `POST /sessions`, `/answer`, and `/resume` return as soon as the session row reflects the request, with the agent turn -running in the background. While it runs: +running in the background. Closing the flow does not cancel the server-side +agent; use `POST /sessions/:id/cancel` (or the dashboard Cancel button) to stop +an in-flight turn while preserving the session as `interrupted`. While it runs: - The engine streams **live progress** through the seam's `onProgress` option (thinking/text deltas + tool start/end markers — a host capability any diff --git a/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-cancel.test.ts b/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-cancel.test.ts new file mode 100644 index 0000000000..9a0d77544d --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-cancel.test.ts @@ -0,0 +1,101 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { InteractiveAiSession } from "@fusion/core"; +import { CE_EVENTS, CeOrchestrator } from "../session/orchestrator.js"; +import { getCeSessionStore, type CeActivityTurn, type CeSessionStatus } from "../session/session-store.js"; +import { makeHarness, type TestHarness } from "./_harness.js"; + +interface OrchestratorInternals { + live: Map; + activity: Map; +} + +function internals(orch: CeOrchestrator): OrchestratorInternals { + return orch as unknown as OrchestratorInternals; +} + +function liveHandle(): InteractiveAiSession { + return { + prompt: vi.fn(), + answer: vi.fn(), + nextEvent: vi.fn(), + dispose: vi.fn(), + }; +} + +describe("CeOrchestrator.cancel", () => { + let h: TestHarness; + + afterEach(() => { + h?.close(); + }); + + it("interrupts an in-flight session with a live handle, flushes progress, disposes, and emits", () => { + h = makeHarness(); + const store = getCeSessionStore(h.ctx); + const orch = new CeOrchestrator({ ctx: h.ctx }); + const session = store.update(store.create({ stage: "brainstorm" }).id, { status: "active" })!; + const handle = liveHandle(); + internals(orch).live.set(session.id, handle); + internals(orch).activity.set(session.id, [ + { kind: "thinking", text: "drafting cancellable progress", at: new Date().toISOString() }, + ]); + + const cancelled = orch.cancel(session.id)!; + + expect(cancelled.status).toBe("interrupted"); + expect(cancelled.error).toBe("Cancelled by user"); + expect(handle.dispose).toHaveBeenCalledTimes(1); + expect(orch.getLiveActivity(session.id)).toEqual([]); + expect(cancelled.conversationHistory.some((t) => t.text.includes("drafting cancellable progress"))).toBe(true); + expect(h.emitted).toContainEqual({ + event: CE_EVENTS.interrupted, + data: { sessionId: session.id, message: "Cancelled by user" }, + }); + }); + + it.each(["launching", "active", "awaiting_input"])( + "interrupts %s without requiring a live handle", + (status) => { + h = makeHarness(); + const store = getCeSessionStore(h.ctx); + const orch = new CeOrchestrator({ ctx: h.ctx }); + const session = store.update(store.create({ stage: "brainstorm" }).id, { status })!; + + const cancelled = orch.cancel(session.id)!; + + expect(cancelled.status).toBe("interrupted"); + expect(cancelled.error).toBe("Cancelled by user"); + expect(h.emitted.map((e) => e.event)).toEqual([CE_EVENTS.interrupted]); + }, + ); + + it.each(["completed", "error", "interrupted"])( + "is idempotent for terminal status %s", + (status) => { + h = makeHarness(); + const store = getCeSessionStore(h.ctx); + const orch = new CeOrchestrator({ ctx: h.ctx }); + const session = store.update(store.create({ stage: "brainstorm" }).id, { + status, + error: status === "completed" ? null : "already settled", + })!; + const handle = liveHandle(); + internals(orch).live.set(session.id, handle); + + const cancelled = orch.cancel(session.id)!; + + expect(cancelled).toEqual(session); + expect(handle.dispose).not.toHaveBeenCalled(); + expect(h.emitted).toEqual([]); + expect(store.get(session.id)!.status).toBe(status); + }, + ); + + it("returns undefined for an unknown session", () => { + h = makeHarness(); + const orch = new CeOrchestrator({ ctx: h.ctx }); + + expect(orch.cancel("missing")).toBeUndefined(); + expect(h.emitted).toEqual([]); + }); +}); 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 2d06285cca..f7906b1ea6 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 @@ -48,6 +48,7 @@ describe("session routes (polling transport)", () => { "POST /sessions", "POST /sessions/:id/answer", "POST /sessions/:id/resume", + "POST /sessions/:id/cancel", "GET /sessions/:id", "GET /sessions", "DELETE /sessions/:id", @@ -70,6 +71,40 @@ describe("session routes (polling transport)", () => { expect(store.get(keep.id)).toBeDefined(); }); + it("POST /sessions/:id/cancel interrupts an in-flight session", async () => { + const { getCeSessionStore } = await import("../session/session-store.js"); + const store = getCeSessionStore(h.ctx); + const created = store.update(store.create({ stage: "brainstorm" }).id, { status: "active" })!; + + const res = await call("POST", "/sessions/:id/cancel", { params: { id: created.id } }, h.ctx); + + expect(res.status).toBe(200); + const session = (res.body as { session: { status: string; error: string | null } }).session; + expect(session.status).toBe("interrupted"); + expect(session.error).toBe("Cancelled by user"); + }); + + it("POST /sessions/:id/cancel returns 404 for an unknown session", async () => { + const res = await call("POST", "/sessions/:id/cancel", { params: { id: "nope" } }, h.ctx); + + expect(res.status).toBe(404); + expect((res.body as { error: string }).error).toMatch(/not found/i); + }); + + it("POST /sessions/:id/cancel is idempotent for terminal sessions", async () => { + const { getCeSessionStore } = await import("../session/session-store.js"); + const store = getCeSessionStore(h.ctx); + const created = store.update(store.create({ stage: "brainstorm" }).id, { status: "completed" })!; + + const res = await call("POST", "/sessions/:id/cancel", { params: { id: created.id } }, h.ctx); + + expect(res.status).toBe(200); + const session = (res.body as { session: { status: string; error: string | null } }).session; + expect(session.status).toBe("completed"); + expect(session.error).toBeNull(); + expect(store.get(created.id)!.status).toBe("completed"); + }); + 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); diff --git a/plugins/fusion-plugin-compound-engineering/src/dashboard/CeFlow.tsx b/plugins/fusion-plugin-compound-engineering/src/dashboard/CeFlow.tsx index 63af5f95b6..b2a24fbbb5 100644 --- a/plugins/fusion-plugin-compound-engineering/src/dashboard/CeFlow.tsx +++ b/plugins/fusion-plugin-compound-engineering/src/dashboard/CeFlow.tsx @@ -33,6 +33,8 @@ export interface CeFlowProps { onAnswer: (questionId: string, response: unknown) => void; /** Resume an interrupted/error session. */ onResume?: () => void; + /** Cancel an in-flight session while preserving it as interrupted. */ + onCancel?: () => void; /** Back to the launcher. */ onClose?: () => void; } @@ -506,7 +508,7 @@ function QuestionPanel({ // ── Flow surface ───────────────────────────────────────────────────────────── export function CeFlow(props: CeFlowProps) { - const { session, busy, error, onAnswer, onResume, onClose } = props; + const { session, busy, error, onAnswer, onResume, onCancel, onClose } = props; const question = session?.currentQuestion ?? undefined; @@ -526,6 +528,7 @@ export function CeFlow(props: CeFlowProps) { const status = session.status; const settledTerminal = status === "completed"; const recoverable = status === "interrupted" || status === "error"; + const cancellable = status === "launching" || status === "active" || status === "awaiting_input"; const working = status === "active" || status === "launching"; return ( @@ -535,6 +538,17 @@ export function CeFlow(props: CeFlowProps) { {status.replace("_", " ")} + {onCancel && cancellable ? ( + + ) : null} {onClose ? ( - ) : null} + ) : ( + + )} ); })} @@ -282,6 +294,7 @@ export function CompoundEngineeringView(props: CompoundEngineeringViewProps) { ...(subscribeList ? { subscribe: subscribeList } : {}), }); const [launcherOpen, setLauncherOpen] = useState(false); + const [sessionActionBusy, setSessionActionBusy] = useState(false); const totalArtifacts = result?.totalArtifacts ?? 0; const totalErrors = result?.totalErrors ?? 0; @@ -310,9 +323,23 @@ export function CompoundEngineeringView(props: CompoundEngineeringViewProps) { [ceSession, projectId], ); + const onCancelSession = useCallback( + (s: CeSession) => { + setSessionActionBusy(true); + void ceSessions + .cancel(s.id) + .then(() => { + if (ceSession.session?.id === s.id) ceSession.reset(); + }) + .finally(() => setSessionActionBusy(false)); + }, + [ceSession, ceSessions], + ); + const onDiscardSession = useCallback( (s: CeSession) => { - void ceSessions.remove(s.id); + setSessionActionBusy(true); + void ceSessions.remove(s.id).finally(() => setSessionActionBusy(false)); }, [ceSessions], ); @@ -336,16 +363,18 @@ export function CompoundEngineeringView(props: CompoundEngineeringViewProps) { onCancelSession(ceSession.session!)} onClose={onCloseFlow} /> @@ -376,8 +405,9 @@ export function CompoundEngineeringView(props: CompoundEngineeringViewProps) { diff --git a/plugins/fusion-plugin-compound-engineering/src/dashboard/__tests__/CeFlow.test.tsx b/plugins/fusion-plugin-compound-engineering/src/dashboard/__tests__/CeFlow.test.tsx index ede600ce43..abfc7898d8 100644 --- a/plugins/fusion-plugin-compound-engineering/src/dashboard/__tests__/CeFlow.test.tsx +++ b/plugins/fusion-plugin-compound-engineering/src/dashboard/__tests__/CeFlow.test.tsx @@ -419,6 +419,26 @@ describe("CeFlow — lifecycle surfaces", () => { expect(screen.getByTestId("ce-activity-tool")).toHaveTextContent("Grep"); }); + it.each(["launching", "active", "awaiting_input"] as const)("offers cancel on a %s session", (status) => { + const onCancel = vi.fn(); + render(); + + fireEvent.click(screen.getByTestId("ce-flow-cancel")); + expect(onCancel).toHaveBeenCalledTimes(1); + }); + + it.each(["completed", "error", "interrupted"] as const)("hides cancel on a terminal %s session", (status) => { + render(); + + expect(screen.queryByTestId("ce-flow-cancel")).not.toBeInTheDocument(); + }); + + it("disables cancel while busy", () => { + render(); + + expect(screen.getByTestId("ce-flow-cancel")).toBeDisabled(); + }); + it("offers resume on an interrupted session", () => { const onResume = vi.fn(); render( 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 0adffa55b9..32ee6e9f68 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 @@ -8,6 +8,9 @@ const listArtifacts = vi.fn(async (): Promise => { }); const listSessions = vi.fn(async (): Promise => []); const deleteSession = vi.fn(async (_id: string, _projectId?: string): Promise => undefined); +const cancelSession = vi.fn(async (_id: string, _projectId?: string): Promise => { + throw new Error("cancelSession mock not configured"); +}); const getSession = vi.fn(async (_id: string, _projectId?: string): Promise => { throw new Error("getSession mock not configured"); }); @@ -16,6 +19,7 @@ vi.mock("../hooks/api.js", () => ({ getArtifactPreviewUrl: (id: string) => `/preview/${id}`, listSessions: () => listSessions(), deleteSession: (id: string, projectId?: string) => deleteSession(id, projectId), + cancelSession: (id: string, projectId?: string) => cancelSession(id, projectId), getSession: (id: string, projectId?: string) => getSession(id, projectId), startSession: vi.fn(), answerSession: vi.fn(), @@ -79,6 +83,8 @@ describe("CompoundEngineeringView", () => { listSessions.mockResolvedValue([]); deleteSession.mockReset(); deleteSession.mockResolvedValue(undefined); + cancelSession.mockReset(); + cancelSession.mockImplementation(async (id: string, projectId?: string) => mkCeSession({ id, projectId: projectId ?? null, status: "interrupted", error: "Cancelled by user" })); getSession.mockReset(); }); @@ -193,8 +199,21 @@ describe("CompoundEngineeringView", () => { ]); // Awaiting sessions advertise that they need the user. expect(rows[0].textContent).toMatch(/needs your input/i); - // Only the terminal session can be discarded. + // Only non-terminal sessions can be cancelled; only terminal sessions can be discarded. + expect(screen.getAllByTestId("ce-session-cancel")).toHaveLength(2); expect(screen.getAllByTestId("ce-session-discard")).toHaveLength(1); + expect(rows[0].querySelector("[data-testid='ce-session-cancel']")).toBeInTheDocument(); + expect(rows[1].querySelector("[data-testid='ce-session-cancel']")).toBeInTheDocument(); + expect(rows[2].querySelector("[data-testid='ce-session-cancel']")).not.toBeInTheDocument(); + }); + + it("renders no cancel affordance for an empty sessions list", async () => { + listArtifacts.mockResolvedValue(makeResult({})); + listSessions.mockResolvedValue([]); + render(); + + await screen.findByTestId("ce-empty-state"); + expect(screen.queryByTestId("ce-session-cancel")).not.toBeInTheDocument(); }); it("opens an existing session from the list into the flow (and back without losing it)", async () => { @@ -233,6 +252,38 @@ describe("CompoundEngineeringView", () => { expect(deleteSession).not.toHaveBeenCalled(); }); + it("cancels an in-flight session via the list", async () => { + listArtifacts.mockResolvedValue(makeResult({})); + listSessions.mockResolvedValue([mkCeSession({ id: "running", stage: "plan", status: "active" })]); + render(); + + await screen.findByTestId("ce-sessions"); + listSessions.mockResolvedValue([mkCeSession({ id: "running", stage: "plan", status: "interrupted", error: "Cancelled by user" })]); + fireEvent.click(screen.getByTestId("ce-session-cancel")); + + await waitFor(() => expect(cancelSession).toHaveBeenCalledWith("running", "p1")); + await waitFor(() => expect(screen.queryByTestId("ce-session-cancel")).not.toBeInTheDocument()); + expect(screen.getByTestId("ce-session-discard")).toBeInTheDocument(); + }); + + it("cancels an open flow and returns to the refreshed sessions overview", async () => { + listArtifacts.mockResolvedValue(makeResult({})); + listSessions.mockResolvedValue([mkCeSession({ id: "flow", stage: "plan", status: "active" })]); + getSession.mockResolvedValue(mkCeSession({ id: "flow", stage: "plan", status: "active" })); + render(); + + await screen.findByTestId("ce-sessions"); + fireEvent.click(screen.getByTestId("ce-session-open")); + await screen.findByTestId("ce-flow"); + listSessions.mockResolvedValue([mkCeSession({ id: "flow", stage: "plan", status: "interrupted", error: "Cancelled by user" })]); + fireEvent.click(screen.getByTestId("ce-flow-cancel")); + + await waitFor(() => expect(cancelSession).toHaveBeenCalledWith("flow", "p1")); + await waitFor(() => expect(screen.queryByTestId("ce-flow")).not.toBeInTheDocument()); + expect(screen.getByTestId("ce-sessions")).toBeInTheDocument(); + expect(screen.getByTestId("ce-session-discard")).toBeInTheDocument(); + }); + it("discards a terminal session via the list", async () => { listArtifacts.mockResolvedValue(makeResult({})); listSessions.mockResolvedValue([mkCeSession({ id: "done", stage: "plan", status: "completed" })]); 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 index da0c740f85..ba26a55d1e 100644 --- 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 @@ -41,6 +41,7 @@ function Harness({ {s.error ?? ""} + ); } @@ -52,7 +53,7 @@ describe("useCeSessions (multi-session list)", () => { 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() }; + const transport: CeSessionsTransport = { list, remove: vi.fn(), cancel: vi.fn() }; render(); await act(async () => {}); @@ -68,6 +69,7 @@ describe("useCeSessions (multi-session list)", () => { remove: vi.fn(async () => { removed = true; }), + cancel: vi.fn(), }; render(); await act(async () => {}); @@ -80,6 +82,43 @@ describe("useCeSessions (multi-session list)", () => { expect(screen.getByTestId("ids")).toHaveTextContent("s2"); }); + it("cancel() cancels via the transport then refreshes the list", async () => { + let cancelled = false; + const transport: CeSessionsTransport = { + list: vi.fn(async () => [mkSession({ id: "s1", status: cancelled ? "interrupted" : "active" })]), + remove: vi.fn(), + cancel: vi.fn(async () => { + cancelled = true; + }), + }; + render(); + await act(async () => {}); + expect(screen.getByTestId("ids")).toHaveTextContent("s1"); + + await act(async () => { + screen.getByText("cancel").click(); + }); + expect(transport.cancel).toHaveBeenCalledWith("s1", "p1"); + expect(transport.list).toHaveBeenCalledTimes(2); + }); + + it("cancel() surfaces a transport error without crashing", async () => { + const transport: CeSessionsTransport = { + list: vi.fn(async () => [mkSession({ id: "s1", status: "active" })]), + remove: vi.fn(), + cancel: vi.fn(async () => { + throw new Error("cancel failed"); + }), + }; + render(); + await act(async () => {}); + + await act(async () => { + screen.getByText("cancel").click(); + }); + expect(screen.getByTestId("err")).toHaveTextContent("cancel failed"); + }); + it("refreshes when a push event fires", async () => { let fire: (() => void) | undefined; const subscribe: CeSessionsSubscribe = (onAnyEvent) => { @@ -92,6 +131,7 @@ describe("useCeSessions (multi-session list)", () => { const transport: CeSessionsTransport = { list: vi.fn(async () => Array.from({ length: n }, (_, i) => mkSession({ id: `s${i + 1}` }))), remove: vi.fn(), + cancel: vi.fn(), }; render(); await act(async () => {}); @@ -114,6 +154,7 @@ describe("useCeSessions (multi-session list)", () => { return [mkSession({ id: "s1", status: calls >= 3 ? "completed" : "active" })]; }), remove: vi.fn(), + cancel: vi.fn(), }; render(); await act(async () => { @@ -140,6 +181,7 @@ describe("useCeSessions (multi-session list)", () => { throw new Error("kaput"); }), remove: vi.fn(), + cancel: vi.fn(), }; render(); await act(async () => {}); 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 b2a0b0433b..d0e931e097 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,16 @@ export async function resumeSession(sessionId: string, projectId?: string): Prom return data.session; } +/** Cancel an in-flight session without deleting it. `projectId` must match start (see answerSession). */ +export async function cancelSession(sessionId: string, projectId?: string): Promise { + const data = await request<{ session: CeSession }>(`/sessions/${encodeURIComponent(sessionId)}/cancel`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ projectId }), + }); + 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 } = {}, diff --git a/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/useCeSessions.ts b/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/useCeSessions.ts index 663b7a5f65..2797a0e807 100644 --- a/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/useCeSessions.ts +++ b/plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/useCeSessions.ts @@ -1,6 +1,6 @@ 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"; +import { cancelSession as cancelSessionApi, deleteSession as deleteSessionApi, listSessions as listSessionsApi } from "./api.js"; /** * Injectable list transport so component tests can drive the session list @@ -9,11 +9,15 @@ import { deleteSession as deleteSessionApi, listSessions as listSessionsApi } fr export interface CeSessionsTransport { list(projectId?: string): Promise; remove(sessionId: string, projectId?: string): Promise; + cancel(sessionId: string, projectId?: string): Promise; } const defaultTransport: CeSessionsTransport = { list: (projectId) => listSessionsApi({ projectId }), remove: (id, projectId) => deleteSessionApi(id, projectId), + cancel: async (id, projectId) => { + await cancelSessionApi(id, projectId); + }, }; /** @@ -41,6 +45,8 @@ export interface UseCeSessionsResult { refresh(): Promise; /** Discard a session and refresh the list. */ remove(sessionId: string): Promise; + /** Cancel an in-flight session and refresh the list. */ + cancel(sessionId: string): Promise; } /** Statuses with an agent turn in flight — the list keeps polling while any exist. */ @@ -123,5 +129,18 @@ export function useCeSessions(options: UseCeSessionsOptions = {}): UseCeSessions [transport, projectId, refresh], ); - return { sessions, loading, error, refresh, remove }; + const cancel = useCallback( + async (sessionId: string) => { + try { + await transport.cancel(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, cancel }; } 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 2c6e9d647d..4f597f1f63 100644 --- a/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts +++ b/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts @@ -104,6 +104,17 @@ export function createSessionRoutes(): PluginRouteDefinition[] { } }, }, + { + method: "POST", + path: "/sessions/:id/cancel", + description: "Cancel an in-flight CE session (stops the agent, keeps the row as interrupted).", + handler: async (req: unknown, ctx: PluginContext): Promise => { + const id = (req as RouteRequest).params.id; + const session = getOrchestrator(ctx).cancel(id); + if (!session) return { status: 404, body: { error: `Session ${id} not found` } }; + return { status: 200, body: { session } }; + }, + }, { method: "GET", path: "/sessions/:id", diff --git a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts index 1f51d00092..229423f3bb 100644 --- a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts +++ b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts @@ -651,6 +651,26 @@ export class CeOrchestrator { return this.store.get(sessionId); } + /** + * Cancel a session: stop any live in-process handle but keep the persisted row + * for inspection/resume by marking it `interrupted`. Unlike discard(), cancel + * preserves the conversation and progress; discard stops the handle AND deletes + * the row. Terminal sessions are idempotent no-ops. + */ + cancel(sessionId: string): CeSession | undefined { + const session = this.store.get(sessionId); + if (!session) return undefined; + if (session.status === "completed" || session.status === "error" || session.status === "interrupted") { + return session; + } + + // Preserve no-silent-loss ordering: interruptSession flushes live activity + // before disposeLive clears the transient buffers (same as runTurn failure). + const interrupted = this.interruptSession(sessionId, new Error("Cancelled by user")); + this.disposeLive(sessionId); + return interrupted; + } + /** * Discard a session: dispose any live in-process handle (so an in-flight * agent doesn't keep running unobserved) and delete the persisted row.