fix(native-chat): retire interrupted compaction ownership

src/main/claude/claude-structured-compaction.test.ts

@@ -1,7 +1,11 @@
 import { afterEach, describe, expect, it, vi } from 'vitest'
 import { StructuredSessionCompaction } from '../native-chat/agent-session-wire/structured-session-compaction'
 import { claudeUnwrittenUserMessageError } from './claude-agent-sdk-user-message-queue'
-import { compactClaudeSession, isClaudeCompactionContent } from './claude-structured-compaction'
+import {
+  compactClaudeSession,
+  isClaudeCompactionContent,
+  observeClaudeCompaction
+} from './claude-structured-compaction'
 import { sessionFor } from './claude-structured-dispatch-test-support'
 
 afterEach(() => {
@@ -107,4 +111,80 @@ describe('Claude compaction transcript content', () => {
     })
     expect(tracker.hasPending('orca-session')).toBe(false)
   })
+
+  it('releases interrupted transcript suppression when a later dispatched turn starts', async () => {
+    const tracker = new StructuredSessionCompaction()
+    const onLateResult = vi.fn(async () => {})
+    const pending = tracker.run(
+      'orca-session',
+      'provider-session',
+      async () => ({}),
+      onLateResult,
+      'compact:operation-1'
+    )
+    tracker.bindClaudeCommand('orca-session', 'compact:operation-1', 'compact-command', true)
+    tracker.interrupted('orca-session')
+    const handle = vi.fn()
+
+    observeClaudeCompaction(
+      tracker,
+      {
+        type: 'message',
+        sessionId: 'orca-session',
+        startsTurn: true,
+        message: {
+          type: 'user',
+          session_id: 'provider-session',
+          uuid: 'later-turn',
+          message: { role: 'user', content: 'continue' }
+        }
+      },
+      { handle }
+    )
+
+    await expect(pending).rejects.toThrow('interrupted')
+    expect(tracker.hasPending('orca-session')).toBe(false)
+    expect(handle).toHaveBeenCalledOnce()
+    expect(onLateResult).toHaveBeenCalledWith({ error: 'Compaction was interrupted.' })
+  })
+
+  it('keeps a late replay of the interrupted compact command suppressed', async () => {
+    const tracker = new StructuredSessionCompaction()
+    const pending = tracker.run(
+      'orca-session',
+      'provider-session',
+      async () => ({}),
+      undefined,
+      'compact:operation-1'
+    )
+    tracker.bindClaudeCommand('orca-session', 'compact:operation-1', 'compact-command', true)
+    tracker.interrupted('orca-session')
+    await expect(pending).rejects.toThrow('interrupted')
+    const handle = vi.fn()
+
+    observeClaudeCompaction(
+      tracker,
+      {
+        type: 'message',
+        sessionId: 'orca-session',
+        startsTurn: true,
+        message: {
+          type: 'user',
+          session_id: 'provider-session',
+          uuid: 'compatibility-replay-id',
+          message: { role: 'user', content: '/compact' }
+        }
+      },
+      { handle }
+    )
+
+    expect(tracker.hasPending('orca-session')).toBe(true)
+    expect(handle).not.toHaveBeenCalled()
+    tracker.claude('orca-session', {
+      type: 'command_lifecycle',
+      state: 'cancelled',
+      command_uuid: 'compact-command',
+      session_id: 'provider-session'
+    })
+  })
 })

src/main/claude/claude-structured-compaction.ts

@@ -39,8 +39,11 @@ export function compactClaudeSession(
 export function observeClaudeCompaction(
   compactions: StructuredSessionCompaction,
   event: ClaudeStructuredSessionEvent,
-  translator: ClaudeSession['translator'] | undefined
+  translator: Pick<NonNullable<ClaudeSession['translator']>, 'handle'> | null | undefined
 ): void {
+  if (event.type === 'message' && event.startsTurn) {
+    compactions.claudeTurnStarted(event.sessionId, event.message)
+  }
   if (!isClaudeCompactionContent(compactions, event)) {
     translator?.handle(event)
   }

src/main/native-chat/agent-session-wire/structured-conversation-command-concurrency.test.ts

@@ -1,6 +1,6 @@
 import { beforeEach, describe, expect, it, vi, type Mock } from 'vitest'
 import type { AgentSessionConversationCommand } from '../../../shared/agent-session-conversation-command'
-import { hostTestMessage } from './structured-agent-session-host-test-data'
+import { HOST_TEST_SESSION, hostTestMessage } from './structured-agent-session-host-test-data'
 import {
   CALLER,
   attach,
@@ -13,6 +13,7 @@ import type { StructuredAgentSessionHost } from './structured-agent-session-host
 const compact = vi.fn<NonNullable<StructuredAgentSessionAdapter['compact']>>()
 let host: StructuredAgentSessionHost
 let dispatch: Mock<StructuredAgentSessionAdapter['dispatch']>
+let store: ReturnType<typeof hostTestState>['store']
 
 function commandParams(command: AgentSessionConversationCommand) {
   return {
@@ -22,7 +23,16 @@ function commandParams(command: AgentSessionConversationCommand) {
 }
 
 beforeEach(async () => {
-  ;({ host, dispatch } = hostTestState())
+  const state = hostTestState()
+  ;({ host, dispatch, store } = state)
+  const acquire = state.acquire.getMockImplementation()
+  if (!acquire) {
+    throw new Error('host harness has no acquisition implementation')
+  }
+  state.acquire.mockImplementation(async (input) => ({
+    ...(await acquire(input)),
+    acquisitionGeneration: 'generation-1'
+  }))
   compact.mockReset().mockResolvedValue({})
   host.deps.adapter.compact = compact
   await attach()
@@ -84,4 +94,36 @@ describe('host conversation command concurrency', () => {
     })
     expect(compact).toHaveBeenCalledTimes(1)
   })
+
+  it('retires an old command after provider recovery advances the session generation', async () => {
+    await host.hold(HOST_TEST_SESSION, 'conversation-surface')
+    let rejectFlush!: (error: Error) => void
+    const flush = vi.spyOn(host, 'flushStreamedEvents').mockImplementationOnce(
+      () =>
+        new Promise<void>((_resolve, reject) => {
+          rejectFlush = reject
+        })
+    )
+    const params = commandParams('compact')
+    const running = host.conversationCommand(CALLER, params)
+    await vi.waitFor(() => expect(rejectFlush).toBeTypeOf('function'))
+    const fence = store.getRecord(params.envelope.sessionId)!.lease.runtimeFence
+
+    await host.handleAdapterEvent({
+      type: 'ended',
+      sessionId: params.envelope.sessionId,
+      reason: 'provider exited',
+      cause: 'unexpected-exit',
+      fence,
+      acquisitionGeneration: 'generation-1'
+    })
+    expect(store.getRecord(params.envelope.sessionId)?.lease.runtimeFence).toBeGreaterThan(fence)
+    rejectFlush(new Error('old event sink failed'))
+
+    await expect(running).resolves.toMatchObject({ ok: true, value: { state: 'unknown' } })
+    flush.mockRestore()
+    await expect(host.conversationCommand(CALLER, commandParams('compact'))).resolves.toMatchObject(
+      { ok: true, value: { state: 'completed' } }
+    )
+  })
 })

src/main/native-chat/agent-session-wire/structured-conversation-command-execution.ts

@@ -232,29 +232,32 @@ export class StructuredConversationCommandExecution {
     retire = false
   ): Promise<void> {
     const execution = entry.execution
-    if (!execution || !this.ownsExecution(entry, execution)) {
+    if (!execution) {
       return
     }
     const error = cause instanceof Error ? cause.message : COMPACTION_UNCONFIRMED
+    const value: AgentSessionConversationCommandResult = {
+      command: entry.command,
+      state: 'unknown',
+      error: entry.command === 'compact' ? COMPACTION_UNCONFIRMED : error.slice(0, 4096)
+    }
+    const result = conversationCommandResult(execution, value)
+    if (!this.ownsExecution(entry, execution)) {
+      if (this.owner.isCurrent(entry)) {
+        this.owner.finish(entry, result)
+      }
+      return
+    }
     try {
       await this.context().deps.store.recordOperationOutcome({
         callerKey: execution.operationCallerKey,
         operationId: execution.prepared.operationId,
         outcome: { status: 'unknown' }
       })
-      await this.publishLifecycle(
-        entry,
-        { ...execution.prepared, error },
-        retire ? 'unverifiable' : 'running'
-      )
+      await this.publishLifecycle(entry, value, retire ? 'unverifiable' : 'running')
     } catch (persistError) {
       this.owner.report(entry, persistError)
     }
-    const result = conversationCommandResult(execution, {
-      command: entry.command,
-      state: 'unknown',
-      error: entry.command === 'compact' ? COMPACTION_UNCONFIRMED : error
-    })
     if (retire) {
       this.owner.finish(entry, result)
     } else {

src/main/native-chat/agent-session-wire/structured-conversation-command.test.ts

@@ -713,11 +713,16 @@ describe('host conversation commands', () => {
   })
 
   it('retires a failed clear that has no provider callback to settle it later', async () => {
-    vi.spyOn(host, 'attach').mockRejectedValueOnce(new Error('replacement transport failed'))
-    await expect(host.conversationCommand(caller, commandParams('clear'))).resolves.toMatchObject({
+    vi.spyOn(host, 'attach').mockRejectedValueOnce(new Error('x'.repeat(5_000)))
+    const result = await host.conversationCommand(caller, commandParams('clear'))
+    expect(result).toMatchObject({
       ok: true,
       value: { state: 'unknown' }
     })
+    if (!result.ok) {
+      throw new Error('clear was refused')
+    }
+    expect(result.value.error).toHaveLength(4_096)
     const status = host
       .history({ sessionId: HOST_TEST_SESSION, direction: 'tail' })
       .page.items.find((item) => item.body.kind === 'status')

src/main/native-chat/agent-session-wire/structured-session-compaction.ts

@@ -56,7 +56,7 @@ export class StructuredSessionCompaction {
           return
         }
         this.pending.delete(sessionId)
-        if (expired && onLateResult) {
+        if ((expired || pending.interrupted) && onLateResult) {
           void onLateResult(result).catch((error) =>
             console.warn('Could not persist late compaction completion', error)
           )
@@ -141,6 +141,26 @@ export class StructuredSessionCompaction {
     this.pending.get(sessionId)?.interrupt()
   }
 
+  /** A provider-acknowledged later turn proves an interrupted command no longer owns output. */
+  claudeTurnStarted(sessionId: string, message: Record<string, unknown>): void {
+    const pending = this.pending.get(sessionId)
+    if (!pending?.interrupted) {
+      return
+    }
+    const envelope = record(message.message)
+    const content = Array.isArray(envelope.content) ? envelope.content : [envelope.content]
+    const last = content.at(-1)
+    const prompt = typeof last === 'string' ? last : record(last).text
+    if (
+      message.user_message_uuid === pending.turnId ||
+      message.uuid === pending.turnId ||
+      prompt === '/compact'
+    ) {
+      return
+    }
+    pending.finish({ error: 'Compaction was interrupted.' })
+  }
+
   codex(sessionId: string, method: string, value: unknown): void {
     const pending = this.pending.get(sessionId)
     const params = record(value)