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 2d044399351..2282ede7c81 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 @@ -162,7 +162,7 @@ export class StructuredConversationCommandExecution { } attachError = null } - if (!this.owner.isCurrent(entry)) { + if (!this.canSettle(entry, execution)) { if (!attachError) { await this.host .close(replacementSessionId) 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 5f6663e7601..a9b335eee6f 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 @@ -538,6 +538,58 @@ describe('host conversation commands', () => { }) }) + it('closes a replacement attached while abandonment lifecycle publication is pending', async () => { + const params = commandParams('clear') + const replacementAcquire = Promise.withResolvers() + const replacementAcquireStarted = Promise.withResolvers() + const abandonmentPublication = Promise.withResolvers() + const abandonmentPersisted = Promise.withResolvers() + const replacementAttached = Promise.withResolvers() + const originalAcquire = vi.mocked(adapter.acquire).getMockImplementation()! + vi.mocked(adapter.acquire).mockImplementation(async (input) => { + if (acquisitions > 0) { + replacementAcquireStarted.resolve() + await replacementAcquire.promise + } + return originalAcquire(input) + }) + const persistOutcome = store.recordOperationOutcome.bind(store) + vi.spyOn(store, 'recordOperationOutcome').mockImplementation(async (input) => { + await persistOutcome(input) + if ( + input.operationId === params.envelope.clientOperationId && + input.outcome.status === 'succeeded' && + input.outcome.conversationCommand?.state === 'unknown' + ) { + abandonmentPersisted.resolve() + await abandonmentPublication.promise + } + }) + const attachReplacement = host.attach.bind(host) + vi.spyOn(host, 'attach').mockImplementation(async (attachCaller, input) => { + const result = await attachReplacement(attachCaller, input) + if (input.envelope.sessionId !== HOST_TEST_SESSION) { + replacementAttached.resolve() + } + return result + }) + + const running = host.conversationCommand(caller, params) + await replacementAcquireStarted.promise + const replacementSessionId = + store.getRecord(HOST_TEST_SESSION)?.conversationCommand?.replacementSessionId + const close = host.close(HOST_TEST_SESSION) + await abandonmentPersisted.promise + + replacementAcquire.resolve() + await replacementAttached.promise + await vi.waitFor(() => expect(host.hasSession(replacementSessionId!)).toBe(false)) + + abandonmentPublication.resolve() + await expect(close).resolves.toBeUndefined() + await expect(running).resolves.toMatchObject({ ok: true, value: { state: 'unknown' } }) + }) + it('does not publish terminal success before its durable command commit', async () => { const persist = store.setConversationCommand.bind(store) let failed = false diff --git a/src/renderer/src/components/native-chat/structured-conversation-command-claim.test.ts b/src/renderer/src/components/native-chat/structured-conversation-command-claim.test.ts index c7da2c1ece2..81194c06150 100644 --- a/src/renderer/src/components/native-chat/structured-conversation-command-claim.test.ts +++ b/src/renderer/src/components/native-chat/structured-conversation-command-claim.test.ts @@ -47,8 +47,10 @@ describe('StructuredConversationCommandClaim', () => { blocked: false, send: neverReplies }) - expect(claim.applyStreamSnapshot([lifecycleItem(command, 'running')])).toBe(false) - expect(claim.applyStreamSnapshot([lifecycleItem(command, 'completed')])).toBe(true) + expect(claim.applyStreamSnapshot([lifecycleItem(command, 'running')])).toEqual([]) + expect(claim.applyStreamSnapshot([lifecycleItem(command, 'completed')])).toEqual([ + OPERATION_ID + ]) await expect(outcome).resolves.toEqual({ accepted: true, error: null }) } ) @@ -59,13 +61,15 @@ describe('StructuredConversationCommandClaim', () => { command: 'compact', operationId: OPERATION_ID, blocked: false, - send: async () => ({ result: { command: 'compact', state: 'unknown' }, unresolved: false }) + send: async () => ({ status: 'unresolved' }) }) await Promise.resolve() - expect(claim.applyStreamSnapshot([lifecycleItem('compact', 'unverifiable')])).toBe(true) + expect(claim.applyStreamSnapshot([lifecycleItem('compact', 'unverifiable')])).toEqual([ + OPERATION_ID + ]) await expect(outcome).resolves.toMatchObject({ accepted: false, - error: expect.stringContaining('Restart the session') + error: expect.stringContaining('Retry the command') }) expect(claim.isRunning).toBe(false) }) @@ -94,13 +98,13 @@ describe('StructuredConversationCommandClaim', () => { command: 'compact', operationId: OPERATION_ID, blocked: false, - send: async () => ({ result: null, unresolved: true }) + send: async () => ({ status: 'unresolved' }) }) ).resolves.toMatchObject({ accepted: false, retrySameOperation: true }) expect(claim.isRunning).toBe(false) }) - it('bounds a missing reply and keeps the obligation until host lifecycle settles it', async () => { + it('releases interaction at the deadline and allows same-operation replay', async () => { vi.useFakeTimers() try { const claim = new StructuredConversationCommandClaim(10) @@ -112,24 +116,46 @@ describe('StructuredConversationCommandClaim', () => { }) const settled = expect(outcome).resolves.toMatchObject({ accepted: false, - error: expect.stringContaining('Restart the session'), + error: expect.stringContaining('Retry the command'), retrySameOperation: true }) await vi.advanceTimersByTimeAsync(11) await settled - expect(claim.hasObligation).toBe(true) + expect(claim.isRunning).toBe(false) await expect( claim.run({ command: 'compact', - operationId: 'op-2', + operationId: OPERATION_ID, blocked: false, - send: neverReplies + send: async () => ({ + status: 'completed', + result: { command: 'compact', state: 'completed' } + }) }) - ).resolves.toMatchObject({ error: expect.stringContaining('Restart the session') }) + ).resolves.toEqual({ accepted: true, error: null }) + } finally { + vi.useRealTimers() + } + }) - expect(claim.applyStreamSnapshot([lifecycleItem('compact', 'completed')])).toBe(true) - expect(claim.hasObligation).toBe(false) + it('drops the deadline observer after a bounded window', async () => { + vi.useFakeTimers() + try { + const claim = new StructuredConversationCommandClaim(10) + const outcome = claim.run({ + command: 'compact', + operationId: OPERATION_ID, + blocked: false, + send: neverReplies + }) + + await vi.advanceTimersByTimeAsync(11) + await outcome + expect(claim.isOperationOutstanding(OPERATION_ID)).toBe(true) + + await vi.advanceTimersByTimeAsync(11) + expect(claim.isOperationOutstanding(OPERATION_ID)).toBe(false) } finally { vi.useRealTimers() } @@ -142,7 +168,10 @@ describe('StructuredConversationCommandClaim', () => { command: 'clear', operationId: OPERATION_ID, blocked: false, - send: async () => ({ result: { command: 'clear', state: 'completed' }, unresolved: false }) + send: async () => ({ + status: 'completed', + result: { command: 'clear', state: 'completed' } + }) }) ).resolves.toEqual({ accepted: true, error: null }) }) @@ -157,4 +186,17 @@ describe('StructuredConversationCommandClaim', () => { expect(send).toHaveBeenCalledTimes(1) claim.reset() }) + + it('returns a definitive refusal without retaining retry ownership', async () => { + const claim = new StructuredConversationCommandClaim() + await expect( + claim.run({ + command: 'compact', + operationId: OPERATION_ID, + blocked: false, + send: async () => ({ status: 'refused', error: 'Wait for pending work.' }) + }) + ).resolves.toEqual({ accepted: false, error: 'Wait for pending work.' }) + expect(claim.isOperationOutstanding(OPERATION_ID)).toBe(false) + }) }) diff --git a/src/renderer/src/components/native-chat/structured-conversation-command-claim.ts b/src/renderer/src/components/native-chat/structured-conversation-command-claim.ts index 8670dff40ff..dbb3f8f19e6 100644 --- a/src/renderer/src/components/native-chat/structured-conversation-command-claim.ts +++ b/src/renderer/src/components/native-chat/structured-conversation-command-claim.ts @@ -12,10 +12,10 @@ export type ConversationCommandOutcome = { accepted: boolean; error: string | nu export type ConversationCommandClaimOutcome = ConversationCommandOutcome & { retrySameOperation?: true } -export type ConversationCommandReply = { - result: AgentSessionConversationCommandResult | null - unresolved: boolean -} +export type ConversationCommandReply = + | { status: 'completed'; result: AgentSessionConversationCommandResult } + | { status: 'refused'; error: string | null } + | { status: 'unresolved' } type Obligation = { command: AgentSessionConversationCommand @@ -28,6 +28,10 @@ type LiveClaim = Obligation & { onLateReply: () => void } +type ClaimObserver = Obligation & { + deadline: ReturnType +} + /** The provider's own completion window is 180s. Keep a small margin for host persistence and * stream delivery before presenting the command as unresolved. */ export const CONVERSATION_COMMAND_DEADLINE_MS = 195_000 @@ -39,7 +43,7 @@ function message(key: string, fallback: string): string { function unresolvedMessage(command: AgentSessionConversationCommand): string { return translate( 'components.native-chat.conversationCommand.mayStillBeRunning', - 'The previous /{{value0}} may still be running. Restart the session before running it again.', + 'The previous /{{value0}} could not be confirmed. Retry the command to check its status.', { value0: command } ) } @@ -83,20 +87,19 @@ export function isUnconfirmedConversationCommand(method: string, value: unknown) /** Correlates one in-flight request with host lifecycle. Durable ownership remains on the host. */ export class StructuredConversationCommandClaim { private live: LiveClaim | null = null - private unresolved: Obligation | null = null + private readonly observers = new Map() constructor(private readonly deadlineMs = CONVERSATION_COMMAND_DEADLINE_MS) {} get isRunning(): boolean { - return this.live !== null || this.unresolved !== null - } - - get hasObligation(): boolean { - return this.live !== null || this.unresolved !== null + return this.live !== null } isOperationOutstanding(operationId: string): boolean { - return this.live?.operationId === operationId || this.unresolved?.operationId === operationId + return ( + this.live?.operationId === operationId || + [...this.observers.values()].some((observer) => observer.operationId === operationId) + ) } run(input: { @@ -106,19 +109,18 @@ export class StructuredConversationCommandClaim { send: () => Promise onLateReply?: () => void }): Promise { - if (this.live || this.unresolved || input.blocked) { + if (this.live || input.blocked) { return Promise.resolve({ accepted: false, - error: this.unresolved - ? unresolvedMessage(this.unresolved.command) - : message( - this.live ? 'running' : 'pendingWork', - this.live - ? 'Wait for the conversation operation to finish.' - : 'Wait for pending work and messages to finish before using this command.' - ) + error: message( + this.live ? 'running' : 'pendingWork', + this.live + ? 'Wait for the conversation operation to finish.' + : 'Wait for pending work and messages to finish before using this command.' + ) }) } + this.removeObserver(input.command, input.operationId) const waiter = Promise.withResolvers() const claim: LiveClaim = { command: input.command, @@ -135,23 +137,24 @@ export class StructuredConversationCommandClaim { return waiter.promise } - applyStreamSnapshot(items: readonly AgentJournalRenderItem[]): boolean { + applyStreamSnapshot(items: readonly AgentJournalRenderItem[]): string[] { + const settledOperationIds: string[] = [] if (this.live) { const outcome = terminalFrameOutcome(items, this.live) - if (!outcome) { - return false - } - this.finish(this.live, outcome) - return true - } - if (this.unresolved) { - const outcome = terminalFrameOutcome(items, this.unresolved) if (outcome) { - this.unresolved = null - return true + settledOperationIds.push(this.live.operationId) + this.finish(this.live, outcome) } } - return false + for (const [key, observer] of this.observers) { + const outcome = terminalFrameOutcome(items, observer) + if (outcome) { + clearTimeout(observer.deadline) + this.observers.delete(key) + settledOperationIds.push(observer.operationId) + } + } + return settledOperationIds } reset(retryPreparedClear = false): void { @@ -164,21 +167,37 @@ export class StructuredConversationCommandClaim { : {}) }) } - this.unresolved = null + for (const observer of this.observers.values()) { + clearTimeout(observer.deadline) + } + this.observers.clear() } private applyReply(claim: LiveClaim, reply: ConversationCommandReply): void { - if (this.unresolved?.operationId === claim.operationId) { - if (!reply.unresolved && reply.result?.state !== 'unknown') { - this.unresolved = null - claim.onLateReply() + if (this.live !== claim) { + if (reply.status === 'unresolved') { + return } + if (this.live?.command === claim.command && this.live.operationId === claim.operationId) { + this.applyReply(this.live, reply) + return + } + this.removeObserver(claim.command, claim.operationId) + claim.onLateReply() return } - if (reply.result?.state === 'unknown') { + if (reply.status === 'unresolved') { + this.finishUnconfirmed(claim) return } - if (reply.unresolved || !reply.result) { + if (reply.status === 'refused') { + this.finish(claim, { + accepted: false, + error: reply.error ?? message('unconfirmed', 'Conversation operation was not confirmed.') + }) + return + } + if (reply.result.state === 'unknown') { this.finishUnconfirmed(claim) return } @@ -189,9 +208,15 @@ export class StructuredConversationCommandClaim { } private finishUnconfirmed(claim: LiveClaim): void { - this.finish(claim, { + if (this.live !== claim) { + return + } + clearTimeout(claim.deadline) + this.live = null + this.observe(claim) + claim.settle({ accepted: false, - error: message('unconfirmed', 'Conversation operation was not confirmed.'), + error: unresolvedMessage(claim.command), retrySameOperation: true }) } @@ -202,7 +227,6 @@ export class StructuredConversationCommandClaim { } clearTimeout(claim.deadline) this.live = null - this.unresolved = null claim.settle(outcome) } @@ -211,11 +235,39 @@ export class StructuredConversationCommandClaim { return } this.live = null - this.unresolved = { command: claim.command, operationId: claim.operationId } + this.observe(claim) claim.settle({ accepted: false, error: unresolvedMessage(claim.command), retrySameOperation: true }) } + + private observe(claim: LiveClaim): void { + const key = this.observerKey(claim.command, claim.operationId) + const observer: ClaimObserver = { + command: claim.command, + operationId: claim.operationId, + deadline: setTimeout(() => { + if (this.observers.get(key) === observer) { + this.observers.delete(key) + } + }, this.deadlineMs) + } + this.observers.set(key, observer) + } + + private observerKey(command: AgentSessionConversationCommand, operationId: string): string { + return `${command}:${operationId}` + } + + private removeObserver(command: AgentSessionConversationCommand, operationId: string): void { + const key = this.observerKey(command, operationId) + const observer = this.observers.get(key) + if (!observer) { + return + } + clearTimeout(observer.deadline) + this.observers.delete(key) + } } diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-mutate.ts b/src/renderer/src/components/native-chat/use-structured-agent-session-mutate.ts index 6d4aa85eef5..1910276d992 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session-mutate.ts +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-mutate.ts @@ -16,11 +16,13 @@ import { structuredSessionOperationId } from './use-structured-agent-session-out export type StructuredAgentSessionMutateOptions = { operationId?: string - /** Called instead of returning a verdict when no reply arrived. A refusal is a decision; a - * missing reply is not, and the request it belongs to may still be running. */ - onUnresolved?: (message: string) => void } +export type StructuredAgentSessionMutationDisposition = + | { status: 'completed'; value: T } + | { status: 'refused'; message: string | null } + | { status: 'unresolved'; message: string } + export type StructuredAgentSessionMutate = ( method: string, fingerprintMethod: string, @@ -28,6 +30,24 @@ export type StructuredAgentSessionMutate = ( options?: StructuredAgentSessionMutateOptions ) => Promise +export type StructuredAgentSessionMutateWithDisposition = ( + method: string, + fingerprintMethod: string, + fields: Record, + options?: StructuredAgentSessionMutateOptions +) => Promise> + +export function structuredAgentSessionMutationScope( + target: RuntimeClientTarget, + sessionId: string +): string { + return JSON.stringify([ + target.kind, + target.kind === 'environment' ? target.environmentId : null, + sessionId + ]) +} + export function useStructuredAgentSessionMutate(args: { sessionId: string target: RuntimeClientTarget @@ -37,6 +57,7 @@ export function useStructuredAgentSessionMutate(args: { stateRef: { current: { fence: number | null } } }): { mutate: StructuredAgentSessionMutate + mutateWithDisposition: StructuredAgentSessionMutateWithDisposition writeError: string | null clearWriteError: (operationId: string) => void } { @@ -47,11 +68,7 @@ export function useStructuredAgentSessionMutate(args: { const latestSettledSequence = useRef(0) const operationIds = useRef(new Map()) const enabledRef = useRef(enabled) - const requestScope = JSON.stringify([ - target.kind, - target.kind === 'environment' ? target.environmentId : null, - sessionId - ]) + const requestScope = structuredAgentSessionMutationScope(target, sessionId) const requestScopeRef = useRef(requestScope) useEffect(() => { // Why: update the gate after commit so render stays free of ref mutations. @@ -65,15 +82,15 @@ export function useStructuredAgentSessionMutate(args: { setWriteError(null) }, [requestScope]) - const mutate = useCallback( + const mutateWithDisposition = useCallback( async ( method: string, fingerprintMethod: string, fields: Record, options?: StructuredAgentSessionMutateOptions - ): Promise => { + ): Promise> => { if (!enabled || !enabledRef.current || stateRef.current.fence === null) { - return null + return { status: 'refused', message: null } } const targetFence = stateRef.current.fence const key = `${sessionId}:${fingerprintMethod}:${JSON.stringify(fields)}` @@ -109,14 +126,14 @@ export function useStructuredAgentSessionMutate(args: { writeErrorOwner.current = { operationId: clientOperationId, sequence } setWriteError(message) } - options?.onUnresolved?.(message) - return null + return { status: 'unresolved', message } } if (!result.ok) { - if ( - agentSessionRefusalOperationState(fingerprintMethod, result.refusal.code) === - 'settled-rejected' - ) { + const operationState = agentSessionRefusalOperationState( + fingerprintMethod, + result.refusal.code + ) + if (operationState === 'settled-rejected') { operationIds.current.delete(key) } if ( @@ -129,14 +146,16 @@ export function useStructuredAgentSessionMutate(args: { writeErrorOwner.current = { operationId: clientOperationId, sequence } setWriteError(result.refusal.message) } - return null + return operationState === 'unknown' + ? { status: 'unresolved', message: result.refusal.message } + : { status: 'refused', message: result.refusal.message } } if ( !enabledRef.current || requestScopeRef.current !== targetScope || stateRef.current.fence !== targetFence ) { - return null + return { status: 'refused', message: null } } if (!isUnconfirmedConversationCommand(fingerprintMethod, result.value)) { operationIds.current.delete(key) @@ -146,11 +165,24 @@ export function useStructuredAgentSessionMutate(args: { writeErrorOwner.current = null setWriteError(null) } - return result.value + return { status: 'completed', value: result.value } }, [enabled, requestScope, sessionId, stateRef, target] ) + const mutate = useCallback( + async ( + method: string, + fingerprintMethod: string, + fields: Record, + options?: StructuredAgentSessionMutateOptions + ): Promise => { + const result = await mutateWithDisposition(method, fingerprintMethod, fields, options) + return result.status === 'completed' ? result.value : null + }, + [mutateWithDisposition] + ) + const clearWriteError = useCallback((operationId: string) => { if (writeErrorOwner.current?.operationId !== operationId) { return @@ -158,5 +190,5 @@ export function useStructuredAgentSessionMutate(args: { writeErrorOwner.current = null setWriteError(null) }, []) - return { mutate, writeError, clearWriteError } + return { mutate, mutateWithDisposition, writeError, clearWriteError } } diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session.ts b/src/renderer/src/components/native-chat/use-structured-agent-session.ts index 45bac237e4d..7425b8c9201 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session.ts +++ b/src/renderer/src/components/native-chat/use-structured-agent-session.ts @@ -29,13 +29,21 @@ export function useStructuredAgentSession(args: { transportEnabled?: boolean }) { const { agent, isVisible, sessionId, target, transportEnabled = true } = args - const { state, loadingOlder, loadOlder, mutate, writeError, clearWriteError, providerVisible } = - useStructuredAgentSessionTransport({ - sessionId, - target, - isVisible, - enabled: transportEnabled - }) + const { + state, + loadingOlder, + loadOlder, + mutate, + mutateWithDisposition, + writeError, + clearWriteError, + providerVisible + } = useStructuredAgentSessionTransport({ + sessionId, + target, + isVisible, + enabled: transportEnabled + }) const transportState = useStructuredAgentSessionTransportState(state, transportEnabled) const { conversationCommands, optionSnapshot, optionSurface, setStructuredOption } = useStructuredAgentSessionOptions({ @@ -64,6 +72,7 @@ export function useStructuredAgentSession(args: { ) const conversationCommand = useStructuredConversationCommand({ sessionId, + target, fence: transportState.fence, items: transportState.journalItems, blocked: Boolean( @@ -72,7 +81,7 @@ export function useStructuredAgentSession(args: { transportState.backgroundTasks.isMonitoring || outbox.length ), - mutate, + mutate: mutateWithDisposition, onReconciled: clearWriteError }) return { diff --git a/src/renderer/src/components/native-chat/use-structured-conversation-command.test.tsx b/src/renderer/src/components/native-chat/use-structured-conversation-command.test.tsx index 1203777c74c..44fb47363da 100644 --- a/src/renderer/src/components/native-chat/use-structured-conversation-command.test.tsx +++ b/src/renderer/src/components/native-chat/use-structured-conversation-command.test.tsx @@ -5,6 +5,7 @@ import { useEffect, useRef } from 'react' import { beforeEach, describe, expect, it, vi } from 'vitest' import { agentJournalSubmissionKey } from '../../../../shared/agent-session-journal-item-key' import type { AgentJournalRenderItem } from '../../../../shared/agent-session-journal-types' +import type { RuntimeClientTarget } from '@/runtime/runtime-rpc-client' const mocks = vi.hoisted(() => ({ call: vi.fn() })) @@ -22,23 +23,26 @@ function useCommandHarness(props: { fence: number items: readonly AgentJournalRenderItem[] sessionId?: string + target?: RuntimeClientTarget }) { const sessionId = props.sessionId ?? 'session-1' + const target = props.target ?? LOCAL_TARGET const stateRef = useRef<{ fence: number | null }>({ fence: props.fence }) useEffect(() => { stateRef.current = { fence: props.fence } }, [props.fence]) const mutation = useStructuredAgentSessionMutate({ sessionId, - target: LOCAL_TARGET, + target, stateRef }) const command = useStructuredConversationCommand({ sessionId, + target, fence: props.fence, items: props.items, blocked: false, - mutate: mutation.mutate, + mutate: mutation.mutateWithDisposition, onReconciled: mutation.clearWriteError }) return { ...command, mutate: mutation.mutate, writeError: mutation.writeError } @@ -106,6 +110,55 @@ describe('useStructuredConversationCommand', () => { view.unmount() }) + it('retires a command when its runtime target changes at the same session and fence', async () => { + mocks.call.mockImplementation(() => new Promise(() => {})) + const initialProps: { + fence: number + items: AgentJournalRenderItem[] + target: RuntimeClientTarget + } = { fence: 1, items: [], target: LOCAL_TARGET } + const view = renderHook((props) => useCommandHarness(props), { initialProps }) + + let local!: ReturnType + act(() => { + local = view.result.current.run('compact') + }) + const localOperationId = mocks.call.mock.calls[0]![2].envelope.clientOperationId + + view.rerender({ + fence: 1, + items: [], + target: { kind: 'environment', environmentId: 'environment-a' } + }) + await expect(local).resolves.toMatchObject({ accepted: false }) + act(() => { + void view.result.current.run('compact') + }) + const environmentAOperationId = mocks.call.mock.calls[1]![2].envelope.clientOperationId + expect(mocks.call.mock.calls[1]![0]).toEqual({ + kind: 'environment', + environmentId: 'environment-a' + }) + expect(environmentAOperationId).not.toBe(localOperationId) + + view.rerender({ + fence: 1, + items: [], + target: { kind: 'environment', environmentId: 'environment-b' } + }) + act(() => { + void view.result.current.run('compact') + }) + expect(mocks.call.mock.calls[2]![0]).toEqual({ + kind: 'environment', + environmentId: 'environment-b' + }) + expect(mocks.call.mock.calls[2]![2].envelope.clientOperationId).not.toBe( + environmentAOperationId + ) + view.unmount() + }) + it('ignores a late transport error after switching sessions at the same fence', async () => { let reject!: (error: Error) => void mocks.call.mockImplementation( @@ -166,6 +219,61 @@ describe('useStructuredConversationCommand', () => { } }) + it('replays an expired command with the same identity', async () => { + vi.useFakeTimers() + try { + mocks.call + .mockImplementationOnce(() => new Promise(() => {})) + .mockResolvedValueOnce({ + ok: true, + value: { command: 'compact', state: 'completed' } + }) + const view = renderHook(() => useCommandHarness({ fence: 1, items: [] })) + + let first!: ReturnType + act(() => { + first = view.result.current.run('compact') + }) + const operationId = mocks.call.mock.calls[0]![2].envelope.clientOperationId + await act(() => vi.advanceTimersByTimeAsync(CONVERSATION_COMMAND_DEADLINE_MS + 1)) + await expect(first).resolves.toMatchObject({ accepted: false }) + + await act(async () => { + await view.result.current.run('compact') + }) + expect(mocks.call.mock.calls[1]![2].envelope.clientOperationId).toBe(operationId) + view.unmount() + } finally { + vi.useRealTimers() + } + }) + + it('retires a definitively refused command identity before retry', async () => { + mocks.call + .mockResolvedValueOnce({ + ok: false, + refusal: { + code: 'agent_session_operation_invalid', + message: 'Wait for pending work.' + } + }) + .mockImplementation(() => new Promise(() => {})) + const view = renderHook(() => useCommandHarness({ fence: 1, items: [] })) + + let refusal!: Awaited> + await act(async () => { + refusal = await view.result.current.run('compact') + }) + expect(refusal).toEqual({ accepted: false, error: 'Wait for pending work.' }) + const refusedOperationId = mocks.call.mock.calls[0]![2].envelope.clientOperationId + + act(() => { + void view.result.current.run('compact') + }) + expect(mocks.call.mock.calls[1]![2].envelope.clientOperationId).not.toBe(refusedOperationId) + view.unmount() + }) + it('retires a completed clear and gives the next command a fresh identity', async () => { mocks.call .mockResolvedValueOnce({ ok: true, value: { command: 'clear', state: 'completed' } }) diff --git a/src/renderer/src/components/native-chat/use-structured-conversation-command.ts b/src/renderer/src/components/native-chat/use-structured-conversation-command.ts index 2d4df6de0c4..3120c0a3fff 100644 --- a/src/renderer/src/components/native-chat/use-structured-conversation-command.ts +++ b/src/renderer/src/components/native-chat/use-structured-conversation-command.ts @@ -1,7 +1,7 @@ // The session hook's half of a conversation command: it owns the claim, feeds it the session's own // stream, and retires it when the session restarts. -import { useEffect, useRef } from 'react' +import { useEffect, useLayoutEffect, useRef } from 'react' import type { AgentSessionConversationCommand, AgentSessionConversationCommandResult @@ -12,34 +12,41 @@ import { type ConversationCommandOutcome, type ConversationCommandReply } from './structured-conversation-command-claim' -import type { StructuredAgentSessionMutate } from './use-structured-agent-session-mutate' +import { + structuredAgentSessionMutationScope, + type StructuredAgentSessionMutateWithDisposition +} from './use-structured-agent-session-mutate' import { structuredSessionOperationId } from './use-structured-agent-session-outbox' +import type { RuntimeClientTarget } from '@/runtime/runtime-rpc-client' export function useStructuredConversationCommand(args: { sessionId: string + target: RuntimeClientTarget fence: number | null items: readonly AgentJournalRenderItem[] /** Work the command would have to interrupt; the host refuses in that state anyway. */ blocked: boolean - mutate: StructuredAgentSessionMutate + mutate: StructuredAgentSessionMutateWithDisposition onReconciled: (operationId: string) => void }): { run: (command: AgentSessionConversationCommand) => Promise isRunning: () => boolean retire: () => void } { - const { blocked, fence, items, mutate, onReconciled, sessionId } = args + const { blocked, fence, items, mutate, onReconciled, sessionId, target } = args const claim = useRef(new StructuredConversationCommandClaim()) const operationIds = useRef(new Map()) + const requestScope = structuredAgentSessionMutationScope(target, sessionId) useEffect(() => { - const operationId = - operationIds.current.get('compact') ?? operationIds.current.get('clear') ?? null - if (claim.current.applyStreamSnapshot(items)) { - operationIds.current.clear() - if (operationId) { - onReconciled(operationId) + const settledOperationIds = claim.current.applyStreamSnapshot(items) + for (const operationId of settledOperationIds) { + for (const [command, candidate] of operationIds.current) { + if (candidate === operationId) { + operationIds.current.delete(command) + } } + onReconciled(operationId) } }, [items, onReconciled]) @@ -53,23 +60,23 @@ export function useStructuredConversationCommand(args: { } }, [fence]) - useEffect(() => { + useLayoutEffect(() => { const current = claim.current const ids = operationIds.current return () => { ids.clear() current.reset() } - }, [sessionId]) + }, [requestScope]) return { run: (command) => { - if (claim.current.hasObligation) { + if (claim.current.isRunning) { return claim.current.run({ command, operationId: '', blocked, - send: async () => ({ result: null, unresolved: false }) + send: async () => ({ status: 'refused', error: null }) }) } // Minted here, not inside `mutate`: the claim needs the id to know which journal item carries @@ -82,22 +89,24 @@ export function useStructuredConversationCommand(args: { operationId, blocked, send: async (): Promise => { - let unresolved = false - const result = await mutate( + const disposition = await mutate( 'agentSession.conversationCommand', 'agentSession.conversationCommand', { command }, - { - operationId, - onUnresolved: () => { - unresolved = true - } - } + { operationId } ) if (!claim.current.isOperationOutstanding(operationId)) { onReconciled(operationId) } - return { result, unresolved } + if (disposition.status === 'unresolved') { + return { status: 'unresolved' } + } + if (disposition.status === 'refused') { + return { status: 'refused', error: disposition.message } + } + return disposition.value.state === 'unknown' + ? { status: 'unresolved' } + : { status: 'completed', result: disposition.value } }, onLateReply: () => { if (operationIds.current.get(command) === operationId) { diff --git a/src/renderer/src/i18n/locales/en.json b/src/renderer/src/i18n/locales/en.json index 56899694a6e..da2ab8eb6c6 100644 --- a/src/renderer/src/i18n/locales/en.json +++ b/src/renderer/src/i18n/locales/en.json @@ -17381,7 +17381,7 @@ "conversationCommand": { "pendingWork": "Wait for pending work and messages to finish before using this command.", "running": "Wait for the conversation operation to finish.", - "mayStillBeRunning": "The previous /{{value0}} may still be running. Restart the session before running it again.", + "mayStillBeRunning": "The previous /{{value0}} could not be confirmed. Retry the command to check its status.", "unconfirmed": "Conversation operation was not confirmed." }, "drop": {