From 258917a0c64567ab96c7440aacbb9cfbc3f2eab9 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Wed, 16 Sep 2026 17:03:37 -0700 Subject: [PATCH] fix(native-chat): retire interrupted compaction ownership --- .../claude-structured-compaction.test.ts | 82 ++++++++++++++++++- .../claude/claude-structured-compaction.ts | 5 +- ...d-conversation-command-concurrency.test.ts | 46 ++++++++++- ...ructured-conversation-command-execution.ts | 25 +++--- .../structured-conversation-command.test.ts | 9 +- .../structured-session-compaction.ts | 22 ++++- 6 files changed, 171 insertions(+), 18 deletions(-) diff --git a/src/main/claude/claude-structured-compaction.test.ts b/src/main/claude/claude-structured-compaction.test.ts index ebb8b4026f1..1e7427a2408 100644 --- a/src/main/claude/claude-structured-compaction.test.ts +++ b/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' + }) + }) }) diff --git a/src/main/claude/claude-structured-compaction.ts b/src/main/claude/claude-structured-compaction.ts index 046ccdd4fb3..2f0727bdef0 100644 --- a/src/main/claude/claude-structured-compaction.ts +++ b/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, 'handle'> | null | undefined ): void { + if (event.type === 'message' && event.startsTurn) { + compactions.claudeTurnStarted(event.sessionId, event.message) + } if (!isClaudeCompactionContent(compactions, event)) { translator?.handle(event) } diff --git a/src/main/native-chat/agent-session-wire/structured-conversation-command-concurrency.test.ts b/src/main/native-chat/agent-session-wire/structured-conversation-command-concurrency.test.ts index 8ac550762dd..9c09a369722 100644 --- a/src/main/native-chat/agent-session-wire/structured-conversation-command-concurrency.test.ts +++ b/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>() let host: StructuredAgentSessionHost let dispatch: Mock +let store: ReturnType['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((_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' } } + ) + }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-conversation-command-execution.ts b/src/main/native-chat/agent-session-wire/structured-conversation-command-execution.ts index 57b66636373..480300ad868 100644 --- a/src/main/native-chat/agent-session-wire/structured-conversation-command-execution.ts +++ b/src/main/native-chat/agent-session-wire/structured-conversation-command-execution.ts @@ -232,29 +232,32 @@ export class StructuredConversationCommandExecution { retire = false ): Promise { 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 { diff --git a/src/main/native-chat/agent-session-wire/structured-conversation-command.test.ts b/src/main/native-chat/agent-session-wire/structured-conversation-command.test.ts index cd7c14d791b..a3e5784b0fb 100644 --- a/src/main/native-chat/agent-session-wire/structured-conversation-command.test.ts +++ b/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') diff --git a/src/main/native-chat/agent-session-wire/structured-session-compaction.ts b/src/main/native-chat/agent-session-wire/structured-session-compaction.ts index 715239707f4..c2e83391c67 100644 --- a/src/main/native-chat/agent-session-wire/structured-session-compaction.ts +++ b/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): 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)