From 48dd3fba5550834d06dfcdbff1a0013e4b9bc0c5 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Wed, 16 Sep 2026 17:19:49 -0700 Subject: [PATCH] fix(native-chat): fence command execution after recovery --- ...d-conversation-command-concurrency.test.ts | 26 +++++++++++++++++++ ...ructured-conversation-command-execution.ts | 24 ++++++++--------- 2 files changed, 37 insertions(+), 13 deletions(-) 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 5f6c598658f..11261808437 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 @@ -127,6 +127,32 @@ describe('host conversation command concurrency', () => { ) }) + it('does not start provider work after recovery advances during event flushing', async () => { + await host.hold(HOST_TEST_SESSION, 'conversation-surface') + const flush = Promise.withResolvers() + const flushing = Promise.withResolvers() + vi.spyOn(host, 'flushStreamedEvents').mockImplementationOnce(() => { + flushing.resolve() + return flush.promise + }) + const running = host.conversationCommand(CALLER, commandParams('compact')) + await flushing.promise + const fence = store.getRecord(HOST_TEST_SESSION)!.lease.runtimeFence + + await host.handleAdapterEvent({ + type: 'ended', + sessionId: HOST_TEST_SESSION, + reason: 'provider exited', + cause: 'unexpected-exit', + fence, + acquisitionGeneration: 'generation-1' + }) + flush.resolve() + + await expect(running).resolves.toMatchObject({ ok: true, value: { state: 'unknown' } }) + expect(compact).not.toHaveBeenCalled() + }) + it('retires a provider completion that arrives after recovery advances the generation', async () => { await host.hold(HOST_TEST_SESSION, 'conversation-surface') const completion = Promise.withResolvers>() 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 efa65c92eee..1134bcac348 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 @@ -16,7 +16,7 @@ import { publishConversationCommandLifecycle } from './structured-conversation-command-lifecycle' import type { StructuredAgentSessionMutationContext } from './structured-agent-session-host-mutations' -import type { StructuredAgentSessionHost } from './structured-agent-session-host' +import type { StructuredAgentSessionHost as Host } from './structured-agent-session-host' type ExecutionOwner = { isCurrent: (entry: PendingConversationCommand) => boolean @@ -24,16 +24,12 @@ type ExecutionOwner = { settleWaiter: (entry: PendingConversationCommand, result: ConversationCommandResult) => void report: (entry: PendingConversationCommand, error: unknown) => void } - -const STALE_COMMAND_COMPLETION = new Error('Conversation operation became stale after recovery.') +type ExecutionHost = Pick export class StructuredConversationCommandExecution { constructor( private readonly context: () => StructuredAgentSessionMutationContext, - private readonly host: Pick< - StructuredAgentSessionHost, - 'attach' | 'close' | 'flushStreamedEvents' | 'hasSession' - >, + private readonly host: ExecutionHost, private readonly owner: ExecutionOwner ) {} @@ -46,7 +42,8 @@ export class StructuredConversationCommandExecution { try { await this.publishLifecycle(entry, execution.prepared, 'running') await this.host.flushStreamedEvents(execution.turn.sessionId) - if (!this.owner.isCurrent(entry)) { + if (!this.canSettle(entry, execution)) { + await this.markUnknown(entry, new Error(CONVERSATION_COMMAND_ABANDONED), true) return } providerMaySettleLate = entry.command === 'compact' @@ -132,7 +129,8 @@ export class StructuredConversationCommandExecution { return } } - if (!this.owner.isCurrent(entry)) { + if (!this.canSettle(entry, execution)) { + await this.markUnknown(entry, new Error(CONVERSATION_COMMAND_ABANDONED), true) return } if (effectiveOptions) { @@ -142,7 +140,8 @@ export class StructuredConversationCommandExecution { } }) } - if (!this.owner.isCurrent(entry)) { + if (!this.canSettle(entry, execution)) { + await this.markUnknown(entry, new Error(CONVERSATION_COMMAND_ABANDONED), true) return } const replacementSessionId = execution.prepared.replacementSessionId! @@ -288,9 +287,8 @@ export class StructuredConversationCommandExecution { entry: PendingConversationCommand, execution: PreparedConversationCommand ): boolean { - const command = this.context().deps.store.getRecord( - execution.turn.sessionId - )?.conversationCommand + const store = this.context().deps.store + const command = store.getRecord(execution.turn.sessionId)?.conversationCommand return this.ownsExecution(entry, execution) && command?.phase === 'prepared' }