fix(agent): serialize non-steering follow-ups (#3963)

Author: tatoalo

packages/agent/src/server/agent-server.test.ts

@@ -2640,6 +2640,59 @@ describe("AgentServer HTTP Mode", () => {
       expect(resetTurnMessages).not.toHaveBeenCalled();
     }, 20000);
 
+    it("does not queue steering behind an active non-steering turn", async () => {
+      const s = createServer();
+      await s.start();
+      let finishTurn!: (result: { stopReason: "end_turn" }) => void;
+      const prompt = vi.fn((params: { _meta?: { steer?: boolean } }) =>
+        params._meta?.steer === true
+          ? Promise.resolve({
+              stopReason: "end_turn" as const,
+              _meta: { steer: true },
+            })
+          : new Promise<{ stopReason: "end_turn" }>((resolve) => {
+              finishTurn = resolve;
+            }),
+      );
+      const serverInternals = s as unknown as {
+        session: { clientConnection: { prompt: typeof prompt } };
+      };
+      serverInternals.session.clientConnection.prompt = prompt;
+
+      const token = createToken();
+      const send = (id: string, steer = false) =>
+        fetch(`http://localhost:${port}/command`, {
+          method: "POST",
+          headers: {
+            Authorization: `Bearer ${token}`,
+            "Content-Type": "application/json",
+          },
+          body: JSON.stringify({
+            jsonrpc: "2.0",
+            id,
+            method: "user_message",
+            params: { content: id, messageId: id, ...(steer && { steer }) },
+          }),
+        });
+
+      const activeTurn = send("normal-turn");
+      await vi.waitFor(() => expect(prompt).toHaveBeenCalledTimes(1));
+
+      const steerResponse = await send("steer-turn", true);
+      await expect(steerResponse.json()).resolves.toMatchObject({
+        result: { stopReason: "steered", steered: true },
+      });
+      expect(prompt).toHaveBeenCalledTimes(2);
+      expect(prompt.mock.calls[1]?.[0]).toEqual(
+        expect.objectContaining({
+          _meta: expect.objectContaining({ steer: true }),
+        }),
+      );
+
+      finishTurn({ stopReason: "end_turn" });
+      await activeTurn;
+    }, 20000);
+
     it("declines steering without blocking on a fallback normal turn", async () => {
       const s = createServer();
       await s.start();
@@ -2951,18 +3004,21 @@ describe("AgentServer HTTP Mode", () => {
       );
       const { relaySpy, send } = await setupRelayEchoServer(prompt);
 
-      // The second message lands while the first turn is still in flight;
-      // each relay carries its own sender's id, not the first turn's.
+      // The second non-steering message lands while the first turn is still in
+      // flight. It waits for exclusive turn ownership, and each relay carries
+      // its own sender's id.
       const first = send("m-first");
       await vi.waitFor(() => expect(prompt).toHaveBeenCalledTimes(1));
       const second = send("m-second");
-      await vi.waitFor(() => expect(prompt).toHaveBeenCalledTimes(2));
+      await new Promise((resolve) => setTimeout(resolve, 25));
+      expect(prompt).toHaveBeenCalledTimes(1);
 
       pendingTurns[0]({ stopReason: "end_turn" });
       await first;
       await vi.waitFor(() => expect(relaySpy).toHaveBeenCalledTimes(1));
       expect(relaySpy.mock.calls[0][4]).toBe("m-first");
 
+      await vi.waitFor(() => expect(prompt).toHaveBeenCalledTimes(2));
       pendingTurns[1]({ stopReason: "end_turn" });
       await second;
       await vi.waitFor(() => expect(relaySpy).toHaveBeenCalledTimes(2));

packages/agent/src/server/agent-server.ts

@@ -407,6 +407,9 @@ export class AgentServer {
   private pendingCompactContinuationMessageIds = new Set<string>();
   private inFlightMessageDeliveries = new Map<string, Promise<unknown>>();
   private activeOwnedTurnCount = 0;
+  // Normal follow-ups own turns in arrival order. Explicit steering bypasses
+  // this tail so it can still reach the active adapter turn immediately.
+  private nonSteerDeliveryTail: Promise<void> = Promise.resolve();
   private pendingPermissions = new Map<
     string,
     {
@@ -1091,6 +1094,7 @@ export class AgentServer {
           this.inFlightMessageDeliveries.set(messageId, deliveryOutcome);
         }
         let deliveryCommitted = retryCompactContinuation;
+        let releaseNonSteerDelivery: (() => void) | undefined;
         const commitDelivery = (): void => {
           deliveryCommitted = true;
           if (!messageId) return;
@@ -1105,6 +1109,9 @@ export class AgentServer {
         };
 
         try {
+          if (params.steer !== true) {
+            releaseNonSteerDelivery = await this.acquireNonSteerDeliveryTurn();
+          }
           this.logger.debug("Received user_message command", {
             hasContent:
               typeof params.content === "string"
@@ -1302,6 +1309,7 @@ export class AgentServer {
           rejectDelivery(error);
           throw error;
         } finally {
+          releaseNonSteerDelivery?.();
           if (
             messageId &&
             this.inFlightMessageDeliveries.get(messageId) === deliveryOutcome
@@ -1930,6 +1938,16 @@ export class AgentServer {
     }
   }
 
+  private async acquireNonSteerDeliveryTurn(): Promise<() => void> {
+    const previous = this.nonSteerDeliveryTail;
+    let release!: () => void;
+    this.nonSteerDeliveryTail = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    await previous;
+    return release;
+  }
+
   /**
    * Send an initial/resume turn prompt, absorbing transient upstream
    * failures with a bounded number of retries. These turns run unattended