fix(ask): a prompt typed into an Ask-this-session turn takes it over on every host

apps/hook/hooks/mod/bridge.test.ts

@@ -5,9 +5,9 @@
  */
 import { afterEach, describe, expect, test } from 'bun:test'
 import { createPullSessionBridge, type PullSessionBridge } from '../../../../packages/ai/session-bridge-pull.ts'
-import { createBridge } from './bridge'
+import { createBridge, TAKEN_OVER_INTERRUPT_TEXT, takenOverFallback } from './bridge'
 import { fakeHost, type FakeHost } from './testing/fake-host'
-import { TurnTracker } from './turns'
+import { TAKEN_OVER_BY_PERSON_TEXT, TurnTracker } from './turns'
 
 const TOKEN = 'k'.repeat(64)
 let server: PullSessionBridge | null = null
@@ -19,7 +19,7 @@ afterEach(() => {
   server = null
 })
 
-function wire(host: FakeHost, bridge: PullSessionBridge) {
+function wire(host: FakeHost, bridge: PullSessionBridge, options: { olderServer?: boolean } = {}) {
   host.onFetch = async (url, body) => {
     const response = await bridge.handle(
       new Request(url, {
@@ -29,7 +29,13 @@ function wire(host: FakeHost, bridge: PullSessionBridge) {
       }),
     )
     if (!response) return { status: 404, ok: false, text: '' }
-    return { status: response.status, ok: response.ok, text: await response.text() }
+    let text = await response.text()
+    if (options.olderServer && url.endsWith('/poll')) {
+      // A server from before the `features` advert.
+      const { features: _features, ...rest } = JSON.parse(text)
+      text = JSON.stringify(rest)
+    }
+    return { status: response.status, ok: response.ok, text }
   }
 }
 
@@ -42,7 +48,7 @@ async function until(check: () => boolean, ms = 3_000) {
 }
 
 function collector() {
-  const seen = { deltas: '', done: null as string | null, error: null as string | null }
+  const seen = { deltas: '', done: null as string | null, error: null as string | null, message: null as string | null }
   return {
     seen,
     sink: {
@@ -52,8 +58,9 @@ function collector() {
       done: (answer: string) => {
         seen.done = answer
       },
-      error: (code: string) => {
+      error: (code: string, message?: string) => {
         seen.error = code
+        seen.message = message ?? null
       },
     },
   }
@@ -135,6 +142,79 @@ describe('Ask this session over the pull bridge', () => {
     await running
   })
 
+  // The failure this guards: the person typed into the question's turn, and
+  // the reply to their prompt streamed into Plannotator as the answer, and
+  // "Interrupt and ask now" then aborted the person's own work.
+  test('a prompt typed into the question\'s turn settles it as taken over, and the turn is never interrupted', async () => {
+    live = true
+    server = createPullSessionBridge({ token: TOKEN, host: 'claude-code', modes: { turn: true, transient: false } })
+    const host = fakeHost()
+    wire(host, server)
+    const turns = new TurnTracker()
+    const client = createBridge({ host, baseUrl: 'http://127.0.0.1:4321', token: TOKEN, turns, isLive: () => live })
+    const running = client.run()
+
+    await until(() => server!.bridge.status() === 'ready')
+    const { seen, sink } = collector()
+    const controller = new AbortController()
+    server.bridge.ask({ askId: 'ask-3', text: '[Plannotator Ask AI] Why step 2?', mode: 'turn' }, sink, controller.signal)
+    await until(() => host.submits.length === 1)
+
+    turns.onTurnStart('turn-3', 'The plannotator plugin sent a message:\n[Plannotator Ask AI] Why step 2?')
+    turns.onStep('turn-3')
+    turns.onText('turn-3', 'Because ')
+    turns.onPromptEntered({ text: 'also fix the tests', fromUs: false, turnId: 'turn-3', originKind: 'composer' })
+    turns.onStep('turn-3')
+    turns.onText('turn-3', 'Fixed the tests.')
+
+    await until(() => seen.error !== null)
+    expect(seen.error).toBe('taken_over')
+    expect(seen.message).toBe(TAKEN_OVER_BY_PERSON_TEXT)
+    expect(seen.deltas).toBe('Because ')
+
+    // A late Stop and "Interrupt and ask now" both leave the person's turn alone.
+    controller.abort()
+    await expect(server.bridge.interrupt!()).rejects.toThrow(TAKEN_OVER_INTERRUPT_TEXT)
+    expect(host.aborted).toEqual([])
+
+    live = false
+    server.dispose()
+    await running
+  })
+
+  // The failure this guards: a newer mod against an older CLI, whose server
+  // reads `taken_over` as `failed` and whose UI then replaces the partial
+  // answer with the error.
+  test('against a server that does not advertise taken_over, a take-over settles as the partial answer plus the note', async () => {
+    live = true
+    server = createPullSessionBridge({ token: TOKEN, host: 'claude-code', modes: { turn: true, transient: false } })
+    const host = fakeHost()
+    wire(host, server, { olderServer: true })
+    const turns = new TurnTracker()
+    const client = createBridge({ host, baseUrl: 'http://127.0.0.1:4321', token: TOKEN, turns, isLive: () => live })
+    const running = client.run()
+
+    await until(() => server!.bridge.status() === 'ready')
+    const { seen, sink } = collector()
+    server.bridge.ask({ askId: 'ask-4', text: '[Plannotator Ask AI] Why step 2?', mode: 'turn' }, sink, new AbortController().signal)
+    await until(() => host.submits.length === 1)
+    turns.onTurnStart('turn-4', 'The plannotator plugin sent a message:\n[Plannotator Ask AI] Why step 2?')
+    turns.onStep('turn-4')
+    turns.onText('turn-4', 'Because ')
+    turns.onPromptEntered({ text: 'also fix the tests', fromUs: false, turnId: 'turn-4', originKind: 'composer' })
+    turns.onStep('turn-4')
+
+    await until(() => seen.done !== null)
+    expect(seen.error).toBeNull()
+    expect(seen.done).toBe(takenOverFallback('Because ', TAKEN_OVER_BY_PERSON_TEXT).answer)
+    expect(seen.deltas).toBe(seen.done)
+    expect(seen.done).toContain(TAKEN_OVER_BY_PERSON_TEXT)
+
+    live = false
+    server.dispose()
+    await running
+  })
+
   test('a wrong token stops the client instead of retrying forever', async () => {
     live = true
     server = createPullSessionBridge({ token: 'x'.repeat(64), host: 'claude-code', modes: { turn: true, transient: false } })

apps/hook/hooks/mod/bridge.ts

@@ -14,7 +14,8 @@
  * (it waits for idle), the turn's streamed text goes back as deltas, and
  * `turn.complete`'s answer as `done`. Busy = Claude is mid-turn: reported as
  * `busy`, so the reviewer chooses wait or interrupt; an interrupt aborts the
- * running turn with `$.turn.abort`. Plan review does not block the session
+ * running turn with `$.turn.abort`, except a question's turn the person typed
+ * into (turns.ts, take-over), which is theirs. Plan review does not block the session
  * under the mod, so the status is never `blocked`.
  */
 
@@ -27,6 +28,28 @@ export const BRIDGE_HOST = 'claude-code'
 export const BRIDGE_MODES = 'turn'
 /** Long-poll wait we ask for; below the server's 25 s cap and any fetch timeout. */
 export const BRIDGE_POLL_WAIT_MS = 15_000
+/**
+ * Why "Interrupt and ask now" refuses a turn another message took over. Same
+ * text as `SESSION_ASK_TAKEN_OVER_INTERRUPT_TEXT` (packages/ai/session-bridge.ts).
+ */
+export const TAKEN_OVER_INTERRUPT_TEXT =
+  'The session is now answering another message, so Plannotator will not stop it. Ask when it finishes instead.'
+
+/** Used when a `taken_over` comes without a message. */
+const TAKEN_OVER_FALLBACK_NOTE =
+  'Another message entered this session while it was answering, so the rest of the reply went to that message.'
+
+/**
+ * A server that does not advertise `taken_over` (poll `features`) reads it as
+ * `failed`, and its UI then replaces the partial answer with the error. Settle
+ * as an answer instead: what streamed, plus the note as its last paragraph.
+ * Mirrors `takenOverFallback` in packages/ai/session-bridge-pull-client.ts.
+ */
+export function takenOverFallback(streamed: string, message: string | undefined): { delta: string; answer: string } {
+  const note = `_${(message || TAKEN_OVER_FALLBACK_NOTE).trim()}_`
+  const delta = streamed ? `\n\n${note}` : note
+  return { delta, answer: `${streamed}${delta}` }
+}
 
 type BridgeCommand =
   | { type: 'ask'; askId: string; text: string; mode: string }
@@ -57,16 +80,17 @@ export function bridgeBaseUrl(port: number): string {
   return `http://127.0.0.1:${port}`
 }
 
-export function parseBridgeCommands(text: string): { commands: BridgeCommand[]; closing: boolean } {
+export function parseBridgeCommands(text: string): { commands: BridgeCommand[]; closing: boolean; features: string[] } {
   try {
-    const body = JSON.parse(text) as { commands?: unknown; closing?: unknown }
+    const body = JSON.parse(text) as { commands?: unknown; closing?: unknown; features?: unknown }
     const commands = Array.isArray(body.commands)
       ? body.commands.filter((command): command is BridgeCommand =>
           !!command && typeof command === 'object' && typeof (command as { type?: unknown }).type === 'string')
       : []
-    return { commands, closing: body.closing === true }
+    const features = Array.isArray(body.features) ? body.features.filter((feature): feature is string => typeof feature === 'string') : []
+    return { commands, closing: body.closing === true, features }
   } catch {
-    return { commands: [], closing: false }
+    return { commands: [], closing: false, features: [] }
   }
 }
 
@@ -87,6 +111,8 @@ export function createBridge(options: BridgeOptions): BridgeHandle {
   let outbox: BridgeEvent[] = []
   let sending: Promise<void> = Promise.resolve()
   let lastStatus: 'ready' | 'busy' = turns.busy ? 'busy' : 'ready'
+  /** The server knows the `taken_over` code (poll `features`). */
+  let serverTakesTakenOver = false
 
   const post = async (events: BridgeEvent[]): Promise<void> => {
     if (events.length === 0) return
@@ -133,11 +159,23 @@ export function createBridge(options: BridgeOptions): BridgeHandle {
     if (seenAsks.has(command.askId)) return
     seenAsks.add(command.askId)
     const askId = command.askId
+    let streamed = ''
     const sink: AskSink = {
-      delta: (text) => emit({ type: 'delta', askId, text }, false),
+      delta: (text) => {
+        streamed += text
+        emit({ type: 'delta', askId, text }, false)
+      },
       tool: (name) => emit({ type: 'tool', askId, name }),
       done: (answer) => emit({ type: 'done', askId, answer }),
-      error: (code, message) => emit({ type: 'error', askId, code, ...(message ? { message } : {}) }),
+      error: (code, message) => {
+        if (code === 'taken_over' && !serverTakesTakenOver) {
+          const fallback = takenOverFallback(streamed, message)
+          emit({ type: 'delta', askId, text: fallback.delta }, false)
+          emit({ type: 'done', askId, answer: fallback.answer })
+          return
+        }
+        emit({ type: 'error', askId, code, ...(message ? { message } : {}) })
+      },
     }
     if (!turns.beginAsk(askId, command.text, sink)) {
       emit({ type: 'error', askId, code: 'busy', message: 'Another question is already running in this session.' })
@@ -157,6 +195,11 @@ export function createBridge(options: BridgeOptions): BridgeHandle {
       emit({ type: 'interrupted', interruptId, ok: true })
       return
     }
+    // A question's turn that the person typed into is theirs now: never stopped from Plannotator.
+    if (turns.isTakenOver(running)) {
+      emit({ type: 'interrupted', interruptId, ok: false, message: TAKEN_OVER_INTERRUPT_TEXT })
+      return
+    }
     try {
       await host.abortTurn(running)
       emit({ type: 'interrupted', interruptId, ok: true })
@@ -215,7 +258,8 @@ export function createBridge(options: BridgeOptions): BridgeHandle {
         continue
       }
       failures = 0
-      const { commands, closing } = parseBridgeCommands(response.text)
+      const { commands, closing, features } = parseBridgeCommands(response.text)
+      serverTakesTakenOver = features.includes('taken_over')
       for (const command of commands) {
         host.debug(`bridge ${base}: ${command.type}`)
         handle(command)

apps/hook/hooks/mod/controller.ts

@@ -52,7 +52,7 @@ import {
   scriptOnlyAnnotateFlag,
   scriptOnlyAnnotateFlagText,
 } from './tool'
-import { TurnTracker } from './turns'
+import { TurnTracker, type EnteredPrompt } from './turns'
 
 /** Persisted in `$.store` so open reviews reattach after a restart or `--resume`. */
 export interface LaunchRecord {
@@ -674,9 +674,34 @@ export class PlannotatorMod {
   }
 
   /** A prompt entered the session (prompt.submit), from register.ts. */
-  onPromptEntered(text: string, fromUs: boolean): void {
-    this.host.debug(`prompt.submit${fromUs ? ' (ours)' : ''}: ${JSON.stringify(text.slice(0, 160))}`)
-    this.turns.onPromptEntered(text, fromUs)
+  onPromptEntered(prompt: EnteredPrompt): void {
+    const { text, fromUs, turnId, originKind } = prompt
+    this.host.debug(
+      `prompt.submit${fromUs ? ' (ours)' : ''}${originKind ? ` [${originKind}]` : ''}${turnId ? ` into ${turnId}` : ''}: ${JSON.stringify(text.slice(0, 160))}`,
+    )
+    const wasOurs = !!turnId && this.turns.ownsTurn(turnId)
+    this.turns.onPromptEntered(prompt)
+    if (wasOurs && turnId && this.turns.isTakenOver(turnId)) this.host.debug(`ask turn ${turnId} taken over`)
+  }
+
+  /** A prompt reached prompt.submit, before the hooks beneath it ran, from register.ts. */
+  onPromptSubmitting(prompt: Omit<EnteredPrompt, 'text'>): void {
+    this.turns.onPromptSubmitting(prompt)
+  }
+
+  /** A prompt announced by onPromptSubmitting did not enter, from register.ts. */
+  onPromptDropped(prompt: Omit<EnteredPrompt, 'text'>): void {
+    this.turns.onPromptDropped(prompt)
+  }
+
+  /** A model request of a turn is about to go out (turn.step), from register.ts. */
+  onTurnStep(turnId: string): void {
+    this.turns.onStep(turnId)
+  }
+
+  /** A model response of a turn finished (turn.step's `stop` chunk), from register.ts. */
+  onTurnStepStop(turnId: string, stopReason: string | null): void {
+    this.turns.onStepStop(turnId, stopReason)
   }
 
   /** Turn events, from register.ts. */

apps/hook/hooks/mod/register.ts

@@ -298,14 +298,34 @@ export function register(on: On) {
   })
 
   on('prompt.submit', async ($: Engine, e: any, next: Next) => {
-    const result = await next(e)
     const instance = allowed ? mod : null
-    if (instance && !instance.isDisposed && result && typeof result.text === 'string') {
+    const live = !!instance && !instance.isDisposed
+    // A prompt typed (or delivered) while a turn ran carries that turn's id:
+    // when the turn is a question's and a person sent it, the rest of the turn
+    // answers this prompt instead (turns.ts, take-over). Streaming stops HERE,
+    // before the hooks beneath run, so a slow one cannot let the next step
+    // through; it resumes if one of them drops the prompt.
+    const origin = e.origin
+    const ref = {
+      fromUs: !!origin && origin.kind === 'plugin' && origin.name === PLUGIN_NAME,
+      ...(typeof e.turnId === 'string' ? { turnId: e.turnId } : {}),
+      ...(origin && typeof origin.kind === 'string' ? { originKind: origin.kind } : {}),
+    }
+    if (live) instance.onPromptSubmitting(ref)
+    let result
+    try {
+      result = await next(e)
+    } catch (error) {
+      if (live) instance.onPromptDropped(ref)
+      throw error
+    }
+    if (!live || instance.isDisposed) return result
+    if (result && typeof result.text === 'string') {
       // Every prompt seen here is someone else's (the engine skips our hooks
       // for prompts our own code submitted): its turn is never a question's.
-      const origin = result.origin ?? e.origin
-      const fromUs = !!origin && origin.kind === 'plugin' && origin.name === PLUGIN_NAME
-      instance.onPromptEntered(result.text, fromUs)
+      instance.onPromptEntered({ ...ref, text: result.text })
+    } else {
+      instance.onPromptDropped(ref)
     }
     return result
   })
@@ -321,12 +341,17 @@ export function register(on: On) {
   on('turn.step', async function* ($: Engine, e: any, next: Next) {
     const instance = allowed ? mod : null
     if (!instance || !instance.turns.ownsTurn(e.turnId)) return yield* next(e)
+    // A question's turn that someone else's prompt entered settles here: this
+    // request carries their prompt, so nothing it says is the question's answer.
+    instance.onTurnStep(e.turnId)
+    if (!instance.turns.ownsTurn(e.turnId)) return yield* next(e)
     const stream = next(e)
     let step = await stream.next()
     while (!step.done) {
       const chunk = step.value
       if (chunk && chunk.kind === 'text' && typeof chunk.text === 'string') instance.turns.onText(e.turnId, chunk.text)
       else if (chunk && chunk.kind === 'tool' && typeof chunk.name === 'string') instance.turns.onTool(e.turnId, chunk.name)
+      else if (chunk && chunk.kind === 'stop') instance.onTurnStepStop(e.turnId, typeof chunk.stopReason === 'string' ? chunk.stopReason : null)
       yield chunk
       step = await stream.next()
     }