feat(acp): steer prompts during active turns

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 `<project>/.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

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<ReturnType<AgentSideConnection['requestPermission']>>
@@ -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<StopReason>((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<void> {
-    // 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({

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<void> {
+    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<void> {
     const res = await this.request({ type: 'abort' })
     if (!res.success) throw new Error(`pi abort failed: ${res.error ?? JSON.stringify(res.data)}`)

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' })

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()
 

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<void> {
+    this.steers.push({ message, attachments })
+  }
+
   async abort(): Promise<void> {
     this.abortCount += 1
   }