From 9b5904b74545f4fa498f9d5c87a831a0c95aa182 Mon Sep 17 00:00:00 2001 From: Neonforge <48338160+Neonforge98@users.noreply.github.com> Date: Tue, 25 Aug 2026 07:51:29 +0800 Subject: [PATCH] feat(runtime): add canonical turn dispatch boundary Map composer and backend-selected turn identities onto one exact lifecycle generation, and reconcile lost responses through durable receipts. Return typed native and CLI acknowledgements while preserving legacy terminal handling until producer migration completes. --- src/api/tauri/rpc/procedures/agentSession.ts | 4 + .../rpc/schemas/__tests__/turnIntent.test.ts | 35 + src/api/tauri/rpc/schemas/cli.ts | 14 +- src/api/tauri/rpc/schemas/index.ts | 1 + src/api/tauri/rpc/schemas/turnIntent.ts | 23 + .../turnIntentDispatchLifecycle.test.ts | 104 ++ .../control/__tests__/turnLifecycle.test.ts | 114 +- .../control/turnIntentDispatchLifecycle.ts | 102 +- .../SessionCore/control/turnLifecycle.ts | 140 ++- .../SessionCore/services/SessionService.ts | 33 +- .../services/TurnDispatchService.test.ts | 1029 +++++++++++++++++ .../services/TurnDispatchService.ts | 630 ++++++++++ src/engines/SessionCore/services/types.ts | 5 + .../__tests__/rustAgentEventLifecycle.test.ts | 77 +- .../__tests__/rustAgentSendPayload.test.ts | 105 +- .../cli/__tests__/cliTransport.test.ts | 126 +- .../sync/adapters/cli/cliTransport.ts | 13 +- .../sync/adapters/createRustAgentAdapter.ts | 45 +- .../sync/adapters/externalHistoryAdapter.ts | 3 +- .../sync/adapters/rustAgentSendPayload.ts | 51 +- src/engines/SessionCore/sync/types.ts | 23 +- 21 files changed, 2619 insertions(+), 58 deletions(-) create mode 100644 src/api/tauri/rpc/schemas/__tests__/turnIntent.test.ts create mode 100644 src/api/tauri/rpc/schemas/turnIntent.ts create mode 100644 src/engines/SessionCore/control/__tests__/turnIntentDispatchLifecycle.test.ts create mode 100644 src/engines/SessionCore/services/TurnDispatchService.test.ts create mode 100644 src/engines/SessionCore/services/TurnDispatchService.ts diff --git a/src/api/tauri/rpc/procedures/agentSession.ts b/src/api/tauri/rpc/procedures/agentSession.ts index 84b496a21c..9bc7a49249 100644 --- a/src/api/tauri/rpc/procedures/agentSession.ts +++ b/src/api/tauri/rpc/procedures/agentSession.ts @@ -11,6 +11,10 @@ export const agentSession = { .input(schemas.agentSession.SessionIdInput) .output(schemas.agentSession.SessionInfoSchema.nullable()) .build(), + getTurnIntentStatus: defineProcedure("agent_turn_intent_status") + .input(schemas.turnIntent.TurnIntentStatusInputSchema) + .output(schemas.turnIntent.TurnIntentStatusReceiptSchema.nullable()) + .build(), manualCompact: defineProcedure("agent_session_manual_compact") .input(schemas.agentSession.ManualCompactInput) .output(schemas.agentSession.ManualCompactResultSchema) diff --git a/src/api/tauri/rpc/schemas/__tests__/turnIntent.test.ts b/src/api/tauri/rpc/schemas/__tests__/turnIntent.test.ts new file mode 100644 index 0000000000..4aceb32f95 --- /dev/null +++ b/src/api/tauri/rpc/schemas/__tests__/turnIntent.test.ts @@ -0,0 +1,35 @@ +import { describe, expect, it } from "vitest"; + +import { + TurnIntentStatusInputSchema, + TurnIntentStatusReceiptSchema, +} from "../turnIntent"; + +describe("turn-intent RPC schemas", () => { + it("accepts an origin lookup and backend-selected effective identity", () => { + expect( + TurnIntentStatusInputSchema.parse({ + sessionId: "sdeagent-1", + turnIntentId: "intent-x", + }) + ).toEqual({ sessionId: "sdeagent-1", turnIntentId: "intent-x" }); + expect( + TurnIntentStatusReceiptSchema.parse({ + status: "queued", + effectiveTurnIntentId: "wir-y", + }) + ).toEqual({ status: "queued", effectiveTurnIntentId: "wir-y" }); + }); + + it("fails closed for unknown status or empty identity", () => { + expect(() => + TurnIntentStatusReceiptSchema.parse({ + status: "maybe-running", + effectiveTurnIntentId: "wir-y", + }) + ).toThrow(); + expect(() => + TurnIntentStatusInputSchema.parse({ sessionId: "", turnIntentId: "" }) + ).toThrow(); + }); +}); diff --git a/src/api/tauri/rpc/schemas/cli.ts b/src/api/tauri/rpc/schemas/cli.ts index 9d118c27bb..1d8ac77b45 100644 --- a/src/api/tauri/rpc/schemas/cli.ts +++ b/src/api/tauri/rpc/schemas/cli.ts @@ -2,6 +2,8 @@ import { z } from "zod/v4"; import { ActivityChunkSchema } from "@src/api/realtime/websocket/schemas"; +import { TurnIntentStatusSchema } from "./turnIntent"; + export const CliMessageRequestSchema = z.object({ sessionId: z.string().min(1), content: z.string(), @@ -24,17 +26,7 @@ export const CliRunReceiptSchema = z.object({ sessionId: z.string(), turnIntentId: z.string(), effectiveTurnIntentId: z.string().min(1), - status: z.enum([ - "optimistic", - "queued", - "running", - "completed", - "failed", - "cancelled", - "stale", - "coalesced", - "rejected", - ]), + status: TurnIntentStatusSchema, duplicate: z.boolean(), }); diff --git a/src/api/tauri/rpc/schemas/index.ts b/src/api/tauri/rpc/schemas/index.ts index ee4d5097f3..00846a46b3 100644 --- a/src/api/tauri/rpc/schemas/index.ts +++ b/src/api/tauri/rpc/schemas/index.ts @@ -26,3 +26,4 @@ export * as flow from "./flow"; export * as humanSession from "./humanSession"; export * as sessionCore from "./sessionCore"; export * as cli from "./cli"; +export * as turnIntent from "./turnIntent"; diff --git a/src/api/tauri/rpc/schemas/turnIntent.ts b/src/api/tauri/rpc/schemas/turnIntent.ts new file mode 100644 index 0000000000..bca8ab04dc --- /dev/null +++ b/src/api/tauri/rpc/schemas/turnIntent.ts @@ -0,0 +1,23 @@ +import { z } from "zod/v4"; + +export const TurnIntentStatusSchema = z.enum([ + "optimistic", + "queued", + "running", + "completed", + "failed", + "cancelled", + "stale", + "coalesced", + "rejected", +]); + +export const TurnIntentStatusInputSchema = z.object({ + sessionId: z.string().min(1), + turnIntentId: z.string().min(1), +}); + +export const TurnIntentStatusReceiptSchema = z.object({ + status: TurnIntentStatusSchema, + effectiveTurnIntentId: z.string().min(1), +}); diff --git a/src/engines/SessionCore/control/__tests__/turnIntentDispatchLifecycle.test.ts b/src/engines/SessionCore/control/__tests__/turnIntentDispatchLifecycle.test.ts new file mode 100644 index 0000000000..7060506e2f --- /dev/null +++ b/src/engines/SessionCore/control/__tests__/turnIntentDispatchLifecycle.test.ts @@ -0,0 +1,104 @@ +import { beforeEach, describe, expect, it } from "vitest"; + +import { + getTurnIntentDispatch, + publishTurnIntentDispatch, + publishTurnIntentDispatchAlias, + resetTurnIntentDispatchLifecycleForTests, + retireSessionTurnIntentDispatches, + retireTurnIntentDispatch, +} from "../turnIntentDispatchLifecycle"; + +describe("turnIntentDispatchLifecycle", () => { + beforeEach(() => resetTurnIntentDispatchLifecycleForTests()); + + it("never evicts a live long-running identity under recent-turn churn", () => { + publishTurnIntentDispatch("long-running", { + sessionId: "long-session", + generation: 1, + }); + + for (let index = 0; index < 250; index += 1) { + const sessionId = `short-session-${index}`; + publishTurnIntentDispatch(`short-${index}`, { + sessionId, + generation: 1, + }); + retireTurnIntentDispatch(sessionId, 1); + } + + expect(getTurnIntentDispatch("long-running")).toEqual({ + sessionId: "long-session", + generation: 1, + }); + }); + + it("retires only the exact generation and keeps it in bounded history", () => { + publishTurnIntentDispatch("first", { + sessionId: "session", + generation: 1, + }); + publishTurnIntentDispatch("second", { + sessionId: "session", + generation: 2, + }); + + retireTurnIntentDispatch("session", 1); + + expect(getTurnIntentDispatch("first")).toEqual({ + sessionId: "session", + generation: 1, + }); + expect(getTurnIntentDispatch("second")).toEqual({ + sessionId: "session", + generation: 2, + }); + }); + + it("removes live identities when their session is deleted", () => { + publishTurnIntentDispatch("intent", { + sessionId: "deleted-session", + generation: 1, + }); + + retireSessionTurnIntentDispatches("deleted-session"); + + expect(getTurnIntentDispatch("intent")).toBeUndefined(); + }); + + it("binds a backend-selected alias to the exact local generation", () => { + publishTurnIntentDispatch("composer-intent", { + sessionId: "session", + generation: 7, + }); + + expect( + publishTurnIntentDispatchAlias("wir_effective", { + sessionId: "session", + generation: 7, + }) + ).toBe(true); + expect(getTurnIntentDispatch("wir_effective")).toEqual({ + sessionId: "session", + generation: 7, + }); + }); + + it("fails closed instead of overwriting a conflicting alias", () => { + publishTurnIntentDispatch("wir_conflict", { + sessionId: "other-session", + generation: 3, + }); + + expect( + publishTurnIntentDispatchAlias("wir_conflict", { + sessionId: "session", + generation: 7, + }) + ).toBe(false); + expect(getTurnIntentDispatch("wir_conflict")).toEqual({ + sessionId: "other-session", + generation: 3, + }); + }); +}); diff --git a/src/engines/SessionCore/control/__tests__/turnLifecycle.test.ts b/src/engines/SessionCore/control/__tests__/turnLifecycle.test.ts index 016dd4564f..029447f8ee 100644 --- a/src/engines/SessionCore/control/__tests__/turnLifecycle.test.ts +++ b/src/engines/SessionCore/control/__tests__/turnLifecycle.test.ts @@ -1,13 +1,20 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + getTurnIntentDispatch, + publishTurnIntentDispatch, + resetTurnIntentDispatchLifecycleForTests, +} from "../turnIntentDispatchLifecycle"; import { beginTurnDispatch, beginTurnStopping, + clearTurnLifecycleSession, confirmTurnRunning, forceTurnIdle, getLastTurnTerminal, getTurnGeneration, getTurnPhase, + getTurnTerminal, isTurnActive, markTurnRunning, markTurnTerminal, @@ -20,11 +27,13 @@ const OTHER_SESSION = "session-2"; describe("turnLifecycle", () => { beforeEach(() => { vi.useFakeTimers(); + resetTurnIntentDispatchLifecycleForTests(); resetTurnLifecycleForTests(); }); afterEach(() => { resetTurnLifecycleForTests(); + resetTurnIntentDispatchLifecycleForTests(); vi.useRealTimers(); }); @@ -35,6 +44,17 @@ describe("turnLifecycle", () => { expect(getLastTurnTerminal(SESSION)).toBeNull(); }); + it("clears recovered intent aliases even before lifecycle state exists", () => { + publishTurnIntentDispatch("recovered-intent", { + sessionId: SESSION, + generation: 7, + }); + + clearTurnLifecycleSession(SESSION); + + expect(getTurnIntentDispatch("recovered-intent")).toBeUndefined(); + }); + it("beginTurnDispatch reserves synchronously and bumps generation", () => { const generation = beginTurnDispatch(SESSION); expect(generation).toBe(1); @@ -82,6 +102,17 @@ describe("turnLifecycle", () => { expect(getTurnPhase(SESSION)).toBe("working"); }); + it("ignores a confirmation for an older generation", () => { + const staleGeneration = beginTurnDispatch(SESSION); + const currentGeneration = beginTurnDispatch(SESSION); + + confirmTurnRunning(SESSION, { generation: staleGeneration }); + expect(getTurnPhase(SESSION)).toBe("dispatching"); + + confirmTurnRunning(SESSION, { generation: currentGeneration }); + expect(getTurnPhase(SESSION)).toBe("working"); + }); + it("beginTurnStopping is a no-op when idle", () => { beginTurnStopping(SESSION); expect(getTurnPhase(SESSION)).toBe("idle"); @@ -96,7 +127,7 @@ describe("turnLifecycle", () => { expect(getLastTurnTerminal(SESSION)?.status).toBe("cancelled"); }); - it("discards a terminal with a stale generation", () => { + it("keeps a stale exact terminal from changing the newer phase", () => { const staleGeneration = beginTurnDispatch(SESSION); markTurnRunning(SESSION); markTurnTerminal(SESSION, "completed"); @@ -113,6 +144,65 @@ describe("turnLifecycle", () => { expect(getTurnPhase(SESSION)).toBe("idle"); }); + it("records a delayed first terminal for an older generation", () => { + const staleGeneration = beginTurnDispatch(SESSION); + const currentGeneration = beginTurnDispatch(SESSION); + + markTurnTerminal(SESSION, "failed", { generation: staleGeneration }); + + expect(getTurnTerminal(SESSION, staleGeneration)).toMatchObject({ + generation: staleGeneration, + status: "failed", + }); + expect(getTurnPhase(SESSION)).toBe("dispatching"); + expect(getTurnGeneration(SESSION)).toBe(currentGeneration); + }); + + it("does not move the last-terminal watermark backwards", () => { + const oldGeneration = beginTurnDispatch(SESSION); + const currentGeneration = beginTurnDispatch(SESSION); + markTurnTerminal(SESSION, "completed", { generation: currentGeneration }); + + markTurnTerminal(SESSION, "failed", { generation: oldGeneration }); + + expect(getTurnTerminal(SESSION, oldGeneration)?.status).toBe("failed"); + expect(getLastTurnTerminal(SESSION)).toMatchObject({ + generation: currentGeneration, + status: "completed", + }); + }); + + it("retains exact terminals by generation when later turns finish", () => { + const firstGeneration = beginTurnDispatch(SESSION); + markTurnTerminal(SESSION, "completed", { generation: firstGeneration }); + const secondGeneration = beginTurnDispatch(SESSION); + markTurnTerminal(SESSION, "cancelled", { generation: secondGeneration }); + + expect(getTurnTerminal(SESSION, firstGeneration)?.status).toBe("completed"); + expect(getTurnTerminal(SESSION, secondGeneration)?.status).toBe( + "cancelled" + ); + }); + + it("keeps the first terminal final for one exact generation", () => { + const generation = beginTurnDispatch(SESSION); + markTurnTerminal(SESSION, "completed", { generation }); + markTurnTerminal(SESSION, "failed", { generation }); + + expect(getTurnTerminal(SESSION, generation)?.status).toBe("completed"); + expect(getLastTurnTerminal(SESSION)?.status).toBe("completed"); + }); + + it("does not reopen a terminal generation from a late exact running signal", () => { + const generation = beginTurnDispatch(SESSION); + markTurnTerminal(SESSION, "failed", { generation }); + + markTurnRunning(SESSION, { generation }); + + expect(getTurnPhase(SESSION)).toBe("idle"); + expect(getTurnGeneration(SESSION)).toBe(generation); + }); + it("discards an unattributed terminal while dispatching", () => { beginTurnDispatch(SESSION); markTurnTerminal(SESSION, "completed"); @@ -138,7 +228,7 @@ describe("turnLifecycle", () => { expect(getLastTurnTerminal(SESSION)?.status).toBe("completed"); }); - it("forceTurnIdle unlocks immediately and invalidates in-flight terminals", () => { + it("forceTurnIdle unlocks immediately and fences in-flight terminals", () => { beginTurnDispatch(SESSION); markTurnRunning(SESSION); const overriddenGeneration = getTurnGeneration(SESSION); @@ -147,24 +237,34 @@ describe("turnLifecycle", () => { expect(getTurnPhase(SESSION)).toBe("idle"); expect(getTurnGeneration(SESSION)).toBe(overriddenGeneration + 1); - // The overridden turn's late terminal is discarded by generation. + // The overridden turn's late exact terminal remains queryable, but cannot + // mutate the fenced idle generation. markTurnTerminal(SESSION, "cancelled", { generation: overriddenGeneration, }); - expect(getLastTurnTerminal(SESSION)).toBeNull(); + expect(getTurnTerminal(SESSION, overriddenGeneration)?.status).toBe( + "cancelled" + ); + expect(getTurnPhase(SESSION)).toBe("idle"); + expect(getTurnGeneration(SESSION)).toBe(overriddenGeneration + 1); }); - it("dead-man: a dispatch that never gets a running ack unlocks eventually", () => { - beginTurnDispatch(SESSION); + it("dead-man: a dispatch that never gets a running ack records failed and fences it", () => { + const generation = beginTurnDispatch(SESSION); vi.advanceTimersByTime(60_000); expect(getTurnPhase(SESSION)).toBe("idle"); + expect(getTurnTerminal(SESSION, generation)?.status).toBe("failed"); + expect(getTurnGeneration(SESSION)).toBe(generation + 1); }); - it("dead-man: a stop that never gets a terminal unlocks after the stop bound", () => { + it("dead-man: a stop that never gets a terminal records cancelled and fences it", () => { markTurnRunning(SESSION); + const generation = getTurnGeneration(SESSION); beginTurnStopping(SESSION); vi.advanceTimersByTime(10_000); expect(getTurnPhase(SESSION)).toBe("idle"); + expect(getTurnTerminal(SESSION, generation)?.status).toBe("cancelled"); + expect(getTurnGeneration(SESSION)).toBe(generation + 1); }); it("dead-man does not fire after the phase already resolved", () => { diff --git a/src/engines/SessionCore/control/turnIntentDispatchLifecycle.ts b/src/engines/SessionCore/control/turnIntentDispatchLifecycle.ts index 44f914bb20..e116a34f12 100644 --- a/src/engines/SessionCore/control/turnIntentDispatchLifecycle.ts +++ b/src/engines/SessionCore/control/turnIntentDispatchLifecycle.ts @@ -12,10 +12,33 @@ export interface TurnIntentDispatch { } const MAX_RECENT_DISPATCHES = 200; +/** + * Active identities are never subject to the recent-history bound. A global + * LRU may see hundreds of short turns while one tool-heavy turn is still + * running; evicting that live mapping would make its exact terminal look + * unattributed and leave the session permanently working. + */ +const activeDispatches = new Map(); +const activeIntentsByTurn = new Map>(); const recentDispatches = new Map(); const waiters = new Map void>>(); -export function publishTurnIntentDispatch( +function turnKey(dispatch: TurnIntentDispatch): string { + return JSON.stringify([dispatch.sessionId, dispatch.generation]); +} + +function removeActiveIntent(turnIntentId: string): TurnIntentDispatch | null { + const dispatch = activeDispatches.get(turnIntentId); + if (!dispatch) return null; + activeDispatches.delete(turnIntentId); + const key = turnKey(dispatch); + const intents = activeIntentsByTurn.get(key); + intents?.delete(turnIntentId); + if (intents?.size === 0) activeIntentsByTurn.delete(key); + return dispatch; +} + +function retainRecent( turnIntentId: string, dispatch: TurnIntentDispatch ): void { @@ -26,18 +49,55 @@ export function publishTurnIntentDispatch( if (!oldest) break; recentDispatches.delete(oldest); } +} + +export function publishTurnIntentDispatch( + turnIntentId: string, + dispatch: TurnIntentDispatch +): void { + removeActiveIntent(turnIntentId); + recentDispatches.delete(turnIntentId); + activeDispatches.set(turnIntentId, dispatch); + const key = turnKey(dispatch); + const intents = activeIntentsByTurn.get(key) ?? new Set(); + intents.add(turnIntentId); + activeIntentsByTurn.set(key, intents); const listeners = waiters.get(turnIntentId); if (!listeners) return; waiters.delete(turnIntentId); for (const listener of listeners) listener(dispatch); } +/** + * Bind a backend-selected intent id to an already-reserved local generation. + * + * Project/Work Item dispatch may replace the composer intent with a durable + * run id. Both ids must resolve to the same generation, but an id that is + * already bound anywhere else is an attribution conflict and must never be + * overwritten. + */ +export function publishTurnIntentDispatchAlias( + turnIntentId: string, + dispatch: TurnIntentDispatch +): boolean { + if (!turnIntentId) return false; + const existing = getTurnIntentDispatch(turnIntentId); + if (existing) { + return ( + existing.sessionId === dispatch.sessionId && + existing.generation === dispatch.generation + ); + } + publishTurnIntentDispatch(turnIntentId, dispatch); + return true; +} + export function waitForTurnIntentDispatch( turnIntentId: string, deadlineMs: number ): Promise { - const recent = recentDispatches.get(turnIntentId); - if (recent) return Promise.resolve(recent); + const known = getTurnIntentDispatch(turnIntentId); + if (known) return Promise.resolve(known); return new Promise((resolve, reject) => { const remainingMs = deadlineMs - Date.now(); if (remainingMs <= 0) { @@ -63,10 +123,44 @@ export function waitForTurnIntentDispatch( export function getTurnIntentDispatch( turnIntentId: string ): TurnIntentDispatch | undefined { - return recentDispatches.get(turnIntentId); + return ( + activeDispatches.get(turnIntentId) ?? recentDispatches.get(turnIntentId) + ); +} + +/** Whether this exact local turn still owns at least one live intent id. */ +export function hasActiveTurnIntentDispatch( + sessionId: string, + generation: number +): boolean { + return activeIntentsByTurn.has(turnKey({ sessionId, generation })); +} + +/** Move every identity for one finalized generation into bounded history. */ +export function retireTurnIntentDispatch( + sessionId: string, + generation: number +): void { + const key = turnKey({ sessionId, generation }); + const intents = activeIntentsByTurn.get(key); + if (!intents) return; + for (const turnIntentId of [...intents]) { + const dispatch = removeActiveIntent(turnIntentId); + if (dispatch) retainRecent(turnIntentId, dispatch); + } +} + +/** Session deletion invalidates all live mappings owned by that session. */ +export function retireSessionTurnIntentDispatches(sessionId: string): void { + for (const [turnIntentId, dispatch] of [...activeDispatches]) { + if (dispatch.sessionId !== sessionId) continue; + removeActiveIntent(turnIntentId); + } } export function resetTurnIntentDispatchLifecycleForTests(): void { + activeDispatches.clear(); + activeIntentsByTurn.clear(); recentDispatches.clear(); waiters.clear(); } diff --git a/src/engines/SessionCore/control/turnLifecycle.ts b/src/engines/SessionCore/control/turnLifecycle.ts index c9bb30c909..3a083f40ea 100644 --- a/src/engines/SessionCore/control/turnLifecycle.ts +++ b/src/engines/SessionCore/control/turnLifecycle.ts @@ -25,9 +25,9 @@ * Invariants: * - Every dispatch bumps `generation` synchronously (the reserve), so two * concurrent submits can never both see "idle". - * - A terminal carrying a generation that does not match the current one is - * discarded — a late terminal from an old turn can never release the - * queue for a newer turn. + * - A terminal carrying an older exact generation is retained for that + * generation's observers, but can never release the queue or mutate the + * phase of a newer turn. * - A terminal without a generation is discarded while "dispatching" * (before the running ack, any unattributed terminal is by definition * from an older turn). @@ -43,6 +43,11 @@ import { isStoreInitialized, } from "@src/util/core/state/instrumentedStore"; +import { + retireSessionTurnIntentDispatches, + retireTurnIntentDispatch, +} from "./turnIntentDispatchLifecycle"; + const log = createLogger("turnLifecycle"); export type TurnPhase = "idle" | "dispatching" | "working" | "stopping"; @@ -73,9 +78,20 @@ interface SessionTurnState { status: TurnTerminalStatus; at: number; } | null; + /** + * Exact terminals retained by generation. A single `lastTerminal` slot is + * insufficient when a later turn finishes before an observer of the prior + * generation attaches (or while that observer is still scheduled). + */ + terminalsByGeneration: Map< + number, + { generation: number; status: TurnTerminalStatus; at: number } + >; deadmanTimer: ReturnType | null; } +const MAX_RETAINED_TERMINALS_PER_SESSION = 32; + /** * If a dispatch never receives a running ack (backend hung before accepting * the turn), unlock after this bound instead of blocking the composer forever. @@ -102,6 +118,7 @@ function getState(sessionId: string): SessionTurnState { phase: "idle", generation: 0, lastTerminal: null, + terminalsByGeneration: new Map(), deadmanTimer: null, }; stateBySession.set(sessionId, state); @@ -142,7 +159,11 @@ function armDeadman( `[turnLifecycle] dead-man: session ${sessionId} stuck in "${phase}" for ` + `${timeoutMs}ms (generation ${armedGeneration}) — forcing idle` ); - forceTurnIdle(sessionId); + forceTurnIdleFromDeadman( + sessionId, + current, + phase === "stopping" ? "cancelled" : "failed" + ); }, timeoutMs); } @@ -162,6 +183,41 @@ function transition( bumpSignal(); } +function recordTurnTerminal( + sessionId: string, + state: SessionTurnState, + generation: number, + status: TurnTerminalStatus +): boolean { + // Terminal finality is monotonic for one exact generation. In particular, + // an IPC response-loss catch must not overwrite a completed provider + // terminal with a synthetic failed terminal for the same dispatch. + if (state.terminalsByGeneration.has(generation)) return false; + retireTurnIntentDispatch(sessionId, generation); + const terminal = { + generation, + status, + at: Date.now(), + }; + // `lastTerminal` is a generation watermark used by follow-up observers. + // A delayed exact terminal for an older turn must remain queryable by its + // generation without moving that watermark backwards. + if (!state.lastTerminal || generation >= state.lastTerminal.generation) { + state.lastTerminal = terminal; + } + state.terminalsByGeneration.set(generation, terminal); + while ( + state.terminalsByGeneration.size > MAX_RETAINED_TERMINALS_PER_SESSION + ) { + const oldestGeneration = state.terminalsByGeneration.keys().next().value as + | number + | undefined; + if (oldestGeneration === undefined) break; + state.terminalsByGeneration.delete(oldestGeneration); + } + return true; +} + /** * Synchronous reserve for a user-initiated dispatch. MUST be called before * the first `await` on every dispatch path so a concurrent submit observes @@ -195,6 +251,12 @@ export function markTurnRunning( ) { return; } + if ( + options.generation !== undefined && + state.terminalsByGeneration.has(options.generation) + ) { + return; + } if (state.phase === "working" || state.phase === "stopping") return; if (state.phase === "idle") { state.generation += 1; @@ -207,8 +269,17 @@ export function markTurnRunning( * never opens a turn from idle. Use for low-trust activity signals (raw * event traffic) that may trail a terminal. */ -export function confirmTurnRunning(sessionId: string): void { +export function confirmTurnRunning( + sessionId: string, + options: { generation?: number } = {} +): void { const state = getState(sessionId); + if ( + options.generation !== undefined && + options.generation !== state.generation + ) { + return; + } if (state.phase !== "dispatching") return; transition(sessionId, state, "working"); } @@ -228,7 +299,8 @@ export function beginTurnStopping(sessionId: string): void { * Provider delivered a turn-final terminal. This is the ONLY natural way a * turn ends. * - * - `generation` provided and stale → discarded (late terminal of old turn). + * - `generation` provided and stale → recorded for that exact generation, + * but never allowed to change the newer generation's phase. * - No `generation` while "dispatching" → discarded (an unattributed * terminal arriving before the running ack belongs to an older turn). */ @@ -240,8 +312,12 @@ export function markTurnTerminal( const state = getState(sessionId); if ( options.generation !== undefined && - options.generation !== state.generation + options.generation > state.generation ) { + log.warn( + `[turnLifecycle] discarding future-generation "${status}" terminal for ` + + `session ${sessionId} (signal ${options.generation}, current ${state.generation})` + ); return; } if (state.phase === "dispatching" && options.generation === undefined) { @@ -251,14 +327,36 @@ export function markTurnTerminal( ); return; } - state.lastTerminal = { - generation: state.generation, - status, - at: Date.now(), - }; + const generation = options.generation ?? state.generation; + if (!recordTurnTerminal(sessionId, state, generation, status)) return; + if (generation !== state.generation) { + // Exact finality belongs to an older reservation. Wake exact-generation + // observers and retire its identities, but preserve every bit of the + // newer turn's phase/timer state. + bumpSignal(); + return; + } + if (state.phase !== "idle") { + transition(sessionId, state, "idle"); + } else { + bumpSignal(); + } +} + +function forceTurnIdleFromDeadman( + sessionId: string, + state: SessionTurnState, + displacedStatus: TurnTerminalStatus +): void { + const displacedGeneration = state.generation; + recordTurnTerminal(sessionId, state, displacedGeneration, displacedStatus); + // Advance the fence after recording the displaced generation so a delayed + // provider signal cannot become the terminal of whatever starts next. + state.generation += 1; if (state.phase !== "idle") { transition(sessionId, state, "idle"); } else { + clearDeadman(state); bumpSignal(); } } @@ -266,10 +364,12 @@ export function markTurnTerminal( /** * Explicit boundary override: rewind boundaries and bounded fallbacks force * the session idle without a provider terminal. The generation is bumped so - * any in-flight terminal of the overridden turn is discarded when it lands. + * any in-flight terminal of the overridden turn cannot mutate the new phase + * when it lands (its exact finality is still retained by generation). */ export function forceTurnIdle(sessionId: string): void { const state = getState(sessionId); + retireTurnIntentDispatch(sessionId, state.generation); state.generation += 1; if (state.phase !== "idle") { transition(sessionId, state, "idle"); @@ -297,8 +397,22 @@ export function getLastTurnTerminal( return stateBySession.get(sessionId)?.lastTerminal ?? null; } +/** Read the immutable terminal for one exact reserved generation. */ +export function getTurnTerminal( + sessionId: string, + generation: number +): { generation: number; status: TurnTerminalStatus; at: number } | null { + return ( + stateBySession.get(sessionId)?.terminalsByGeneration.get(generation) ?? null + ); +} + /** Release all retained lifecycle state when a session is permanently removed. */ export function clearTurnLifecycleSession(sessionId: string): void { + // Intent aliases can be published before any lifecycle state is observed + // (for example, a recovered receipt during cold startup). Always clear + // those identities even when this process has no SessionTurnState yet. + retireSessionTurnIntentDispatches(sessionId); const state = stateBySession.get(sessionId); if (!state) return; clearDeadman(state); diff --git a/src/engines/SessionCore/services/SessionService.ts b/src/engines/SessionCore/services/SessionService.ts index 763a16260d..e3b3d63292 100644 --- a/src/engines/SessionCore/services/SessionService.ts +++ b/src/engines/SessionCore/services/SessionService.ts @@ -26,7 +26,10 @@ import { } from "@src/api/tauri/agent"; import { rpc } from "@src/api/tauri/rpc"; import { ROUTES } from "@src/config/routes"; -import { getAdapterForSession } from "@src/engines/SessionCore/sync/types"; +import { + type AdapterSendReceipt, + getAdapterForSession, +} from "@src/engines/SessionCore/sync/types"; import { buildPendingForkHandoff, markForkHandoffConsumed, @@ -288,7 +291,9 @@ export const SessionService = { * keeps adding new IDEs (Trae, Windsurf, ...) to a single new file * under `sync/adapters/` instead of patching the service. */ - async sendMessage(params: SessionSendMessageParams): Promise { + async sendMessage( + params: SessionSendMessageParams + ): Promise { const { sessionId, content, @@ -296,6 +301,7 @@ export const SessionService = { model, accountId, mode, + workspacePath, imageDataUrls, isResume, clientMessageId, @@ -303,13 +309,13 @@ export const SessionService = { turnIntentSource, directUserIntent, } = params; - // Gate ADE context on the session row's persisted repo so a session - // on repo A doesn't ship repo B's editor / git / LSP state when the - // toolbar happens to point elsewhere. Legacy rows with no - // `repoPath` fall through to the unconstrained path. + // Gate ADE context on an explicit trusted execution workspace when one + // is supplied, otherwise on the session row's persisted repo. const sessionRow = getInstrumentedStore().get(sessionByIdAtom(sessionId)); + const executionWorkspacePath = + workspacePath ?? sessionRow?.repoPath ?? null; const adeContext = collectAdeContext({ - expectedRepoPath: sessionRow?.repoPath ?? null, + expectedRepoPath: executionWorkspacePath, sessionId, }); const adapter = getAdapterForSession(sessionId); @@ -348,7 +354,7 @@ export const SessionService = { } try { - await adapter.sendMessage({ + const receipt = await adapter.sendMessage({ sessionId, content: effectiveContent, displayText: effectiveDisplayText, @@ -362,7 +368,7 @@ export const SessionService = { turnIntentSource, directUserIntent, adeContext, - sessionRepoPath: sessionRow?.repoPath ?? null, + sessionRepoPath: executionWorkspacePath, }); if (forkHandoffArmed) { markForkHandoffConsumed(sessionId); @@ -373,11 +379,20 @@ export const SessionService = { // which simply overwrites this local stamp — no drift. markSessionActive(sessionId); logger.info(`Sent message to ${adapter.category} session: ${sessionId}`); + return receipt; } catch (error) { throwServiceError(`Failed to send message to ${sessionId}`, error); } }, + /** Durable receipt used to disambiguate a lost send IPC response. */ + async getTurnIntentStatus(sessionId: string, turnIntentId: string) { + return rpc.agentSession.getTurnIntentStatus({ + sessionId, + turnIntentId, + }); + }, + // ========================================== // Questions // ========================================== diff --git a/src/engines/SessionCore/services/TurnDispatchService.test.ts b/src/engines/SessionCore/services/TurnDispatchService.test.ts new file mode 100644 index 0000000000..e913c4ea48 --- /dev/null +++ b/src/engines/SessionCore/services/TurnDispatchService.test.ts @@ -0,0 +1,1029 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { clearRecentOptimisticTurn } from "@src/engines/SessionCore/control/optimisticTurnStatus"; +import { + getTurnIntentDispatch, + publishTurnIntentDispatch, + resetTurnIntentDispatchLifecycleForTests, +} from "@src/engines/SessionCore/control/turnIntentDispatchLifecycle"; +import { + getLastTurnTerminal, + getTurnPhase, + getTurnTerminal, + markTurnTerminal, + resetTurnLifecycleForTests, +} from "@src/engines/SessionCore/control/turnLifecycle"; +import { + sessionRuntimeStatusAtom, + setSessionRuntimeStatusAtom, +} from "@src/store/session/cliSessionStatusAtom"; +import { + createInstrumentedStore, + getInstrumentedStore, +} from "@src/util/core/state/instrumentedStore"; + +import { + dispatchTurn, + failReservedTurn, + reserveTurnDispatch, + resetTurnDispatchMonitorsForTests, + sendReservedTurn, + waitForTurnOutcome, +} from "./TurnDispatchService"; + +createInstrumentedStore(); + +const mocks = vi.hoisted(() => ({ + getTurnIntentStatus: vi.fn(), + markSessionActive: vi.fn(), + sendMessage: vi.fn(), +})); + +vi.mock("@src/store/session", () => ({ + markSessionActive: mocks.markSessionActive, +})); + +vi.mock("./SessionService", () => ({ + SessionService: { + getTurnIntentStatus: mocks.getTurnIntentStatus, + sendMessage: mocks.sendMessage, + }, +})); + +const SESSION = "sdeagent-session-1"; + +describe("TurnDispatchService", () => { + beforeEach(() => { + resetTurnDispatchMonitorsForTests(); + resetTurnIntentDispatchLifecycleForTests(); + resetTurnLifecycleForTests(); + mocks.markSessionActive.mockReset(); + mocks.getTurnIntentStatus.mockReset().mockResolvedValue(null); + mocks.sendMessage.mockReset().mockResolvedValue({ duplicate: false }); + getInstrumentedStore().set(setSessionRuntimeStatusAtom, { + sessionId: SESSION, + status: "idle", + source: "dispatch", + }); + }); + + afterEach(() => { + resetTurnDispatchMonitorsForTests(); + clearRecentOptimisticTurn(SESSION); + clearRecentOptimisticTurn("cursoride-session-1"); + resetTurnLifecycleForTests(); + }); + + it("reserves the generation and publishes the intent synchronously", () => { + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-1", + }); + + expect(dispatch).toMatchObject({ + sessionId: SESSION, + turnIntentId: "intent-1", + generation: 1, + }); + expect(getTurnPhase(SESSION)).toBe("dispatching"); + expect(getTurnIntentDispatch("intent-1")).toEqual({ + sessionId: SESSION, + generation: 1, + }); + }); + + it("sends the reserved identity and confirms running after acceptance", async () => { + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-2", + }); + + await expect( + sendReservedTurn({ + dispatch, + content: "hello", + turnIntentSource: "user_submit", + }) + ).resolves.toMatchObject({ accepted: true, turnIntentId: "intent-2" }); + expect(mocks.sendMessage).toHaveBeenCalledWith( + expect.objectContaining({ + sessionId: SESSION, + content: "hello", + turnIntentId: "intent-2", + clientMessageId: "intent-2", + }) + ); + expect(mocks.markSessionActive).toHaveBeenCalledWith(SESSION); + expect(getTurnPhase(SESSION)).toBe("working"); + }); + + it("reserves and forwards headless dispatch options through one call", async () => { + const accepted = await dispatchTurn({ + sessionId: SESSION, + content: "execute plan", + workspacePath: "/workspace/repo-a", + turnIntentSource: "user_submit", + }); + + expect(accepted).toMatchObject({ + accepted: true, + sessionId: SESSION, + generation: 1, + }); + expect(accepted.turnIntentId).toEqual(expect.any(String)); + expect(mocks.sendMessage).toHaveBeenCalledWith( + expect.objectContaining({ + sessionId: SESSION, + content: "execute plan", + workspacePath: "/workspace/repo-a", + turnIntentId: accepted.turnIntentId, + }) + ); + expect(getTurnIntentDispatch(accepted.turnIntentId)).toEqual({ + sessionId: SESSION, + generation: accepted.generation, + }); + }); + + it("closes the exact reservation when transport rejects", async () => { + mocks.sendMessage.mockRejectedValueOnce(new Error("transport down")); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-3", + }); + + await expect( + sendReservedTurn({ + dispatch, + content: "hello", + turnIntentSource: "user_submit", + }) + ).rejects.toThrow("transport down"); + expect(getTurnPhase(SESSION)).toBe("idle"); + expect(getLastTurnTerminal(SESSION)).toMatchObject({ + generation: dispatch.generation, + status: "failed", + }); + }); + + it("does not reopen a fast terminal that arrives before send resolves", async () => { + let resolveSend!: (receipt: { duplicate: boolean }) => void; + mocks.sendMessage.mockReturnValueOnce( + new Promise<{ duplicate: boolean }>((resolve) => { + resolveSend = resolve; + }) + ); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-fast", + }); + const accepted = sendReservedTurn({ + dispatch, + content: "hello", + turnIntentSource: "user_submit", + }); + + markTurnTerminal(SESSION, "completed", { + generation: dispatch.generation, + }); + resolveSend({ duplicate: false }); + await accepted; + + expect(getTurnPhase(SESSION)).toBe("idle"); + expect(getLastTurnTerminal(SESSION)).toMatchObject({ + generation: dispatch.generation, + status: "completed", + }); + }); + + it("does not let an old send acknowledgement promote a newer generation", async () => { + let resolveFirst!: (receipt: { duplicate: boolean }) => void; + mocks.sendMessage.mockReturnValueOnce( + new Promise<{ duplicate: boolean }>((resolve) => { + resolveFirst = resolve; + }) + ); + const first = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-old-ack", + }); + const firstSend = sendReservedTurn({ + dispatch: first, + content: "first", + turnIntentSource: "user_submit", + }); + const second = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-current", + }); + + resolveFirst({ duplicate: false }); + await firstSend; + + expect(second.generation).toBe(first.generation + 1); + expect(getTurnPhase(SESSION)).toBe("dispatching"); + }); + + it("does not let an old terminal receipt clear a newer optimistic mirror", async () => { + let resolveFirst!: (receipt: { duplicate: boolean }) => void; + mocks.sendMessage.mockReturnValueOnce( + new Promise<{ duplicate: boolean }>((resolve) => { + resolveFirst = resolve; + }) + ); + mocks.getTurnIntentStatus.mockResolvedValueOnce({ + status: "completed", + effectiveTurnIntentId: "intent-old-terminal-receipt", + }); + const first = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-old-terminal-receipt", + }); + const firstSend = sendReservedTurn({ + dispatch: first, + content: "first", + turnIntentSource: "user_submit", + }); + const firstOutcome = waitForTurnOutcome(first, Date.now() + 1_000); + const newer = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-newer-optimistic", + }); + await sendReservedTurn({ + dispatch: newer, + content: "newer", + turnIntentSource: "user_submit", + }); + + resolveFirst({ duplicate: true }); + await firstSend; + await expect(firstOutcome).resolves.toMatchObject({ status: "completed" }); + + expect(getTurnPhase(SESSION)).toBe("working"); + expect(getInstrumentedStore().get(sessionRuntimeStatusAtom)).toBe( + "running" + ); + }); + + it("records a delayed older pre-transport failure while a newer turn stays working", async () => { + const older = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-old-pretransport-failure", + }); + const olderOutcome = waitForTurnOutcome(older, Date.now() + 1_000); + const newer = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-newer-working", + }); + await sendReservedTurn({ + dispatch: newer, + content: "newer", + turnIntentSource: "user_submit", + }); + + failReservedTurn(older); + + await expect(olderOutcome).resolves.toMatchObject({ status: "failed" }); + expect(getTurnPhase(SESSION)).toBe("working"); + }); + + it("does not let an old rejected send roll back a newer optimistic mirror", async () => { + let rejectFirst!: (error: Error) => void; + mocks.sendMessage.mockReturnValueOnce( + new Promise<{ duplicate: boolean }>((_resolve, reject) => { + rejectFirst = reject; + }) + ); + const first = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-old-rejection", + }); + const firstSend = sendReservedTurn({ + dispatch: first, + content: "first", + turnIntentSource: "user_submit", + }); + reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-newer-after-rejection", + }); + + rejectFirst(new Error("old IPC rejection")); + await expect(firstSend).rejects.toThrow("old IPC rejection"); + + expect(getTurnPhase(SESSION)).toBe("dispatching"); + expect(getInstrumentedStore().get(sessionRuntimeStatusAtom)).toBe( + "running" + ); + }); + + it("does not overwrite a fast exact terminal when the send response is lost", async () => { + let rejectSend!: (error: Error) => void; + mocks.sendMessage.mockReturnValueOnce( + new Promise((_resolve, reject) => { + rejectSend = reject; + }) + ); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-response-lost", + }); + const sent = sendReservedTurn({ + dispatch, + content: "hello", + turnIntentSource: "user_submit", + }); + + markTurnTerminal(SESSION, "completed", { + generation: dispatch.generation, + }); + rejectSend(new Error("IPC response lost")); + await expect(sent).rejects.toThrow("IPC response lost"); + + expect(getLastTurnTerminal(SESSION)).toMatchObject({ + generation: dispatch.generation, + status: "completed", + }); + }); + + it.each(["queued", "running"])( + "settles a response-loss steering intent after durable %s becomes completed", + async (receiptStatus) => { + vi.useFakeTimers(); + try { + const turnIntentId = `intent-${receiptStatus}-steering-response-loss`; + mocks.sendMessage.mockRejectedValueOnce(new Error("IPC response lost")); + mocks.getTurnIntentStatus + .mockResolvedValueOnce({ + status: receiptStatus, + effectiveTurnIntentId: turnIntentId, + }) + .mockResolvedValueOnce({ + status: "completed", + effectiveTurnIntentId: turnIntentId, + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId, + }); + + await expect( + sendReservedTurn({ + dispatch, + content: "steer existing turn", + turnIntentSource: "user_submit", + }) + ).resolves.toMatchObject({ + accepted: true, + turnIntentId, + }); + const outcome = waitForTurnOutcome(dispatch, Date.now() + 1_000); + expect(getTurnPhase(SESSION)).toBe("working"); + // One monitor timer plus waitForTurnOutcome's deadline. + expect(vi.getTimerCount()).toBe(2); + + await vi.advanceTimersByTimeAsync(100); + + await expect(outcome).resolves.toMatchObject({ status: "completed" }); + expect(getTurnPhase(SESSION)).toBe("idle"); + expect(getInstrumentedStore().get(sessionRuntimeStatusAtom)).toBe( + "running" + ); + expect(vi.getTimerCount()).toBe(0); + } finally { + resetTurnDispatchMonitorsForTests(); + vi.useRealTimers(); + } + } + ); + + it.each(["queued", "running"])( + "settles a duplicate steering receipt after durable %s becomes completed", + async (receiptStatus) => { + vi.useFakeTimers(); + try { + const turnIntentId = `intent-${receiptStatus}-steering-duplicate`; + mocks.sendMessage.mockResolvedValueOnce({ + duplicate: true, + turnIntentStatus: receiptStatus, + effectiveTurnIntentId: turnIntentId, + }); + mocks.getTurnIntentStatus.mockResolvedValueOnce({ + status: "completed", + effectiveTurnIntentId: turnIntentId, + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId, + }); + + await sendReservedTurn({ + dispatch, + content: "duplicate steering retry", + turnIntentSource: "user_submit", + }); + const outcome = waitForTurnOutcome(dispatch, Date.now() + 1_000); + expect(mocks.getTurnIntentStatus).not.toHaveBeenCalled(); + // One monitor timer plus waitForTurnOutcome's deadline. + expect(vi.getTimerCount()).toBe(2); + + await vi.advanceTimersByTimeAsync(100); + + await expect(outcome).resolves.toMatchObject({ status: "completed" }); + expect(getInstrumentedStore().get(sessionRuntimeStatusAtom)).toBe( + "running" + ); + expect(vi.getTimerCount()).toBe(0); + } finally { + resetTurnDispatchMonitorsForTests(); + vi.useRealTimers(); + } + } + ); + + it("does not poll an ordinary exact acknowledgement", async () => { + vi.useFakeTimers(); + try { + mocks.sendMessage.mockResolvedValueOnce({ + duplicate: false, + turnIntentStatus: "running", + effectiveTurnIntentId: "intent-ordinary-ack", + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-ordinary-ack", + }); + + await sendReservedTurn({ + dispatch, + content: "ordinary turn", + turnIntentSource: "user_submit", + }); + + expect(mocks.getTurnIntentStatus).not.toHaveBeenCalled(); + expect(getTurnPhase(SESSION)).toBe("working"); + expect(vi.getTimerCount()).toBe(0); + } finally { + resetTurnDispatchMonitorsForTests(); + vi.useRealTimers(); + } + }); + + it("stops an ambiguous exact-X monitor on a live terminal", async () => { + vi.useFakeTimers(); + try { + const turnIntentId = "intent-ambiguous-live-terminal"; + mocks.sendMessage.mockRejectedValueOnce(new Error("IPC response lost")); + mocks.getTurnIntentStatus.mockResolvedValue({ + status: "running", + effectiveTurnIntentId: turnIntentId, + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId, + }); + + await sendReservedTurn({ + dispatch, + content: "ambiguous turn", + turnIntentSource: "user_submit", + }); + const callsBeforeTerminal = mocks.getTurnIntentStatus.mock.calls.length; + expect(vi.getTimerCount()).toBe(1); + + markTurnTerminal(SESSION, "completed", { + generation: dispatch.generation, + }); + expect(vi.getTimerCount()).toBe(0); + await vi.advanceTimersByTimeAsync(60_000); + expect(mocks.getTurnIntentStatus).toHaveBeenCalledTimes( + callsBeforeTerminal + ); + } finally { + resetTurnDispatchMonitorsForTests(); + vi.useRealTimers(); + } + }); + + it("stops an ambiguous exact-X monitor when its generation is superseded", async () => { + vi.useFakeTimers(); + try { + const turnIntentId = "intent-ambiguous-superseded"; + mocks.sendMessage.mockRejectedValueOnce(new Error("IPC response lost")); + mocks.getTurnIntentStatus.mockResolvedValue({ + status: "queued", + effectiveTurnIntentId: turnIntentId, + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId, + }); + + await sendReservedTurn({ + dispatch, + content: "ambiguous turn", + turnIntentSource: "user_submit", + }); + const callsBeforeSupersession = + mocks.getTurnIntentStatus.mock.calls.length; + expect(vi.getTimerCount()).toBe(1); + + const newer = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-after-ambiguous", + }); + expect(newer.generation).toBe(dispatch.generation + 1); + // Only the newer dispatch dead-man remains; the old status monitor was + // cancelled synchronously by generation supersession. + expect(vi.getTimerCount()).toBe(1); + await vi.advanceTimersByTimeAsync(10_000); + expect(mocks.getTurnIntentStatus).toHaveBeenCalledTimes( + callsBeforeSupersession + ); + failReservedTurn(newer); + expect(vi.getTimerCount()).toBe(0); + } finally { + resetTurnDispatchMonitorsForTests(); + vi.useRealTimers(); + } + }); + + it.each([ + ["completed", "completed"], + ["failed", "failed"], + ["cancelled", "cancelled"], + ] as const)( + "projects a durable %s receipt onto the exact terminal", + async (receiptStatus, expectedTerminal) => { + mocks.sendMessage.mockRejectedValueOnce(new Error("IPC response lost")); + mocks.getTurnIntentStatus.mockResolvedValueOnce({ + status: receiptStatus, + effectiveTurnIntentId: `intent-${receiptStatus}-receipt`, + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: `intent-${receiptStatus}-receipt`, + }); + + await sendReservedTurn({ + dispatch, + content: "hello", + turnIntentSource: "user_submit", + }); + + await expect( + waitForTurnOutcome(dispatch, Date.now() + 1_000) + ).resolves.toMatchObject({ status: expectedTerminal }); + expect(getInstrumentedStore().get(sessionRuntimeStatusAtom)).toBe( + expectedTerminal + ); + } + ); + + it("reconciles a successful duplicate acknowledgement instead of assuming a terminal will arrive", async () => { + mocks.sendMessage.mockResolvedValueOnce({ duplicate: true }); + mocks.getTurnIntentStatus.mockResolvedValueOnce({ + status: "coalesced", + effectiveTurnIntentId: "intent-coalesced-receipt", + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-coalesced-receipt", + }); + + await expect( + sendReservedTurn({ + dispatch, + content: "hello", + turnIntentSource: "user_submit", + }) + ).resolves.toMatchObject({ accepted: true }); + await expect( + waitForTurnOutcome(dispatch, Date.now() + 1_000) + ).resolves.toMatchObject({ status: "failed" }); + expect(mocks.getTurnIntentStatus).toHaveBeenCalledWith( + SESSION, + "intent-coalesced-receipt" + ); + expect(getTurnPhase(SESSION)).toBe("idle"); + expect(getInstrumentedStore().get(sessionRuntimeStatusAtom)).toBe("failed"); + }); + + it("prefers an exact status carried by the duplicate acknowledgement", async () => { + mocks.sendMessage.mockResolvedValueOnce({ + duplicate: true, + turnIntentStatus: "completed", + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-acknowledged-terminal", + }); + + await sendReservedTurn({ + dispatch, + content: "hello", + turnIntentSource: "user_submit", + }); + + expect(mocks.getTurnIntentStatus).not.toHaveBeenCalled(); + await expect( + waitForTurnOutcome(dispatch, Date.now() + 1_000) + ).resolves.toMatchObject({ status: "completed" }); + }); + + it("settles a steered augmentation immediately without projecting the provider idle", async () => { + mocks.sendMessage.mockResolvedValueOnce({ + duplicate: false, + steered: true, + turnIntentStatus: "queued", + effectiveTurnIntentId: "intent-steered", + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-steered", + }); + + await sendReservedTurn({ + dispatch, + content: "adjust course", + turnIntentSource: "user_submit", + }); + + await expect( + waitForTurnOutcome(dispatch, Date.now() + 1_000) + ).resolves.toMatchObject({ status: "completed" }); + expect(getTurnPhase(SESSION)).toBe("idle"); + expect(getInstrumentedStore().get(sessionRuntimeStatusAtom)).toBe( + "running" + ); + expect(mocks.getTurnIntentStatus).not.toHaveBeenCalled(); + }); + + it("aliases a normal Project acknowledgement before reconciling its effective run", async () => { + mocks.sendMessage.mockResolvedValueOnce({ + duplicate: false, + turnIntentStatus: "queued", + effectiveTurnIntentId: "wir_effective-normal", + }); + mocks.getTurnIntentStatus.mockResolvedValueOnce({ + status: "running", + effectiveTurnIntentId: "wir_effective-normal", + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-project-normal", + }); + + await sendReservedTurn({ + dispatch, + content: "project task", + turnIntentSource: "user_submit", + }); + + expect(getTurnIntentDispatch("wir_effective-normal")).toEqual({ + sessionId: SESSION, + generation: dispatch.generation, + }); + expect(mocks.getTurnIntentStatus).toHaveBeenCalledWith( + SESSION, + "intent-project-normal" + ); + expect(getTurnPhase(SESSION)).toBe("working"); + }); + + it("polls durable Project ownership and records failure before runtime starts", async () => { + vi.useFakeTimers(); + try { + mocks.sendMessage.mockResolvedValueOnce({ + duplicate: false, + turnIntentStatus: "queued", + effectiveTurnIntentId: "wir_dead-letter-before-runtime", + }); + mocks.getTurnIntentStatus + .mockResolvedValueOnce({ + status: "queued", + effectiveTurnIntentId: "wir_dead-letter-before-runtime", + }) + .mockResolvedValueOnce({ + status: "failed", + effectiveTurnIntentId: "wir_dead-letter-before-runtime", + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-dead-letter-before-runtime", + }); + + await sendReservedTurn({ + dispatch, + content: "project task", + turnIntentSource: "user_submit", + }); + const outcome = waitForTurnOutcome(dispatch, Date.now() + 1_000); + expect(getTurnPhase(SESSION)).toBe("working"); + + await vi.advanceTimersByTimeAsync(100); + + await expect(outcome).resolves.toMatchObject({ status: "failed" }); + expect(getTurnPhase(SESSION)).toBe("idle"); + expect(getInstrumentedStore().get(sessionRuntimeStatusAtom)).toBe( + "failed" + ); + expect(vi.getTimerCount()).toBe(0); + } finally { + resetTurnDispatchMonitorsForTests(); + vi.useRealTimers(); + } + }); + + it("does not synthesize failure after more than the 20-minute claim horizon", async () => { + vi.useFakeTimers(); + try { + mocks.sendMessage.mockResolvedValueOnce({ + duplicate: false, + turnIntentStatus: "queued", + effectiveTurnIntentId: "wir_stuck-queued", + }); + mocks.getTurnIntentStatus.mockResolvedValue({ + status: "queued", + effectiveTurnIntentId: "wir_stuck-queued", + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-stuck-queued", + }); + + await sendReservedTurn({ + dispatch, + content: "project task", + turnIntentSource: "user_submit", + }); + const outcome = waitForTurnOutcome(dispatch, Date.now() + 22 * 60_000); + expect(getTurnPhase(SESSION)).toBe("working"); + + await vi.advanceTimersByTimeAsync(20 * 60_000 + 1); + + expect(getTurnPhase(SESSION)).toBe("working"); + expect(getTurnTerminal(SESSION, dispatch.generation)).toBeNull(); + expect(mocks.getTurnIntentStatus.mock.calls.length).toBeLessThanOrEqual( + 50 + ); + mocks.getTurnIntentStatus.mockResolvedValue({ + status: "failed", + effectiveTurnIntentId: "wir_stuck-queued", + }); + await vi.advanceTimersByTimeAsync(30_000); + await expect(outcome).resolves.toMatchObject({ status: "failed" }); + expect(vi.getTimerCount()).toBe(0); + } finally { + resetTurnDispatchMonitorsForTests(); + vi.useRealTimers(); + } + }); + + it("cancels the Project safety poll and lifecycle subscription on reset", async () => { + vi.useFakeTimers(); + try { + mocks.sendMessage.mockResolvedValueOnce({ + duplicate: false, + turnIntentStatus: "queued", + effectiveTurnIntentId: "wir_reset-monitor", + }); + mocks.getTurnIntentStatus.mockResolvedValue({ + status: "queued", + effectiveTurnIntentId: "wir_reset-monitor", + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-reset-monitor", + }); + + await sendReservedTurn({ + dispatch, + content: "project task", + turnIntentSource: "user_submit", + }); + const callsBeforeReset = mocks.getTurnIntentStatus.mock.calls.length; + expect(vi.getTimerCount()).toBe(1); + + resetTurnDispatchMonitorsForTests(); + expect(vi.getTimerCount()).toBe(0); + await vi.advanceTimersByTimeAsync(60 * 60_000); + expect(mocks.getTurnIntentStatus).toHaveBeenCalledTimes(callsBeforeReset); + } finally { + resetTurnDispatchMonitorsForTests(); + vi.useRealTimers(); + } + }); + + it("recovers a Project terminal that happened before its acknowledgement alias arrived", async () => { + let resolveSend!: (receipt: { + duplicate: boolean; + turnIntentStatus: string; + effectiveTurnIntentId: string; + }) => void; + mocks.sendMessage.mockReturnValueOnce( + new Promise((resolve) => { + resolveSend = resolve; + }) + ); + mocks.getTurnIntentStatus.mockResolvedValueOnce({ + status: "completed", + effectiveTurnIntentId: "wir_finished-before-ack", + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-before-alias", + }); + const sent = sendReservedTurn({ + dispatch, + content: "fast project task", + turnIntentSource: "user_submit", + }); + + expect(getTurnIntentDispatch("wir_finished-before-ack")).toBeUndefined(); + resolveSend({ + duplicate: false, + turnIntentStatus: "queued", + effectiveTurnIntentId: "wir_finished-before-ack", + }); + await sent; + + expect(getTurnIntentDispatch("wir_finished-before-ack")).toEqual({ + sessionId: SESSION, + generation: dispatch.generation, + }); + await expect( + waitForTurnOutcome(dispatch, Date.now() + 1_000) + ).resolves.toMatchObject({ status: "completed" }); + }); + + it("installs the effective alias from an ambiguous response-loss lookup", async () => { + vi.useFakeTimers(); + try { + mocks.sendMessage.mockRejectedValueOnce(new Error("IPC response lost")); + mocks.getTurnIntentStatus + .mockResolvedValueOnce({ + status: "running", + effectiveTurnIntentId: "wir_response-loss", + }) + .mockResolvedValueOnce({ + status: "completed", + effectiveTurnIntentId: "wir_response-loss", + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-response-loss-project", + }); + + await sendReservedTurn({ + dispatch, + content: "project task", + turnIntentSource: "user_submit", + }); + + expect(getTurnIntentDispatch("wir_response-loss")).toEqual({ + sessionId: SESSION, + generation: dispatch.generation, + }); + expect(getTurnPhase(SESSION)).toBe("working"); + const outcome = waitForTurnOutcome(dispatch, Date.now() + 1_000); + + await vi.advanceTimersByTimeAsync(100); + + await expect(outcome).resolves.toMatchObject({ status: "completed" }); + expect(getInstrumentedStore().get(sessionRuntimeStatusAtom)).toBe( + "completed" + ); + expect(vi.getTimerCount()).toBe(0); + } finally { + resetTurnDispatchMonitorsForTests(); + vi.useRealTimers(); + } + }); + + it("reconciles a duplicate effective run against the original generation", async () => { + mocks.sendMessage.mockResolvedValueOnce({ + duplicate: true, + turnIntentStatus: "completed", + effectiveTurnIntentId: "wir_duplicate-effective", + }); + mocks.getTurnIntentStatus.mockResolvedValueOnce({ + status: "completed", + effectiveTurnIntentId: "wir_duplicate-effective", + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-project-duplicate", + }); + + await sendReservedTurn({ + dispatch, + content: "retry project task", + turnIntentSource: "user_submit", + }); + + expect(getTurnIntentDispatch("wir_duplicate-effective")).toEqual({ + sessionId: SESSION, + generation: dispatch.generation, + }); + await expect( + waitForTurnOutcome(dispatch, Date.now() + 1_000) + ).resolves.toMatchObject({ status: "completed" }); + }); + + it("fails closed without overwriting a conflicting effective alias", async () => { + publishTurnIntentDispatch("wir_conflicting", { + sessionId: "sdeagent-other", + generation: 9, + }); + mocks.sendMessage.mockResolvedValueOnce({ + duplicate: false, + turnIntentStatus: "queued", + effectiveTurnIntentId: "wir_conflicting", + }); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-alias-conflict", + }); + + await expect( + sendReservedTurn({ + dispatch, + content: "project task", + turnIntentSource: "user_submit", + }) + ).rejects.toThrow(/conflicts/); + + expect(getTurnIntentDispatch("wir_conflicting")).toEqual({ + sessionId: "sdeagent-other", + generation: 9, + }); + await expect( + waitForTurnOutcome(dispatch, Date.now() + 1_000) + ).resolves.toMatchObject({ status: "failed" }); + }); + + it.each(["optimistic", "stale", "rejected", "future_status", null])( + "fails closed for a non-executable durable receipt %s", + async (receiptStatus) => { + mocks.sendMessage.mockRejectedValueOnce(new Error("IPC response lost")); + mocks.getTurnIntentStatus.mockResolvedValueOnce( + receiptStatus === null + ? null + : { + status: receiptStatus, + effectiveTurnIntentId: `intent-non-executable-${String(receiptStatus)}`, + } + ); + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: `intent-non-executable-${String(receiptStatus)}`, + }); + + await expect( + sendReservedTurn({ + dispatch, + content: "hello", + turnIntentSource: "user_submit", + }) + ).rejects.toThrow("IPC response lost"); + expect(getTurnPhase(SESSION)).toBe("idle"); + expect(getInstrumentedStore().get(sessionRuntimeStatusAtom)).toBe("idle"); + } + ); + + it("settles Cursor handoffs immediately because they have no terminal stream", async () => { + const dispatch = reserveTurnDispatch({ + sessionId: "cursoride-session-1", + turnIntentId: "intent-cursor", + }); + + await sendReservedTurn({ + dispatch, + content: "hello", + turnIntentSource: "user_submit", + }); + expect(getLastTurnTerminal("cursoride-session-1")).toMatchObject({ + generation: dispatch.generation, + status: "completed", + }); + }); + + it("waits for and returns the exact generation terminal", async () => { + const dispatch = reserveTurnDispatch({ + sessionId: SESSION, + turnIntentId: "intent-4", + }); + const outcomePromise = waitForTurnOutcome(dispatch, Date.now() + 1_000); + + markTurnTerminal(SESSION, "cancelled", { + generation: dispatch.generation, + }); + + await expect(outcomePromise).resolves.toMatchObject({ + turnIntentId: "intent-4", + generation: dispatch.generation, + status: "cancelled", + }); + }); +}); diff --git a/src/engines/SessionCore/services/TurnDispatchService.ts b/src/engines/SessionCore/services/TurnDispatchService.ts new file mode 100644 index 0000000000..012d60bbf6 --- /dev/null +++ b/src/engines/SessionCore/services/TurnDispatchService.ts @@ -0,0 +1,630 @@ +/** + * Canonical frontend turn dispatch/finality boundary. + * + * Feature callers reserve synchronously before their first await, then send + * through the shared transport path and optionally await the exact generation + * terminal. UI-specific transcript writes stay outside this service. + */ +import { + beginOptimisticTurn, + clearRecentOptimisticTurn, + failOptimisticTurn, +} from "@src/engines/SessionCore/control/optimisticTurnStatus"; +import { + publishTurnIntentDispatch, + publishTurnIntentDispatchAlias, + waitForTurnIntentDispatch, +} from "@src/engines/SessionCore/control/turnIntentDispatchLifecycle"; +import { + type TurnTerminalStatus, + beginTurnDispatch, + confirmTurnRunning, + getTurnGeneration, + getTurnTerminal, + markTurnTerminal, + turnLifecycleSignalAtom, +} from "@src/engines/SessionCore/control/turnLifecycle"; +import { mintTurnIntentId } from "@src/engines/SessionCore/sync/adapters/shared/eventFactories"; +import { createLogger } from "@src/hooks/logger"; +import { markSessionActive } from "@src/store/session"; +import { + type SessionRuntimeStatusSource, + setSessionRuntimeStatusAtom, +} from "@src/store/session/cliSessionStatusAtom"; +import { getInstrumentedStore } from "@src/util/core/state/instrumentedStore"; +import { isCursorIdeSession } from "@src/util/session/sessionDispatch"; + +import { SessionService } from "./SessionService"; +import type { SessionSendMessageParams } from "./types"; + +export interface ReservedTurnDispatch { + sessionId: string; + turnIntentId: string; + generation: number; + optimisticSource: SessionRuntimeStatusSource; +} + +export interface TurnDispatchAccepted extends ReservedTurnDispatch { + accepted: true; +} + +export interface TurnOutcome extends ReservedTurnDispatch { + status: TurnTerminalStatus; + at: number; +} + +export interface ReserveTurnDispatchInput { + sessionId: string; + turnIntentId?: string; + optimisticSource?: SessionRuntimeStatusSource; +} + +export type SendReservedTurnInput = Omit< + SessionSendMessageParams, + "sessionId" | "turnIntentId" +> & { + dispatch: ReservedTurnDispatch; +}; + +export type DispatchTurnInput = Omit & + ReserveTurnDispatchInput; + +interface DurableReceiptReconcileOptions { + acknowledgedStatus?: string | null; + acknowledgedEffectiveTurnIntentId?: string | null; + forceAuthoritativeRead?: boolean; + monitorNonterminalUntilFinality?: boolean; +} + +interface EffectiveTurnStatusMonitor { + timer: ReturnType | null; + unsubscribe: (() => void) | null; + backoffIndex: number; + effectiveTurnIntentId: string; + preserveRuntimePresentationOnTerminal: boolean; +} + +const log = createLogger("TurnDispatchService"); +const EFFECTIVE_STATUS_BACKOFF_MS = [ + 100, 250, 500, 1_000, 2_000, 5_000, 10_000, 30_000, +] as const; +const effectiveTurnStatusMonitors = new Map< + string, + EffectiveTurnStatusMonitor +>(); + +function effectiveMonitorKey(dispatch: ReservedTurnDispatch): string { + return JSON.stringify([dispatch.sessionId, dispatch.generation]); +} + +function stopEffectiveTurnStatusMonitor(dispatch: ReservedTurnDispatch): void { + const key = effectiveMonitorKey(dispatch); + const monitor = effectiveTurnStatusMonitors.get(key); + if (!monitor) return; + if (monitor.timer !== null) clearTimeout(monitor.timer); + monitor.unsubscribe?.(); + effectiveTurnStatusMonitors.delete(key); +} + +export function resetTurnDispatchMonitorsForTests(): void { + for (const monitor of effectiveTurnStatusMonitors.values()) { + if (monitor.timer !== null) clearTimeout(monitor.timer); + monitor.unsubscribe?.(); + } + effectiveTurnStatusMonitors.clear(); +} + +function failAttribution( + dispatch: ReservedTurnDispatch, + message: string +): never { + failReservedTurn(dispatch); + throw new Error(message); +} + +function bindEffectiveTurnIntent( + dispatch: ReservedTurnDispatch, + effectiveTurnIntentId: string | null | undefined +): string { + const effective = effectiveTurnIntentId ?? dispatch.turnIntentId; + if (!effective) { + return failAttribution( + dispatch, + `empty effective turn intent for ${dispatch.turnIntentId}` + ); + } + if ( + effective !== dispatch.turnIntentId && + !publishTurnIntentDispatchAlias(effective, { + sessionId: dispatch.sessionId, + generation: dispatch.generation, + }) + ) { + return failAttribution( + dispatch, + `effective turn intent ${effective} conflicts with ${dispatch.turnIntentId}` + ); + } + return effective; +} + +function settleDurableTerminalReceipt( + dispatch: ReservedTurnDispatch, + receiptStatus: "completed" | "failed" | "cancelled" | "coalesced", + options: { preserveRuntimePresentation?: boolean } = {} +): TurnDispatchAccepted { + stopEffectiveTurnStatusMonitor(dispatch); + // A delayed receipt for an older reservation is still a successful + // transport reconciliation. Its exact terminal/mapping must close, but it + // must not clear or overwrite the optimistic mirror owned by a newer + // generation. + const ownsCurrentGeneration = + getTurnGeneration(dispatch.sessionId) === dispatch.generation; + const terminalStatus = + receiptStatus === "coalesced" ? "failed" : receiptStatus; + // Always clear the dispatch-only optimistic marker. Exact-X ambiguity may + // be a consumed steering augmentation while its underlying provider turn + // is still running, so that monitor settles lifecycle finality without + // overwriting the provider-owned runtime presentation. Project Y owns a + // standalone run and continues to project its durable terminal here. + if (ownsCurrentGeneration) { + clearRecentOptimisticTurn(dispatch.sessionId); + if (!options.preserveRuntimePresentation) { + getInstrumentedStore().set(setSessionRuntimeStatusAtom, { + sessionId: dispatch.sessionId, + status: terminalStatus, + source: dispatch.optimisticSource, + }); + } + } + markTurnTerminal(dispatch.sessionId, terminalStatus, { + generation: dispatch.generation, + }); + markSessionActive(dispatch.sessionId); + return { ...dispatch, accepted: true }; +} + +function startEffectiveTurnStatusMonitor( + dispatch: ReservedTurnDispatch, + effectiveTurnIntentId: string, + options: { preserveRuntimePresentationOnTerminal?: boolean } = {} +): void { + if ( + getTurnGeneration(dispatch.sessionId) !== dispatch.generation || + getTurnTerminal(dispatch.sessionId, dispatch.generation) + ) { + return; + } + const key = effectiveMonitorKey(dispatch); + if (effectiveTurnStatusMonitors.has(key)) return; + const monitor: EffectiveTurnStatusMonitor = { + timer: null, + unsubscribe: null, + backoffIndex: 0, + effectiveTurnIntentId, + preserveRuntimePresentationOnTerminal: + options.preserveRuntimePresentationOnTerminal ?? false, + }; + effectiveTurnStatusMonitors.set(key, monitor); + + const stop = (): void => { + if (effectiveTurnStatusMonitors.get(key) !== monitor) return; + if (monitor.timer !== null) clearTimeout(monitor.timer); + monitor.unsubscribe?.(); + monitor.timer = null; + monitor.unsubscribe = null; + effectiveTurnStatusMonitors.delete(key); + }; + const scheduleNext = (): void => { + if (effectiveTurnStatusMonitors.get(key) !== monitor) return; + if ( + getTurnGeneration(dispatch.sessionId) !== dispatch.generation || + getTurnTerminal(dispatch.sessionId, dispatch.generation) + ) { + stop(); + return; + } + const backoffMs = + EFFECTIVE_STATUS_BACKOFF_MS[ + Math.min(monitor.backoffIndex, EFFECTIVE_STATUS_BACKOFF_MS.length - 1) + ]; + monitor.backoffIndex += 1; + monitor.timer = setTimeout(() => { + monitor.timer = null; + void poll(); + }, backoffMs); + }; + const poll = async (): Promise => { + if (effectiveTurnStatusMonitors.get(key) !== monitor) return; + if ( + getTurnGeneration(dispatch.sessionId) !== dispatch.generation || + getTurnTerminal(dispatch.sessionId, dispatch.generation) + ) { + stop(); + return; + } + + let durableReceipt: Awaited< + ReturnType + >; + try { + durableReceipt = await SessionService.getTurnIntentStatus( + dispatch.sessionId, + dispatch.turnIntentId + ); + } catch { + scheduleNext(); + return; + } + if (effectiveTurnStatusMonitors.get(key) !== monitor) return; + if (!durableReceipt) { + scheduleNext(); + return; + } + + try { + const observedEffectiveTurnIntentId = bindEffectiveTurnIntent( + dispatch, + durableReceipt.effectiveTurnIntentId + ); + if (observedEffectiveTurnIntentId !== monitor.effectiveTurnIntentId) { + failAttribution( + dispatch, + `durable effective turn intent ${observedEffectiveTurnIntentId} ` + + `conflicts with monitor ${monitor.effectiveTurnIntentId}` + ); + } + switch (durableReceipt.status) { + case "queued": + scheduleNext(); + return; + case "running": + confirmTurnRunning(dispatch.sessionId, { + generation: dispatch.generation, + }); + markSessionActive(dispatch.sessionId); + scheduleNext(); + return; + case "completed": + case "failed": + case "cancelled": + case "coalesced": + settleDurableTerminalReceipt(dispatch, durableReceipt.status, { + preserveRuntimePresentation: + monitor.preserveRuntimePresentationOnTerminal, + }); + return; + case "optimistic": + case "stale": + case "rejected": + default: + failReservedTurn(dispatch); + stop(); + return; + } + } catch (error) { + stop(); + log.warn( + `[TurnDispatchService] effective turn status monitor stopped: ${String(error)}` + ); + } + }; + + // Live exact terminals and generation supersession are the primary stop + // signal. The recursive status read is only a low-frequency safety net for + // a dropped terminal event or a pre-runtime dead-letter. + monitor.unsubscribe = getInstrumentedStore().sub( + turnLifecycleSignalAtom, + () => { + if ( + getTurnGeneration(dispatch.sessionId) !== dispatch.generation || + getTurnTerminal(dispatch.sessionId, dispatch.generation) + ) { + stop(); + } + } + ); + scheduleNext(); +} + +async function reconcileDurableTurnIntentReceipt( + dispatch: ReservedTurnDispatch, + transportError: unknown, + options: DurableReceiptReconcileOptions = {} +): Promise { + const hasAcknowledgedEffectiveTurnIntent = + options.acknowledgedEffectiveTurnIntentId != null; + let effectiveTurnIntentId = bindEffectiveTurnIntent( + dispatch, + options.acknowledgedEffectiveTurnIntentId + ); + let receiptStatus = options.acknowledgedStatus; + if (options.forceAuthoritativeRead || receiptStatus == null) { + const durableReceipt = await SessionService.getTurnIntentStatus( + dispatch.sessionId, + dispatch.turnIntentId + ).catch(() => null); + if (durableReceipt) { + const durableEffectiveTurnIntentId = bindEffectiveTurnIntent( + dispatch, + durableReceipt.effectiveTurnIntentId + ); + if ( + hasAcknowledgedEffectiveTurnIntent && + durableEffectiveTurnIntentId !== effectiveTurnIntentId + ) { + return failAttribution( + dispatch, + `durable effective turn intent ${durableEffectiveTurnIntentId} ` + + `conflicts with acknowledgement ${effectiveTurnIntentId}` + ); + } + effectiveTurnIntentId = durableEffectiveTurnIntentId; + receiptStatus = durableReceipt.status; + } else { + receiptStatus = null; + } + } + + // Keep this switch explicit and fail closed. A new backend status must not + // silently become an unbounded frontend `working` phase. + const shouldMonitorNonterminal = + effectiveTurnIntentId !== dispatch.turnIntentId || + options.monitorNonterminalUntilFinality === true; + const preserveRuntimePresentationOnTerminal = + effectiveTurnIntentId === dispatch.turnIntentId && + options.monitorNonterminalUntilFinality === true; + switch (receiptStatus) { + case "queued": + if (shouldMonitorNonterminal) { + // X→Y is a durable backend ownership transfer. A legitimate Y can + // remain queued behind another turn/setup/path lock for longer than + // any sound frontend elapsed-time bound, so clear the pre-accept + // dead-man and reconcile durable finality until terminal or exact- + // generation supersession. Exact-X response-loss/duplicate receipts + // also need this path because a consumed steering intent has no + // standalone provider terminal. + confirmTurnRunning(dispatch.sessionId, { + generation: dispatch.generation, + }); + startEffectiveTurnStatusMonitor(dispatch, effectiveTurnIntentId, { + preserveRuntimePresentationOnTerminal, + }); + markSessionActive(dispatch.sessionId); + return { ...dispatch, accepted: true }; + } + confirmTurnRunning(dispatch.sessionId, { + generation: dispatch.generation, + }); + markSessionActive(dispatch.sessionId); + return { ...dispatch, accepted: true }; + case "running": + confirmTurnRunning(dispatch.sessionId, { + generation: dispatch.generation, + }); + if (shouldMonitorNonterminal) { + startEffectiveTurnStatusMonitor(dispatch, effectiveTurnIntentId, { + preserveRuntimePresentationOnTerminal, + }); + } + markSessionActive(dispatch.sessionId); + return { ...dispatch, accepted: true }; + case "completed": + case "failed": + case "cancelled": + case "coalesced": + return settleDurableTerminalReceipt(dispatch, receiptStatus); + case "optimistic": + case "stale": + case "rejected": + case null: + default: + failReservedTurn(dispatch); + throw transportError; + } +} + +/** Reserve the session generation and intent synchronously before any await. */ +export function reserveTurnDispatch( + input: ReserveTurnDispatchInput +): ReservedTurnDispatch { + const turnIntentId = input.turnIntentId ?? mintTurnIntentId(); + const generation = beginTurnDispatch(input.sessionId); + publishTurnIntentDispatch(turnIntentId, { + sessionId: input.sessionId, + generation, + }); + const optimisticSource = input.optimisticSource ?? "dispatch"; + beginOptimisticTurn(input.sessionId, optimisticSource); + return { + sessionId: input.sessionId, + turnIntentId, + generation, + optimisticSource, + }; +} + +/** Close a reservation that failed before transport dispatch began. */ +export function failReservedTurn(dispatch: ReservedTurnDispatch): void { + stopEffectiveTurnStatusMonitor(dispatch); + // A response from an older reservation may arrive after another submit has + // installed a new optimistic mirror. Roll back presentation only for the + // current generation, but always finalize the exact failed reservation so + // its waiter and intent mapping cannot leak. + if (getTurnGeneration(dispatch.sessionId) === dispatch.generation) { + failOptimisticTurn(dispatch.sessionId, dispatch.optimisticSource); + } + if (getTurnTerminal(dispatch.sessionId, dispatch.generation)) return; + markTurnTerminal(dispatch.sessionId, "failed", { + generation: dispatch.generation, + }); +} + +/** Send an already-reserved turn through the category adapter. */ +export async function sendReservedTurn( + input: SendReservedTurnInput +): Promise { + const { dispatch, ...params } = input; + let receipt: Awaited>; + try { + receipt = await SessionService.sendMessage({ + ...params, + sessionId: dispatch.sessionId, + turnIntentId: dispatch.turnIntentId, + // The logical intent is also the default transport idempotency key. + // Callers may provide a stable domain-specific key, but no canonical + // dispatch is allowed to fall back to an un-deduplicated send. + clientMessageId: params.clientMessageId ?? dispatch.turnIntentId, + }); + } catch (error) { + // Tauri can lose a successful command response after the backend has + // already persisted/enqueued the exact intent. Read the durable receipt + // before declaring rejection; this keeps response loss from becoming a + // second logical turn on retry. + return reconcileDurableTurnIntentReceipt(dispatch, error, { + forceAuthoritativeRead: true, + monitorNonterminalUntilFinality: true, + }); + } + + const effectiveTurnIntentId = bindEffectiveTurnIntent( + dispatch, + receipt.effectiveTurnIntentId + ); + if (effectiveTurnIntentId !== dispatch.turnIntentId) { + // The effective WorkItemRun may have reached terminal before the enqueue + // acknowledgement crossed IPC and before this window could install Y as + // an alias for composer intent X. Re-read X after alias publication so + // durable finality closes the original reserved generation even when the + // live Y terminal was presentation-only in this window. + return reconcileDurableTurnIntentReceipt( + dispatch, + new Error( + `effective turn ${effectiveTurnIntentId} has no executable durable receipt` + ), + { + acknowledgedStatus: receipt.turnIntentStatus, + acknowledgedEffectiveTurnIntentId: effectiveTurnIntentId, + forceAuthoritativeRead: true, + } + ); + } + + if (receipt.duplicate) { + return reconcileDurableTurnIntentReceipt( + dispatch, + new Error( + `duplicate send for ${dispatch.turnIntentId} has no executable durable receipt` + ), + { + acknowledgedStatus: receipt.turnIntentStatus, + acknowledgedEffectiveTurnIntentId: effectiveTurnIntentId, + monitorNonterminalUntilFinality: true, + } + ); + } + + if (receipt.steered) { + // Mid-turn steering is an accepted augmentation of a provider turn, not + // a standalone turn that will emit its own provider terminal. Settle this + // reservation exactly while leaving the authoritative runtime-status + // presentation untouched (the underlying turn is still running). + if (getTurnGeneration(dispatch.sessionId) === dispatch.generation) { + clearRecentOptimisticTurn(dispatch.sessionId); + } + markTurnTerminal(dispatch.sessionId, "completed", { + generation: dispatch.generation, + }); + markSessionActive(dispatch.sessionId); + return { ...dispatch, accepted: true }; + } + + confirmTurnRunning(dispatch.sessionId, { generation: dispatch.generation }); + markSessionActive(dispatch.sessionId); + if (isCursorIdeSession(dispatch.sessionId)) { + if (getTurnGeneration(dispatch.sessionId) !== dispatch.generation) { + return { ...dispatch, accepted: true }; + } + getInstrumentedStore().set(setSessionRuntimeStatusAtom, { + sessionId: dispatch.sessionId, + status: "idle", + source: dispatch.optimisticSource, + }); + markTurnTerminal(dispatch.sessionId, "completed", { + generation: dispatch.generation, + }); + } + return { ...dispatch, accepted: true }; +} + +/** Convenience for headless callers that do not need a pre-send UI write. */ +export async function dispatchTurn( + input: DispatchTurnInput +): Promise { + const { sessionId, turnIntentId, optimisticSource, ...sendParams } = input; + const dispatch = reserveTurnDispatch({ + sessionId, + turnIntentId, + optimisticSource, + }); + return sendReservedTurn({ dispatch, ...sendParams }); +} + +/** Wait for the provider terminal belonging to this exact reservation. */ +export function waitForTurnOutcome( + dispatch: ReservedTurnDispatch, + deadlineMs: number +): Promise { + const readOutcome = (): TurnOutcome | null => { + const terminal = getTurnTerminal(dispatch.sessionId, dispatch.generation); + if (!terminal) return null; + return { ...dispatch, status: terminal.status, at: terminal.at }; + }; + const immediate = readOutcome(); + if (immediate) return Promise.resolve(immediate); + + return new Promise((resolve, reject) => { + const remainingMs = deadlineMs - Date.now(); + if (remainingMs <= 0) { + reject(new Error("turn outcome timed out")); + return; + } + const store = getInstrumentedStore(); + let unsubscribe: (() => void) | null = null; + const timer = setTimeout(() => { + unsubscribe?.(); + reject(new Error("turn outcome timed out")); + }, remainingMs); + const check = (): void => { + const outcome = readOutcome(); + if (!outcome) return; + clearTimeout(timer); + unsubscribe?.(); + resolve(outcome); + }; + unsubscribe = store.sub(turnLifecycleSignalAtom, check); + check(); + }); +} + +/** + * Resolve a caller-owned intent to its exact reserved generation, then await + * that generation's terminal. Queued and direct turns share this rendezvous; + * feature modules must not reimplement terminal subscriptions or timestamps. + */ +export async function waitForTurnIntentOutcome( + turnIntentId: string, + deadlineMs: number +): Promise { + const dispatch = await waitForTurnIntentDispatch(turnIntentId, deadlineMs); + return waitForTurnOutcome( + { + ...dispatch, + turnIntentId, + // This field is relevant only to pre-transport rollback. The intent has + // already dispatched by the time this observer resolves it. + optimisticSource: "dispatch", + }, + deadlineMs + ); +} diff --git a/src/engines/SessionCore/services/types.ts b/src/engines/SessionCore/services/types.ts index 7c7a8ee0c9..f04f56db77 100644 --- a/src/engines/SessionCore/services/types.ts +++ b/src/engines/SessionCore/services/types.ts @@ -92,6 +92,11 @@ export interface SessionSendMessageParams { accountId?: string; /** Optional agent mode for SDE sessions (build/plan/explore) */ mode?: string; + /** + * Explicit execution workspace override for trusted non-composer sends. + * Ordinary chat sends omit this and use the session row's persisted repo. + */ + workspacePath?: string; /** Base64 image data URLs attached to this message. */ imageDataUrls?: string[]; /** Client-side idempotency key used to suppress duplicate sends. */ diff --git a/src/engines/SessionCore/sync/adapters/__tests__/rustAgentEventLifecycle.test.ts b/src/engines/SessionCore/sync/adapters/__tests__/rustAgentEventLifecycle.test.ts index fd262a666f..09374159b7 100644 --- a/src/engines/SessionCore/sync/adapters/__tests__/rustAgentEventLifecycle.test.ts +++ b/src/engines/SessionCore/sync/adapters/__tests__/rustAgentEventLifecycle.test.ts @@ -1,8 +1,25 @@ -import { describe, expect, it } from "vitest"; +import { beforeEach, describe, expect, it } from "vitest"; -import { isRustAgentTurnNeutralEvent } from "../createRustAgentAdapter"; +import { + publishTurnIntentDispatch, + resetTurnIntentDispatchLifecycleForTests, +} from "@src/engines/SessionCore/control/turnIntentDispatchLifecycle"; +import { + beginTurnDispatch, + resetTurnLifecycleForTests, +} from "@src/engines/SessionCore/control/turnLifecycle"; + +import { + isRustAgentTurnNeutralEvent, + shouldAcceptRustAgentTerminalAttribution, +} from "../createRustAgentAdapter"; describe("Rust agent event lifecycle classification", () => { + beforeEach(() => { + resetTurnIntentDispatchLifecycleForTests(); + resetTurnLifecycleForTests(); + }); + it.each([ "agent:snapshot_created", "agent:file_change", @@ -21,4 +38,60 @@ describe("Rust agent event lifecycle classification", () => { ])("still treats substantive event %s as turn activity", (eventType) => { expect(isRustAgentTurnNeutralEvent(eventType)).toBe(false); }); + + it("fails closed for an unknown or cross-session terminal during a canonical turn", () => { + const generation = beginTurnDispatch("session-1"); + publishTurnIntentDispatch("active-intent", { + sessionId: "session-1", + generation, + }); + expect( + shouldAcceptRustAgentTerminalAttribution( + { turnIntentId: "unknown-intent" }, + "session-1" + ) + ).toBe(false); + + expect( + shouldAcceptRustAgentTerminalAttribution( + { details: { turnIntentId: "unknown-scheduler-error" } }, + "session-1" + ) + ).toBe(false); + + publishTurnIntentDispatch("known-intent", { + sessionId: "session-other", + generation: 4, + }); + expect( + shouldAcceptRustAgentTerminalAttribution( + { turnIntentId: "known-intent" }, + "session-1" + ) + ).toBe(false); + }); + + it("accepts an exact attributed terminal and the legacy unattributed path", () => { + const generation = beginTurnDispatch("session-1"); + publishTurnIntentDispatch("known-intent", { + sessionId: "session-1", + generation, + }); + + expect( + shouldAcceptRustAgentTerminalAttribution( + { turnIntentId: "known-intent" }, + "session-1" + ) + ).toBe(true); + expect(shouldAcceptRustAgentTerminalAttribution({}, "session-legacy")).toBe( + true + ); + expect( + shouldAcceptRustAgentTerminalAttribution( + { turnIntentId: "backend-minted-legacy-intent" }, + "session-legacy" + ) + ).toBe(true); + }); }); diff --git a/src/engines/SessionCore/sync/adapters/__tests__/rustAgentSendPayload.test.ts b/src/engines/SessionCore/sync/adapters/__tests__/rustAgentSendPayload.test.ts index c5244cbac8..185ba047a0 100644 --- a/src/engines/SessionCore/sync/adapters/__tests__/rustAgentSendPayload.test.ts +++ b/src/engines/SessionCore/sync/adapters/__tests__/rustAgentSendPayload.test.ts @@ -1,6 +1,9 @@ import { describe, expect, it } from "vitest"; -import { buildRustAgentSendMessageArgs } from "../rustAgentSendPayload"; +import { + buildRustAgentSendMessageArgs, + parseRustAgentSendReceipt, +} from "../rustAgentSendPayload"; describe("buildRustAgentSendMessageArgs", () => { it("preserves force-send as an explicit wire source", () => { @@ -35,3 +38,103 @@ describe("buildRustAgentSendMessageArgs", () => { }); }); }); + +describe("parseRustAgentSendReceipt", () => { + it.each([false, true])("preserves duplicate=%s", (duplicate) => { + expect( + parseRustAgentSendReceipt({ + content: JSON.stringify({ + queued: true, + duplicate, + turnIntentStatus: "queued", + effectiveTurnIntentId: "intent-1", + }), + sessionId: "sdeagent-1", + model: "test-model", + }) + ).toEqual({ + duplicate, + turnIntentStatus: "queued", + effectiveTurnIntentId: "intent-1", + }); + }); + + it("preserves an exact durable status from a duplicate ack", () => { + expect( + parseRustAgentSendReceipt({ + content: JSON.stringify({ + queued: false, + duplicate: true, + turnIntentStatus: "completed", + effectiveTurnIntentId: "wir_effective", + }), + sessionId: "sdeagent-1", + model: "test-model", + }) + ).toEqual({ + duplicate: true, + turnIntentStatus: "completed", + effectiveTurnIntentId: "wir_effective", + }); + }); + + it("preserves an explicit mid-turn steering receipt", () => { + expect( + parseRustAgentSendReceipt({ + content: JSON.stringify({ + duplicate: false, + steered: true, + turnIntentStatus: "queued", + effectiveTurnIntentId: "intent-steered", + }), + sessionId: "sdeagent-1", + model: "test-model", + }) + ).toEqual({ + duplicate: false, + steered: true, + turnIntentStatus: "queued", + effectiveTurnIntentId: "intent-steered", + }); + }); + + it.each(["not-json", JSON.stringify({ queued: true })])( + "fails closed for malformed acknowledgement %s", + (content) => { + expect(() => + parseRustAgentSendReceipt({ + content, + sessionId: "sdeagent-1", + model: "test-model", + }) + ).toThrow(/acknowledgement/); + } + ); + + it("fails closed for a malformed durable status", () => { + expect(() => + parseRustAgentSendReceipt({ + content: JSON.stringify({ + duplicate: true, + turnIntentStatus: 42, + effectiveTurnIntentId: "intent-1", + }), + sessionId: "sdeagent-1", + model: "test-model", + }) + ).toThrow(/turnIntentStatus/); + }); + + it.each([ + { duplicate: false, effectiveTurnIntentId: "intent-1" }, + { duplicate: false, turnIntentStatus: "queued" }, + ])("fails closed when a required receipt identity is missing", (payload) => { + expect(() => + parseRustAgentSendReceipt({ + content: JSON.stringify(payload), + sessionId: "sdeagent-1", + model: "test-model", + }) + ).toThrow(/turnIntentStatus|effectiveTurnIntentId/); + }); +}); diff --git a/src/engines/SessionCore/sync/adapters/cli/__tests__/cliTransport.test.ts b/src/engines/SessionCore/sync/adapters/cli/__tests__/cliTransport.test.ts index f884191400..6df4c82fd9 100644 --- a/src/engines/SessionCore/sync/adapters/cli/__tests__/cliTransport.test.ts +++ b/src/engines/SessionCore/sync/adapters/cli/__tests__/cliTransport.test.ts @@ -28,7 +28,9 @@ describe("sendCliMessage acceptance boundary", () => { mocks.message.mockResolvedValue({ sessionId: "cliagent-worker", turnIntentId: "intent-1", + effectiveTurnIntentId: "intent-1", status: "running", + duplicate: false, }); mocks.enterIntervention.mockReturnValue(new Promise(() => undefined)); }); @@ -43,7 +45,11 @@ describe("sendCliMessage acceptance boundary", () => { turnIntentSource: "user_submit", directUserIntent: true, }) - ).resolves.toBeUndefined(); + ).resolves.toEqual({ + duplicate: false, + turnIntentStatus: "running", + effectiveTurnIntentId: "intent-1", + }); expect(mocks.message).toHaveBeenCalledWith({ request: { @@ -56,11 +62,110 @@ describe("sendCliMessage acceptance boundary", () => { expect(mocks.registerReceipt).toHaveBeenCalledWith({ sessionId: "cliagent-worker", turnIntentId: "intent-1", + effectiveTurnIntentId: "intent-1", status: "running", + duplicate: false, }); expect(mocks.enterIntervention).toHaveBeenCalledWith("cliagent-worker"); }); + it("returns an exact running replay without repeating intervention", async () => { + mocks.message.mockResolvedValue({ + sessionId: "cliagent-worker", + turnIntentId: "intent-running", + effectiveTurnIntentId: "intent-running", + status: "running", + duplicate: true, + }); + + await expect( + sendCliMessage({ + sessionId: "cliagent-worker", + content: "retry after response loss", + turnIntentId: "intent-running", + clientMessageId: "message-running", + turnIntentSource: "user_submit", + directUserIntent: true, + }) + ).resolves.toEqual({ + duplicate: true, + turnIntentStatus: "running", + effectiveTurnIntentId: "intent-running", + }); + + expect(mocks.registerReceipt).toHaveBeenCalledWith({ + sessionId: "cliagent-worker", + turnIntentId: "intent-running", + effectiveTurnIntentId: "intent-running", + status: "running", + duplicate: true, + }); + expect(mocks.enterIntervention).not.toHaveBeenCalled(); + }); + + it("returns an exact completed replay without reopening the turn", async () => { + mocks.message.mockResolvedValue({ + sessionId: "cliagent-worker", + turnIntentId: "intent-completed", + effectiveTurnIntentId: "intent-completed", + status: "completed", + duplicate: true, + }); + + await expect( + sendCliMessage({ + sessionId: "cliagent-worker", + content: "late retry", + turnIntentId: "intent-completed", + clientMessageId: "message-completed", + turnIntentSource: "user_submit", + }) + ).resolves.toEqual({ + duplicate: true, + turnIntentStatus: "completed", + effectiveTurnIntentId: "intent-completed", + }); + + expect(mocks.registerReceipt).toHaveBeenCalledWith({ + sessionId: "cliagent-worker", + turnIntentId: "intent-completed", + effectiveTurnIntentId: "intent-completed", + status: "completed", + duplicate: true, + }); + expect(mocks.enterIntervention).not.toHaveBeenCalled(); + }); + + it("returns a backend-selected effective Project intent", async () => { + mocks.message.mockResolvedValue({ + sessionId: "cliagent-worker", + turnIntentId: "intent-project-x", + effectiveTurnIntentId: "wir_project-y", + status: "queued", + duplicate: false, + }); + + await expect( + sendCliMessage({ + sessionId: "cliagent-worker", + content: "project task", + turnIntentId: "intent-project-x", + clientMessageId: "message-project-x", + turnIntentSource: "user_submit", + }) + ).resolves.toEqual({ + duplicate: false, + turnIntentStatus: "queued", + effectiveTurnIntentId: "wir_project-y", + }); + expect(mocks.registerReceipt).toHaveBeenCalledWith( + expect.objectContaining({ + turnIntentId: "intent-project-x", + effectiveTurnIntentId: "wir_project-y", + }) + ); + }); + it("rejects only when the backend command rejects", async () => { mocks.message.mockRejectedValue(new Error("ipc unavailable")); @@ -77,4 +182,23 @@ describe("sendCliMessage acceptance boundary", () => { expect(mocks.registerReceipt).not.toHaveBeenCalled(); expect(mocks.enterIntervention).not.toHaveBeenCalled(); }); + + it("rejects when the effective receipt cannot be attributed exactly", async () => { + mocks.registerReceipt.mockImplementationOnce(() => { + throw new Error("CLI effective turn intent wir-conflict conflicts"); + }); + + await expect( + sendCliMessage({ + sessionId: "cliagent-worker", + content: "project task", + turnIntentId: "intent-conflict", + clientMessageId: "message-conflict", + turnIntentSource: "user_submit", + directUserIntent: true, + }) + ).rejects.toThrow(/conflicts/); + + expect(mocks.enterIntervention).not.toHaveBeenCalled(); + }); }); diff --git a/src/engines/SessionCore/sync/adapters/cli/cliTransport.ts b/src/engines/SessionCore/sync/adapters/cli/cliTransport.ts index e6612cff40..6cb2a49db6 100644 --- a/src/engines/SessionCore/sync/adapters/cli/cliTransport.ts +++ b/src/engines/SessionCore/sync/adapters/cli/cliTransport.ts @@ -4,7 +4,7 @@ import { rpc } from "@src/api/tauri/rpc"; import { cliTurnLifecycleCoordinator } from "@src/hooks/cliSession/cliTurnLifecycleCoordinator"; import { createLogger } from "@src/hooks/logger"; -import type { AdapterSendInput } from "../../types"; +import type { AdapterSendInput, AdapterSendReceipt } from "../../types"; const log = createLogger("CliTransport"); @@ -12,7 +12,9 @@ function newMessageId(): string { return crypto.randomUUID(); } -export async function sendCliMessage(input: AdapterSendInput): Promise { +export async function sendCliMessage( + input: AdapterSendInput +): Promise { const { sessionId, content, @@ -42,7 +44,7 @@ export async function sendCliMessage(input: AdapterSendInput): Promise { }); cliTurnLifecycleCoordinator.registerReceipt(receipt); - if (directUserIntent) { + if (!receipt.duplicate && directUserIntent) { void enterAgentOrgSessionIntervention(sessionId).catch((error) => { log.warn( "[sendCliMessage] accepted CLI turn but failed to persist intervention:", @@ -50,6 +52,11 @@ export async function sendCliMessage(input: AdapterSendInput): Promise { ); }); } + return { + duplicate: receipt.duplicate, + turnIntentStatus: receipt.status, + effectiveTurnIntentId: receipt.effectiveTurnIntentId, + }; } export async function stopCliSession( diff --git a/src/engines/SessionCore/sync/adapters/createRustAgentAdapter.ts b/src/engines/SessionCore/sync/adapters/createRustAgentAdapter.ts index ad88e8e66d..864905a5f3 100644 --- a/src/engines/SessionCore/sync/adapters/createRustAgentAdapter.ts +++ b/src/engines/SessionCore/sync/adapters/createRustAgentAdapter.ts @@ -18,7 +18,12 @@ import { getSessionInfo, loadMessages, } from "@src/api/tauri/agent"; -import type { CancelReason } from "@src/api/tauri/agent"; +import type { AgentMessageResponse, CancelReason } from "@src/api/tauri/agent"; +import { + getTurnIntentDispatch, + hasActiveTurnIntentDispatch, +} from "@src/engines/SessionCore/control/turnIntentDispatchLifecycle"; +import { getTurnGeneration } from "@src/engines/SessionCore/control/turnLifecycle"; import { eventStoreProxy } from "@src/engines/SessionCore/core/store/EventStoreProxy"; import type { SessionEvent } from "@src/engines/SessionCore/core/types"; import { @@ -60,7 +65,10 @@ import { applyToolUsageToEvents, loadUsageTelemetry, } from "./rustAgent/toolUsageCache"; -import { buildRustAgentSendMessageArgs } from "./rustAgentSendPayload"; +import { + buildRustAgentSendMessageArgs, + parseRustAgentSendReceipt, +} from "./rustAgentSendPayload"; import type { AgentTokenUsage, AgentWSEvent, @@ -136,6 +144,21 @@ export function isRustAgentTurnNeutralEvent(eventType: string): boolean { return TURN_NEUTRAL_EVENTS.has(eventType); } +export function shouldAcceptRustAgentTerminalAttribution( + event: Pick, + sessionId: string +): boolean { + const turnIntentId = event.turnIntentId ?? event.details?.turnIntentId; + if (!turnIntentId) return true; + const dispatch = getTurnIntentDispatch(turnIntentId); + if (dispatch) return dispatch.sessionId === sessionId; + // Rollout compatibility: legacy producers still let the backend mint the + // intent. Accept that terminal only when this session has no canonical + // active mapping; once a mapped generation exists, an unknown id is stale + // or misrouted and must fail closed. + return !hasActiveTurnIntentDispatch(sessionId, getTurnGeneration(sessionId)); +} + const PLAN_SUBMITTED_END_TURN_PREFIX = "PLAN_SUBMITTED_END_TURN:"; const LIVE_STREAM_EVENTS_IGNORED_AFTER_STOP = new Set([ "agent:message_delta", @@ -458,6 +481,7 @@ export function createRustAgentAdapter( errorMessage?: string, meta?: { turnId?: string; + turnIntentId?: string; turnStatus?: string; intermediate?: boolean; } @@ -543,6 +567,18 @@ export function createRustAgentAdapter( event.result.startsWith(PLAN_SUBMITTED_END_TURN_PREFIX); const isTerminal = TERMINAL_EVENTS.has(event.type) || isPlanReadyTerminal; + if ( + isTerminal && + !shouldAcceptRustAgentTerminalAttribution(event, sessionId) + ) { + const rejectedTurnIntentId = + event.turnIntentId ?? event.details?.turnIntentId; + logger.warn( + `[${category}] ignored terminal with unknown or misrouted ` + + `turn intent ${rejectedTurnIntentId} for ${sessionId}` + ); + return; + } const isQueueStatus = event.type === "agent:queue_status"; const queueIsProcessing = event.isProcessing === true; const isActiveQueueStatus = isQueueStatus && queueIsProcessing; @@ -677,14 +713,15 @@ export function createRustAgentAdapter( }; }, - async sendMessage(input: AdapterSendInput): Promise { + async sendMessage(input: AdapterSendInput) { const { sessionId } = input; clearSessionStreamingStopped(sessionId); - await retryInvokeTauri( + const response = await retryInvokeTauri( "agent_send_message", buildRustAgentSendMessageArgs(input), sessionId ); + return parseRustAgentSendReceipt(response); }, async stopSession(sessionId: string, reason: CancelReason): Promise { diff --git a/src/engines/SessionCore/sync/adapters/externalHistoryAdapter.ts b/src/engines/SessionCore/sync/adapters/externalHistoryAdapter.ts index 57dd842fda..20c5af83a8 100644 --- a/src/engines/SessionCore/sync/adapters/externalHistoryAdapter.ts +++ b/src/engines/SessionCore/sync/adapters/externalHistoryAdapter.ts @@ -10,6 +10,7 @@ import { } from "../externalHistoryTranscriptSignatures"; import type { AdapterSendInput, + AdapterSendReceipt, EventHandlerCallbacks, SessionAdapter, SessionEventHandler, @@ -192,7 +193,7 @@ export const externalHistoryAdapter: ExternalHistorySessionAdapter = { return createNoopEventHandler(); }, - async sendMessage(input: AdapterSendInput): Promise { + async sendMessage(input: AdapterSendInput): Promise { throw new Error( `External history sessions are read-only and cannot receive messages (${input.sessionId}).` ); diff --git a/src/engines/SessionCore/sync/adapters/rustAgentSendPayload.ts b/src/engines/SessionCore/sync/adapters/rustAgentSendPayload.ts index b95c56f71d..ae20660142 100644 --- a/src/engines/SessionCore/sync/adapters/rustAgentSendPayload.ts +++ b/src/engines/SessionCore/sync/adapters/rustAgentSendPayload.ts @@ -1,4 +1,6 @@ -import type { AdapterSendInput } from "../types"; +import type { AgentMessageResponse } from "@src/api/tauri/agent"; + +import type { AdapterSendInput, AdapterSendReceipt } from "../types"; /** Build the exact Tauri payload for a Rust-native agent turn. */ export function buildRustAgentSendMessageArgs( @@ -41,3 +43,50 @@ export function buildRustAgentSendMessageArgs( turnIntentSource, }; } + +/** Parse the typed acknowledgement embedded in Rust's AgentResponse content. */ +export function parseRustAgentSendReceipt( + response: AgentMessageResponse +): AdapterSendReceipt { + let payload: unknown; + try { + payload = JSON.parse(response.content); + } catch { + throw new Error("agent_send_message returned a non-JSON acknowledgement"); + } + if ( + typeof payload !== "object" || + payload === null || + typeof (payload as { duplicate?: unknown }).duplicate !== "boolean" + ) { + throw new Error( + "agent_send_message acknowledgement is missing boolean duplicate" + ); + } + const turnIntentStatus = (payload as { turnIntentStatus?: unknown }) + .turnIntentStatus; + if (typeof turnIntentStatus !== "string" || !turnIntentStatus) { + throw new Error( + "agent_send_message acknowledgement has invalid turnIntentStatus" + ); + } + const steered = (payload as { steered?: unknown }).steered; + if (steered !== undefined && typeof steered !== "boolean") { + throw new Error( + "agent_send_message acknowledgement has invalid steered flag" + ); + } + const effectiveTurnIntentId = (payload as { effectiveTurnIntentId?: unknown }) + .effectiveTurnIntentId; + if (typeof effectiveTurnIntentId !== "string" || !effectiveTurnIntentId) { + throw new Error( + "agent_send_message acknowledgement has invalid effectiveTurnIntentId" + ); + } + return { + duplicate: (payload as { duplicate: boolean }).duplicate, + ...(steered !== undefined ? { steered } : {}), + turnIntentStatus, + effectiveTurnIntentId, + }; +} diff --git a/src/engines/SessionCore/sync/types.ts b/src/engines/SessionCore/sync/types.ts index 9b0d609e2a..970e99a4bc 100644 --- a/src/engines/SessionCore/sync/types.ts +++ b/src/engines/SessionCore/sync/types.ts @@ -179,6 +179,27 @@ export interface AdapterSendInput { sessionRepoPath?: string | null; } +/** + * Transport acknowledgement for one logical send. + * + * `duplicate` means the adapter observed an idempotency collision. The + * canonical dispatcher must then reconcile the exact durable turn-intent + * receipt instead of assuming this reservation will emit a provider terminal. + */ +export interface AdapterSendReceipt { + duplicate: boolean; + /** Backend diverted this message into an already-running native turn. */ + steered?: boolean; + /** Exact durable status when the transport can return it in the ack. */ + turnIntentStatus?: string | null; + /** + * Backend-selected durable identity for the execution. Project dispatch + * replaces a composer intent with its WorkItemRun id; both identities alias + * the same frontend generation. + */ + effectiveTurnIntentId?: string | null; +} + /** * Adapter interface for session-type-specific logic. * Each session type (SDE Agent, OS Agent, CLI, Cursor IDE) implements this. @@ -214,7 +235,7 @@ export interface SessionAdapter { * previous switch on `isAgentSession` / `isCliSession`. New IDE * adapters slot in here without touching `SessionService`. */ - sendMessage(input: AdapterSendInput): Promise; + sendMessage(input: AdapterSendInput): Promise; /** Stop the running agent/session with an explicit control-flow reason. */ stopSession(sessionId: string, reason: CancelReason): Promise;