feat(daemon): deliver web-shell mid-turn messages into the running turn

packages/acp-bridge/src/bridge.test.ts

@@ -10301,3 +10301,241 @@ describe('close on last client detach', () => {
     await bridge.shutdown();
   });
 });
+
+/**
+ * `enqueueMidTurnMessage` backs the web-shell mid-turn drain: the browser
+ * pushes a message typed during a turn, the ACP child drains it via
+ * `craft/drainMidTurnQueue` (answered by BridgeClient.extMethod). The accept
+ * gate is the exactly-once linchpin — it must reject when the session is idle
+ * so the browser's own next-turn queue stays the single delivery path in the
+ * settle-window race.
+ */
+describe('createAcpSessionBridge — mid-turn message queue (enqueueMidTurnMessage)', () => {
+  function hangingPromptFactory(): {
+    factory: ChannelFactory;
+    release: () => void;
+  } {
+    let release: (() => void) | undefined;
+    const factory: ChannelFactory = async () =>
+      makeChannel({
+        promptImpl: async () => {
+          await new Promise<void>((res) => {
+            release = res;
+          });
+          return { stopReason: 'end_turn' };
+        },
+      }).channel;
+    return { factory, release: () => release?.() };
+  }
+
+  it('rejects (accepted:false) when the session is idle', async () => {
+    const bridge = makeBridge({
+      channelFactory: async () => makeChannel().channel,
+    });
+    const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
+    expect(bridge.enqueueMidTurnMessage(session.sessionId, 'later')).toEqual({
+      accepted: false,
+    });
+    await bridge.shutdown();
+  });
+
+  it('accepts while a turn is in flight, then rejects again once it settles', async () => {
+    const { factory, release } = hangingPromptFactory();
+    const bridge = makeBridge({ channelFactory: factory });
+    const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
+    const promptPromise = bridge
+      .sendPrompt(
+        session.sessionId,
+        {
+          sessionId: session.sessionId,
+          prompt: [{ type: 'text', text: 'run tools' }],
+        },
+        undefined,
+        { clientId: session.clientId },
+      )
+      .catch(() => {});
+
+    // Let the FIFO worker pick up the (hanging) prompt before asserting.
+    await new Promise((r) => setTimeout(r, 10));
+    expect(
+      bridge.enqueueMidTurnMessage(session.sessionId, 'also check tests'),
+    ).toEqual({ accepted: true });
+
+    // Settle the turn → the queue flips back to idle and the undrained copy is
+    // dropped server-side (the browser resends it as the next turn).
+    release();
+    await promptPromise;
+    expect(
+      bridge.enqueueMidTurnMessage(session.sessionId, 'next time'),
+    ).toEqual({ accepted: false });
+    await bridge.shutdown();
+  });
+
+  it('rejects a whitespace-only message even while busy', async () => {
+    const { factory, release } = hangingPromptFactory();
+    const bridge = makeBridge({ channelFactory: factory });
+    const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
+    const promptPromise = bridge
+      .sendPrompt(
+        session.sessionId,
+        {
+          sessionId: session.sessionId,
+          prompt: [{ type: 'text', text: 'x' }],
+        },
+        undefined,
+        { clientId: session.clientId },
+      )
+      .catch(() => {});
+    await new Promise((r) => setTimeout(r, 10));
+    expect(bridge.enqueueMidTurnMessage(session.sessionId, '   ')).toEqual({
+      accepted: false,
+    });
+    release();
+    await promptPromise;
+    await bridge.shutdown();
+  });
+
+  it('throws SessionNotFoundError for an unknown session', async () => {
+    const bridge = makeBridge({
+      channelFactory: async () => makeChannel().channel,
+    });
+    expect(() => bridge.enqueueMidTurnMessage('nope', 'hi')).toThrow(
+      SessionNotFoundError,
+    );
+    await bridge.shutdown();
+  });
+
+  it('drains the queue through the child connection; a second drain is empty', async () => {
+    let release: (() => void) | undefined;
+    const handle = makeChannel({
+      promptImpl: async () => {
+        await new Promise<void>((r) => {
+          release = r;
+        });
+        return { stopReason: 'end_turn' };
+      },
+    });
+    const bridge = makeBridge({ channelFactory: async () => handle.channel });
+    const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
+    const prompt = bridge
+      .sendPrompt(
+        session.sessionId,
+        { sessionId: session.sessionId, prompt: [{ type: 'text', text: 'go' }] },
+        undefined,
+        { clientId: session.clientId },
+      )
+      .catch(() => {});
+    await new Promise((r) => setTimeout(r, 10));
+
+    bridge.enqueueMidTurnMessage(session.sessionId, 'm1');
+    bridge.enqueueMidTurnMessage(session.sessionId, 'm2');
+
+    // The child pulls them via the ext-method the real Session calls between
+    // tool batches — exercising bridge queue ⇄ BridgeClient.extMethod end to end.
+    const drained = await handle.agentConnection.extMethod(
+      'craft/drainMidTurnQueue',
+      { sessionId: session.sessionId },
+    );
+    expect(drained).toEqual({ messages: ['m1', 'm2'] });
+    // Spliced out, so the next batch's drain is empty.
+    expect(
+      await handle.agentConnection.extMethod('craft/drainMidTurnQueue', {
+        sessionId: session.sessionId,
+      }),
+    ).toEqual({ messages: [] });
+
+    release?.();
+    await prompt;
+    await bridge.shutdown();
+  });
+
+  it('clears undrained messages at settle — not re-drained on the next turn', async () => {
+    const releases: Array<() => void> = [];
+    const handle = makeChannel({
+      promptImpl: async () => {
+        await new Promise<void>((r) => {
+          releases.push(r);
+        });
+        return { stopReason: 'end_turn' };
+      },
+    });
+    const bridge = makeBridge({ channelFactory: async () => handle.channel });
+    const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
+    const send = (text: string) =>
+      bridge
+        .sendPrompt(
+          session.sessionId,
+          { sessionId: session.sessionId, prompt: [{ type: 'text', text }] },
+          undefined,
+          { clientId: session.clientId },
+        )
+        .catch(() => {});
+    const drain = () =>
+      handle.agentConnection.extMethod('craft/drainMidTurnQueue', {
+        sessionId: session.sessionId,
+      });
+
+    // Turn 1: enqueue, do NOT drain, then settle.
+    const t1 = send('t1');
+    await new Promise((r) => setTimeout(r, 10));
+    expect(bridge.enqueueMidTurnMessage(session.sessionId, 'leftover')).toEqual({
+      accepted: true,
+    });
+    releases[0]!();
+    await t1;
+
+    // Turn 2: the drain must be empty — the leftover was dropped at turn-1
+    // settle, NOT carried into turn 2's first batch. (Deleting the settle-clear
+    // line makes this return ['leftover'].)
+    const t2 = send('t2');
+    await new Promise((r) => setTimeout(r, 10));
+    expect(await drain()).toEqual({ messages: [] });
+
+    releases[1]!();
+    await t2;
+    await bridge.shutdown();
+  });
+
+  it('keeps the queue across a back-to-back prompt FIFO, clearing only at true idle', async () => {
+    const releases: Array<() => void> = [];
+    const handle = makeChannel({
+      promptImpl: async () => {
+        await new Promise<void>((r) => {
+          releases.push(r);
+        });
+        return { stopReason: 'end_turn' };
+      },
+    });
+    const bridge = makeBridge({ channelFactory: async () => handle.channel });
+    const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
+    const send = (text: string) =>
+      bridge
+        .sendPrompt(
+          session.sessionId,
+          { sessionId: session.sessionId, prompt: [{ type: 'text', text }] },
+          undefined,
+          { clientId: session.clientId },
+        )
+        .catch(() => {});
+
+    const p1 = send('p1');
+    await new Promise((r) => setTimeout(r, 10));
+    const p2 = send('p2'); // queued behind p1 ⇒ pendingPromptCount = 2
+    expect(bridge.enqueueMidTurnMessage(session.sessionId, 'x')).toEqual({
+      accepted: true,
+    });
+
+    releases[0]!(); // settle p1 — session still busy (p2 pending), so NOT idle
+    await new Promise((r) => setTimeout(r, 10));
+    // Survived the p1→p2 boundary: the clear only fires at true idle.
+    expect(
+      await handle.agentConnection.extMethod('craft/drainMidTurnQueue', {
+        sessionId: session.sessionId,
+      }),
+    ).toEqual({ messages: ['x'] });
+
+    releases[1]!();
+    await Promise.all([p1, p2]);
+    await bridge.shutdown();
+  });
+});

packages/acp-bridge/src/bridge.ts

@@ -229,6 +229,19 @@ interface SessionEntry {
   promptQueue: Promise<void>;
   /** Accepted prompts that have not settled yet (queued + active). */
   pendingPromptCount: number;
+  /**
+   * Mid-turn user messages pushed by the browser (`POST
+   * /session/:id/mid-turn-message`) while a turn is running. The ACP child
+   * drains these between tool batches via the `craft/drainMidTurnQueue`
+   * ext-method so the model sees them before the turn ends. The queue is
+   * accepted into only while the session is busy (`pendingPromptCount > 0`)
+   * and emptied when the session next goes idle — see the settle handler in
+   * `sendPrompt`. The browser keeps its own copy as the next-turn fallback,
+   * so a message left undrained here is NOT lost: it is dropped server-side
+   * (preventing a stale next-turn re-injection) and resent by the browser as
+   * a fresh prompt.
+   */
+  midTurnMessageQueue: string[];
   /**
    * Per-session model-change FIFO. Prevents two concurrent
    * `applyModelServiceId` calls (e.g. simultaneous attach-with-different-
@@ -650,6 +663,12 @@ const SESSION_RECAP_TIMEOUT_MS = 60_000;
 const SESSION_BTW_TIMEOUT_MS = 60_000;
 const SHELL_COMMAND_TIMEOUT_MS = 120_000;
 const MAX_SHELL_OUTPUT_FOR_HISTORY = 10_000;
+// Per-session cap on undrained mid-turn messages. Mirrors the bound `/prompt`
+// gets from `maxPendingPromptsPerSession`: a busy turn with no drain point (a
+// long tool-free generation) must not let a client pin unbounded messages in
+// the in-memory queue. Past the cap, `enqueueMidTurnMessage` returns
+// `{ accepted: false }` and the browser keeps the message for the next turn.
+const MAX_MID_TURN_QUEUE_DEPTH = 20;
 const DEFAULT_MAX_SESSIONS = 20;
 // Keep in sync with CLI serve/server.ts and SDK DaemonClient.ts.
 const DEFAULT_MAX_PENDING_PROMPTS_PER_SESSION = 5;
@@ -2002,6 +2021,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
       events,
       promptQueue: Promise.resolve(),
       pendingPromptCount: 0,
+      midTurnMessageQueue: [],
       modelChangeQueue: Promise.resolve(),
       approvalModeQueue: Promise.resolve(),
       modelPublishGeneration: 0,
@@ -2941,7 +2961,26 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
         () => undefined,
         () => undefined,
       );
-      result.finally(releasePromptSlot).catch(() => {});
+      result
+        .finally(() => {
+          releasePromptSlot();
+          // Mid-turn messages are scoped to the turn the user typed them
+          // during. Once the session goes fully idle with some still
+          // undrained, drop the server-side copy: the browser still holds
+          // them in its own queue and resends them as the next turn. Leaving
+          // them here would let the NEXT turn's first tool batch inject a
+          // stale message the browser ALSO resends — double delivery. The
+          // `pendingPromptCount === 0` guard keeps queued messages intact
+          // across a back-to-back FIFO of prompts (still "one turn" to the
+          // user) and only clears at the true idle boundary.
+          if (
+            entry.pendingPromptCount === 0 &&
+            entry.midTurnMessageQueue.length > 0
+          ) {
+            entry.midTurnMessageQueue.length = 0;
+          }
+        })
+        .catch(() => {});
       return result;
     },
 
@@ -4032,6 +4071,31 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
       };
     },
 
+    enqueueMidTurnMessage(sessionId, message) {
+      const entry = byId.get(sessionId);
+      if (!entry) throw new SessionNotFoundError(sessionId);
+      const trimmed = message.trim();
+      // Reject empty messages and — critically — messages that arrive while
+      // the session is idle. The browser only pushes here when it believes a
+      // turn is running, but the turn may have settled in the small window
+      // before its turn-complete frame landed. Accepting an idle message
+      // would strand it until the NEXT turn's first tool batch drained it,
+      // by which point the browser has already resent it as a fresh prompt —
+      // double delivery. Rejecting keeps the browser's next-turn fallback the
+      // single delivery path in that race.
+      if (trimmed.length === 0 || entry.pendingPromptCount === 0) {
+        return { accepted: false };
+      }
+      // Bound queue depth (see MAX_MID_TURN_QUEUE_DEPTH). Full → reject so the
+      // browser keeps the message in its own queue for the next turn rather than
+      // pinning it here unboundedly when the turn has no drain point.
+      if (entry.midTurnMessageQueue.length >= MAX_MID_TURN_QUEUE_DEPTH) {
+        return { accepted: false };
+      }
+      entry.midTurnMessageQueue.push(trimmed);
+      return { accepted: true };
+    },
+
     async generateSessionBtw(sessionId, question, signal, _context) {
       const entry = byId.get(sessionId);
       if (!entry) throw new SessionNotFoundError(sessionId);