Merge pull request #728 from rowboatlabs/fix/session-stop-concurrent-advances

fix(x): make stopTurn reliable under concurrent advances on one turn

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
+- `stopTurn` cancels via the turn runtime: aborting the signal of every
-  signal if one is running, else advancing a suspended turn with a `cancel`
+  live advance the layer has started for the turn, else advancing the turn
-  input.
+  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
 

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<TurnOutcome> };
 
 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<TurnOutcome>((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)", () => {

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 {
+import {
-    ITurnRuntime,
+    type ITurnRuntime,
-    Turn,
+    type Turn,
-    TurnExecution,
+    type TurnExecution,
-    TurnExternalInput,
+    type TurnExternalInput,
-    TurnOutcome,
+    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<
+    // A turn can legally have more than one live advance at once — a running
-        string,
+    // invocation plus external-input invocations queued on the turn lock —
-        { sessionId: string | null; controller: AbortController; execution: TurnExecution }
+    // so entries accumulate per turn and each advance removes only its own.
-    >();
+    private readonly active = new Map<string, Set<ActiveAdvance>>();
 
     constructor({
         sessionRepo,
@@ -270,15 +277,35 @@ export class SessionsImpl implements ISessions {
 
     async stopTurn(turnId: string, reason?: string): Promise<void> {
         const running = this.active.get(turnId);
-        if (running) {
+        if (running && running.size > 0) {
-            running.controller.abort();
+            // Abort every live advance for this turn: the running invocation
-            await running.execution.outcome.catch(() => undefined);
+            // 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, {
+        try {
-            type: "cancel",
+            await this.advanceWithInput(turnId, {
-            ...(reason === undefined ? {} : { reason }),
+                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<ActiveAdvance>();
+        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;
     }