diff --git a/apps/x/packages/core/docs/session-design.md b/apps/x/packages/core/docs/session-design.md index f92acd47..82dd665d 100644 --- a/apps/x/packages/core/docs/session-design.md +++ b/apps/x/packages/core/docs/session-design.md @@ -338,9 +338,14 @@ runtime's job; the session layer passes its errors through. ### 9.3 stopTurn and resumeTurn -- `stopTurn` cancels via the turn runtime: aborting the active invocation's - signal if one is running, else advancing a suspended turn with a `cancel` - input. +- `stopTurn` cancels via the turn runtime: aborting the signal of every + live advance the layer has started for the turn, else advancing the turn + with a `cancel` input. A turn can legally have several live advances at + once — one running invocation plus external-input invocations queued on + the turn lock — so the layer tracks them per turn and stop aborts them + all; the queued ones observe their aborted signal when the lock frees. A + cancel input that loses the race with a concurrent settle (the turn is + already terminal by the time it applies) counts as a successful stop. - `resumeTurn` re-enters the latest turn with no input — the turn spec's recovery entry point — for turns left `idle` by a crash. There is no automatic resume sweep at startup: recovery re-issues interrupted model @@ -499,6 +504,11 @@ All tests use the in-memory/mocked turn runtime and repo fakes. correct turn with the correct input type. - Turn-runtime rejections (unknown call, terminal turn) pass through. - `sendMessage` never routes to ask-human. +- `stopTurn` aborts every live advance when a turn has concurrent + invocations, and an earlier advance settling does not untrack a later + one. +- A `stopTurn` cancel input that lost the race with a concurrent settle + resolves as a successful stop; a non-terminal rejection still surfaces. ### 13.6 Index diff --git a/apps/x/packages/core/src/runtime/sessions/sessions.test.ts b/apps/x/packages/core/src/runtime/sessions/sessions.test.ts index 6ce9332b..80a929d8 100644 --- a/apps/x/packages/core/src/runtime/sessions/sessions.test.ts +++ b/apps/x/packages/core/src/runtime/sessions/sessions.test.ts @@ -178,7 +178,8 @@ type AdvanceScript = (call: { }) => | { events?: TurnStreamEvent[]; outcome: TurnOutcome } | { error: unknown } - | { untilAbort: true }; + | { untilAbort: true } + | { pending: Promise }; class FakeTurnRuntime implements ITurnRuntime { createTurnInputs: CreateTurnInput[] = []; @@ -229,6 +230,8 @@ class FakeTurnRuntime implements ITurnRuntime { }; if ("error" in result) { stream.fail(result.error); + } else if ("pending" in result) { + void result.pending.then((outcome) => stream.end(outcome)); } else if ("untilAbort" in result) { const signal = options?.signal; if (!signal) { @@ -641,6 +644,101 @@ describe("stopTurn and resumeTurn", () => { /no turns to resume/, ); }); + + it("aborts every live advance when a turn has concurrent invocations", async () => { + // A suspended turn with two pending externals gets a permission + // response (advance 1, running) and an async result (advance 2, + // queued on the turn lock in the real runtime). Stop must abort + // both — aborting only the latest leaves the first streaming. + const { sessions, fake } = makeSessions(); + const signals: AbortSignal[] = []; + fake.script = ({ signal }) => { + signals.push(signal as AbortSignal); + return { untilAbort: true }; + }; + const sessionId = await sessions.createSession(); + const { turnId } = await sessions.sendMessage(sessionId, user("go"), { + agent: { agentId: "copilot" }, + }); + const delivery = sessions + .deliverAsyncToolResult(turnId, "B", { output: "done", isError: false }) + .catch(() => undefined); + await flush(); + expect(fake.advanceCalls).toHaveLength(2); + await sessions.stopTurn(turnId); + await delivery; + expect(signals.map((s) => s.aborted)).toEqual([true, true]); + // Abort path only — no third advance carrying a cancel input. + expect(fake.advanceCalls).toHaveLength(2); + }); + + it("an earlier advance settling does not untrack a later one", async () => { + // Advance 1 settles while advance 2 is still live. Its cleanup must + // remove only its own entry: stop must still find and abort + // advance 2 instead of falling back to a cancel input. + const { sessions, fake } = makeSessions(); + let releaseFirst!: (outcome: TurnOutcome) => void; + const firstOutcome = new Promise((resolve) => { + releaseFirst = resolve; + }); + const signals: AbortSignal[] = []; + fake.script = ({ signal }) => { + signals.push(signal as AbortSignal); + if (signals.length === 1) { + return { pending: firstOutcome }; + } + if (signals.length === 2) { + return { untilAbort: true }; + } + // A cancel-input advance would land here; it must never happen. + return { outcome: completedOutcome() }; + }; + const sessionId = await sessions.createSession(); + const { turnId } = await sessions.sendMessage(sessionId, user("go"), { + agent: { agentId: "copilot" }, + }); + const delivery = sessions + .deliverAsyncToolResult(turnId, "B", { output: "done", isError: false }) + .catch(() => undefined); + await flush(); + releaseFirst(completedOutcome()); + await flush(); + await sessions.stopTurn(turnId); + await delivery; + expect(signals).toHaveLength(2); + expect(signals[1].aborted).toBe(true); + expect(fake.advanceCalls).toHaveLength(2); + }); + + it("treats a cancel input that lost the race with a settle as a successful stop", async () => { + const { sessions, fake } = makeSessions(); + const sessionId = await sessions.createSession(); + const { turnId } = await sessions.sendMessage(sessionId, user("go"), { + agent: { agentId: "copilot" }, + }); + await flush(); + fake.setLog(turnId, turnLog(turnId, sessionId, "completed")); + fake.script = () => ({ + error: new TurnInputError(`turn ${turnId} is terminal; input rejected`), + }); + await expect(sessions.stopTurn(turnId)).resolves.toBeUndefined(); + }); + + it("rethrows a cancel-input rejection when the turn is not terminal", async () => { + const { sessions, fake } = makeSessions(); + const sessionId = await sessions.createSession(); + const { turnId } = await sessions.sendMessage(sessionId, user("go"), { + agent: { agentId: "copilot" }, + }); + await flush(); + fake.setLog(turnId, turnLog(turnId, sessionId, "suspended")); + fake.script = () => ({ + error: new TurnInputError("no pending async tool call B"), + }); + await expect(sessions.stopTurn(turnId)).rejects.toThrowError( + /no pending async tool call/, + ); + }); }); describe("event forwarding and index maintenance (13.6, 13.7)", () => { diff --git a/apps/x/packages/core/src/runtime/sessions/sessions.ts b/apps/x/packages/core/src/runtime/sessions/sessions.ts index 5487735c..c929c56c 100644 --- a/apps/x/packages/core/src/runtime/sessions/sessions.ts +++ b/apps/x/packages/core/src/runtime/sessions/sessions.ts @@ -21,12 +21,13 @@ import { } from "@x/shared/dist/turns.js"; import type { IMonotonicallyIncreasingIdGenerator } from "../../application/lib/id-gen.js"; import { chatActivity } from "../../application/lib/chat-activity.js"; -import type { - ITurnRuntime, - Turn, - TurnExecution, - TurnExternalInput, - TurnOutcome, +import { + type ITurnRuntime, + type Turn, + type TurnExecution, + type TurnExternalInput, + TurnInputError, + type TurnOutcome, } from "../turns/api.js"; // traits.js, not registry.js: the registry's builders transitively reach // di/container, which imports this module — a cycle. The traits leaf exists @@ -50,6 +51,12 @@ export interface SessionsDependencies { sessionBus: ISessionBus; } +interface ActiveAdvance { + sessionId: string | null; + controller: AbortController; + execution: TurnExecution; +} + // The session layer per session-design.md: owns conversations as ordered // chains of turn references, enforces one active turn per session, assembles // context as a reference to the previous turn, and maintains the in-memory @@ -64,10 +71,10 @@ export class SessionsImpl implements ISessions { private readonly index = new SessionIndex(); // Ephemeral: executions this process started, for stopTurn's abort path. - private readonly active = new Map< - string, - { sessionId: string | null; controller: AbortController; execution: TurnExecution } - >(); + // A turn can legally have more than one live advance at once — a running + // invocation plus external-input invocations queued on the turn lock — + // so entries accumulate per turn and each advance removes only its own. + private readonly active = new Map>(); constructor({ sessionRepo, @@ -270,15 +277,35 @@ export class SessionsImpl implements ISessions { async stopTurn(turnId: string, reason?: string): Promise { const running = this.active.get(turnId); - if (running) { - running.controller.abort(); - await running.execution.outcome.catch(() => undefined); + if (running && running.size > 0) { + // Abort every live advance for this turn: the running invocation + // cancels, and queued ones observe their aborted signal once the + // turn lock frees. Await them all so stop returns settled. + const advances = [...running]; + for (const advance of advances) { + advance.controller.abort(); + } + await Promise.all( + advances.map((a) => a.execution.outcome.catch(() => undefined)), + ); return; } - await this.advanceWithInput(turnId, { - type: "cancel", - ...(reason === undefined ? {} : { reason }), - }); + try { + await this.advanceWithInput(turnId, { + type: "cancel", + ...(reason === undefined ? {} : { reason }), + }); + } catch (error) { + // A cancel input that loses the race with a concurrent settle is + // a successful stop: the turn is already terminal. + if (error instanceof TurnInputError) { + const turn = await this.turnRuntime.getTurn(turnId); + if (reduceTurn(turn.events).terminal) { + return; + } + } + throw error; + } } // Recovery entry for idle (crash-interrupted) turns. Deliberately not run @@ -372,7 +399,10 @@ export class SessionsImpl implements ISessions { if (sessionId !== null) { void execution.outcome.catch(() => undefined).finally(() => chatActivity.exit()); } - this.active.set(turnId, { sessionId, controller, execution }); + const advance: ActiveAdvance = { sessionId, controller, execution }; + const live = this.active.get(turnId) ?? new Set(); + live.add(advance); + this.active.set(turnId, live); void (async () => { try { @@ -387,7 +417,18 @@ export class SessionsImpl implements ISessions { void execution.outcome .then((outcome) => this.onSettled(sessionId, turnId, outcome)) .catch(() => undefined) - .finally(() => this.active.delete(turnId)); + .finally(() => { + // Remove only this advance's entry; a sibling advance for the + // same turn may still be live and must stay stoppable. + const set = this.active.get(turnId); + if (!set) { + return; + } + set.delete(advance); + if (set.size === 0) { + this.active.delete(turnId); + } + }); return execution; }