diff --git a/packages/cli/src/serve/httpAcpBridge.test.ts b/packages/cli/src/serve/httpAcpBridge.test.ts index 40139d9400..a9b5c85693 100644 --- a/packages/cli/src/serve/httpAcpBridge.test.ts +++ b/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; } class FakeAgent implements Agent { newSessionCalls: NewSessionRequest[] = []; + promptCalls: PromptRequest[] = []; + cancelCalls: CancelNotification[] = []; constructor(private readonly opts: FakeAgentOpts = {}) {} async initialize(_p: InitializeRequest): Promise { @@ -72,10 +83,16 @@ class FakeAgent implements Agent { async authenticate(_p: AuthenticateRequest): Promise { throw new Error('not implemented in test fake'); } - async prompt(_p: PromptRequest): Promise { + async prompt(p: PromptRequest): Promise { + this.promptCalls.push(p); + if (this.opts.promptImpl) { + return this.opts.promptImpl(p, this); + } return { stopReason: 'end_turn' }; } - async cancel(_p: CancelNotification): Promise {} + async cancel(p: CancelNotification): Promise { + this.cancelCalls.push(p); + } async setSessionMode( _p: SetSessionModeRequest, ): Promise { @@ -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((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, + ); + }); + }); }); diff --git a/packages/cli/src/serve/httpAcpBridge.ts b/packages/cli/src/serve/httpAcpBridge.ts index 74b027d59c..415554c78f 100644 --- a/packages/cli/src/serve/httpAcpBridge.ts +++ b/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; + /** + * 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; + + /** + * 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; + /** Test/inspection hook: number of live sessions. */ readonly sessionCount: number; @@ -70,6 +89,19 @@ export interface HttpAcpBridge { shutdown(): Promise; } +/** + * 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; } /** @@ -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(); diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index 3bfdc6a022..f3d6fd710d 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/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; + promptImpl?: ( + sessionId: string, + req: PromptRequest, + ) => Promise; + cancelImpl?: (sessionId: string, req?: CancelNotification) => Promise; } -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); diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index e1f63f32ff..3663cc5808 100644 --- a/packages/cli/src/serve/server.ts +++ b/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) + : {}; + 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[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) + : {}; + try { + await bridge.cancelSession(sessionId, { + ...(body as object), + sessionId, + } as Parameters[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) }); +}