feat(cli): wire POST /session/:id/prompt + /cancel for qwen serve (#3803)

Stage 1 follow-up after the bridge scaffold. Adds the two routes a client
needs to actually run a turn against the daemon.

Bridge:
  - `sendPrompt(sessionId, req)` looks up the session, FIFO-queues the
    call against the per-session prompt queue, and forwards through the
    SDK `ClientSideConnection.prompt`. Concurrent calls observe ACP's
    "one active prompt per session" invariant — second waits for first.
  - A failed prompt does NOT poison the queue; the tail catches and
    keeps draining so the next caller still runs (the original caller
    still sees its own rejection).
  - `cancelSession(sessionId, req?)` bypasses the queue and forwards
    the ACP notification immediately. ACP semantics: the agent winds
    down the *currently active* prompt; queued work is unaffected.
  - Both methods throw `SessionNotFoundError` (a typed Error subclass)
    when the id is unknown so route handlers can map cleanly to 404
    without brittle message matching.
  - Both methods overwrite the `sessionId` field in the request body
    with the routing id — a stale or spoofed body would otherwise be
    dispatched to the wrong agent process.

Routes:
  - `POST /session/:id/prompt` → 200 with PromptResponse, 400 on
    missing/non-array prompt, 404 on unknown session, 500 on agent
    error.
  - `POST /session/:id/cancel` → 204 always (cancel is a notification),
    404 on unknown session.

Tests (14 new — 7 bridge + 7 route, 0 regressions in the 4981 baseline):
  - sendPrompt: success forwards & returns response · routing-id
    overrides body sessionId · concurrent prompts FIFO-serialize
    (verified via per-prompt start/end ordering with a release latch) ·
    failed prompt doesn't block subsequent prompts · 404 for unknown id.
  - cancelSession: forwards with routing id · 404 for unknown id.
  - Routes: 200/400/404/500 paths for prompt; 204 with body or empty +
    404 for cancel.

Verified end-to-end against a real `qwen --acp` child:
  - POST /session/:id/prompt with `[{type:'text',text:'hi'}]` → 200
    `{"stopReason":"end_turn"}` in ~3.4s.
  - POST /session/:id/cancel → 204.
  - POST /session/does-not-exist/prompt → 404 with the unknown id
    surfaced in the body.

packages/cli/src/serve/httpAcpBridge.test.ts

@@ -30,6 +30,7 @@ import type {
 } from '@agentclientprotocol/sdk';
 import {
   createHttpAcpBridge,
+  SessionNotFoundError,
   type AcpChannel,
   type ChannelFactory,
 } from './httpAcpBridge.js';
@@ -41,10 +42,20 @@ interface FakeAgentOpts {
   initializeDelayMs?: number;
   /** Force `initialize` to throw. */
   initializeThrows?: Error;
+  /**
+   * Custom prompt handler. Default returns `end_turn` synchronously. Useful
+   * for test cases that want to observe prompt ordering.
+   */
+  promptImpl?: (
+    p: PromptRequest,
+    self: FakeAgent,
+  ) => Promise<PromptResponse> | PromptResponse;
 }
 
 class FakeAgent implements Agent {
   newSessionCalls: NewSessionRequest[] = [];
+  promptCalls: PromptRequest[] = [];
+  cancelCalls: CancelNotification[] = [];
   constructor(private readonly opts: FakeAgentOpts = {}) {}
 
   async initialize(_p: InitializeRequest): Promise<InitializeResponse> {
@@ -72,10 +83,16 @@ class FakeAgent implements Agent {
   async authenticate(_p: AuthenticateRequest): Promise<AuthenticateResponse> {
     throw new Error('not implemented in test fake');
   }
-  async prompt(_p: PromptRequest): Promise<PromptResponse> {
+  async prompt(p: PromptRequest): Promise<PromptResponse> {
+    this.promptCalls.push(p);
+    if (this.opts.promptImpl) {
+      return this.opts.promptImpl(p, this);
+    }
     return { stopReason: 'end_turn' };
   }
-  async cancel(_p: CancelNotification): Promise<void> {}
+  async cancel(p: CancelNotification): Promise<void> {
+    this.cancelCalls.push(p);
+  }
   async setSessionMode(
     _p: SetSessionModeRequest,
   ): Promise<SetSessionModeResponse> {
@@ -307,4 +324,188 @@ describe('createHttpAcpBridge', () => {
     expect(handles.every((h) => h.killed)).toBe(true);
     expect(bridge.sessionCount).toBe(0);
   });
+
+  describe('sendPrompt', () => {
+    it('forwards a prompt and returns the agent response', async () => {
+      const handles: ChannelHandle[] = [];
+      const factory: ChannelFactory = async () => {
+        const h = makeChannel({
+          promptImpl: () => ({ stopReason: 'max_tokens' }),
+        });
+        handles.push(h);
+        return h.channel;
+      };
+      const bridge = createHttpAcpBridge({ channelFactory: factory });
+      const session = await bridge.spawnOrAttach({ workspaceCwd: '/work/a' });
+
+      const result = await bridge.sendPrompt(session.sessionId, {
+        sessionId: session.sessionId,
+        prompt: [{ type: 'text', text: 'hi' }],
+      });
+      expect(result).toEqual({ stopReason: 'max_tokens' });
+      expect(handles[0]?.agent.promptCalls).toHaveLength(1);
+
+      await bridge.shutdown();
+    });
+
+    it('overrides a stale sessionId in the body with the routing id', async () => {
+      const handles: ChannelHandle[] = [];
+      const factory: ChannelFactory = async () => {
+        const h = makeChannel();
+        handles.push(h);
+        return h.channel;
+      };
+      const bridge = createHttpAcpBridge({ channelFactory: factory });
+      const session = await bridge.spawnOrAttach({ workspaceCwd: '/work/a' });
+
+      await bridge.sendPrompt(session.sessionId, {
+        // Body claims a different sessionId — bridge must not honor it.
+        sessionId: 'spoofed',
+        prompt: [{ type: 'text', text: 'hi' }],
+      });
+      expect(handles[0]?.agent.promptCalls[0]?.sessionId).toBe(
+        session.sessionId,
+      );
+
+      await bridge.shutdown();
+    });
+
+    it('FIFO-serializes concurrent prompts on the same session', async () => {
+      const order: string[] = [];
+      let resolveFirst: (() => void) | undefined;
+      const handles: ChannelHandle[] = [];
+      const factory: ChannelFactory = async () => {
+        const h = makeChannel({
+          promptImpl: async (p) => {
+            const tag =
+              (p.prompt[0] as { text?: string } | undefined)?.text ?? '?';
+            order.push(`start:${tag}`);
+            if (tag === 'first') {
+              await new Promise<void>((res) => {
+                resolveFirst = res;
+              });
+            }
+            order.push(`end:${tag}`);
+            return { stopReason: 'end_turn' };
+          },
+        });
+        handles.push(h);
+        return h.channel;
+      };
+      const bridge = createHttpAcpBridge({ channelFactory: factory });
+      const session = await bridge.spawnOrAttach({ workspaceCwd: '/work/a' });
+
+      const p1 = bridge.sendPrompt(session.sessionId, {
+        sessionId: session.sessionId,
+        prompt: [{ type: 'text', text: 'first' }],
+      });
+      const p2 = bridge.sendPrompt(session.sessionId, {
+        sessionId: session.sessionId,
+        prompt: [{ type: 'text', text: 'second' }],
+      });
+
+      // Give the event loop a chance to run the agent's start handler.
+      await new Promise((r) => setTimeout(r, 10));
+      // The second prompt MUST NOT have started before the first ended.
+      expect(order).toEqual(['start:first']);
+
+      resolveFirst!();
+      await Promise.all([p1, p2]);
+      expect(order).toEqual([
+        'start:first',
+        'end:first',
+        'start:second',
+        'end:second',
+      ]);
+
+      await bridge.shutdown();
+    });
+
+    it('a failed prompt does not poison the queue for subsequent prompts', async () => {
+      let promptCount = 0;
+      const handles: ChannelHandle[] = [];
+      const factory: ChannelFactory = async () => {
+        const h = makeChannel({
+          promptImpl: async () => {
+            promptCount += 1;
+            if (promptCount === 1) {
+              throw new Error('first prompt boom');
+            }
+            return { stopReason: 'end_turn' };
+          },
+        });
+        handles.push(h);
+        return h.channel;
+      };
+      const bridge = createHttpAcpBridge({ channelFactory: factory });
+      const session = await bridge.spawnOrAttach({ workspaceCwd: '/work/a' });
+
+      const failed = await bridge
+        .sendPrompt(session.sessionId, {
+          sessionId: session.sessionId,
+          prompt: [{ type: 'text', text: 'a' }],
+        })
+        .then(
+          () => null,
+          (e: unknown) => e,
+        );
+      expect(failed).not.toBeNull();
+
+      const ok = await bridge.sendPrompt(session.sessionId, {
+        sessionId: session.sessionId,
+        prompt: [{ type: 'text', text: 'b' }],
+      });
+      expect(ok).toEqual({ stopReason: 'end_turn' });
+
+      await bridge.shutdown();
+    });
+
+    it('throws SessionNotFoundError for unknown session ids', async () => {
+      const bridge = createHttpAcpBridge({
+        channelFactory: async () => {
+          throw new Error('factory should not be called');
+        },
+      });
+      await expect(
+        bridge.sendPrompt('unknown', {
+          sessionId: 'unknown',
+          prompt: [{ type: 'text', text: 'x' }],
+        }),
+      ).rejects.toBeInstanceOf(SessionNotFoundError);
+    });
+  });
+
+  describe('cancelSession', () => {
+    it('forwards a cancel notification with the routing id', async () => {
+      const handles: ChannelHandle[] = [];
+      const factory: ChannelFactory = async () => {
+        const h = makeChannel();
+        handles.push(h);
+        return h.channel;
+      };
+      const bridge = createHttpAcpBridge({ channelFactory: factory });
+      const session = await bridge.spawnOrAttach({ workspaceCwd: '/work/a' });
+
+      await bridge.cancelSession(session.sessionId);
+      // Cancel is a notification — let it propagate before observing.
+      await new Promise((r) => setTimeout(r, 10));
+      expect(handles[0]?.agent.cancelCalls).toHaveLength(1);
+      expect(handles[0]?.agent.cancelCalls[0]?.sessionId).toBe(
+        session.sessionId,
+      );
+
+      await bridge.shutdown();
+    });
+
+    it('throws SessionNotFoundError for unknown session ids', async () => {
+      const bridge = createHttpAcpBridge({
+        channelFactory: async () => {
+          throw new Error('factory should not be called');
+        },
+      });
+      await expect(bridge.cancelSession('unknown')).rejects.toBeInstanceOf(
+        SessionNotFoundError,
+      );
+    });
+  });
 });

packages/cli/src/serve/httpAcpBridge.ts

@@ -14,7 +14,10 @@ import {
   ndJsonStream,
 } from '@agentclientprotocol/sdk';
 import type {
+  CancelNotification,
   Client,
+  PromptRequest,
+  PromptResponse,
   ReadTextFileRequest,
   ReadTextFileResponse,
   RequestPermissionRequest,
@@ -63,6 +66,22 @@ export interface HttpAcpBridge {
    */
   spawnOrAttach(req: BridgeSpawnRequest): Promise<BridgeSession>;
 
+  /**
+   * Forward a prompt to the agent. Concurrent prompts against the same
+   * session FIFO-serialize through a per-session queue (ACP guarantees
+   * "one active prompt per session"). Throws `SessionNotFoundError` when
+   * the id is unknown.
+   */
+  sendPrompt(sessionId: string, req: PromptRequest): Promise<PromptResponse>;
+
+  /**
+   * Cancel the in-flight prompt on the session. ACP-side this is a
+   * notification, not a request — the agent acknowledges by resolving the
+   * active `prompt()` with a `cancelled` stop reason. Throws
+   * `SessionNotFoundError` when the id is unknown.
+   */
+  cancelSession(sessionId: string, req?: CancelNotification): Promise<void>;
+
   /** Test/inspection hook: number of live sessions. */
   readonly sessionCount: number;
 
@@ -70,6 +89,19 @@ export interface HttpAcpBridge {
   shutdown(): Promise<void>;
 }
 
+/**
+ * Routes catch this to map to HTTP 404. Distinct from generic Error so the
+ * route layer doesn't have to brittle-match on message text.
+ */
+export class SessionNotFoundError extends Error {
+  readonly sessionId: string;
+  constructor(sessionId: string) {
+    super(`No session with id "${sessionId}"`);
+    this.name = 'SessionNotFoundError';
+    this.sessionId = sessionId;
+  }
+}
+
 /**
  * One ACP NDJSON channel to a single agent. Tests inject a fake by replacing
  * the channel factory; production uses `defaultSpawnChannelFactory`.
@@ -102,6 +134,14 @@ interface SessionEntry {
   connection: ClientSideConnection;
   /** Stage 1 buffer; consumed by SSE wiring in the next PR. */
   notifications: SessionNotification[];
+  /**
+   * Tail of the per-session prompt queue. Each new prompt chains off the
+   * resolved (or rejected) state of this promise so prompts run one at a
+   * time in arrival order. Always resolves — failures are swallowed at the
+   * tail so a prior failure doesn't block subsequent prompts; the original
+   * caller still observes the rejection on its own returned promise.
+   */
+  promptQueue: Promise<void>;
 }
 
 /**
@@ -218,6 +258,7 @@ export function createHttpAcpBridge(opts: BridgeOptions = {}): HttpAcpBridge {
           channel,
           connection,
           notifications: [],
+          promptQueue: Promise.resolve(),
         };
         byWorkspace.set(workspaceKey, entry);
         byId.set(entry.sessionId, entry);
@@ -233,6 +274,37 @@ export function createHttpAcpBridge(opts: BridgeOptions = {}): HttpAcpBridge {
       }
     },
 
+    async sendPrompt(sessionId, req) {
+      const entry = byId.get(sessionId);
+      if (!entry) throw new SessionNotFoundError(sessionId);
+      // Force the body's sessionId to match the routing id — a client that
+      // sent a stale id in the body would otherwise be dispatched to the
+      // wrong agent process.
+      const normalized: PromptRequest = { ...req, sessionId };
+      const result = entry.promptQueue.then(() =>
+        entry.connection.prompt(normalized),
+      );
+      // Tail swallows failures so subsequent prompts still run. The caller
+      // still sees rejections on its own `result` reference.
+      entry.promptQueue = result.then(
+        () => undefined,
+        () => undefined,
+      );
+      return result;
+    },
+
+    async cancelSession(sessionId, req) {
+      const entry = byId.get(sessionId);
+      if (!entry) throw new SessionNotFoundError(sessionId);
+      // Cancel intentionally bypasses the prompt queue: it's a notification
+      // that the agent uses to wind down the *currently active* prompt, not
+      // something to wait behind queued work.
+      const notif: CancelNotification = req
+        ? { ...req, sessionId }
+        : { sessionId };
+      await entry.connection.cancel(notif);
+    },
+
     async shutdown() {
       const entries = Array.from(byId.values());
       byWorkspace.clear();

packages/cli/src/serve/server.test.ts

@@ -9,9 +9,15 @@ import request from 'supertest';
 import { createServeApp } from './server.js';
 import { runQwenServe, type RunHandle } from './runQwenServe.js';
 import type {
-  BridgeSession,
-  BridgeSpawnRequest,
-  HttpAcpBridge,
+  CancelNotification,
+  PromptRequest,
+  PromptResponse,
+} from '@agentclientprotocol/sdk';
+import {
+  SessionNotFoundError,
+  type BridgeSession,
+  type BridgeSpawnRequest,
+  type HttpAcpBridge,
 } from './httpAcpBridge.js';
 import {
   CAPABILITIES_SCHEMA_VERSION,
@@ -27,23 +33,39 @@ const baseOpts: ServeOptions = {
 
 interface FakeBridgeOpts {
   spawnImpl?: (req: BridgeSpawnRequest) => Promise<BridgeSession>;
+  promptImpl?: (
+    sessionId: string,
+    req: PromptRequest,
+  ) => Promise<PromptResponse>;
+  cancelImpl?: (sessionId: string, req?: CancelNotification) => Promise<void>;
 }
 
-function fakeBridge(opts: FakeBridgeOpts = {}): HttpAcpBridge & {
+interface FakeBridge extends HttpAcpBridge {
   calls: BridgeSpawnRequest[];
+  promptCalls: Array<{ sessionId: string; req: PromptRequest }>;
+  cancelCalls: Array<{ sessionId: string; req?: CancelNotification }>;
   shutdownCalls: number;
-} {
+}
+
+function fakeBridge(opts: FakeBridgeOpts = {}): FakeBridge {
   const calls: BridgeSpawnRequest[] = [];
+  const promptCalls: FakeBridge['promptCalls'] = [];
+  const cancelCalls: FakeBridge['cancelCalls'] = [];
   let shutdownCalls = 0;
-  const impl =
+  const spawnImpl =
     opts.spawnImpl ??
     (async (req) => ({
       sessionId: `fake-${calls.length}`,
       workspaceCwd: req.workspaceCwd,
       attached: false,
     }));
+  const promptImpl =
+    opts.promptImpl ?? (async () => ({ stopReason: 'end_turn' }));
+  const cancelImpl = opts.cancelImpl ?? (async () => {});
   return {
     calls,
+    promptCalls,
+    cancelCalls,
     get shutdownCalls() {
       return shutdownCalls;
     },
@@ -51,10 +73,18 @@ function fakeBridge(opts: FakeBridgeOpts = {}): HttpAcpBridge & {
       return calls.length;
     },
     async spawnOrAttach(req) {
-      const result = await impl(req);
+      const result = await spawnImpl(req);
       calls.push(req);
       return result;
     },
+    async sendPrompt(sessionId, req) {
+      promptCalls.push({ sessionId, req });
+      return promptImpl(sessionId, req);
+    },
+    async cancelSession(sessionId, req) {
+      cancelCalls.push({ sessionId, req });
+      return cancelImpl(sessionId, req);
+    },
     async shutdown() {
       shutdownCalls += 1;
     },
@@ -162,6 +192,108 @@ describe('createServeApp', () => {
     });
   });
 
+  describe('POST /session/:id/prompt', () => {
+    it('200 with PromptResponse on success; route :id wins over body sessionId', async () => {
+      const bridge = fakeBridge({
+        promptImpl: async () => ({ stopReason: 'end_turn' }),
+      });
+      const app = createServeApp(baseOpts, undefined, { bridge });
+      const res = await request(app)
+        .post('/session/session-A/prompt')
+        .set('Host', `127.0.0.1:${baseOpts.port}`)
+        .send({
+          sessionId: 'spoofed-session-B',
+          prompt: [{ type: 'text', text: 'hi' }],
+        });
+      expect(res.status).toBe(200);
+      expect(res.body).toEqual({ stopReason: 'end_turn' });
+      expect(bridge.promptCalls).toHaveLength(1);
+      expect(bridge.promptCalls[0]?.sessionId).toBe('session-A');
+      expect(bridge.promptCalls[0]?.req.sessionId).toBe('session-A');
+    });
+
+    it('400 when prompt body is missing', async () => {
+      const bridge = fakeBridge();
+      const app = createServeApp(baseOpts, undefined, { bridge });
+      const res = await request(app)
+        .post('/session/session-A/prompt')
+        .set('Host', `127.0.0.1:${baseOpts.port}`)
+        .send({});
+      expect(res.status).toBe(400);
+      expect(bridge.promptCalls).toHaveLength(0);
+    });
+
+    it('404 when bridge reports unknown session', async () => {
+      const bridge = fakeBridge({
+        promptImpl: async (sessionId) => {
+          throw new SessionNotFoundError(sessionId);
+        },
+      });
+      const app = createServeApp(baseOpts, undefined, { bridge });
+      const res = await request(app)
+        .post('/session/missing/prompt')
+        .set('Host', `127.0.0.1:${baseOpts.port}`)
+        .send({ prompt: [{ type: 'text', text: 'hi' }] });
+      expect(res.status).toBe(404);
+      expect(res.body.sessionId).toBe('missing');
+    });
+
+    it('500 on generic bridge errors', async () => {
+      const bridge = fakeBridge({
+        promptImpl: async () => {
+          throw new Error('agent crashed');
+        },
+      });
+      const app = createServeApp(baseOpts, undefined, { bridge });
+      const res = await request(app)
+        .post('/session/session-A/prompt')
+        .set('Host', `127.0.0.1:${baseOpts.port}`)
+        .send({ prompt: [{ type: 'text', text: 'hi' }] });
+      expect(res.status).toBe(500);
+      expect(res.body).toEqual({ error: 'agent crashed' });
+    });
+  });
+
+  describe('POST /session/:id/cancel', () => {
+    it('204 on success and forwards routing id', async () => {
+      const bridge = fakeBridge();
+      const app = createServeApp(baseOpts, undefined, { bridge });
+      const res = await request(app)
+        .post('/session/session-A/cancel')
+        .set('Host', `127.0.0.1:${baseOpts.port}`)
+        .send({ sessionId: 'spoofed-B' });
+      expect(res.status).toBe(204);
+      expect(res.body).toEqual({});
+      expect(bridge.cancelCalls).toHaveLength(1);
+      expect(bridge.cancelCalls[0]?.sessionId).toBe('session-A');
+      expect(bridge.cancelCalls[0]?.req?.sessionId).toBe('session-A');
+    });
+
+    it('204 with empty body', async () => {
+      const bridge = fakeBridge();
+      const app = createServeApp(baseOpts, undefined, { bridge });
+      const res = await request(app)
+        .post('/session/session-A/cancel')
+        .set('Host', `127.0.0.1:${baseOpts.port}`);
+      expect(res.status).toBe(204);
+      expect(bridge.cancelCalls).toHaveLength(1);
+    });
+
+    it('404 on unknown session', async () => {
+      const bridge = fakeBridge({
+        cancelImpl: async (sessionId) => {
+          throw new SessionNotFoundError(sessionId);
+        },
+      });
+      const app = createServeApp(baseOpts, undefined, { bridge });
+      const res = await request(app)
+        .post('/session/missing/cancel')
+        .set('Host', `127.0.0.1:${baseOpts.port}`);
+      expect(res.status).toBe(404);
+      expect(res.body.sessionId).toBe('missing');
+    });
+  });
+
   describe('bearer auth', () => {
     it('is open by default (loopback developer convenience)', async () => {
       const app = createServeApp(baseOpts);

packages/cli/src/serve/server.ts

@@ -8,7 +8,11 @@ import * as path from 'node:path';
 import express from 'express';
 import type { Application } from 'express';
 import { bearerAuth, denyBrowserOriginCors, hostAllowlist } from './auth.js';
-import { createHttpAcpBridge, type HttpAcpBridge } from './httpAcpBridge.js';
+import {
+  createHttpAcpBridge,
+  SessionNotFoundError,
+  type HttpAcpBridge,
+} from './httpAcpBridge.js';
 import {
   CAPABILITIES_SCHEMA_VERSION,
   STAGE1_FEATURES,
@@ -89,5 +93,59 @@ export function createServeApp(
     }
   });
 
+  app.post('/session/:id/prompt', async (req, res) => {
+    const sessionId = req.params['id'];
+    const body =
+      typeof req.body === 'object' && req.body !== null
+        ? (req.body as Record<string, unknown>)
+        : {};
+    const prompt = body['prompt'];
+    if (!Array.isArray(prompt)) {
+      res
+        .status(400)
+        .json({
+          error: '`prompt` is required and must be an array of content blocks',
+        });
+      return;
+    }
+    try {
+      const result = await bridge.sendPrompt(sessionId, {
+        ...(body as object),
+        sessionId,
+        prompt,
+      } as Parameters<HttpAcpBridge['sendPrompt']>[1]);
+      res.status(200).json(result);
+    } catch (err) {
+      sendBridgeError(res, err);
+    }
+  });
+
+  app.post('/session/:id/cancel', async (req, res) => {
+    const sessionId = req.params['id'];
+    const body =
+      typeof req.body === 'object' && req.body !== null
+        ? (req.body as Record<string, unknown>)
+        : {};
+    try {
+      await bridge.cancelSession(sessionId, {
+        ...(body as object),
+        sessionId,
+      } as Parameters<HttpAcpBridge['cancelSession']>[1]);
+      res.status(204).end();
+    } catch (err) {
+      sendBridgeError(res, err);
+    }
+  });
+
   return app;
 }
+
+function sendBridgeError(res: import('express').Response, err: unknown): void {
+  if (err instanceof SessionNotFoundError) {
+    res.status(404).json({ error: err.message, sessionId: err.sessionId });
+    return;
+  }
+  res
+    .status(500)
+    .json({ error: err instanceof Error ? err.message : String(err) });
+}