CR: executeOrQueueSteeringRequest

Rizzen · Rizzen · commit fc7f6670b251 · 2026-07-20T16:19:46.000+02:00
diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts
@@ -35,6 +35,7 @@ import {
 import type {TokenCount} from "./TokenCount";
 import {toPromptUsage} from "./TokenCount";
 import {CodexCommands} from "./CodexCommands";
+import {SteeringQueue} from "./SteeringQueue";
 import type {QuotaMeta} from "./QuotaMeta";
 import {logger} from "./Logger";
 import {sanitizeMcpServerName} from "./McpServerName";
@@ -166,7 +167,7 @@ export class CodexAcpServer {
     private readonly pendingMcpStartupSessions: Map<string, PendingMcpStartupSession>;
     private readonly pendingTurnStarts: Map<string, PendingTurnStart>;
     private readonly activePrompts: Map<string, ActivePrompt>;
-    private readonly steeringRequests: Map<string, Promise<void>>;
+    private readonly steeringQueues: Map<string, SteeringQueue>;
     private readonly closingSessions: Map<string, number>;
     private readonly sessionGenerations: Map<string, number>;
     private readonly sessionOpenGenerations: Map<string, number>;
@@ -182,7 +183,7 @@ export class CodexAcpServer {
         this.pendingMcpStartupSessions = new Map();
         this.pendingTurnStarts = new Map();
         this.activePrompts = new Map();
-        this.steeringRequests = new Map();
+        this.steeringQueues = new Map();
         this.closingSessions = new Map();
         this.sessionGenerations = new Map();
         this.sessionOpenGenerations = new Map();
@@ -266,7 +267,7 @@ export class CodexAcpServer {
             case LEGACY_SET_SESSION_MODEL_METHOD:
                 return await this.unstable_setSessionModel(this.parseLegacySetSessionModelParams(methodRequest.params));
             case SESSION_STEERING_METHOD:
-                return await this.steerSessionWithFallback(this.parseSessionSteerParams(methodRequest.params));
+                return await this.executeOrQueueSteeringRequest(this.parseSessionSteerParams(methodRequest.params));
             case GOAL_CONTROL_METHOD: {
                 const sessionState = this.sessions.get(methodRequest.params.sessionId);
                 if (!sessionState) {
@@ -601,6 +602,7 @@ export class CodexAcpServer {
                 this.pendingMcpStartupSessions.delete(params.sessionId);
                 this.pendingTurnStarts.delete(params.sessionId);
                 this.activePrompts.delete(params.sessionId);
+                this.steeringQueues.delete(params.sessionId);
             }
             this.endSessionCloseFence(params.sessionId);
         }
@@ -873,26 +875,49 @@ export class CodexAcpServer {
         };
     }
 
-    async steerSessionWithFallback(params: SessionSteerRequest): Promise<SessionSteeringResponse> {
-        const previousRequest = this.steeringRequests.get(params.sessionId) ?? Promise.resolve();
-        let releaseRequest: () => void = () => {};
-        const requestCompleted = new Promise<void>((resolve) => {
-            releaseRequest = resolve;
-        });
-        const requestQueue = previousRequest.then(() => requestCompleted);
-        this.steeringRequests.set(params.sessionId, requestQueue);
-
-        await previousRequest;
+    /**
+     * Handles one incoming steering request, serialising it against any other
+     * steer already in flight for the same session.
+     *
+     * Every session gets its own {@link SteeringQueue}: the request is enqueued
+     * and awaited, so concurrent steers for one session run strictly one at a
+     * time, in arrival order, and can never race to inject into — or start —
+     * rival turns. Steers for different sessions use different queues and run
+     * concurrently. Once the queue drains to idle it is removed from the map,
+     * so no per-session entry leaks after the session goes quiet (the identity
+     * check guards against deleting a queue a later request has since reused).
+     *
+     * @param params The target session id and the prompt to steer with.
+     * @returns Whether the prompt joined the active turn ("injected") or started
+     *     a new one ("startedNewTurn"); see {@link performSteeringRequest}.
+     */
+    async executeOrQueueSteeringRequest(params: SessionSteerRequest): Promise<SessionSteeringResponse> {
+        const queue = this.getSteeringQueue(params.sessionId);
         try {
-            return await this.performSteeringRequest(params);
+            return await queue.enqueue(params);
         } finally {
-            releaseRequest();
-            if (this.steeringRequests.get(params.sessionId) === requestQueue) {
-                this.steeringRequests.delete(params.sessionId);
+            if (queue.isIdle && this.steeringQueues.get(params.sessionId) === queue) {
+                this.steeringQueues.delete(params.sessionId);
             }
         }
     }
 
+    /**
+     * Returns the steering queue for a session, creating and registering it on
+     * first use.
+     *
+     * @param sessionId The session whose steering queue is required.
+     * @returns The session's existing queue, or a freshly created one.
+     */
+    private getSteeringQueue(sessionId: string): SteeringQueue {
+        let queue = this.steeringQueues.get(sessionId);
+        if (!queue) {
+            queue = new SteeringQueue((params) => this.performSteeringRequest(params));
+            this.steeringQueues.set(sessionId, queue);
+        }
+        return queue;
+    }
+
     private async performSteeringRequest(params: SessionSteerRequest): Promise<SessionSteeringResponse> {
         logger.log("Steering session requested", {
             sessionId: params.sessionId,
diff --git a/src/SteeringQueue.ts b/src/SteeringQueue.ts
@@ -0,0 +1,56 @@
+import type {SessionSteerRequest, SessionSteeringResponse} from "./AcpExtensions";
+
+interface QueuedSteering {
+    params: SessionSteerRequest;
+    resolve: (response: SessionSteeringResponse) => void;
+    reject: (error: unknown) => void;
+}
+
+/**
+ * Serialises steering requests for a single session. Callers add a request via
+ * enqueue(); a single consumer loop runs them one at a time, in arrival order,
+ * so two concurrent steers can never race to start rival turns.
+ */
+export class SteeringQueue {
+    private readonly pending: QueuedSteering[] = [];
+    private processing = false;
+
+    constructor(
+        private readonly handle: (params: SessionSteerRequest) => Promise<SessionSteeringResponse>,
+    ) {}
+
+    enqueue(params: SessionSteerRequest): Promise<SessionSteeringResponse> {
+        return new Promise<SessionSteeringResponse>((resolve, reject) => {
+            this.pending.push({params, resolve, reject});
+            this.startConsumer();
+        });
+    }
+
+    /** No request is queued and the consumer is not running. */
+    get isIdle(): boolean {
+        return !this.processing && this.pending.length === 0;
+    }
+
+    private startConsumer(): void {
+        if (this.processing) {
+            return; // consumer already draining the queue
+        }
+        this.processing = true;
+        void this.consume();
+    }
+
+    private async consume(): Promise<void> {
+        try {
+            while (this.pending.length > 0) {
+                const next = this.pending.shift()!;
+                try {
+                    next.resolve(await this.handle(next.params));
+                } catch (error) {
+                    next.reject(error); // one failed steer must not stall the rest
+                }
+            }
+        } finally {
+            this.processing = false;
+        }
+    }
+}
diff --git a/src/__tests__/SteeringQueue.test.ts b/src/__tests__/SteeringQueue.test.ts
@@ -0,0 +1,115 @@
+import {describe, expect, it} from "vitest";
+import type {SessionSteerRequest, SessionSteeringResponse} from "../AcpExtensions";
+import {SteeringQueue} from "../SteeringQueue";
+
+function request(text: string): SessionSteerRequest {
+    return {sessionId: "session-id", prompt: [{type: "text", text}]};
+}
+
+function deferred<T>(): {promise: Promise<T>, resolve: (value: T) => void, reject: (error: unknown) => void} {
+    let resolve: (value: T) => void = () => {};
+    let reject: (error: unknown) => void = () => {};
+    const promise = new Promise<T>((innerResolve, innerReject) => {
+        resolve = innerResolve;
+        reject = innerReject;
+    });
+    return {promise, resolve, reject};
+}
+
+describe("SteeringQueue", () => {
+    it("runs enqueued requests one at a time in arrival order", async () => {
+        const order: string[] = [];
+        const queue = new SteeringQueue(async (params) => {
+            const text = (params.prompt[0] as {text: string}).text;
+            order.push(`start:${text}`);
+            await Promise.resolve();
+            order.push(`end:${text}`);
+            return {outcome: "injected"};
+        });
+
+        await Promise.all([
+            queue.enqueue(request("a")),
+            queue.enqueue(request("b")),
+            queue.enqueue(request("c")),
+        ]);
+
+        // Each request fully completes before the next one starts.
+        expect(order).toEqual([
+            "start:a", "end:a",
+            "start:b", "end:b",
+            "start:c", "end:c",
+        ]);
+    });
+
+    it("never overlaps two handlers", async () => {
+        let active = 0;
+        let maxActive = 0;
+        const queue = new SteeringQueue(async () => {
+            active++;
+            maxActive = Math.max(maxActive, active);
+            await Promise.resolve();
+            active--;
+            return {outcome: "injected"};
+        });
+
+        await Promise.all(Array.from({length: 5}, (_, i) => queue.enqueue(request(`${i}`))));
+
+        expect(maxActive).toBe(1);
+    });
+
+    it("delivers each handler result to its own caller", async () => {
+        const outcomes: SessionSteeringResponse["outcome"][] = ["injected", "startedNewTurn", "injected"];
+        let call = 0;
+        const queue = new SteeringQueue(async () => ({outcome: outcomes[call++]!}));
+
+        const results = await Promise.all([
+            queue.enqueue(request("a")),
+            queue.enqueue(request("b")),
+            queue.enqueue(request("c")),
+        ]);
+
+        expect(results).toEqual([
+            {outcome: "injected"},
+            {outcome: "startedNewTurn"},
+            {outcome: "injected"},
+        ]);
+    });
+
+    it("rejects only the failing caller and keeps draining the rest", async () => {
+        const seen: string[] = [];
+        const queue = new SteeringQueue(async (params) => {
+            const text = (params.prompt[0] as {text: string}).text;
+            seen.push(text);
+            if (text === "boom") {
+                throw new Error("steer failed");
+            }
+            return {outcome: "injected"};
+        });
+
+        const first = queue.enqueue(request("ok"));
+        const failing = queue.enqueue(request("boom"));
+        const third = queue.enqueue(request("after"));
+
+        await expect(first).resolves.toEqual({outcome: "injected"});
+        await expect(failing).rejects.toThrow("steer failed");
+        await expect(third).resolves.toEqual({outcome: "injected"});
+        expect(seen).toEqual(["ok", "boom", "after"]);
+    });
+
+    it("reports isIdle before, during, and after processing", async () => {
+        const gate = deferred<void>();
+        const queue = new SteeringQueue(async () => {
+            await gate.promise;
+            return {outcome: "injected"};
+        });
+
+        expect(queue.isIdle).toBe(true);
+
+        const inFlight = queue.enqueue(request("a"));
+        expect(queue.isIdle).toBe(false);
+
+        gate.resolve();
+        await inFlight;
+        expect(queue.isIdle).toBe(true);
+    });
+});
