fix(acp): keep OMP replies that follow its reasoning

config/scripts/vitest-sqlite-runtime-files.mjs

@@ -3,6 +3,9 @@ import { SQLITE_DATABASE_RUNTIME_INCLUDE } from './vitest-sqlite-database-runtim
 // Real SQLite fixtures keep Node publication and close semantics.
 export const SQLITE_RUNTIME_INCLUDE = [
   ...SQLITE_DATABASE_RUNTIME_INCLUDE,
+  'src/main/acp/acp-structured-omp-recordings.test.ts',
+  'src/main/acp/acp-text-stream-identity.test.ts',
+  'src/main/acp/acp-text-message-ownership.test.ts',
   'src/main/native-chat/agent-session-wire/structured-agent-session-clear-queued-command.test.ts',
   'src/main/acp/acp-structured-sign-in.test.ts',
   'src/main/acp/acp-structured-auth.test.ts',

src/main/acp/acp-session-update.ts

@@ -8,6 +8,7 @@ import type { AcpDialect } from './acp-dialects/acp-dialect'
 import type { AcpBackgroundTaskTimeline } from './acp-background-task-timeline'
 import type { AcpToolTimeline } from './acp-tool-timeline'
 import type { SessionNotification } from './generated/acp-protocol.generated'
+import { acpNamedTextKey } from './acp-turn-messages'
 
 export function acpSessionUpdate(
   notification: SessionNotification,
@@ -24,22 +25,24 @@ export function acpSessionUpdate(
   const join = { join: { thread: notification.sessionId, ...(turn === undefined ? {} : { turn }) } }
   switch (update.sessionUpdate) {
     case 'agent_message_chunk':
-    case 'agent_thought_chunk':
+    case 'agent_thought_chunk': {
+      const channel = update.sessionUpdate === 'agent_thought_chunk' ? 'reasoning' : 'assistant'
       return update.content.type === 'text'
         ? [
             {
               type: 'text.delta',
               item: context.messageKey
                 ? { id: context.messageKey }
                 : update.messageId
-                  ? { id: `message:${update.messageId}` }
+                  ? { id: acpNamedTextKey(update.messageId, channel) }
                   : { stream: update.sessionUpdate },
-              channel: update.sessionUpdate === 'agent_thought_chunk' ? 'reasoning' : 'assistant',
+              channel,
               text: update.content.text,
               ...join
             }
           ]
         : [{ type: 'provider.frame', frameKind: update.sessionUpdate, payload: update, ...join }]
+    }
     case 'user_message_chunk':
       // A live echo of the send is the send's own row.
       return []

src/main/acp/acp-structured-acquire.ts

@@ -44,6 +44,7 @@ import {
 import type { AcpStructuredSessionAdapterDeps } from './acp-structured-session-adapter-deps'
 import { AcpStructuredTurns, type AcpStructuredTurnsDeps } from './acp-structured-turns'
 import { RequestPermissionResponseSchema } from './generated/acp-protocol.generated'
+import { createStructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger'
 
 /** Frames an agent may send before its session exists; past this they are dropped. */
 const MAX_EARLY_FRAMES = 2_048
@@ -196,6 +197,7 @@ export async function acquireAcpStructuredSession(input: {
       generation,
       providerSessionId,
       dialect: spec.dialect,
+      logger: deps.logger ?? createStructuredAgentSessionLogger(),
       onInputAccepted: (clientMessageId) => session?.turns.accept(clientMessageId),
       onFailed: () => input.forceClose(sessionId)
     })

src/main/acp/acp-structured-lane.ts

@@ -12,6 +12,11 @@ import { createLegacyProviderTimelineIdentityScheme } from '../native-chat/agent
 import type { ProviderTimelineSink } from '../native-chat/agent-session-timeline/provider-timeline-plan'
 import type { AcpDialect } from './acp-dialects/acp-dialect'
 import { AcpTimelineTranslator } from './acp-timeline-translator'
+import {
+  neverThrowingStructuredAgentSessionLogger,
+  type StructuredAgentSessionLogger
+} from '../native-chat/agent-session-wire/structured-agent-session-logger'
+import type { AcpTextDrop } from './acp-turn-messages'
 
 /** How long a held event waits for the sink to say it drained before trying again on its own. */
 const BACKPRESSURE_RETRY_MS = 250
@@ -25,6 +30,7 @@ export type AcpStructuredLaneDeps = {
   generation: string
   providerSessionId: string
   dialect: AcpDialect
+  logger: StructuredAgentSessionLogger
   /** A send of Orca's reached the agent: the turn it opened has its first provider event. */
   onInputAccepted: (clientMessageId: string) => void
   /** The sink refused for good; nothing more this child says can be journaled. */
@@ -40,12 +46,15 @@ export class AcpStructuredLane {
   private retryTimer: ReturnType<typeof setTimeout> | null = null
   private failed = false
   private disposed = false
+  private readonly logger: StructuredAgentSessionLogger
 
   constructor(private readonly deps: AcpStructuredLaneDeps) {
+    this.logger = neverThrowingStructuredAgentSessionLogger(deps.logger)
     this.translator = new AcpTimelineTranslator({
       sessionId: deps.providerSessionId,
       dialect: deps.dialect,
-      agentName: deps.agentName
+      agentName: deps.agentName,
+      onTextDropped: (drop) => this.reportTextDrop(drop)
     })
     this.requestIdentity = createLegacyProviderTimelineIdentityScheme({
       agent: deps.agent,
@@ -123,7 +132,17 @@ export class AcpStructuredLane {
     this.clearRetry()
     while (this.backlog.length > 0 && !this.failed && !this.disposed) {
       const event = this.backlog[0]
-      const { admission } = this.assembler.apply(event)
+      const { admission, dropped } = this.assembler.apply(event)
+      if (dropped === 'stream-mismatch' && event.type === 'text.delta') {
+        this.reportTextDrop({
+          reason: dropped,
+          itemId: 'id' in event.item ? event.item.id : undefined,
+          channel: event.channel,
+          threadId: event.join?.thread,
+          turnId: event.join?.turn,
+          producerAgentId: event.producer?.agentId
+        })
+      }
       if (!admission.accepted) {
         if (admission.reason === 'backpressure') {
           this.retryTimer = setTimeout(() => this.drain(), BACKPRESSURE_RETRY_MS)
@@ -149,6 +168,17 @@ export class AcpStructuredLane {
     }
   }
 
+  private reportTextDrop(drop: AcpTextDrop): void {
+    this.logger.warn('ACP text chunk rejected by its message stream', {
+      scope: `acp-text-${drop.reason}`,
+      sessionId: this.deps.sessionId,
+      providerSessionId: this.deps.providerSessionId,
+      agent: this.deps.agent,
+      generation: this.deps.generation,
+      ...drop
+    })
+  }
+
   private clearRetry(): void {
     if (this.retryTimer) {
       clearTimeout(this.retryTimer)

src/main/acp/acp-structured-omp-recordings.test.ts

@@ -16,6 +16,7 @@ import {
 } from './acp-structured-adapter.test-support'
 import { GrokFixtureReplay } from './acp-structured-fixture-replay.test-support'
 import { readAcpFixture } from './acp-timeline-fixture.test-support'
+import { SessionNotificationSchema } from './generated/acp-protocol.generated'
 
 afterEach(async () => {
   await closeProviderTimelineRigs()
@@ -67,6 +68,67 @@ async function allowOnce(rig: AcpAdapterRig, replay: GrokFixtureReplay) {
 }
 
 describe('OMP recordings through the adapter', () => {
+  it.each(['omp-v17-reply-without-thought', 'omp-v17-reply-after-thought'])(
+    'keeps both ordinary answers from %s',
+    async (name) => {
+      const frames = await readAcpFixture(name)
+      const replay = new GrokFixtureReplay(frames)
+      const rig = await openAcpAdapterRig({
+        spec: acpLaunchSpecFor('omp')!,
+        script: (scripted) => replay.attach(scripted)
+      })
+      await rig.acquire()
+      await rig.adapter.dispatch({
+        sessionId: SESSION,
+        clientMessageId: 'send-1',
+        body: ask,
+        fence: 1
+      })
+      await waitFor(async () => expect(await turns(rig)).toHaveLength(1))
+      await rig.adapter.dispatch({
+        sessionId: SESSION,
+        clientMessageId: 'send-2',
+        body: ask,
+        fence: 1
+      })
+      await waitFor(() => expect(replay.awaiting).toBeNull())
+      await waitFor(async () => expect(await turns(rig)).toHaveLength(2))
+      const recordedAnswers = frames
+        .flatMap((frame) => {
+          const parsed = SessionNotificationSchema.safeParse(frame.message.params)
+          const update = parsed.data?.update
+          return update?.sessionUpdate === 'agent_message_chunk' && update.content.type === 'text'
+            ? [update.content.text]
+            : []
+        })
+        .join('')
+      const rows = await rig.rig.rows()
+      const answers = rows
+        .flatMap((row) =>
+          row.body.kind === 'message' && row.body.role === 'assistant'
+            ? row.body.blocks.flatMap((block) => (block.type === 'text' ? [block.text] : []))
+            : []
+        )
+        .join('')
+      expect(await turns(rig)).toMatchObject([
+        { state: 'completed', outcome: 'success' },
+        { state: 'completed', outcome: 'success' }
+      ])
+      expect(answers).toBe(recordedAnswers)
+      if (name === 'omp-v17-reply-after-thought') {
+        expect(
+          rows.filter((row) => row.body.kind === 'message' && row.body.role === 'reasoning')
+        ).toHaveLength(1)
+      }
+      await rig.adapter.closeSession(SESSION)
+      const savedMessages = (await rig.rig.rows()).filter((row) => row.body.kind === 'message')
+      const reopened = await rig.rig.reopenJournal()
+      expect(reopened.snapshot().items.filter((row) => row.body.kind === 'message')).toEqual(
+        savedMessages
+      )
+    }
+  )
+
   it("a failed command's row shows its output and exit code, not the command again", async () => {
     const { rig, replay } = await replaying('omp-v17-shell-exit-3')
     await allowOnce(rig, replay)

src/main/acp/acp-text-message-ownership.test.ts

@@ -0,0 +1,114 @@
+import { afterEach, describe, expect, it } from 'vitest'
+import {
+  closeProviderTimelineRigs,
+  SESSION
+} from '../native-chat/agent-session-timeline/provider-timeline-assembler-test-support'
+import { recordingStructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger-test-support'
+import { acpLaunchSpecFor } from './acp-launch-specs'
+import {
+  openAcpAdapterRig,
+  PROVIDER_SESSION,
+  sendHello,
+  waitFor
+} from './acp-structured-adapter.test-support'
+
+afterEach(closeProviderTimelineRigs)
+
+describe.each(['omp', 'opencode', 'grok'] as const)('%s provider message ownership', (agent) => {
+  it.each([
+    { stopReason: 'end_turn', nextPrompt: false },
+    { stopReason: 'cancelled', nextPrompt: false },
+    { stopReason: 'end_turn', nextPrompt: true },
+    { stopReason: 'cancelled', nextPrompt: true }
+  ] as const)(
+    'refuses a late answer after $stopReason (next prompt: $nextPrompt)',
+    async ({ stopReason, nextPrompt }) => {
+      const recording = recordingStructuredAgentSessionLogger()
+      let prompts = 0
+      const rig = await openAcpAdapterRig({
+        spec: acpLaunchSpecFor(agent)!,
+        deps: { logger: recording.logger },
+        script: (scripted) => {
+          scripted.on('session/prompt', (frame) => {
+            prompts += 1
+            scripted.notify('session/update', {
+              sessionId: PROVIDER_SESSION,
+              _meta: { promptId: `prompt:send-${prompts}` },
+              update: {
+                sessionUpdate: 'agent_thought_chunk',
+                messageId: prompts === 1 ? 'first-message' : 'next-message',
+                content: { type: 'text', text: `Thought ${prompts}` }
+              }
+            })
+            if (prompts === 1) {
+              scripted.reply(frame, { stopReason })
+            }
+          })
+        }
+      })
+      await rig.acquire()
+      await sendHello(rig, 'send-1')
+      await waitFor(async () =>
+        expect(await rig.rig.turns()).toMatchObject([
+          { state: stopReason === 'cancelled' ? 'interrupted' : 'completed' }
+        ])
+      )
+      if (nextPrompt) {
+        await sendHello(rig, 'send-2')
+        await waitFor(async () => expect(await rig.rig.turns()).toHaveLength(2))
+      }
+      rig.child().agent.notify('session/update', {
+        sessionId: PROVIDER_SESSION,
+        update: {
+          sessionUpdate: 'agent_message_chunk',
+          messageId: 'first-message',
+          content: { type: 'text', text: 'Private late answer' }
+        }
+      })
+      await waitFor(() =>
+        expect(recording.entries).toContainEqual({
+          level: 'warn',
+          message: 'ACP text chunk rejected by its message stream',
+          fields: {
+            scope: 'acp-text-turn-settled',
+            reason: 'turn-settled',
+            sessionId: SESSION,
+            providerSessionId: PROVIDER_SESSION,
+            agent,
+            generation: 'gen-acp',
+            itemId: 'message:["first-message","assistant"]',
+            channel: 'assistant',
+            threadId: PROVIDER_SESSION,
+            turnId: 'prompt:send-1'
+          }
+        })
+      )
+      if (nextPrompt) {
+        rig.child().agent.notify('session/update', {
+          sessionId: PROVIDER_SESSION,
+          _meta: { promptId: 'prompt:send-2' },
+          update: {
+            sessionUpdate: 'agent_message_chunk',
+            messageId: 'next-message',
+            content: { type: 'text', text: 'Next answer' }
+          }
+        })
+        rig.child().agent.reply(await rig.frame('session/prompt', 1), { stopReason: 'end_turn' })
+        await waitFor(async () =>
+          expect(await rig.rig.turns()).toMatchObject([
+            { state: stopReason === 'cancelled' ? 'interrupted' : 'completed' },
+            { state: 'completed', outcome: 'success' }
+          ])
+        )
+      }
+      const rows = await rig.rig.rows()
+      expect(
+        rows
+          .filter((row) => row.body.kind === 'message' && row.body.role === 'assistant')
+          .map((row) => row.body)
+      ).toMatchObject(nextPrompt ? [{ blocks: [{ text: 'Next answer' }] }] : [])
+      expect(rows.filter((row) => row.body.kind === 'status')).toEqual([])
+      await rig.adapter.closeSession(SESSION)
+    }
+  )
+})