diff --git a/README.md b/README.md index 4de4f41a..02145da8 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,7 @@ Expect some minor breaking changes. - Adds a small set of built-in commands for headless/editor usage - Supports skill commands (if enabled in pi settings, they appear as `/skill:skill-name` in the ACP client) - Skills are loaded by pi directly and are available in ACP sessions +- Prompts received during an active turn use Pi's native steering queue instead of waiting for the turn to finish - (Zed) `pi-acp` emits “startup info” block into the session (pi version, context, skills, prompts, extensions - similar to `pi` in the terminal). You can disable it by setting `quietStartup: true` in pi settings (`~/.pi/agent/settings.json` or `/.pi/settings.json`). When `quietStartup` is enabled, `pi-acp` will still emit a 'New version available' message if the installed pi version is outdated. - (Zed) Session history is supported in Zed starting with [`v0.225.0`](https://zed.dev/releases/preview/0.225.0). Session loading / history maps to pi's session files. Sessions can be resumed both in `pi` and in the ACP client. @@ -197,7 +198,6 @@ Project layout: - No ACP filesystem delegation (`fs/*`) and no ACP terminal delegation (`terminal/*`). pi reads/writes and executes locally. - MCP servers are accepted in ACP params and stored in session state, but not wired through to pi in this adapter. If you use [pi MCP adapter](https://github.com/nicobailon/pi-mcp-adapter) it will be available in the ACP client. - Assistant streaming is currently sent as `agent_message_chunk` (no separate thought stream). -- Queue is implemented client-side and should work like pi's `one-at-a-time` - ~~ACP clients don't yet suport session history, but ACP sessions from `pi-acp` can be `/resume`d in pi directly~~ ## License diff --git a/src/acp/session.ts b/src/acp/session.ts index 2f40e46f..eded8298 100644 --- a/src/acp/session.ts +++ b/src/acp/session.ts @@ -43,11 +43,9 @@ type PendingTurn = { reject: (err: unknown) => void } -type QueuedTurn = { +type TurnRequest = PendingTurn & { message: string images: unknown[] - resolve: (reason: StopReason) => void - reject: (err: unknown) => void } type PermissionResponse = Awaited> @@ -271,9 +269,10 @@ export class PiAcpSession { // Applies to the currently running turn. private cancelRequested = false - // Current in-flight turn (if any). Additional prompts are queued. + // Current in-flight turn, prompts steered into it, and startup/settlement-race prompts queued after it. private pendingTurn: PendingTurn | null = null - private readonly turnQueue: QueuedTurn[] = [] + private readonly steeredTurns: PendingTurn[] = [] + private readonly turnQueue: TurnRequest[] = [] // Track tool call statuses and ensure they are monotonic (pending -> in_progress -> completed). // Some pi events can arrive out of order (e.g. late toolcall_* deltas after execution starts), // and clients may hide progress if we ever downgrade back to `pending`. @@ -338,46 +337,49 @@ export class PiAcpSession { const expandedMessage = expandSlashCommand(message, this.fileCommands) const turnPromise = new Promise((resolve, reject) => { - const queued: QueuedTurn = { message: expandedMessage, images, resolve, reject } + const turn: TurnRequest = { message: expandedMessage, images, resolve, reject } - // If a turn is already running, enqueue. if (this.pendingTurn) { - this.turnQueue.push(queued) - - // Best-effort: notify client that a prompt was queued. - // This doesn't work in Zed yet, needs to be revisited - this.emit({ - sessionUpdate: 'agent_message_chunk', - content: { - type: 'text', - text: `Queued message (position ${this.turnQueue.length}).` - } - }) - - // Also publish queue depth via session info metadata. - // This also not visible in the client - this.emit({ - sessionUpdate: 'session_info_update', - _meta: { piAcp: { queueDepth: this.turnQueue.length, running: true } } - }) - + if (this.inAgentLoop) { + const steeredTurn: PendingTurn = { resolve, reject } + this.steeredTurns.push(steeredTurn) + this.proc.steer(expandedMessage, images).catch(err => { + const index = this.steeredTurns.indexOf(steeredTurn) + if (index >= 0) this.steeredTurns.splice(index, 1) + reject(err) + }) + } else { + this.turnQueue.push(turn) + this.emit({ + sessionUpdate: 'agent_message_chunk', + content: { + type: 'text', + text: `Queued message (position ${this.turnQueue.length}).` + } + }) + this.emit({ + sessionUpdate: 'session_info_update', + _meta: { piAcp: { queueDepth: this.turnQueue.length, running: true } } + }) + } return } - // No turn is running; start immediately. - this.startTurn(queued) + this.startTurn(turn) }) return turnPromise } async cancel(): Promise { - // Cancel current and clear any queued prompts. this.cancelRequested = true + const steeredTurns = this.steeredTurns.splice(0, this.steeredTurns.length) + for (const turn of steeredTurns) turn.resolve('cancelled') + if (this.turnQueue.length) { - const queued = this.turnQueue.splice(0, this.turnQueue.length) - for (const t of queued) t.resolve('cancelled') + const queuedTurns = this.turnQueue.splice(0, this.turnQueue.length) + for (const turn of queuedTurns) turn.resolve('cancelled') this.emit({ sessionUpdate: 'agent_message_chunk', @@ -470,13 +472,13 @@ export class PiAcpSession { this.bashOutputSnapshots.delete(toolCallId) } - private startTurn(t: QueuedTurn): void { + private startTurn(t: TurnRequest): void { this.cancelRequested = false this.inAgentLoop = false this.pendingTurn = { resolve: t.resolve, reject: t.reject } - // Publish queue depth (0 because we're starting the turn now). + // Publish queue depth (0 unless a prompt arrived during startup/settlement). this.emit({ sessionUpdate: 'session_info_update', _meta: { piAcp: { queueDepth: this.turnQueue.length, running: true } } @@ -491,18 +493,20 @@ export class PiAcpSession { void this.flushEmits().finally(() => { // If this looks like an auth/config issue, surface AUTH_REQUIRED so clients can offer terminal login. const authErr = maybeAuthRequiredError(err) + const turns = [this.pendingTurn, ...this.steeredTurns.splice(0, this.steeredTurns.length)].filter( + (turn): turn is PendingTurn => turn !== null + ) if (authErr) { - this.pendingTurn?.reject(authErr) + for (const turn of turns) turn.reject(authErr) } else { const reason: StopReason = this.cancelRequested ? 'cancelled' : 'error' - this.pendingTurn?.resolve(reason) + for (const turn of turns) turn.resolve(reason) } this.pendingTurn = null this.inAgentLoop = false // If the prompt failed, do not automatically proceed—pi may be unhealthy. - // But we still clear the queueDepth metadata. this.emit({ sessionUpdate: 'session_info_update', _meta: { piAcp: { queueDepth: this.turnQueue.length, running: false } } @@ -833,11 +837,13 @@ export class PiAcpSession { // the ACP `session/prompt` request. void this.flushEmits().finally(() => { const reason: StopReason = this.cancelRequested ? 'cancelled' : 'end_turn' - this.pendingTurn?.resolve(reason) + const turns = [this.pendingTurn, ...this.steeredTurns.splice(0, this.steeredTurns.length)].filter( + (turn): turn is PendingTurn => turn !== null + ) + for (const turn of turns) turn.resolve(reason) this.pendingTurn = null this.inAgentLoop = false - // Start next queued prompt, if any. const next = this.turnQueue.shift() if (next) { this.emit({ diff --git a/src/pi-rpc/process.ts b/src/pi-rpc/process.ts index 92f7959c..ae58d8de 100644 --- a/src/pi-rpc/process.ts +++ b/src/pi-rpc/process.ts @@ -29,6 +29,7 @@ function stripAnsi(s: string): string { type PiRpcCommand = | { type: 'prompt'; id?: string; message: string; images?: unknown[] } + | { type: 'steer'; id?: string; message: string; images?: unknown[] } | { type: 'abort'; id?: string } | { type: 'get_state'; id?: string } // Model @@ -235,6 +236,11 @@ export class PiRpcProcess { if (!res.success) throw new Error(`pi prompt failed: ${res.error ?? JSON.stringify(res.data)}`) } + async steer(message: string, images: unknown[] = []): Promise { + const res = await this.request({ type: 'steer', message, images }) + if (!res.success) throw new Error(`pi steer failed: ${res.error ?? JSON.stringify(res.data)}`) + } + async abort(): Promise { const res = await this.request({ type: 'abort' }) if (!res.success) throw new Error(`pi abort failed: ${res.error ?? JSON.stringify(res.data)}`) diff --git a/test/component/session-events.test.ts b/test/component/session-events.test.ts index de921a00..cb9d98ff 100644 --- a/test/component/session-events.test.ts +++ b/test/component/session-events.test.ts @@ -721,7 +721,7 @@ test('PiAcpSession: cancel flips stopReason to cancelled', async () => { assert.equal(reason, 'cancelled') }) -test('PiAcpSession: queues concurrent prompt and starts it after agent_end', async () => { +test('PiAcpSession: queues a prompt received before the agent loop starts', async () => { const conn = new FakeAgentSideConnection() const proc = new FakePiRpcProcess() @@ -738,27 +738,50 @@ test('PiAcpSession: queues concurrent prompt and starts it after agent_end', asy const second = session.prompt('two') assert.equal(proc.prompts.length, 1) - assert.equal(proc.prompts[0]!.message, 'one') + assert.equal(proc.steers.length, 0) proc.emit({ type: 'agent_start' }) - proc.emit({ type: 'turn_end' }) proc.emit({ type: 'agent_end' }) - const r1 = await first - assert.equal(r1, 'end_turn') - + assert.equal(await first, 'end_turn') assert.equal(proc.prompts.length, 2) assert.equal(proc.prompts[1]!.message, 'two') + proc.emit({ type: 'agent_start' }) + proc.emit({ type: 'agent_end' }) + assert.equal(await second, 'end_turn') +}) + +test('PiAcpSession: steers a concurrent prompt into the active agent loop', async () => { + const conn = new FakeAgentSideConnection() + const proc = new FakePiRpcProcess() + + const session = new PiAcpSession({ + sessionId: 's1', + cwd: process.cwd(), + mcpServers: [], + proc: proc as any, + conn: asAgentConn(conn), + fileCommands: [] + }) + + const first = session.prompt('one') + proc.emit({ type: 'agent_start' }) + const second = session.prompt('two') + + assert.deepEqual(proc.prompts, [{ message: 'one', attachments: [] }]) + assert.deepEqual(proc.steers, [{ message: 'two', attachments: [] }]) + proc.emit({ type: 'agent_start' }) proc.emit({ type: 'turn_end' }) proc.emit({ type: 'agent_end' }) - const r2 = await second - assert.equal(r2, 'end_turn') + assert.equal(await first, 'end_turn') + assert.equal(await second, 'end_turn') + assert.equal(proc.prompts.length, 1) }) -test('PiAcpSession: cancel clears queued prompts', async () => { +test('PiAcpSession: cancel resolves a steered prompt as cancelled', async () => { const conn = new FakeAgentSideConnection() const proc = new FakePiRpcProcess() @@ -772,9 +795,11 @@ test('PiAcpSession: cancel clears queued prompts', async () => { }) const first = session.prompt('one') + proc.emit({ type: 'agent_start' }) const second = session.prompt('two') assert.equal(proc.prompts.length, 1) + assert.equal(proc.steers.length, 1) await session.cancel() proc.emit({ type: 'agent_start' }) diff --git a/test/component/session-queue-cancel.test.ts b/test/component/session-queue-cancel.test.ts index dd519f5d..125de7ea 100644 --- a/test/component/session-queue-cancel.test.ts +++ b/test/component/session-queue-cancel.test.ts @@ -22,6 +22,7 @@ test('PiAcpSession: cancel clears queued prompts', async () => { // first started, second+third queued assert.equal(proc.prompts.length, 1) + assert.equal(proc.steers.length, 0) await session.cancel() diff --git a/test/helpers/fakes.ts b/test/helpers/fakes.ts index e0a7b32a..00288712 100644 --- a/test/helpers/fakes.ts +++ b/test/helpers/fakes.ts @@ -27,6 +27,7 @@ export class FakePiRpcProcess { // spies readonly prompts: Array<{ message: string; attachments: unknown[] }> = [] + readonly steers: Array<{ message: string; attachments: unknown[] }> = [] readonly extensionUiResponses: unknown[] = [] abortCount = 0 @@ -45,6 +46,10 @@ export class FakePiRpcProcess { this.prompts.push({ message, attachments }) } + async steer(message: string, attachments: unknown[] = []): Promise { + this.steers.push({ message, attachments }) + } + async abort(): Promise { this.abortCount += 1 }