diff --git a/config/scripts/vitest-sqlite-runtime-files.mjs b/config/scripts/vitest-sqlite-runtime-files.mjs index 3d2245334426..1d89111c6f49 100644 --- a/config/scripts/vitest-sqlite-runtime-files.mjs +++ b/config/scripts/vitest-sqlite-runtime-files.mjs @@ -88,6 +88,7 @@ export const SQLITE_RUNTIME_INCLUDE = [ 'src/main/native-chat/agent-session-wire/structured-agent-session-close-verdict.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-codex-stop-note-end.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-codex-stop-row.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-codex-submission-stop.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-command-publication.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-conversation-close.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts', diff --git a/mobile/src/session/mobile-structured-agent-session-cancel.ts b/mobile/src/session/mobile-structured-agent-session-cancel.ts index 49232afeb326..440bdc379925 100644 --- a/mobile/src/session/mobile-structured-agent-session-cancel.ts +++ b/mobile/src/session/mobile-structured-agent-session-cancel.ts @@ -2,6 +2,7 @@ import type { AgentSessionCancelResult } from '../../../src/shared/agent-session import type { AgentJournalRenderItem } from '../../../src/shared/agent-session-journal-types' import type { StructuredAgentSessionState } from '../../../src/shared/structured-agent-session-reducer' import { runningStructuredAgentSessionTurnId } from '../../../src/shared/structured-agent-session-live-turn' +import { agentSessionStopTarget } from '../../../src/shared/agent-session-stop-target' import type { RpcClient } from '../transport/rpc-client' import { requestStructuredAgentSessionMutation, @@ -30,6 +31,7 @@ export async function requestMobileStructuredAgentSessionCancel(args: { prompt?: PromptIdentity /** Whether the host answers a repeated Stop of a turn quietly; null until the status probe answers. */ hostAnswersRepeatedStops: boolean | null + targetedStopSupported?: boolean /** Stops still on their way, by what they stop; against a host that does not answer a repeat * quietly, a press for the same one joins it here. */ inFlight: Map> @@ -38,13 +40,17 @@ export async function requestMobileStructuredAgentSessionCancel(args: { const { client, enabled, inFlight, onSendError, sessionId, stateRef } = args const current = stateRef.current const turnId = runningStructuredAgentSessionTurnId(current) - if (!client || !sessionId || !enabled || current.fence === null || !turnId) { + const stopTarget = args.targetedStopSupported + ? agentSessionStopTarget(turnId, current.submissions, current.fence) + : undefined + if (!client || !sessionId || !enabled || current.fence === null || (!turnId && !stopTarget)) { onSendError('Stop not sent') return false } // Check the capability before fields enter the fingerprint. const fields = { - turnId, + ...(turnId ? { turnId } : {}), + ...(stopTarget ? { stopTarget } : {}), ...(args.prompt && args.promptCancelSupported === true ? { prompt: args.prompt } : {}) } // Every press is its own Stop: a kept id would be answered from the last one and stop nothing. diff --git a/mobile/src/session/mobile-structured-agent-session-host-support.test.ts b/mobile/src/session/mobile-structured-agent-session-host-support.test.ts index e5debf14376a..12a969584ca5 100644 --- a/mobile/src/session/mobile-structured-agent-session-host-support.test.ts +++ b/mobile/src/session/mobile-structured-agent-session-host-support.test.ts @@ -17,6 +17,7 @@ describe('structuredAgentSessionHostSupport', () => { queuedMessages: false, queuedCommands: false, quietRepeatedStop: false, + targetedStop: false, statusFeed: false } expect(structuredAgentSessionHostSupport([])).toEqual(none) diff --git a/mobile/src/session/mobile-structured-agent-session-host-support.ts b/mobile/src/session/mobile-structured-agent-session-host-support.ts index 84c329e63738..b72309e16e1b 100644 --- a/mobile/src/session/mobile-structured-agent-session-host-support.ts +++ b/mobile/src/session/mobile-structured-agent-session-host-support.ts @@ -4,7 +4,8 @@ import { AGENT_SESSION_QUEUED_COMMANDS_RUNTIME_CAPABILITY, AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY, AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY, - AGENT_SESSION_REPEATED_STOP_RUNTIME_CAPABILITY + AGENT_SESSION_REPEATED_STOP_RUNTIME_CAPABILITY, + AGENT_SESSION_TARGETED_STOP_RUNTIME_CAPABILITY } from '../../../src/shared/protocol-version' /** Structured-session features the connected host advertised; null until the status probe answers. */ @@ -20,6 +21,7 @@ export type StructuredAgentSessionHostSupport = { statusFeed: boolean /** A Stop that stopped nothing adds no row, so a repeated Stop is quiet. */ quietRepeatedStop: boolean + targetedStop?: boolean } export function structuredAgentSessionHostSupport( @@ -31,6 +33,7 @@ export function structuredAgentSessionHostSupport( queuedMessages: capabilities.includes(AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY), queuedCommands: capabilities.includes(AGENT_SESSION_QUEUED_COMMANDS_RUNTIME_CAPABILITY), statusFeed: capabilities.includes(AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY), - quietRepeatedStop: capabilities.includes(AGENT_SESSION_REPEATED_STOP_RUNTIME_CAPABILITY) + quietRepeatedStop: capabilities.includes(AGENT_SESSION_REPEATED_STOP_RUNTIME_CAPABILITY), + targetedStop: capabilities.includes(AGENT_SESSION_TARGETED_STOP_RUNTIME_CAPABILITY) } } diff --git a/mobile/src/session/mobile-structured-agent-session-stop-target.test.ts b/mobile/src/session/mobile-structured-agent-session-stop-target.test.ts new file mode 100644 index 000000000000..9d44b9049412 --- /dev/null +++ b/mobile/src/session/mobile-structured-agent-session-stop-target.test.ts @@ -0,0 +1,109 @@ +import { beforeEach, expect, it, vi } from 'vitest' +import { EMPTY_STRUCTURED_AGENT_SESSION } from '../../../src/shared/structured-agent-session-reducer' +import { isStructuredAgentSessionMainAgentWorking } from '../../../src/shared/structured-agent-session-main-agent-working' +import type { StructuredAgentSessionState } from '../../../src/shared/structured-agent-session-reducer' +import type { RpcClient } from '../transport/rpc-client' +import { requestMobileStructuredAgentSessionCancel } from './mobile-structured-agent-session-cancel' +import { requestStructuredAgentSessionMutation } from './mobile-structured-agent-session-rpc' + +vi.mock('./mobile-structured-agent-session-rpc', () => ({ + requestStructuredAgentSessionMutation: vi.fn(async () => ({ + status: 'accepted', + value: { cancelled: true } + })) +})) + +beforeEach(() => vi.clearAllMocks()) + +const client: RpcClient = { + sendRequest: vi.fn(), + subscribe: () => () => {}, + updateTerminalSubscriptionViewport: () => {}, + getState: () => 'connected', + getReconnectAttempt: () => 0, + getLastConnectedAt: () => null, + onStateChange: () => () => {}, + notifyForeground: () => {}, + close: () => {} +} + +function stop(state: StructuredAgentSessionState, targetedStopSupported: boolean) { + const onSendError = vi.fn() + return { + onSendError, + result: requestMobileStructuredAgentSessionCancel({ + client, + sessionId: 'phone-chat', + enabled: true, + stateRef: { current: state }, + promptCancelSupported: false, + hostAnswersRepeatedStops: true, + targetedStopSupported, + inFlight: new Map(), + onSendError + }) + } +} + +it.each([false, true])('keeps turn targets capability-gated: %s', async (supported) => { + const state: StructuredAgentSessionState = { + ...EMPTY_STRUCTURED_AGENT_SESSION, + fence: 1, + latestTurn: { + itemId: 'turn-item', + observedAt: 1, + turn: { turnId: 'one', state: 'running' } + } + } + expect(await stop(state, supported).result).toBe(true) + expect(vi.mocked(requestStructuredAgentSessionMutation).mock.calls[0]?.[0].fields).toEqual( + supported ? { turnId: 'one', stopTarget: { kind: 'turn', turnId: 'one' } } : { turnId: 'one' } + ) +}) + +const pendingState: StructuredAgentSessionState = { + ...EMPTY_STRUCTURED_AGENT_SESSION, + fence: 1, + submissions: [ + { + clientMessageId: 'host-published-send', + fence: 1, + payloadFingerprint: 'fp', + dispatchState: 'pending', + providerItemId: null, + reason: null, + submittedAt: 1, + resolvedAt: null + } + ] +} + +it('stops the host-published unanswered submission before a turn opens', async () => { + expect(isStructuredAgentSessionMainAgentWorking(null, pendingState.submissions, 1)).toBe(true) + const request = stop(pendingState, true) + expect(await request.result).toBe(true) + expect(requestStructuredAgentSessionMutation).toHaveBeenCalledWith( + expect.objectContaining({ + fields: { stopTarget: { kind: 'submission', clientMessageId: 'host-published-send' } } + }) + ) + expect(request.onSendError).not.toHaveBeenCalled() +}) + +it('keeps the no-turn refusal and feedback for an older host', async () => { + const request = stop(pendingState, false) + expect(await request.result).toBe(false) + expect(requestStructuredAgentSessionMutation).not.toHaveBeenCalled() + expect(request.onSendError).toHaveBeenCalledWith('Stop not sent') +}) + +it.each([ + { ...pendingState, fence: 2 }, + { ...pendingState, fence: null }, + { ...EMPTY_STRUCTURED_AGENT_SESSION, fence: 1 } +])('does not invent a target without current unanswered work', async (state) => { + const request = stop(state, true) + expect(await request.result).toBe(false) + expect(requestStructuredAgentSessionMutation).not.toHaveBeenCalled() + expect(request.onSendError).toHaveBeenCalledWith('Stop not sent') +}) diff --git a/mobile/src/session/mobile-structured-operation-id-retirement.test.ts b/mobile/src/session/mobile-structured-operation-id-retirement.test.ts index 03181db4100b..98a768cf0f96 100644 --- a/mobile/src/session/mobile-structured-operation-id-retirement.test.ts +++ b/mobile/src/session/mobile-structured-operation-id-retirement.test.ts @@ -63,6 +63,28 @@ function cancelArgs(client: RpcClient, hostAnswersRepeatedStops: boolean | null } describe('structured mutation id retirement', () => { + it.each([false, true])( + 'uses a receipt-free turn target only for a capable host: %s', + async (targetedStopSupported) => { + const sendRequest = vi.fn(async (_method: string, _params: SentParams) => ({ + ok: true, + result: { ok: true, value: { cancelled: true } }, + _meta: { runtimeId: 'runtime-1' } + })) + expect( + await requestMobileStructuredAgentSessionCancel({ + ...cancelArgs(fakeClient(sendRequest)), + targetedStopSupported + }) + ).toBe(true) + const params = sendRequest.mock.calls[0]?.[1] + expect(params).toMatchObject( + targetedStopSupported + ? { stopTarget: { kind: 'turn', turnId: 'turn-1' } } + : expect.not.objectContaining({ stopTarget: expect.anything() }) + ) + } + ) it('reports a host that cannot say what became of the id as unknown', async () => { const result = await requestStructuredAgentSessionMutation({ client: fakeClient(async () => operationRefusedAsUnknown()), diff --git a/mobile/src/session/use-mobile-structured-agent-session-background-tasks.test.tsx b/mobile/src/session/use-mobile-structured-agent-session-background-tasks.test.tsx index aef38e82e6e9..c61cc5adf713 100644 --- a/mobile/src/session/use-mobile-structured-agent-session-background-tasks.test.tsx +++ b/mobile/src/session/use-mobile-structured-agent-session-background-tasks.test.tsx @@ -77,7 +77,13 @@ describe('mobile structured session background tasks', () => { close: () => {} } - function Harness({ connected }: { connected: boolean }): null { + function Harness({ + connected, + targetedStop = false + }: { + connected: boolean + targetedStop?: boolean + }): null { hook = useMobileStructuredAgentSession({ client, sessionId: 'session-1', @@ -85,15 +91,15 @@ describe('mobile structured session background tasks', () => { enabled: true, connected, agent: 'claude', - hostSupport: CAPABLE, + hostSupport: { ...CAPABLE, targetedStop }, onSendError }) return null } - async function mount(event: AgentSessionSubscribeEvent): Promise { + async function mount(event: AgentSessionSubscribeEvent, targetedStop = false): Promise { act(() => { - renderer = create(createElement(Harness, { connected: true })) + renderer = create(createElement(Harness, { connected: true, targetedStop })) }) await vi.waitFor(() => expect(listener).toEqual(expect.any(Function))) act(() => listener?.(event)) @@ -181,4 +187,18 @@ describe('mobile structured session background tasks', () => { taskId: 'task-a' }) }) + + it('names the host child invocation when targeted Stop is supported', async () => { + await mount(withChildren(), true) + await act(async () => { + await hook?.backgroundTasks.stop('task-a') + }) + const cancel = sendRequest.mock.calls.find(([method]) => method === 'agentSession.cancel') + expect(fieldsOf(cancel?.[1])).toMatchObject({ + stopTarget: { + kind: 'background-tasks', + tasks: [{ id: 'a', invocation: { invocationId: 'spawn-a', generation: 1 } }] + } + }) + }) }) diff --git a/mobile/src/session/use-mobile-structured-agent-session-send.test.tsx b/mobile/src/session/use-mobile-structured-agent-session-send.test.tsx index 1af770c669c0..25400ba1276f 100644 --- a/mobile/src/session/use-mobile-structured-agent-session-send.test.tsx +++ b/mobile/src/session/use-mobile-structured-agent-session-send.test.tsx @@ -4,6 +4,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { AgentJournalDispatchState } from '../../../src/shared/agent-session-journal-types' import { DISPATCH_REJECTED_CANCELLED } from '../../../src/shared/structured-agent-session-dispatch-rejection' import type { AgentSessionSubscribeEvent } from '../../../src/shared/agent-session-wire' +import { SendParams } from '../../../src/shared/rpc-contract/structured-agent-session-params' import type { RpcClient } from '../transport/rpc-client' import { fieldsOf, ok } from './use-mobile-structured-agent-session-queued.test-fixture' import { markRpcDeliveryUnknown } from '../transport/rpc-delivery-ambiguity' @@ -197,6 +198,60 @@ describe('mobile structured send actions', () => { expect(calls()).toHaveLength(1) }) + it.each(['accepted', 'rejected'] as const)( + 'an expired reply and later %s history never automatically replace the Send', + async (dispatchState) => { + sendRequest.mockImplementation(async (method) => { + if (method !== 'agentSession.send') { + return ok({ models: [], current: {} }) + } + return calls().length === 1 + ? ok({ + ok: false, + refusal: { code: 'agent_session_operation_expired', message: 'Operation expired' } + }) + : sendResult('accepted') + }) + await mountSession() + await act(async () => { + expect(await hook!.sendWithOutcome('same text')).toBe('rejected') + }) + expect(calls()).toHaveLength(1) + const request = SendParams.parse(calls()[0]?.[1]) + const original = structuredSendResultFixture(dispatchState) + if (!('submission' in original)) { + throw new Error('expected a submission answer') + } + const event = snapshotEvent() + event.page.submissions = [ + { + ...original.submission, + clientMessageId: request.envelope.clientOperationId, + payloadFingerprint: request.envelope.payloadFingerprint + } + ] + event.page.items = [ + { + itemId: 'original-message', + revision: 1, + sequence: 1, + observedAt: 10, + body: request.body + } + ] + act(() => listener?.(event)) + expect(hook!.session.messages).toHaveLength(1) + expect(calls()).toHaveLength(1) + expect(new Set(sentIds()).size).toBe(1) + await act(async () => { + expect(await hook!.sendWithOutcome('same text')).toBe('accepted') + }) + expect(calls()).toHaveLength(2) + expect(sentIds()[1]).not.toBe(request.envelope.clientOperationId) + expect(asyncStorage.getItem).not.toHaveBeenCalled() + } + ) + it.each(['reply-before-row', 'row-before-commit'] as const)( 'states delivery once and leaves auth guidance in the transcript for %s', async (order) => { diff --git a/mobile/src/session/use-mobile-structured-agent-session.ts b/mobile/src/session/use-mobile-structured-agent-session.ts index fb3c69c51aa6..7877cc1714d9 100644 --- a/mobile/src/session/use-mobile-structured-agent-session.ts +++ b/mobile/src/session/use-mobile-structured-agent-session.ts @@ -159,6 +159,7 @@ export function useMobileStructuredAgentSession(args: { const thinking = isStructuredAgentSessionThinking(state) const isWorking = isStructuredAgentSessionMainAgentWorking(turnId, state.submissions, state.fence) const backgroundTasks = useMobileStructuredBackgroundTasks({ + targetedStopSupported: hostSupport?.targetedStop === true, sessionKey, state, turnId, @@ -234,6 +235,7 @@ export function useMobileStructuredAgentSession(args: { client, enabled, hostAnswersRepeatedStops, + targetedStopSupported: hostSupport?.targetedStop === true, inFlight: inFlightStopsRef.current, onSendError, prompt, @@ -245,6 +247,7 @@ export function useMobileStructuredAgentSession(args: { client, enabled, hostAnswersRepeatedStops, + hostSupport?.targetedStop, onSendError, promptCancelSupported, sessionId, diff --git a/mobile/src/session/use-mobile-structured-background-tasks.ts b/mobile/src/session/use-mobile-structured-background-tasks.ts index bab8597b30c5..c89cc65bc518 100644 --- a/mobile/src/session/use-mobile-structured-background-tasks.ts +++ b/mobile/src/session/use-mobile-structured-background-tasks.ts @@ -12,6 +12,7 @@ import { } from '../../../src/shared/structured-session-background-tasks-view' import type { StructuredAgentSessionMutationResult } from './mobile-structured-agent-session-rpc' import type { MobileStructuredAgentMutate } from './use-mobile-structured-agent-mutation' +import { agentSessionBackgroundStopTarget } from '../../../src/shared/agent-session-background-stop-target' export type MobileStructuredBackgroundTasks = { /** The conversation these tasks belong to; the strip's open and stopping state is kept per one. */ @@ -33,6 +34,7 @@ export function useMobileStructuredBackgroundTasks(args: { turnId: string | null /** Live transport; without it the kept roster is the last thing heard, not what runs now. */ connected: boolean + targetedStopSupported?: boolean mutate: MobileStructuredAgentMutate }): MobileStructuredBackgroundTasks { const { connected, mutate, sessionKey, state, turnId } = args @@ -58,9 +60,14 @@ export function useMobileStructuredBackgroundTasks(args: { mutate('agentSession.cancel', 'agentSession.cancel', { turnId: 'background-tasks', scope: 'background-tasks', - ...(taskId ? { taskId } : {}) + ...(taskId ? { taskId } : {}), + ...(args.targetedStopSupported + ? { + stopTarget: agentSessionBackgroundStopTarget(view.children, taskId) + } + : {}) }), - [mutate] + [args.targetedStopSupported, mutate, view.children] ) return useMemo( () => ({ sessionKey, view, rowContext, stop }), diff --git a/src/main/codex/codex-structured-dispatch-echo.test.ts b/src/main/codex/codex-structured-dispatch-echo.test.ts index 1eeb8186ad98..a810d19324aa 100644 --- a/src/main/codex/codex-structured-dispatch-echo.test.ts +++ b/src/main/codex/codex-structured-dispatch-echo.test.ts @@ -81,6 +81,39 @@ describe('codex dispatch echoes', () => { expect(echoes.settle('client-1')).toBe(false) }) + it('exposes exact admitted bindings before the echo, including coalesced sends', () => { + const echoes = createCodexDispatchEchoes() + echoes.arm('client-1') + echoes.arm('client-2') + expect(echoes.boundTurn('client-1')).toBeNull() + echoes.bindTurn('client-1', 'thread-1', 'turn-1', 'start') + echoes.bindTurn('client-2', 'thread-1', 'turn-1', 'steer') + expect(echoes.boundTurn('client-1')).toEqual({ + threadId: 'thread-1', + turnId: 'turn-1', + ended: false + }) + expect(echoes.boundTurn('client-2')).toEqual(echoes.boundTurn('client-1')) + echoes.settle('client-1') + expect(echoes.boundTurn('client-1')).toBeNull() + expect(echoes.boundTurn('client-2')?.turnId).toBe('turn-1') + }) + + it('observes the positively ended binding while a completed send still awaits its echo', () => { + const echoes = createCodexDispatchEchoes() + echoes.arm('client-1') + echoes.bindTurn('client-1', 'thread-1', 'turn-1', 'start') + echoes.endTurn('thread-1', 'turn-1', { status: 'completed' }) + expect(echoes.boundTurn('client-1')).toEqual({ + threadId: 'thread-1', + turnId: 'turn-1', + ended: true + }) + echoes.arm('client-2') + echoes.bindTurn('client-2', 'thread-1', 'turn-2', 'start') + expect(echoes.boundTurn('client-2')?.ended).toBe(false) + }) + it('refuses new correlations at capacity without dropping an older send', () => { const echoes = createCodexDispatchEchoes() for (let index = 0; index < MAX_CODEX_PENDING_DISPATCH_ECHOES; index += 1) { diff --git a/src/main/codex/codex-structured-dispatch-echo.ts b/src/main/codex/codex-structured-dispatch-echo.ts index b582ba8715b1..493db7c97f66 100644 --- a/src/main/codex/codex-structured-dispatch-echo.ts +++ b/src/main/codex/codex-structured-dispatch-echo.ts @@ -71,6 +71,10 @@ export type CodexDispatchEchoes = { ) => { clientMessageId: string; via: AgentJournalTurnJoin }[] /** Submission origin for this exact send, retained until its echo settles it. */ requestOrigin: (clientMessageId: string) => CodexDispatchRequestOrigin | null + /** The exact answered binding, including a turn that ended before its message echo. */ + boundTurn: ( + clientMessageId: string + ) => { threadId: string; turnId: string; ended: boolean } | null /** Highest causal sequence assigned to a dispatch in this session. */ latestSequence: () => number clear: () => void @@ -187,6 +191,16 @@ export function createCodexDispatchEchoes(): CodexDispatchEchoes { ? null : { requestedAt: origin.requestedAt, sequence: origin.sequence } }, + boundTurn: (clientMessageId) => { + const turn = armed.get(clientMessageId)?.turn + return turn + ? { + threadId: turn.threadId, + turnId: turn.turnId, + ended: endedTurns.has(turnKey(turn.threadId, turn.turnId)) + } + : null + }, latestSequence: () => nextSequence - 1, clear: () => { armed.clear() diff --git a/src/main/codex/codex-structured-session-adapter.ts b/src/main/codex/codex-structured-session-adapter.ts index 73c3ad14a09b..c6b204dc98ef 100644 --- a/src/main/codex/codex-structured-session-adapter.ts +++ b/src/main/codex/codex-structured-session-adapter.ts @@ -11,6 +11,7 @@ import type { } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { CodexJournalTranslationAdmission } from './codex-structured-journal-translation' import { dispatchCodexTurn, isCodexTurnOptionKey } from './codex-structured-turn-start' +import { codexSubmissionTurnObserver } from './codex-structured-submission-turn' import { agentSessionFailureFact, providerDiagnosticOf } from '../../shared/agent-session-failure' import { supportsCodexStructuredLocation } from './codex-structured-location-support' import { CodexStructuredSessionTeardown } from './codex-structured-session-teardown' @@ -237,6 +238,8 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap requestTimeoutMs: this.deps.requestTimeoutMs }) + observeSubmissionTurn = codexSubmissionTurnObserver(this.sessions) + rewindSupport: NonNullable = (sessionId) => this.sessions.get(sessionId)?.historyMode === 'legacy' ? { supported: false, reason: 'history-not-paginated' } diff --git a/src/main/codex/codex-structured-submission-turn.ts b/src/main/codex/codex-structured-submission-turn.ts new file mode 100644 index 000000000000..e94fe3434d29 --- /dev/null +++ b/src/main/codex/codex-structured-submission-turn.ts @@ -0,0 +1,31 @@ +import type { StructuredAgentSessionAdapter } from '../native-chat/agent-session-wire/structured-agent-session-adapter' +import type { CodexSession } from './codex-structured-session-state' + +type SubmissionTurnObserver = NonNullable + +/** Admission binds the request before its user echo; an absent binding proves no exit. */ +export function observeCodexSubmissionTurn( + session: CodexSession | undefined, + input: Parameters[0] +): ReturnType { + if (!session) { + return { verdict: 'unverifiable' } + } + if (session.ended || session.exitObservedAt !== undefined || session.fence !== input.fence) { + return { verdict: 'exited' } + } + const turn = session.dispatchEchoes.boundTurn(input.clientMessageId) + if (!turn || turn.threadId !== session.threadId) { + return { verdict: 'unverifiable' } + } + return turn.ended || session.abortedTurnIds?.has(turn.turnId) + ? { verdict: 'exited' } + : { verdict: 'live', turnId: turn.turnId } +} + +/** The adapter's observer over its own sessions. */ +export function codexSubmissionTurnObserver( + sessions: ReadonlyMap +): SubmissionTurnObserver { + return (input) => observeCodexSubmissionTurn(sessions.get(input.sessionId), input) +} diff --git a/src/main/native-chat/agent-session-journal/journal-stop-marks.ts b/src/main/native-chat/agent-session-journal/journal-stop-marks.ts index 18c4becacbe1..9642cbf38c17 100644 --- a/src/main/native-chat/agent-session-journal/journal-stop-marks.ts +++ b/src/main/native-chat/agent-session-journal/journal-stop-marks.ts @@ -53,7 +53,7 @@ export class JournalStopMarks { } /** Closes a settle `beginSettle` opened, binding `turnId` when the Stop stopped one, or marking - * `failedOn` (display only) when it failed to stop that turn. */ + * `failedOn` when it failed to stop that turn. */ settled(settle: JournalStopSettle | null, turnId?: string, failedOn?: JournalStopFailedOn): void { if (!settle) { return diff --git a/src/main/native-chat/agent-session-journal/queued-message-pause.ts b/src/main/native-chat/agent-session-journal/queued-message-pause.ts index b0289cbfe33c..f36bb8a0f254 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-pause.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-pause.ts @@ -25,7 +25,8 @@ export type QueuePauseReason = 'stopped' | 'cleared' | 'restarted' * reopen binds nothing: while it settles, every turn that ends; once settled, the turn it stopped. * `failedOn`: the turn a Stop that failed could not stop, which reads "Stopping…" until it ends: * the one running when it failed, or with none, the first that opens after the journal position - * it failed at. Display only: no turn-end rule reads it, so that turn's own end stays its own. */ + * it failed at. No turn-end rule reads it, so that turn's own end stays its own; it only keeps the + * Stop from counting as having taken effect (`structuredAgentSessionStopTookEffect`). */ export type JournalStopSettle = { settling: boolean turnId?: string diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts index 25e700960780..1e3b60dd7f14 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts @@ -102,6 +102,13 @@ export class StructuredAgentSessionAdapterRouter implements StructuredAgentSessi cancelTurn: StructuredAgentSessionAdapter['cancelTurn'] = (input) => this.owner(input.sessionId).cancelTurn(input) + observeSubmissionTurn: NonNullable = ( + input + ) => + this.liveOwnerOrNull(input.sessionId)?.observeSubmissionTurn?.(input) ?? { + verdict: 'unverifiable' + } + changeThreadGoal: NonNullable = (input) => { const change = this.owner(input.sessionId).changeThreadGoal if (!change) { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts index 958c71d80fcc..ca1b05525b8c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts @@ -305,6 +305,12 @@ export type StructuredAgentSessionAdapter = StructuredAgentSessionAdapterStop & /** Revalidate after preparation, immediately before writing to the provider. */ beforeDispatch?: () => Promise }): Promise + /** Execution-host binding before the user-message echo reaches the journal. */ + observeSubmissionTurn?(input: { + sessionId: string + clientMessageId: string + fence: number + }): { verdict: 'live'; turnId: string } | { verdict: 'exited' } | { verdict: 'unverifiable' } /** How this session narrows its agent's declared rewind; `agent` answers for one with no child * running. The router applies the declaration first, so an adapter's answer never widens it. */ rewindSupport?(sessionId: string, agent?: string): AgentSessionRewindSupport diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-background-target-lifetime.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-background-target-lifetime.test.ts new file mode 100644 index 000000000000..6dd0c0493df6 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-background-target-lifetime.test.ts @@ -0,0 +1,151 @@ +import { afterEach, expect, it, vi } from 'vitest' +import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view' +import { agentSessionBackgroundStopTarget } from '../../../shared/agent-session-background-stop-target' +import { + createQueuedMessageTestRig, + QUEUED_RIG_CALLER, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { + HOST_TEST_NOW as NOW, + HOST_TEST_SESSION as SESSION, + hostTestOperationId +} from './structured-agent-session-host-test-data' +import { + AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS, + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS +} from '../../../shared/agent-session-host-authority' + +const child: AgentChildWorkView = { + id: 'child-1', + providerId: 'task-1', + kind: 'agent', + state: 'working', + membership: 'live', + firstObservedAt: NOW, + observedAt: NOW, + stoppable: true, + invocation: { invocationId: 'invocation-1', generation: 1 } +} + +let rig: QueuedMessageTestRig | undefined +afterEach(async () => { + vi.restoreAllMocks() + await rig?.dispose() + rig = undefined +}) + +it.each([false, true])( + 'retries a lost task Stop reply without restarting a released provider: %s', + async (restart) => { + let now = NOW + rig = await createQueuedMessageTestRig({ now: () => now }) + let children = [child] + rig.host.deps.statusSink = { + publish: () => {}, + forget: () => {}, + readChildWork: () => children + } + await rig.workingSend() + const stopBackgroundTasks = vi.fn(async () => { + children = [{ ...child, state: 'done', membership: 'settled' }] + return { cancelled: true } + }) + rig.host.deps.adapter.stopBackgroundTasks = stopBackgroundTasks + const fields = { + scope: 'background-tasks' as const, + taskId: 'task-1', + stopTarget: agentSessionBackgroundStopTarget(children) + } + const id = hostTestOperationId() + const params = { ...fields, envelope: rig.envelope(fields, 'agentSession.cancel', id) } + expect(await rig.host.cancel(QUEUED_RIG_CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: true } + }) + expect(stopBackgroundTasks).toHaveBeenCalledWith( + expect.objectContaining({ taskIds: ['task-1'] }) + ) + await rig.host.close(SESSION, 'user-close') + now += AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS + 1 + if (restart) { + await rig.restartHostProcess() + } + const acquire = vi.spyOn(rig.host.deps.adapter, 'acquire') + expect(await rig.host.cancel(QUEUED_RIG_CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: false } + }) + expect(stopBackgroundTasks).toHaveBeenCalledOnce() + expect(acquire).not.toHaveBeenCalled() + expect(rig.store.getOperationRow(QUEUED_RIG_CALLER.callerKey, id)).toBeNull() + } +) + +it.each(['new-invocation', 'new-child'])( + 'cannot stop later work reusing a task handle: %s', + async (change) => { + rig = await createQueuedMessageTestRig() + const children = [ + { + ...child, + ...(change === 'new-child' + ? { id: 'child-2' } + : { + invocation: { ...child.invocation, generation: 2 } + }) + } + ] + rig.host.deps.statusSink = { + publish: () => {}, + forget: () => {}, + readChildWork: () => children + } + await rig.workingSend() + const stopBackgroundTasks = vi.fn(async () => ({ cancelled: true })) + rig.host.deps.adapter.stopBackgroundTasks = stopBackgroundTasks + const fields = { + scope: 'background-tasks' as const, + taskId: 'task-1', + stopTarget: agentSessionBackgroundStopTarget([child]) + } + expect( + await rig.host.cancel(QUEUED_RIG_CALLER, { + ...fields, + envelope: rig.envelope(fields, 'agentSession.cancel', hostTestOperationId()) + }) + ).toMatchObject({ ok: true, value: { cancelled: false } }) + expect(stopBackgroundTasks).not.toHaveBeenCalled() + } +) + +it.each([false, true])( + 'requires the execution writer for a live task (targeted: %s)', + async (targeted) => { + rig = await createQueuedMessageTestRig() + rig.host.deps.statusSink = { + publish: () => {}, + forget: () => {}, + readChildWork: () => [child] + } + await rig.workingSend() + const stopBackgroundTasks = vi.fn(async () => ({ cancelled: true })) + rig.host.deps.adapter.stopBackgroundTasks = stopBackgroundTasks + await rig.store.transitionHandoff(SESSION, (record) => ({ + ...record, + lease: { ...record.lease, claimStatus: 'released', ownerProcess: null } + })) + const fields = { + scope: 'background-tasks' as const, + taskId: 'task-1', + ...(targeted ? { stopTarget: agentSessionBackgroundStopTarget([child]) } : {}) + } + expect( + await rig.host.cancel(QUEUED_RIG_CALLER, { + ...fields, + envelope: rig.envelope(fields, 'agentSession.cancel', hostTestOperationId()) + }) + ).toMatchObject({ ok: false, refusal: { code: 'agent_session_ownership_unknown' } }) + expect(stopBackgroundTasks).not.toHaveBeenCalled() + } +) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-background-target-unverifiable.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-background-target-unverifiable.test.ts new file mode 100644 index 000000000000..a34a44c11650 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-background-target-unverifiable.test.ts @@ -0,0 +1,250 @@ +import { afterEach, expect, it, vi } from 'vitest' +import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view' +import { agentSessionBackgroundStopTarget } from '../../../shared/agent-session-background-stop-target' +import { + createQueuedMessageTestRig, + QUEUED_RIG_CALLER, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { + HOST_TEST_NOW as NOW, + HOST_TEST_SESSION as SESSION, + hostTestOperationId +} from './structured-agent-session-host-test-data' + +const child: AgentChildWorkView = { + id: 'child-1', + providerId: 'task-1', + kind: 'agent', + state: 'working', + membership: 'live', + firstObservedAt: NOW, + observedAt: NOW, + stoppable: true, + invocation: { invocationId: 'invocation-1', generation: 1 } +} +const fields = { + scope: 'background-tasks' as const, + taskId: 'task-1', + stopTarget: agentSessionBackgroundStopTarget([child]) +} + +function stopParams(testRig: QueuedMessageTestRig) { + return { + ...fields, + envelope: testRig.envelope(fields, 'agentSession.cancel', hostTestOperationId()) + } +} + +let rig: QueuedMessageTestRig | undefined +afterEach(async () => { + vi.restoreAllMocks() + await rig?.dispose() + rig = undefined +}) + +it('does not answer an unavailable execution-host child roster as a successfully ended task', async () => { + rig = await createQueuedMessageTestRig() + let readFails = false + rig.host.deps.statusSink = { + publish: () => {}, + forget: () => {}, + readChildWork: () => { + if (readFails) { + throw new Error('host child store temporarily unreadable') + } + return [child] + } + } + await rig.workingSend() + const stopBackgroundTasks = vi.fn(async () => ({ cancelled: true })) + rig.host.deps.adapter.stopBackgroundTasks = stopBackgroundTasks + const params = stopParams(rig) + const before = await rig.host.journalSnapshot(SESSION) + readFails = true + const result = await rig.host.cancel(QUEUED_RIG_CALLER, params) + expect(result).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_unknown', details: { reason: 'outcomeUnknown' } } + }) + expect(stopBackgroundTasks).not.toHaveBeenCalled() + expect(await rig.host.journalSnapshot(SESSION)).toEqual(before) + expect( + rig.store.getOperationRow(QUEUED_RIG_CALLER.callerKey, params.envelope.clientOperationId) + ).toBeNull() + readFails = false + expect(await rig.host.cancel(QUEUED_RIG_CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: true } + }) + expect(stopBackgroundTasks).toHaveBeenCalledOnce() + expect(stopBackgroundTasks).toHaveBeenCalledWith(expect.objectContaining({ taskIds: ['task-1'] })) + expect( + rig.store.getOperationRow(QUEUED_RIG_CALLER.callerKey, params.envelope.clientOperationId) + ).toBeNull() +}) + +it('does not bypass the execution writer if an unavailable child roster recovers during admission', async () => { + rig = await createQueuedMessageTestRig() + let failNextRead = false + rig.host.deps.statusSink = { + publish: () => {}, + forget: () => {}, + readChildWork: () => { + if (failNextRead) { + failNextRead = false + throw new Error('one transient host child read failure') + } + return [child] + } + } + await rig.workingSend() + const stopBackgroundTasks = vi.fn(async () => ({ cancelled: true })) + rig.host.deps.adapter.stopBackgroundTasks = stopBackgroundTasks + await rig.store.transitionHandoff(SESSION, (record) => ({ + ...record, + lease: { ...record.lease, claimStatus: 'released', ownerProcess: null } + })) + failNextRead = true + const result = await rig.host.cancel(QUEUED_RIG_CALLER, stopParams(rig)) + expect(stopBackgroundTasks).not.toHaveBeenCalled() + expect(result).toMatchObject({ ok: false, refusal: { code: 'agent_session_ownership_unknown' } }) +}) + +it.each(['live', 'new-invocation', 'ended'])( + 're-derives unavailable evidence after writer admission: %s', + async (recovered) => { + rig = await createQueuedMessageTestRig() + const readChildWork = vi.fn(() => [child]) + rig.host.deps.statusSink = { publish: () => {}, forget: () => {}, readChildWork } + await rig.workingSend() + const stopBackgroundTasks = vi.fn(async () => ({ cancelled: true })) + rig.host.deps.adapter.stopBackgroundTasks = stopBackgroundTasks + readChildWork.mockClear() + readChildWork + .mockImplementationOnce(() => { + throw new Error('one transient host child read failure') + }) + .mockReturnValue( + recovered === 'ended' + ? [] + : recovered === 'new-invocation' + ? [{ ...child, invocation: { ...child.invocation, generation: 2 } }] + : [child] + ) + const params = stopParams(rig) + expect(await rig.host.cancel(QUEUED_RIG_CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: recovered === 'live' } + }) + if (recovered === 'live') { + expect(stopBackgroundTasks).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ taskIds: ['task-1'] }) + ) + } else { + expect(stopBackgroundTasks).not.toHaveBeenCalled() + } + expect(readChildWork).toHaveBeenCalledTimes(2) + expect( + rig.store.getOperationRow(QUEUED_RIG_CALLER.callerKey, params.envelope.clientOperationId) + ).toBeNull() + } +) + +it('keeps a positively ended admission as a no-op when a later roster read would be live', async () => { + rig = await createQueuedMessageTestRig() + const readChildWork = vi.fn(() => [child]) + rig.host.deps.statusSink = { publish: () => {}, forget: () => {}, readChildWork } + await rig.workingSend() + const stopBackgroundTasks = vi.fn(async () => ({ cancelled: true })) + rig.host.deps.adapter.stopBackgroundTasks = stopBackgroundTasks + await rig.store.transitionHandoff(SESSION, (record) => ({ + ...record, + lease: { ...record.lease, claimStatus: 'released', ownerProcess: null } + })) + readChildWork.mockClear().mockReturnValueOnce([]) + expect(await rig.host.cancel(QUEUED_RIG_CALLER, stopParams(rig))).toMatchObject({ + ok: true, + value: { cancelled: false } + }) + expect(stopBackgroundTasks).not.toHaveBeenCalled() + expect(readChildWork).toHaveBeenCalledOnce() +}) + +it('uses the admitted live roster without another bookkeeping read before the provider Stop', async () => { + rig = await createQueuedMessageTestRig() + const readChildWork = vi.fn(() => [child]) + rig.host.deps.statusSink = { publish: () => {}, forget: () => {}, readChildWork } + await rig.workingSend() + const stopBackgroundTasks = vi.fn(async () => ({ cancelled: true })) + rig.host.deps.adapter.stopBackgroundTasks = stopBackgroundTasks + readChildWork + .mockClear() + .mockReturnValueOnce([child]) + .mockImplementation(() => { + throw new Error('host child store became unreadable') + }) + expect(await rig.host.cancel(QUEUED_RIG_CALLER, stopParams(rig))).toMatchObject({ + ok: true, + value: { cancelled: true } + }) + expect(stopBackgroundTasks).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ taskIds: ['task-1'] }) + ) + expect(readChildWork).toHaveBeenCalledOnce() +}) + +it('reports an unpublished parent roster as unconfirmed and allows the unchanged Stop after publication', async () => { + rig = await createQueuedMessageTestRig() + let publishFails = true + rig.host.deps.statusSink = { + publish: () => { + if (publishFails) { + throw new Error('parent publication unavailable') + } + }, + forget: () => {}, + readChildWork: () => [child] + } + await rig.workingSend() + const stopBackgroundTasks = vi.fn(async () => ({ cancelled: true })) + rig.host.deps.adapter.stopBackgroundTasks = stopBackgroundTasks + const params = stopParams(rig) + expect(await rig.host.cancel(QUEUED_RIG_CALLER, params)).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_unknown' } + }) + expect(stopBackgroundTasks).not.toHaveBeenCalled() + publishFails = false + const unsubscribe = rig.host.subscribeStatus({ id: 'republish', emit: () => {} }) + try { + expect(await rig.host.cancel(QUEUED_RIG_CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: true } + }) + expect(stopBackgroundTasks).toHaveBeenCalledOnce() + } finally { + unsubscribe() + } +}) + +it('reports a provider Stop write failure without silently claiming the task ended', async () => { + rig = await createQueuedMessageTestRig() + rig.host.deps.statusSink = { publish: () => {}, forget: () => {}, readChildWork: () => [child] } + await rig.workingSend() + const stopBackgroundTasks = vi.fn(async () => { + throw new Error('provider Stop failed') + }) + rig.host.deps.adapter.stopBackgroundTasks = stopBackgroundTasks + const params = stopParams(rig) + expect(await rig.host.cancel(QUEUED_RIG_CALLER, params)).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_unknown' } + }) + expect(stopBackgroundTasks).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ taskIds: ['task-1'] }) + ) + expect( + rig.store.getOperationRow(QUEUED_RIG_CALLER.callerKey, params.envelope.clientOperationId) + ).toBeNull() +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-cancel-plan.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-cancel-plan.ts new file mode 100644 index 000000000000..1e1395478e1f --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-cancel-plan.ts @@ -0,0 +1,103 @@ +import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view' +import type { AgentSessionStopTarget } from '../../../shared/agent-session-stop-target' +import { targetedAgentSessionBackgroundTaskIds } from '../../../shared/agent-session-background-stop-target' +import type { + AgentSessionCancelResult, + AgentSessionMutationEnvelope +} from '../../../shared/agent-session-wire' +import type { MutationPlan } from './structured-agent-session-mutation-plans' +import { performCancel } from './structured-agent-session-turns' +import { agentSessionOperationOutcomeUnknown } from './structured-agent-session-replay-outcome' + +type BackgroundStopObservation = + | { verdict: 'live'; children: readonly AgentChildWorkView[] } + | { verdict: 'exited' } + | { verdict: 'unverifiable' } + +export function cancelPlan(params: { + envelope: AgentSessionMutationEnvelope + turnId?: string + scope?: 'background-tasks' + taskId?: string + stopTarget?: AgentSessionStopTarget + prompt?: { itemId: string; expectedRevision: number } + /** The session's child records, which name the tasks a background Stop reaches. */ + childWork?: () => readonly AgentChildWorkView[] | undefined +}): MutationPlan { + const observeBackgroundTarget = (): BackgroundStopObservation | undefined => { + if (params.scope !== 'background-tasks' || params.stopTarget?.kind !== 'background-tasks') { + return undefined + } + const children = params.childWork?.() + if (children === undefined) { + return { verdict: 'unverifiable' } + } + return targetedAgentSessionBackgroundTaskIds(children, params.stopTarget, params.taskId) + .length > 0 + ? { verdict: 'live', children } + : { verdict: 'exited' } + } + // Admission and execution share the observation; an exempt no-op can never become a provider write. + let backgroundObservation: BackgroundStopObservation | undefined + const admissionObservation = () => (backgroundObservation ??= observeBackgroundTarget()) + return { + method: 'agentSession.cancel', + // An ended target needs no writer; a real background stop still checks execution ownership. + get conversationWrite() { + return (!params.scope && !params.prompt) || admissionObservation()?.verdict === 'exited' + ? true + : undefined + }, + // A Stop must reach the agent even when storage refuses the row recording it. + runsWithoutLedgerRow: true, + ...(params.stopTarget ? { receiptPolicy: 'none' as const } : {}), + fields: { + ...(params.turnId !== undefined ? { turnId: params.turnId } : {}), + ...(params.scope ? { scope: params.scope } : {}), + ...(params.taskId ? { taskId: params.taskId } : {}), + ...(params.stopTarget ? { stopTarget: params.stopTarget } : {}), + ...(params.prompt ? { prompt: params.prompt } : {}) + }, + run: async (ctx) => { + const admitted = admissionObservation() + // Unavailable evidence already paid the writer check; retry through the same host authority. + const observed = admitted?.verdict === 'unverifiable' ? observeBackgroundTarget() : admitted + if (observed?.verdict === 'unverifiable') { + return { + ok: false, + refusal: agentSessionOperationOutcomeUnknown(params.envelope.clientOperationId) + } + } + if (observed?.verdict === 'exited') { + return { + ok: true, + value: { + ...(params.turnId !== undefined ? { turnId: params.turnId } : {}), + cancelled: false + } + } + } + return performCancel(ctx, { + clientOperationId: params.envelope.clientOperationId, + ...(params.turnId !== undefined ? { turnId: params.turnId } : {}), + ...(params.scope ? { scope: params.scope } : {}), + ...(params.taskId ? { taskId: params.taskId } : {}), + ...(params.stopTarget?.kind === 'background-tasks' + ? { backgroundStopTarget: params.stopTarget } + : {}), + ...(params.prompt ? { prompt: params.prompt } : {}), + ...(observed?.verdict === 'live' + ? { childWork: () => observed.children } + : params.childWork + ? { childWork: params.childWork } + : {}) + }) + }, + // Interrupting twice would kill a turn the client never asked to stop, so a + // replay reports the turn as already handled. + replay: () => ({ + ...(params.turnId !== undefined ? { turnId: params.turnId } : {}), + cancelled: false + }) + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts index b6cd948fa4dc..310ce8cf5f0e 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts @@ -48,7 +48,7 @@ export function mutateWithChatStop( plan: MutationPlan, run: ( ctx: AgentSessionTurnContext, - stop: () => Promise + stop: (turnId?: string) => Promise ) => Promise> ): Promise> { const { envelope, turnId } = params @@ -59,13 +59,14 @@ export function mutateWithChatStop( | undefined // The Stop's event, still landing when its session ends: the next step holds the lane for it. let eventAfterEnd: Promise | undefined - const named = turnId !== undefined ? { turnId } : {} // Its own step wrote the Stop's event first. const stopChild = () => context.stopAgent(sessionId, { recorded: 'user-stop' }) // The same for every client: once the Stop takes effect its event is written, and the queue's // pause follows from it. The cards stay published; no text rides the answer. - const stop = (ctx: AgentSessionTurnContext): Promise => - runRecordedStop( + const stop = (ctx: AgentSessionTurnContext, stoppedTurnId = turnId): Promise => { + const turnId = stoppedTurnId + const named = turnId !== undefined ? { turnId } : {} + return runRecordedStop( ctx, { reason: 'user-stop', @@ -170,6 +171,7 @@ export function mutateWithChatStop( } } ) + } const result = mutateStructuredAgentSession( context, caller, @@ -177,7 +179,10 @@ export function mutateWithChatStop( { ...plan, run: (ctx) => - run(ctx, async () => ({ outcome: await stop(ctx), endsSession: windDown !== undefined })) + run(ctx, async (turnId) => ({ + outcome: await stop(ctx, turnId), + endsSession: windDown !== undefined + })) }, openForWrite(context, envelope) ) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-close-storage.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-close-storage.test.ts new file mode 100644 index 000000000000..121009ec11f2 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-close-storage.test.ts @@ -0,0 +1,144 @@ +import { afterEach, expect, it, vi } from 'vitest' +import { + call, + clearStructuredHostStub, + hostCalls, + installStructuredHostStub, + SESSION, + STRUCTURED_CLIENT +} from '../../runtime/rpc/methods/structured-agent-session-rpc.test-fixture' +import * as unsent from '../agent-session-journal/journal-unsent-send-hold' +import { setStructuredAgentSessionHost } from './structured-agent-session-registry' +import { HOST_TEST_SESSION } from './structured-agent-session-host-test-data' +import { + createQueuedMessageTestRig, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' + +let rig: QueuedMessageTestRig | undefined +afterEach(async () => { + clearStructuredHostStub() + vi.restoreAllMocks() + await rig?.dispose() + rig = undefined +}) + +it('Close stops the host and reports a failed tab-hide write', async () => { + installStructuredHostStub() + hostCalls.setSessionTabVisibility.mockRejectedValueOnce(new Error('visibility write failed')) + expect(await call('agentSession.close', { sessionId: SESSION }, STRUCTURED_CLIENT)).toMatchObject( + { + ok: false, + error: { message: 'visibility write failed' } + } + ) + expect(hostCalls.close).toHaveBeenCalledExactlyOnceWith(SESSION, 'user-close') +}) + +it('Close reaches the host while the tab-hide write is still pending', async () => { + installStructuredHostStub() + let finishHide = (): void => {} + hostCalls.setSessionTabVisibility.mockImplementationOnce( + () => new Promise((resolve) => (finishHide = resolve)) + ) + const response = call('agentSession.close', { sessionId: SESSION }, STRUCTURED_CLIENT) + try { + await vi.waitFor(() => expect(hostCalls.setSessionTabVisibility).toHaveBeenCalledOnce()) + expect(hostCalls.close).toHaveBeenCalledExactlyOnceWith(SESSION, 'user-close') + } finally { + finishHide() + await response + } + expect(await response).toMatchObject({ ok: true, result: { ok: true } }) +}) + +it('a failed child close still attempts to hide the tab and reports the close failure', async () => { + installStructuredHostStub() + hostCalls.close.mockRejectedValueOnce(new Error('child close failed')) + expect(await call('agentSession.close', { sessionId: SESSION }, STRUCTURED_CLIENT)).toMatchObject( + { + ok: false, + error: { message: 'child close failed' } + } + ) + expect(hostCalls.setSessionTabVisibility).toHaveBeenCalledExactlyOnceWith(SESSION, false) +}) + +it('RPC Close ends a live provider when hiding its tab cannot persist', async () => { + rig = await createQueuedMessageTestRig() + await rig.workingSend() + setStructuredAgentSessionHost(rig.host) + vi.spyOn(rig.store, 'setSessionTabVisibility').mockRejectedValueOnce( + new Error('visibility write failed') + ) + expect( + await call('agentSession.close', { sessionId: HOST_TEST_SESSION }, STRUCTURED_CLIENT) + ).toMatchObject({ ok: false, error: { message: 'visibility write failed' } }) + expect(rig.closeSession).toHaveBeenCalledOnce() + expect(rig.store.getRecord(HOST_TEST_SESSION)?.lease.ownerProcess).toBeNull() +}) + +it('hold-sends and reopen-marker failures still allow provider shutdown', async () => { + rig = await createQueuedMessageTestRig() + await rig.workingSend() + await rig.send('queued', 'queue-if-active').result + const session = rig.host.collaboratorsForTests().sessions.get(HOST_TEST_SESSION) + if (!session) { + throw new Error('expected attached conversation') + } + const hold = vi + .spyOn(unsent, 'holdUnsentSends') + .mockRejectedValueOnce(new Error('hold write failed')) + const mark = vi + .spyOn(session.journal, 'markQueueReopen') + .mockRejectedValueOnce(new Error('marker write failed')) + await rig.host.close(HOST_TEST_SESSION, 'user-close') + expect(hold).toHaveBeenCalled() + expect(mark).toHaveBeenCalled() + expect(rig.closeSession).toHaveBeenCalledOnce() +}) + +it.each(['settling', 'reopen-marking'] as const)( + 'Close ends the provider while %s its waiting message is still pending', + async (step) => { + // A start that never answers holds the message queued; a running turn queues it behind. + rig = await createQueuedMessageTestRig( + step === 'settling' ? { starting: true, startUnanswered: true } : {} + ) + if (step === 'reopen-marking') { + await rig.workingSend() + } + const waiting = rig.send('waiting at Close', 'queue-if-active') + expect(await waiting.result).toMatchObject({ ok: true }) + setStructuredAgentSessionHost(rig.host) + const journal = rig.host.collaboratorsForTests().sessions.get(HOST_TEST_SESSION)?.journal + if (!journal) { + throw new Error('expected attached conversation') + } + let release = (): void => {} + const held = new Promise((resolve) => (release = resolve)) + const settle = journal.resolveDispatch.bind(journal) + const mark = journal.appendQueueReopen.bind(journal) + const write = + step === 'settling' + ? vi + .spyOn(journal, 'resolveDispatch') + .mockImplementationOnce((...args) => held.then(() => settle(...args))) + : vi + .spyOn(journal, 'appendQueueReopen') + .mockImplementationOnce((...args) => held.then(() => mark(...args))) + const dispatched = rig.dispatch.mock.calls.length + const closing = call('agentSession.close', { sessionId: HOST_TEST_SESSION }, STRUCTURED_CLIENT) + try { + await vi.waitFor(() => expect(write).toHaveBeenCalled()) + expect(rig.closeSession).toHaveBeenCalledOnce() + } finally { + release() + } + expect(await closing).toMatchObject({ ok: true }) + expect(rig.closeSession).toHaveBeenCalledOnce() + // Nothing handed over after Close: the message stays a card for the chat's next turn. + expect(rig.dispatch).toHaveBeenCalledTimes(dispatched) + expect(await rig.drafts()).toEqual([{ messageId: waiting.id, state: 'waiting' }]) + } +) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-codex-submission-stop.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-codex-submission-stop.test.ts new file mode 100644 index 000000000000..dccdd7b5a6d8 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-codex-submission-stop.test.ts @@ -0,0 +1,174 @@ +import { afterEach, expect, it, vi } from 'vitest' +import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' +import { agentSessionStopTarget } from '../../../shared/agent-session-stop-target' +import { + adapterFor, + fakeCodex, + THREAD_ID as THREAD +} from '../../codex/codex-structured-session-adapter-fixture' +import { codexTurnLifecycleFake } from '../../codex/codex-turn-lifecycle-fake' +import { + createRestTestRig, + REST_TEST_CALLER as CALLER, + restTestSend, + type RestTestRig +} from './structured-agent-session-rest-test-rig' +import { + HOST_TEST_SESSION as SESSION, + hostTestAttachParams, + hostTestOperationId +} from './structured-agent-session-host-test-data' +import type { CodexStructuredSessionAdapter } from '../../codex/codex-structured-session-adapter' + +let rig: RestTestRig | undefined +let adapter: CodexStructuredSessionAdapter | undefined +afterEach(async () => { + await adapter?.closeAll() + await rig?.dispose() + rig = undefined + adapter = undefined +}) + +async function submittedTurn() { + const codex = fakeCodex() + const notify = (method: string, params: unknown) => + codex.connections.at(-1)?.handlers.onNotification?.(method, params) + const turns = codexTurnLifecycleFake(THREAD, () => notify) + Object.assign(codex.routes, turns.routes) + const provider = adapterFor(codex, {}, [], { + onDispatchSettledLate: (settlement) => void rig?.host.settleLateDispatch(settlement), + onEvent: (event) => { + if (event.type === 'ended' && 'cause' in event) { + void rig?.host.handleAdapterEvent(event) + } + } + }) + adapter = provider + const current = await createRestTestRig({ adapter: provider }) + rig = current + expect(await current.host.attach(CALLER, hostTestAttachParams(null))).toMatchObject({ ok: true }) + const send = restTestSend('work on this') + expect(await current.host.send(CALLER, send)).toMatchObject({ ok: true }) + const id = send.envelope.clientOperationId + await vi.waitFor(() => + expect( + provider.observeSubmissionTurn({ sessionId: SESSION, clientMessageId: id, fence: 1 }) + ).toEqual({ verdict: 'live', turnId: 'turn-1' }) + ) + const snapshot = await current.host.journalSnapshot(SESSION) + const stopTarget = agentSessionStopTarget(null, snapshot.submissions, 1) + expect(stopTarget).toEqual({ kind: 'submission', clientMessageId: id }) + const envelope = { + sessionId: SESSION, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: 1, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.cancel', + sessionId: SESSION, + fields: { stopTarget } + }) + } + const interrupts = () => + codex.connections[0]!.calls.filter((call) => call.method === 'turn/interrupt') + return { current, provider, turns, notify, id, params: { stopTarget, envelope }, interrupts } +} + +it('interrupts the bound turn when it opens after the client captures a submission Stop', async () => { + const { current, turns, id, params, interrupts } = await submittedTurn() + turns.start() + await current.host.flushStreamedEvents(SESSION) + expect( + (await current.host.journalSnapshot(SESSION)).submissions.find( + (row) => row.clientMessageId === id + ) + ).toMatchObject({ providerItemId: null, dispatchState: 'pending' }) + expect(await current.host.cancel(CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: true } + }) + expect(interrupts()).toMatchObject([{ params: { threadId: THREAD, turnId: 'turn-1' } }]) + expect( + current.store.getOperationRow(CALLER.callerKey, params.envelope.clientOperationId) + ).toBeNull() +}) + +it('interrupts the admitted turn before its started notification or user echo', async () => { + const { current, turns, params, interrupts } = await submittedTurn() + turns.run() + expect(await current.host.cancel(CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: true } + }) + expect(interrupts()).toMatchObject([{ params: { turnId: 'turn-1' } }]) +}) + +it('reports an unconfirmed Stop immediately when the admitted turn cannot yet take an interrupt', async () => { + const { current, turns, params, interrupts } = await submittedTurn() + expect(await current.host.cancel(CALLER, params)).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_unknown' } + }) + expect(interrupts()).toHaveLength(1) + expect(turns.turnId).toBe('turn-1') + expect(current.host.collaboratorsForTests().sessions.get(SESSION)?.child).toBeDefined() +}) + +it('reports unavailable binding without interrupting a turn it cannot identify', async () => { + const { current, provider, turns, id, params, interrupts } = await submittedTurn() + turns.start() + await current.host.flushStreamedEvents(SESSION) + vi.spyOn(provider, 'observeSubmissionTurn').mockReturnValue({ verdict: 'unverifiable' }) + expect(await current.host.cancel(CALLER, params)).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_unknown' } + }) + expect(interrupts()).toHaveLength(0) + expect( + (await current.host.journalSnapshot(SESSION)).submissions.find( + (row) => row.clientMessageId === id + )?.providerItemId + ).toBeNull() +}) + +it('does not interrupt later work after the bound turn ends without a user echo', async () => { + const { current, turns, params, interrupts } = await submittedTurn() + turns.start() + turns.end('completed') + await current.host.flushStreamedEvents(SESSION) + expect(await current.host.send(CALLER, restTestSend('later work'))).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(turns.turnId).toBe('turn-2')) + turns.start() + await current.host.flushStreamedEvents(SESSION) + expect(await current.host.cancel(CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: false } + }) + expect(interrupts()).toHaveLength(0) + expect(turns.turnId).toBe('turn-2') +}) + +it('does not interrupt a later journal turn while an older submission binding remains', async () => { + const { current, notify, params, interrupts } = await submittedTurn() + notify('turn/started', { threadId: THREAD, turn: { id: 'unrelated' } }) + await current.host.flushStreamedEvents(SESSION) + expect(await current.host.cancel(CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: false } + }) + expect(interrupts()).toHaveLength(0) +}) + +it('keeps a positively interrupted target quiet when its Stop is retried', async () => { + const { current, turns, params, interrupts } = await submittedTurn() + turns.start() + await current.host.flushStreamedEvents(SESSION) + expect(await current.host.cancel(CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: true } + }) + expect(await current.host.cancel(CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: false } + }) + expect(interrupts()).toHaveLength(1) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime.ts index 431ca9e761de..f2b1f7025f16 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime.ts @@ -178,15 +178,23 @@ export function createStructuredAgentSessionConversationLifetime(host: { host.context().runtimeState.acquireAborts.abort(sessionId, 'closed while starting') return serialize(sessionId, async () => { readRefusals.forget(sessionId) - const session = sessions.get(sessionId) - if (session) { - // Settled and marked before the stop, so no start delivers it, and nothing it closed with - // sends by itself, even if the handle stays open. - await holdClosedStructuredAgentSessionSends(deps(), sessionId, session.journal, { - mark: 'always' + // The agent stops first, so no settlement write can keep it running. Nothing queued is + // handed over meanwhile: every send and start waits behind this step, and what a person's + // close failed to settle is settled before the next start (`closeWhatTheUserClosed`). + try { + await stopStructuredAgentSessionAgentUnderSerialize(host.context(), sessionId, { + cause, + closing: true }) + } finally { + // After the child's end, so the reopen mark also holds the cards its end kept. + const session = sessions.get(sessionId) + if (session) { + await holdClosedStructuredAgentSessionSends(deps(), sessionId, session.journal, { + mark: 'always' + }) + } } - await stopStructuredAgentSessionAgentUnderSerialize(host.context(), sessionId, { cause }) await closeConversation(sessionId) }) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-failed-stop-repeat.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-failed-stop-repeat.test.ts new file mode 100644 index 000000000000..c1113e41f1a7 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-failed-stop-repeat.test.ts @@ -0,0 +1,53 @@ +import { afterEach, expect, it, vi } from 'vitest' +import { + createQueuedMessageTestRig, + QUEUED_RIG_CALLER, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { + HOST_TEST_SESSION as SESSION, + hostTestOperationId +} from './structured-agent-session-host-test-data' + +let rig: QueuedMessageTestRig | undefined +afterEach(async () => { + vi.restoreAllMocks() + await rig?.dispose() + rig = undefined +}) + +function stopParams(r: QueuedMessageTestRig, clientMessageId?: string) { + const fields = + clientMessageId === undefined + ? {} + : { stopTarget: { kind: 'submission' as const, clientMessageId } } + return { ...fields, envelope: r.envelope(fields, 'agentSession.cancel', hostTestOperationId()) } +} + +for (const repeat of ['targeted', 'unnamed'] as const) { + it(`a ${repeat} Stop after a Stop that failed still tries to stop the waiting send`, async () => { + rig = await createQueuedMessageTestRig({ restartable: true }) + const first = await rig.workingSend() + // The interrupt fails and the child's exit cannot be proven: the send may run on. + rig.cancelTurn.mockImplementationOnce(async () => { + throw new Error('interrupt timed out') + }) + rig.closeSession.mockImplementationOnce(async () => { + throw new Error('exit not proven') + }) + expect(await rig.host.cancel(QUEUED_RIG_CALLER, stopParams(rig, first))).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_unknown' } + }) + expect(rig.host.collaboratorsForTests().sessions.get(SESSION)?.child).toBeTruthy() + expect(await rig.submission(first)).toMatchObject({ dispatchState: 'pending' }) + const tries = rig.cancelTurn.mock.calls.length + rig.closeSession.mock.calls.length + await rig.host.cancel( + QUEUED_RIG_CALLER, + stopParams(rig, repeat === 'targeted' ? first : undefined) + ) + expect(rig.cancelTurn.mock.calls.length + rig.closeSession.mock.calls.length).toBeGreaterThan( + tries + ) + }) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts index cb9cd10fea68..aae6e7a09a08 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts @@ -31,17 +31,23 @@ import { structuredAgentSessionSendBlock } from './structured-agent-session-send-preparation' import { - cancelPlan, promptPlan, sendPlan, setOptionPlan, type MutationPlan } from './structured-agent-session-mutation-plans' +import { cancelPlan } from './structured-agent-session-cancel-plan' import { agentSessionMutationAdmitsNow } from './structured-agent-session-mutation-admits-now' import { runQueueableStructuredAgentSessionSend } from './structured-agent-session-queued-send' import { cancelStructuredAgentSessionPrompt } from './structured-agent-session-prompt-cancel' -import { mutateWithChatStop } from './structured-agent-session-chat-stop' +import { + mutateWithChatStop, + type StructuredAgentSessionChatStopRun +} from './structured-agent-session-chat-stop' import { performSetOption } from './structured-agent-session-turns-options' +import type { AgentSessionStopTarget } from '../../../shared/agent-session-stop-target' +import { observeStructuredAgentSessionStopTarget } from './structured-agent-session-stop-target-observation' +import { agentSessionOperationOutcomeUnknown } from './structured-agent-session-replay-outcome' export type { StructuredAgentSessionMutationContext } from './structured-agent-session-mutation-context' import type { StructuredAgentSessionCaller } from './structured-agent-session-host-types' import { @@ -99,6 +105,7 @@ export function cancelStructuredAgentSessionTurn( turnId?: string scope?: 'background-tasks' taskId?: string + stopTarget?: AgentSessionStopTarget prompt?: { itemId: string; expectedRevision: number } } ): Promise> { @@ -114,20 +121,62 @@ export function cancelStructuredAgentSessionTurn( const plan = cancelPlan(params) const { prompt } = params if (!prompt && params.turnId === undefined) { - abortAcquireForStop(context, caller, params.envelope, plan) + const journal = context.sessions.get(params.envelope.sessionId)?.journal + if ( + !params.stopTarget || + (journal && + observeStructuredAgentSessionStopTarget( + { + journal, + sessionId: params.envelope.sessionId, + adapter: context.deps.adapter, + fence: context.deps.store.getRecord(params.envelope.sessionId)?.lease.runtimeFence ?? 0 + }, + params.stopTarget + ).verdict === 'live') + ) { + abortAcquireForStop(context, caller, params.envelope, plan) + } } // A card's Cancel stops whatever the chat has in flight, as the Stop button does; it reaches the // Stop only for a card the live turn raised (`cancelStructuredAgentSessionPrompt`). const stopped = prompt ? { envelope: params.envelope } : params - return mutateWithChatStop(context, caller, stopped, plan, (ctx, stop) => - prompt - ? cancelStructuredAgentSessionPrompt( - ctx, - { ...(params.turnId !== undefined ? { turnId: params.turnId } : {}), prompt }, - { stop, interrupt: () => plan.run(ctx) } - ) - : stop().then(({ outcome }) => outcome) - ) + return mutateWithChatStop(context, caller, stopped, plan, async (ctx, stop) => { + const target = params.stopTarget + const unconfirmed = () => ({ + ok: false as const, + refusal: agentSessionOperationOutcomeUnknown(params.envelope.clientOperationId) + }) + // The target bounds only the Stop: a card names itself, so its own route still settles it. + const stopTarget = async (): Promise => { + const observed = target ? observeStructuredAgentSessionStopTarget(ctx, target) : undefined + if (observed?.verdict === 'unverifiable') { + return { outcome: unconfirmed(), endsSession: false } + } + if (observed?.verdict === 'exited') { + return { outcome: { ok: true, value: { cancelled: false } }, endsSession: false } + } + const run = await stop(observed?.turnId) + const { outcome } = run + if ( + target?.kind === 'submission' && + outcome.ok && + !outcome.value.cancelled && + observeStructuredAgentSessionStopTarget(ctx, target).verdict !== 'exited' + ) { + return { ...run, outcome: unconfirmed() } + } + return run + } + if (prompt) { + return cancelStructuredAgentSessionPrompt( + ctx, + { ...(params.turnId !== undefined ? { turnId: params.turnId } : {}), prompt }, + { stop: stopTarget, interrupt: () => plan.run(ctx) } + ) + } + return (await stopTarget()).outcome + }) } /** A Stop must not wait behind a start the provider may never answer: the start the session's diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-stop-event.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-stop-event.ts index d9b8e068c37e..9f3fd3664879 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-stop-event.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-stop-event.ts @@ -2,6 +2,7 @@ // record, and the write itself, issued before the kill. import { isStructuredAgentSessionMainAgentWorking } from '../../../shared/structured-agent-session-main-agent-working' +import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission' import { withTimeout } from '../../../shared/promise-timeout-fallback' import type { StructuredAgentSessionStopCause } from './structured-agent-session-adapter' import type { StructuredAgentSessionLifetimeContext } from './structured-agent-session-host-lifetime' @@ -24,6 +25,8 @@ export type StructuredAgentSessionStopEnding = /** The idle sweep judged the agent resting (`owesWork`): a send it retires unanswered is * no work its event records. */ resting?: true + /** The chat's close settles what is queued once its agent stopped: none of it is work. */ + closing?: true } /** How long a host stop waits for the session's sink before it judges whether the stop ends work. */ @@ -62,13 +65,12 @@ export async function stopEndsWork( STOP_EVENT_DRAIN_TIMEOUT_MS, 'slow' as const ) + const submissions = ending.closing + ? journal.submissions().filter((entry) => !isQueuedAgentJournalSubmission(entry)) + : journal.submissions() const working = (drain === 'slow' && journal.stopMarks.latestAcceptedSendUnopened()) || - isStructuredAgentSessionMainAgentWorking( - journal.activeTurnId(), - journal.submissions(), - child.fence - ) + isStructuredAgentSessionMainAgentWorking(journal.activeTurnId(), submissions, child.fence) return working && (ending.cause === 'user-close' || !defersToPersonsStop(session)) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admission.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admission.ts index aec938b85fe6..a0a551925515 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admission.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admission.ts @@ -107,6 +107,7 @@ export async function admitAndRunAgentSessionMutation( envelope, hostFingerprint, now: request.now(), + ...(plan.receiptPolicy ? { receiptPolicy: plan.receiptPolicy } : {}), ...(plan.operationIdScope ? { operationIdScope: plan.operationIdScope } : {}) }) if (!ledger) { @@ -143,13 +144,17 @@ export async function admitAndRunAgentSessionMutation( envelope, hostFingerprint, now: request.now(), + ...(plan.receiptPolicy ? { receiptPolicy: plan.receiptPolicy } : {}), ...(plan.operationIdScope ? { operationIdScope: plan.operationIdScope } : {}), ...(plan.conversationWrite ? { conversationWrite: true } : {}) } let admitted: AgentSessionMutationOperationDecision let ledgerRowWritten = true try { - admitted = await request.store.admitMutationOperation(operation) + admitted = + plan.receiptPolicy === 'none' + ? admitWithoutLedgerRow(request, operation) + : await request.store.admitMutationOperation(operation) } catch (error) { if (plan.runsWithoutLedgerRow) { admitted = admitWithoutLedgerRow(request, operation, error) @@ -168,6 +173,7 @@ export async function admitAndRunAgentSessionMutation( return refuseAgentSessionMutation(AGENT_SESSION_NOT_ATTACHED) } const { admission, record } = admitted + ledgerRowWritten = ledgerRowWritten && admitted.receiptRecorded !== false if (admission.decision === 'refused') { return refuseAgentSessionMutation(admission.refusal) } @@ -295,13 +301,15 @@ function replayRecordedOperation( function admitWithoutLedgerRow( { store, logger }: Pick, 'store' | 'logger'>, operation: AgentSessionMutationOperationAdmission, - error: unknown + error?: unknown ): AgentSessionMutationOperationDecision { - logger.warn("writing Stop's ledger row failed; Stop runs without it", { - scope: 'stop-ledger-row', - sessionId: operation.envelope.sessionId, - error - }) + if (error !== undefined) { + logger.warn("writing Stop's ledger row failed; Stop runs without it", { + scope: 'stop-ledger-row', + sessionId: operation.envelope.sessionId, + error + }) + } const evaluated = store.evaluateMutationOperation(operation) if (!evaluated) { return null @@ -313,5 +321,5 @@ function admitWithoutLedgerRow( lease: evaluated.record.lease, ...(operation.conversationWrite ? { conversationWrite: true } : {}) }) - return { admission, record: evaluated.record } + return { admission, record: evaluated.record, receiptRecorded: false } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admits-now.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admits-now.ts index cc1e2bd5076c..36b62918c4d4 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admits-now.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admits-now.ts @@ -9,7 +9,10 @@ import type { MutationPlan } from './structured-agent-session-mutation-plans' * must not wait in the session's queue for its turn. */ export function agentSessionMutationAdmitsNow( request: Pick, 'store' | 'callerKey' | 'envelope' | 'now'> & { - plan: Pick, 'method' | 'fields' | 'operationIdScope' | 'conversationWrite'> + plan: Pick< + MutationPlan, + 'method' | 'fields' | 'operationIdScope' | 'conversationWrite' | 'receiptPolicy' + > } ): boolean { const { plan, envelope } = request @@ -23,6 +26,7 @@ export function agentSessionMutationAdmitsNow( envelope, hostFingerprint, now: request.now(), + ...(plan.receiptPolicy ? { receiptPolicy: plan.receiptPolicy } : {}), ...(plan.operationIdScope ? { operationIdScope: plan.operationIdScope } : {}) }) return ( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts index 6d4520f9c9d8..325d2ba49cba 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts @@ -10,10 +10,8 @@ import { USER_MESSAGE_SOURCE, type AgentSessionMessageSource } from '../../../shared/agent-session-message-source' -import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view' import type { AgentSessionOperationOutcome } from '../../../shared/agent-session-operation-ledger' import type { - AgentSessionCancelResult, AgentSessionMutationEnvelope, AgentSessionOptionResult, AgentSessionPromptResult, @@ -31,7 +29,6 @@ import { structuredAgentSessionCompactBody } from './structured-agent-session-command-turn' import { - performCancel, performPrompt, performSend, performSetOption, @@ -50,6 +47,7 @@ export type MutationPlan = { conversationWrite?: true /** Still runs, decided from the committed ledger, when its ledger row cannot be written. */ runsWithoutLedgerRow?: true + receiptPolicy?: 'none' run: (ctx: AgentSessionTurnContext) => Promise> replay: (ctx: AgentSessionTurnContext, outcome: AgentSessionOperationOutcome) => TValue | null rerunWhenReplayMissing?: (ctx: AgentSessionTurnContext) => boolean @@ -230,45 +228,6 @@ export function conversationCommandPlan(params: { } } -export function cancelPlan(params: { - envelope: AgentSessionMutationEnvelope - turnId?: string - scope?: 'background-tasks' - taskId?: string - prompt?: { itemId: string; expectedRevision: number } - /** The session's child records, which name the tasks a background Stop reaches. */ - childWork?: () => readonly AgentChildWorkView[] | undefined -}): MutationPlan { - return { - method: 'agentSession.cancel', - // Stop is a conversation write; a prompt or background-task cancel needs the live child. - ...(params.scope || params.prompt ? {} : { conversationWrite: true as const }), - // A Stop must reach the agent even when storage refuses the row recording it. - runsWithoutLedgerRow: true, - fields: { - ...(params.turnId !== undefined ? { turnId: params.turnId } : {}), - ...(params.scope ? { scope: params.scope } : {}), - ...(params.taskId ? { taskId: params.taskId } : {}), - ...(params.prompt ? { prompt: params.prompt } : {}) - }, - run: (ctx) => - performCancel(ctx, { - clientOperationId: params.envelope.clientOperationId, - ...(params.turnId !== undefined ? { turnId: params.turnId } : {}), - ...(params.scope ? { scope: params.scope } : {}), - ...(params.taskId ? { taskId: params.taskId } : {}), - ...(params.prompt ? { prompt: params.prompt } : {}), - ...(params.childWork ? { childWork: params.childWork } : {}) - }), - // Interrupting twice would kill a turn the client never asked to stop, so a - // replay reports the turn as already handled. - replay: () => ({ - ...(params.turnId !== undefined ? { turnId: params.turnId } : {}), - cancelled: false - }) - } -} - export function promptPlan( params: AgentSessionPromptRequest ): MutationPlan { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-cancel-ended-target.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-cancel-ended-target.test.ts new file mode 100644 index 000000000000..e65db6b504ab --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-cancel-ended-target.test.ts @@ -0,0 +1,45 @@ +import { beforeEach, expect, it, vi, type Mock } from 'vitest' +import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' +import type { AgentSessionPromptCancelRoute } from './structured-agent-session-adapter-stop' +import type { StructuredAgentSessionHost } from './structured-agent-session-host' +import { + attach, + CALLER, + envelope, + hostTestState, + seedApproval +} from './structured-agent-session-host-test-harness' +import { HOST_TEST_SESSION as SESSION } from './structured-agent-session-host-test-data' + +let host: StructuredAgentSessionHost +let cancelTurn: Mock + +beforeEach(() => { + ;({ host, cancelTurn } = hostTestState()) +}) + +it.each(['stop', 'dismiss'])( + "a card's Cancel still settles the card when the turn its Stop named already ended (%s route)", + async (kind) => { + await attach() + const prompt = await seedApproval() + const dismissPrompt = vi.fn>( + async ({ commit }) => commit() + ) + Object.assign(host.deps.adapter, { routePromptCancel: () => ({ kind }), dismissPrompt }) + // Pressed while turn-2 ran; it ended before the request was processed. + const fields = { + turnId: 'turn-2', + stopTarget: { kind: 'turn' as const, turnId: 'turn-2' }, + prompt: { itemId: prompt.itemId, expectedRevision: prompt.revision } + } + expect( + await host.cancel(CALLER, { envelope: envelope('agentSession.cancel', fields), ...fields }) + ).toMatchObject({ ok: true, value: { cancelled: true } }) + expect(dismissPrompt).toHaveBeenCalledWith(expect.objectContaining({ answer: true })) + expect(cancelTurn).not.toHaveBeenCalled() + const page = await host.history({ sessionId: SESSION, direction: 'tail' }) + const card = page.ok ? page.page.items.find((item) => item.itemId === prompt.itemId) : null + expect(card?.body).toMatchObject({ resolution: { state: 'cancelled' } }) + } +) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-message-rig.test-fixture.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-message-rig.test-fixture.ts index ef7e36953bd3..d5f9f0064c79 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-message-rig.test-fixture.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-message-rig.test-fixture.ts @@ -49,6 +49,7 @@ export async function createQueuedMessageTestRig( idleSweep?: { idleMs: number; intervalMs: number } /** Teardown records its restart offers, which `restartOffers` lists. */ recoveryCapsule?: true + now?: () => number } = {} ) { const root = await mkdtemp(join(tmpdir(), 'orca-queued-messages-')) @@ -65,7 +66,7 @@ export async function createQueuedMessageTestRig( journalDatabase: openTestJournalHostDatabase(root), claimKeyId: 'key-1', mintSpawnToken: () => 'spawn-1', - now: () => NOW, + now: options.now ?? (() => NOW), ...(options.idleSweep ? { idleSweep: options.idleSweep } : {}), ...(options.recoveryCapsule ? { recoveryCapsule: new AgentSessionRecoveryCapsule(root) } : {}) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts index 2a26af994dcb..76544f661a5e 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts @@ -12,6 +12,7 @@ import type { StructuredAgentSessionStatusObserverOptions } from './structured-a // republishes them. import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import { agentSessionLeaseOwnerVerdict } from '../../../shared/agent-session-lease-adjudication' import type { AgentSessionStatusEvent, AgentSessionStatusSummary @@ -320,11 +321,20 @@ export class StructuredAgentSessionStatusFeed { * surface of them shares. */ readChildWork(sessionId: string): AgentChildWorkView[] | undefined { try { - return this.ownership.readChildWork(sessionId) + const children = this.ownership.readChildWork(sessionId) + if (children !== undefined) { + return children + } } catch (error) { this.logFailure('child-work-read', 'child work read failed', sessionId, error) - return undefined } + // A released provider with proven exit has no live tasks, including after the roster is forgotten. + const lease = this.deps.getRecord(sessionId)?.lease + return !this.deps.sessions.get(sessionId)?.child && + lease && + agentSessionLeaseOwnerVerdict(lease) === 'exited' + ? [] + : undefined } private admitChildWork( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-capacity.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-capacity.test.ts new file mode 100644 index 000000000000..f66bd563c020 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-capacity.test.ts @@ -0,0 +1,154 @@ +import { afterEach, expect, it, vi } from 'vitest' +import { + agentSessionOperationKey, + pendingAgentSessionOperationRow, + type AgentSessionOperationOutcome +} from '../../../shared/agent-session-operation-ledger' +import { + createQueuedMessageTestRig, + QUEUED_RIG_CALLER, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { + HOST_TEST_NOW as NOW, + HOST_TEST_SESSION as SESSION, + hostTestMessage +} from './structured-agent-session-host-test-data' + +let rig: QueuedMessageTestRig | undefined +afterEach(async () => { + vi.restoreAllMocks() + await rig?.dispose() + rig = undefined +}) + +const outcomes: AgentSessionOperationOutcome[] = [ + { status: 'succeeded', sessionId: SESSION }, + { status: 'failed', code: 'fixture-failure' }, + { status: 'pending' }, + { status: 'unknown' } +] + +it.each(['caller', 'global'] as const)( + 'Stop and send replay survive the former %s receipt quota', + async (limit) => { + for (const outcome of outcomes) { + rig = await createQueuedMessageTestRig() + const sentId = await rig.workingSend() + const receipt = rig.store.getOperationRow(QUEUED_RIG_CALLER.callerKey, sentId) + const count = limit === 'caller' ? 512 : 4_096 + await rig.store['transactions'].transact(({ operations: rows }) => { + for (let index = 0; index < count; index += 1) { + const callerKey = limit === 'caller' ? QUEUED_RIG_CALLER.callerKey : `other-${index}` + const operationId = `${NOW}-1${index.toString(16).padStart(31, '0')}` + const row = pendingAgentSessionOperationRow({ + callerKey, + operationId, + fingerprint: 'retained-fixture', + now: NOW + }) + rows.set(agentSessionOperationKey(callerKey, operationId), { ...row, outcome }) + } + }) + const stopId = `${NOW}-${'f'.repeat(32)}` + expect(await rig.stop(stopId)).toMatchObject({ ok: true, replayed: false }) + expect(rig.cancelTurn).toHaveBeenCalledOnce() + expect(await rig.stop(stopId)).toMatchObject({ ok: true, replayed: true }) + expect(rig.cancelTurn).toHaveBeenCalledOnce() + expect(rig.store.getOperationRow(QUEUED_RIG_CALLER.callerKey, sentId)).toEqual(receipt) + const body = hostTestMessage('work on this') + expect( + await rig.host.send(QUEUED_RIG_CALLER, { + envelope: rig.envelope({ body }, 'agentSession.send', sentId), + body, + userSend: true + }) + ).toMatchObject({ ok: true, replayed: true }) + expect(rig.dispatch).toHaveBeenCalledOnce() + await rig.dispose() + rig = undefined + } + } +) + +it('Stop still runs when its receipt cannot be written above the former quota', async () => { + rig = await createQueuedMessageTestRig() + await rig.workingSend() + await rig.store['transactions'].transact(({ operations: rows }) => { + for (let index = 0; index < 512; index += 1) { + const row = pendingAgentSessionOperationRow({ + callerKey: QUEUED_RIG_CALLER.callerKey, + operationId: `${NOW}-1${index.toString(16).padStart(31, '0')}`, + fingerprint: 'retained-fixture', + now: NOW + }) + rows.set(agentSessionOperationKey(row.callerKey, row.operationId), row) + } + }) + vi.spyOn(rig.store, 'admitMutationOperation').mockRejectedValueOnce(new Error('write failed')) + expect(await rig.stop()).toMatchObject({ ok: true }) + expect(rig.cancelTurn).toHaveBeenCalledOnce() +}) + +it.each(['caller', 'global'] as const)( + 'retains idle Stop and Send receipts above the former %s limit', + async (limit) => { + rig = await createQueuedMessageTestRig() + const workCount = limit === 'caller' ? 512 : 4_096 + await rig.store['transactions'].transact(({ operations: rows }) => { + for (let index = 0; index < workCount; index += 1) { + const callerKey = limit === 'caller' ? QUEUED_RIG_CALLER.callerKey : `other-${index}` + const operationId = `${NOW}-1${index.toString(16).padStart(31, '0')}` + const row = pendingAgentSessionOperationRow({ + callerKey, + operationId, + fingerprint: 'retained-fixture', + now: NOW + }) + rows.set(agentSessionOperationKey(callerKey, operationId), { + ...row, + outcome: { status: 'succeeded', sessionId: SESSION } + }) + } + }) + const protectedRows = rig.store.listOperationRows() + const before = rig.store.listOperationRows().length + for (let index = 0; index < 4; index += 1) { + expect(await rig.stop(`${NOW}-f${index.toString(16).padStart(31, '0')}`)).toMatchObject({ + ok: true + }) + } + expect(rig.store.listOperationRows()).toHaveLength(before + 4) + expect(rig.store.listOperationRows()).toEqual(expect.arrayContaining(protectedRows)) + expect(rig.cancelTurn).not.toHaveBeenCalled() + expect(await rig.send('new send above the former limit').result).toMatchObject({ ok: true }) + expect(rig.store.listOperationRows()).toHaveLength(before + 5) + } +) + +it.each(['pending', 'unknown'] as const)( + 'Stop retains its receipt when prior cleanup receipts are %s', + async (status) => { + rig = await createQueuedMessageTestRig() + await rig.workingSend() + await rig.store['transactions'].transact(({ operations: rows }) => { + for (let index = 0; index < 64; index += 1) { + const row = pendingAgentSessionOperationRow({ + callerKey: QUEUED_RIG_CALLER.callerKey, + operationId: `${NOW}-1${index.toString(16).padStart(31, '0')}`, + fingerprint: 'in-flight', + now: NOW + }) + rows.set(agentSessionOperationKey(row.callerKey, row.operationId), { + ...row, + outcome: { status } + }) + } + }) + const before = rig.store.listOperationRows() + expect(await rig.stop()).toMatchObject({ ok: true }) + expect(rig.cancelTurn).toHaveBeenCalledOnce() + expect(rig.store.listOperationRows()).toHaveLength(before.length + 1) + expect(rig.store.listOperationRows()).toEqual(expect.arrayContaining(before)) + } +) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-target-observation.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-target-observation.ts new file mode 100644 index 000000000000..f4382da50c0e --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-target-observation.ts @@ -0,0 +1,109 @@ +import { + agentJournalSubmissionKey, + parseAgentJournalItemKey +} from '../../../shared/agent-session-journal-item-key' +import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission' +import { + agentSessionStopTargetIsLive, + type AgentSessionStopTarget +} from '../../../shared/agent-session-stop-target' +import { structuredAgentSessionStopTookEffect } from './structured-agent-session-stopping' +import type { AgentSessionTurnContext } from './structured-agent-session-turns' + +type StopTargetObservation = + | { verdict: 'live'; turnId?: string } + | { verdict: 'exited' } + | { verdict: 'unverifiable' } + +/** Missing message linkage is uncertainty; only the execution host can bind an opening turn. */ +export function observeStructuredAgentSessionStopTarget( + ctx: Pick, + target: AgentSessionStopTarget +): StopTargetObservation { + const activeTurnId = ctx.journal.activeTurnId() + const submissions = ctx.journal.submissions() + if (target.kind !== 'submission') { + return agentSessionStopTargetIsLive(target, activeTurnId, submissions, ctx.fence) + ? { verdict: 'live' } + : { verdict: 'exited' } + } + const submission = submissions.find((entry) => entry.clientMessageId === target.clientMessageId) + if (!submission) { + return { verdict: 'unverifiable' } + } + if ( + submission.dispatchState === 'rejected' || + submission.recovered === true || + (submission.fence < ctx.fence && !isQueuedAgentJournalSubmission(submission)) + ) { + return { verdict: 'exited' } + } + if ( + activeTurnId !== null && + agentSessionStopTargetIsLive( + target, + activeTurnId, + submissions, + ctx.fence, + ctx.journal.newestTurn()?.userItemId + ) + ) { + return { verdict: 'live', turnId: activeTurnId } + } + const identity = submission.providerItemId + ? parseAgentJournalItemKey(submission.providerItemId) + : null + const linkedTurnId = + identity?.provider === 'codex' + ? identity.turnId + : identity?.provider === 'claude' + ? identity.uuid + : undefined + const observed = ctx.adapter.observeSubmissionTurn?.({ + sessionId: ctx.sessionId, + clientMessageId: target.clientMessageId, + fence: ctx.fence + }) + const turnId = linkedTurnId ?? (observed?.verdict === 'live' ? observed.turnId : undefined) + if (turnId !== undefined) { + const turn = ctx.journal + .snapshot() + .items.findLast((item) => item.body.kind === 'turn' && item.body.turnId === turnId)?.body + if ( + (activeTurnId !== null && activeTurnId !== turnId) || + (turn?.kind === 'turn' && turn.state !== 'running') + ) { + return { verdict: 'exited' } + } + if (observed?.verdict === 'live') { + return { verdict: 'live', turnId } + } + } + if (observed?.verdict === 'exited') { + return observed + } + if (activeTurnId !== null) { + return { verdict: 'unverifiable' } + } + // A Stop of everything in flight that took effect after the send reached the agent ended it. + const latestStop = ctx.journal.stopMarks.latest() + const reachedAgentAt = + ctx.journal.item(agentJournalSubmissionKey(submission.clientMessageId))?.sequence ?? + submission.submittedSequence + if ( + latestStop !== null && + latestStop.event.turnId === undefined && + structuredAgentSessionStopTookEffect(latestStop) && + reachedAgentAt !== undefined && + latestStop.sequence > reachedAgentAt + ) { + return { verdict: 'exited' } + } + if (agentSessionStopTargetIsLive(target, null, submissions, ctx.fence)) { + return { verdict: 'live' } + } + // Answered, and no turn runs now: its own turn already ended. + return submission.dispatchState === 'accepted' + ? { verdict: 'exited' } + : { verdict: 'unverifiable' } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stopping.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stopping.ts index 284a474b9c65..24c370535b90 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stopping.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stopping.ts @@ -11,7 +11,10 @@ import type { import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-live-turn' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' -import { stopIsAPersons } from '../agent-session-journal/journal-stop-turn-end' +import { + stopIsAPersons, + type JournalLatestStop +} from '../agent-session-journal/journal-stop-turn-end' import type { JournalStopFailedOn } from '../agent-session-journal/queued-message-pause' /** Whether the latest person's Stop is still settling, or is bound to the turn running now: the @@ -61,7 +64,13 @@ function firstSendHandedOverAfter( return first } -/** Where a Stop that failed leaves "Stopping…" (display only): the turn running now, or, with +/** Whether `stop` took effect: settled, and not failed. While it settles or after it failed, the + * chat reads "Stopping…" and what it was sent for may still run. */ +export function structuredAgentSessionStopTookEffect(stop: JournalLatestStop): boolean { + return stop.settle?.settling !== true && stop.settle?.failedOn === undefined +} + +/** Where a Stop that failed leaves "Stopping…": the turn running now, or, with * none, the first that opens after this position. */ export function structuredAgentSessionFailedStopMark( journal: Pick diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-submission-stop-opening.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-submission-stop-opening.test.ts new file mode 100644 index 000000000000..537fbfa3264d --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-submission-stop-opening.test.ts @@ -0,0 +1,76 @@ +import { afterEach, expect, it } from 'vitest' +import { agentSessionStopTarget } from '../../../shared/agent-session-stop-target' +import { publishCodexTurnLifecycle } from '../../codex/codex-structured-journal-translation-turns' +import { + createQueuedMessageTestRig, + QUEUED_RIG_CALLER, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { + HOST_TEST_SESSION as SESSION, + HOST_TEST_THREAD as THREAD, + hostTestOperationId +} from './structured-agent-session-host-test-data' + +let rig: QueuedMessageTestRig | undefined +afterEach(async () => { + await rig?.dispose() + rig = undefined +}) + +async function openingSubmission() { + const current = await createQueuedMessageTestRig() + rig = current + const id = await current.workingSend() + const stopTarget = agentSessionStopTarget( + null, + (await current.host.journalSnapshot(SESSION)).submissions, + 1 + ) + expect(stopTarget).toEqual({ kind: 'submission', clientMessageId: id }) + const envelope = current.envelope({ stopTarget }, 'agentSession.cancel', hostTestOperationId()) + publishCodexTurnLifecycle({ + sink: current.providerEvents(), + primaryThreadId: THREAD, + sessionId: SESSION, + threadId: THREAD, + turnId: 'turn-opening', + state: 'running' + }) + await current.host.flushStreamedEvents(SESSION) + expect((await current.submission(id))?.providerItemId).toBeNull() + expect((await current.host.journalSnapshot(SESSION)).items).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + body: expect.objectContaining({ turnId: 'turn-opening', state: 'running' }) + }) + ]) + ) + return { current, id, params: { stopTarget, envelope } } +} + +it('does not silently discard a captured submission Stop before its user echo', async () => { + const { current, params } = await openingSubmission() + const response = await current.host.cancel(QUEUED_RIG_CALLER, params) + expect(response).not.toMatchObject({ ok: true, value: { cancelled: false } }) + expect(response).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_unknown' } + }) + expect(current.cancelTurn).not.toHaveBeenCalled() + expect(current.closeSession).not.toHaveBeenCalled() +}) + +it('the captured submission Stop reaches the turn after its user echo links it', async () => { + const { current, id, params } = await openingSubmission() + await current.host.settleLateDispatch({ + sessionId: SESSION, + clientMessageId: id, + providerIdentity: { provider: 'codex', threadId: THREAD, turnId: 'turn-opening', ordinal: 0 } + }) + expect(await current.host.cancel(QUEUED_RIG_CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: true } + }) + expect(current.cancelTurn).toHaveBeenCalledOnce() +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-target-expiry.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-target-expiry.test.ts new file mode 100644 index 000000000000..92c0a7c8cd77 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-target-expiry.test.ts @@ -0,0 +1,117 @@ +import { afterEach, expect, it, vi } from 'vitest' +import { + AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS, + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS +} from '../../../shared/agent-session-host-authority' +import { + createQueuedMessageTestRig, + QUEUED_RIG_CALLER, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { + HOST_TEST_NOW as NOW, + HOST_TEST_SESSION as SESSION, + hostTestOperationId +} from './structured-agent-session-host-test-data' +let rig: QueuedMessageTestRig | undefined + +afterEach(async () => { + vi.restoreAllMocks() + await rig?.dispose() + rig = undefined +}) + +it('delivers a delayed Stop to its still-unanswered host target beyond the work receipt window', async () => { + let now = NOW + rig = await createQueuedMessageTestRig({ now: () => now }) + const sentId = await rig.workingSend() + const stopTarget = { kind: 'submission' as const, clientMessageId: sentId } + const params = { + stopTarget, + envelope: rig.envelope({ stopTarget }, 'agentSession.cancel', hostTestOperationId()) + } + now += AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS + 1 + const result = await rig.host.cancel(QUEUED_RIG_CALLER, params) + expect(result).toMatchObject({ ok: true, value: { cancelled: true } }) + expect(rig.cancelTurn).toHaveBeenCalledOnce() +}) +it.each([false, true])( + 'answers an expired ended target quietly after restart: %s', + async (restart) => { + let now = NOW + rig = await createQueuedMessageTestRig({ + now: () => now, + stopEndsSession: true, + restartable: true + }) + const sentId = await rig.workingSend() + const stopTarget = { kind: 'submission' as const, clientMessageId: sentId } + const params = { + stopTarget, + envelope: rig.envelope({ stopTarget }, 'agentSession.cancel', hostTestOperationId()) + } + expect(await rig.host.cancel(QUEUED_RIG_CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: true } + }) + await rig.host.close(SESSION, 'user-close') + now += AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS + 1 + if (restart) { + await rig.restartHostProcess() + } + const acquire = vi.spyOn(rig.host.deps.adapter, 'acquire') + const closes = rig.closeSession.mock.calls.length + const result = await rig.host.cancel(QUEUED_RIG_CALLER, params) + expect(result).toMatchObject({ ok: true, value: { cancelled: false } }) + expect(rig.cancelTurn).toHaveBeenCalledOnce() + expect(rig.closeSession).toHaveBeenCalledTimes(closes) + expect(acquire).not.toHaveBeenCalled() + } +) + +it('answers an ended background target as a no-op after the provider was closed', async () => { + rig = await createQueuedMessageTestRig() + await rig.workingSend() + await rig.host.close(SESSION, 'user-close') + const acquire = vi.spyOn(rig.host.deps.adapter, 'acquire') + const stopBackgroundTasks = vi.fn(async () => ({ cancelled: true })) + rig.host.deps.adapter.stopBackgroundTasks = stopBackgroundTasks + const fields = { + scope: 'background-tasks' as const, + turnId: 'background-tasks', + stopTarget: { + kind: 'background-tasks' as const, + tasks: [{ id: 'ended-child', invocation: { invocationId: 'old-invocation', generation: 1 } }] + } + } + const params = { + ...fields, + envelope: rig.envelope(fields, 'agentSession.cancel', hostTestOperationId()) + } + const result = await rig.host.cancel(QUEUED_RIG_CALLER, params) + expect(result).toMatchObject({ ok: true, value: { cancelled: false } }) + expect(acquire).not.toHaveBeenCalled() + expect(stopBackgroundTasks).not.toHaveBeenCalled() +}) + +it('still validates an expired targeted Stop before any effect', async () => { + let now = NOW + rig = await createQueuedMessageTestRig({ now: () => now }) + const sentId = await rig.workingSend() + const stopTarget = { kind: 'submission' as const, clientMessageId: sentId } + const envelope = rig.envelope({ stopTarget }, 'agentSession.cancel', hostTestOperationId()) + now += AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS + 1 + expect( + await rig.host.cancel(QUEUED_RIG_CALLER, { + stopTarget, + envelope: { ...envelope, payloadFingerprint: 'wrong' } + }) + ).toMatchObject({ ok: false, refusal: { code: 'agent_session_operation_conflict' } }) + expect( + await rig.host.cancel(QUEUED_RIG_CALLER, { + stopTarget, + envelope: { ...envelope, clientOperationId: 'invalid' } + }) + ).toMatchObject({ ok: false, refusal: { code: 'agent_session_operation_invalid' } }) + expect(rig.cancelTurn).not.toHaveBeenCalled() +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-targeted-stop.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-targeted-stop.test.ts new file mode 100644 index 000000000000..584ea0a16ae5 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-targeted-stop.test.ts @@ -0,0 +1,126 @@ +import { afterEach, expect, it, vi } from 'vitest' +import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' +import { + createQueuedMessageTestRig, + QUEUED_RIG_CALLER, + eventually, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { + HOST_TEST_SESSION as SESSION, + hostTestOperationId +} from './structured-agent-session-host-test-data' + +let rig: QueuedMessageTestRig | undefined +afterEach(async () => { + vi.restoreAllMocks() + await rig?.dispose() + rig = undefined +}) + +it('targets a turn without a receipt and cannot stop the next turn after retry or restart', async () => { + rig = await createQueuedMessageTestRig({ stopEndsSession: true, restartable: true }) + await rig.workingSend() + const identity = { + provider: 'legacy' as const, + agent: 'codex', + sessionId: SESSION, + recordId: 'turn:one' + } + rig + .providerEvents() + .appendItem( + identity, + { kind: 'turn', turnId: 'one', state: 'running' }, + { turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + await eventually(async () => + expect((await rig!.host.journalSnapshot(SESSION)).items).toEqual( + expect.arrayContaining([ + expect.objectContaining({ body: expect.objectContaining({ turnId: 'one' }) }) + ]) + ) + ) + const stopTarget = { kind: 'turn' as const, turnId: 'one' } + const fields = { stopTarget } + const id = hostTestOperationId() + const params = { ...fields, envelope: rig.envelope(fields, 'agentSession.cancel', id) } + expect(await rig.host.cancel(QUEUED_RIG_CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: true } + }) + expect(rig.store.getOperationRow(QUEUED_RIG_CALLER.callerKey, id)).toBeNull() + await rig.restartHostProcess() + await rig.workingSend() + rig + .providerEvents() + .appendItem( + { ...identity, recordId: 'turn:two' }, + { kind: 'turn', turnId: 'two', state: 'running' }, + { turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + await rig.host.flushAllStreamedEvents() + const calls = rig.cancelTurn.mock.calls.length + const closes = rig.closeSession.mock.calls.length + expect(await rig.host.cancel(QUEUED_RIG_CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: false } + }) + expect(rig.cancelTurn).toHaveBeenCalledTimes(calls) + expect(rig.closeSession).toHaveBeenCalledTimes(closes) +}) + +it('targets an unanswered submission without growing receipts or interrupting a later send', async () => { + rig = await createQueuedMessageTestRig({ restartable: true }) + const first = await rig.workingSend() + const stopTarget = { kind: 'submission' as const, clientMessageId: first } + const id = hostTestOperationId() + const params = { stopTarget, envelope: rig.envelope({ stopTarget }, 'agentSession.cancel', id) } + const admission = vi.spyOn(rig.store, 'recordOperationOutcome') + expect(await rig.host.cancel(QUEUED_RIG_CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: true } + }) + expect(rig.store.getOperationRow(QUEUED_RIG_CALLER.callerKey, id)).toBeNull() + expect(admission).not.toHaveBeenCalled() + const next = rig.send('later send') + expect(await next.result).toMatchObject({ ok: true }) + const calls = rig.cancelTurn.mock.calls.length + expect(await rig.host.cancel(QUEUED_RIG_CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: false } + }) + expect(rig.cancelTurn).toHaveBeenCalledTimes(calls) +}) + +it('stops a waiting submission even when a later send is queued before the Stop lands', async () => { + rig = await createQueuedMessageTestRig({ restartable: true }) + const first = await rig.workingSend() + const stopTarget = { kind: 'submission' as const, clientMessageId: first } + const later = rig.send('sent just before Stop') + expect(await later.result).toMatchObject({ ok: true }) + const queued = await rig.submission(later.id) + expect(queued).toMatchObject({ handoverRecorded: true, dispatchState: 'pending' }) + expect(queued?.handedOverAt).toBeUndefined() + const id = hostTestOperationId() + const params = { stopTarget, envelope: rig.envelope({ stopTarget }, 'agentSession.cancel', id) } + expect(await rig.host.cancel(QUEUED_RIG_CALLER, params)).toMatchObject({ + ok: true, + value: { cancelled: true } + }) + expect(rig.cancelTurn).toHaveBeenCalled() +}) + +it('keeps target and fingerprint validation when a Stop cannot write its receipt', async () => { + rig = await createQueuedMessageTestRig({ restartable: true }) + await rig.workingSend() + const stopTarget = { kind: 'turn' as const, turnId: 'gone' } + const envelope = rig.envelope({ stopTarget }, 'agentSession.cancel', hostTestOperationId()) + expect( + await rig.host.cancel(QUEUED_RIG_CALLER, { + envelope: { ...envelope, payloadFingerprint: 'wrong' }, + stopTarget + }) + ).toMatchObject({ ok: false }) + expect(rig.cancelTurn).not.toHaveBeenCalled() +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts index a588fa7168af..1401fac920cc 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts @@ -1,4 +1,8 @@ import { agentChildWorkStopTargets } from '../../../shared/agent-child-work-stop-targets' +import { + targetedAgentSessionBackgroundTaskIds, + type AgentSessionBackgroundStopTarget +} from '../../../shared/agent-session-background-stop-target' import { agentSessionFailureFact } from '../../../shared/agent-session-failure' import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view' import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words' @@ -129,6 +133,7 @@ type PerformCancelInput = { turnId?: string scope?: 'background-tasks' taskId?: string + backgroundStopTarget?: AgentSessionBackgroundStopTarget prompt?: { itemId: string; expectedRevision: number } /** Ends the provider child, for a running command the provider did not take the Stop on, or a * turn whose interrupt failed. */ @@ -200,7 +205,13 @@ async function cancelAndNote( ? await ctx.adapter.stopBackgroundTasks?.({ sessionId: ctx.sessionId, fence: ctx.fence, - taskIds: agentChildWorkStopTargets(input.childWork?.(), input.taskId) + taskIds: input.backgroundStopTarget + ? targetedAgentSessionBackgroundTaskIds( + input.childWork?.(), + input.backgroundStopTarget, + input.taskId + ) + : agentChildWorkStopTargets(input.childWork?.(), input.taskId) }) : undefined // A background Stop writes no row, so a task the agent shows still running after it goes back diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-work-receipt-window.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-work-receipt-window.test.ts new file mode 100644 index 000000000000..22921046f38b --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-work-receipt-window.test.ts @@ -0,0 +1,111 @@ +import { afterEach, expect, it } from 'vitest' +import { + AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS, + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS +} from '../../../shared/agent-session-host-authority' +import { + agentSessionOperationKey, + pendingAgentSessionOperationRow +} from '../../../shared/agent-session-operation-ledger' +import { + createQueuedMessageTestRig, + eventually, + QUEUED_RIG_CALLER, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { + HOST_TEST_NOW as NOW, + HOST_TEST_SESSION as SESSION, + hostTestMessage +} from './structured-agent-session-host-test-data' + +let rig: QueuedMessageTestRig | undefined +afterEach(async () => { + await rig?.dispose() + rig = undefined +}) + +it('allows a fresh Send after 600 settled operations over a day', async () => { + rig = await createQueuedMessageTestRig() + for (let index = 0; index < 600; index += 1) { + const now = NOW - (599 - index) * 138_000 + const operationId = `${now}-e${index.toString(16).padStart(31, '0')}` + expect( + await rig.store.admitOperation({ + callerKey: QUEUED_RIG_CALLER.callerKey, + operationId, + fingerprint: 'fp-history', + now + }) + ).toMatchObject({ decision: 'admit' }) + await rig.store.recordOperationOutcome({ + callerKey: QUEUED_RIG_CALLER.callerKey, + operationId, + outcome: { status: 'succeeded', sessionId: SESSION } + }) + } + expect(await rig.send('fresh work after a heavy day').result).toMatchObject({ ok: true }) + await eventually(() => expect(rig?.dispatch).toHaveBeenCalledOnce()) + expect(rig.store.listOperationRows().length).toBeGreaterThan(600) +}) + +it('replays a lost Send reply from its receipt, past the former operation age, without dispatching again', async () => { + let now = NOW + rig = await createQueuedMessageTestRig({ now: () => now }) + const id = await rig.workingSend() + await rig.settleAccepted(id, 'accepted-item') + const body = hostTestMessage('work on this') + const args = { + envelope: rig.envelope({ body }, 'agentSession.send', id), + body, + userSend: true as const + } + now += AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS - 1 + expect(await rig.host.send(QUEUED_RIG_CALLER, args)).toMatchObject({ + ok: true, + replayed: true + }) + // A send's receipt, not operation age, answers its retry. + now += 1 + expect(await rig.host.send(QUEUED_RIG_CALLER, args)).toMatchObject({ + ok: true, + replayed: true + }) + expect(rig.dispatch).toHaveBeenCalledOnce() + const history = await rig.host.history({ sessionId: SESSION, direction: 'tail' }) + expect(history).toMatchObject({ + ok: true, + page: { + submissions: expect.arrayContaining([ + expect.objectContaining({ clientMessageId: id, dispatchState: 'accepted' }) + ]) + } + }) +}) + +it.each(['pending', 'unknown'] as const)( + 'admits new Send with 512 %s receipts while Stop and Close still succeed', + async (status) => { + rig = await createQueuedMessageTestRig() + await rig.store['transactions'].transact(({ operations: rows }) => { + for (let index = 0; index < 512; index += 1) { + const row = pendingAgentSessionOperationRow({ + callerKey: QUEUED_RIG_CALLER.callerKey, + operationId: `${NOW}-e${index.toString(16).padStart(31, '0')}`, + fingerprint: 'fp-in-flight', + now: NOW + }) + rows.set(agentSessionOperationKey(row.callerKey, row.operationId), { + ...row, + outcome: { status } + }) + } + }) + expect(await rig.send('fresh work above the former limit').result).toMatchObject({ ok: true }) + await eventually(() => expect(rig?.dispatch).toHaveBeenCalledOnce()) + expect(await rig.stop()).toMatchObject({ ok: true }) + expect(rig.cancelTurn).toHaveBeenCalledOnce() + await rig.host.close(SESSION, 'user-close') + expect(rig.closeSession).toHaveBeenCalledOnce() + } +) diff --git a/src/main/runtime/agent-session-operation-admission.ts b/src/main/runtime/agent-session-operation-admission.ts index a37b263d72a4..a4970f7e3093 100644 --- a/src/main/runtime/agent-session-operation-admission.ts +++ b/src/main/runtime/agent-session-operation-admission.ts @@ -31,6 +31,7 @@ export type AgentSessionOperationAdmission = { operationId: string fingerprint: string now: number + skipReceipt?: true } export type AgentSessionOperationAcceptance = AgentSessionOperationAdmission & { @@ -82,11 +83,13 @@ export type AgentSessionMutationOperationAdmission = { now: number operationIdScope?: 'global' conversationWrite?: true + receiptPolicy?: 'none' } export type AgentSessionMutationOperationDecision = { admission: AgentSessionMutationAdmission record: AgentSessionRecord + receiptRecorded?: false } | null type EvaluatedOperationRows = { rows: OperationRows; decision: AgentSessionOperationDecision } @@ -96,6 +99,9 @@ export function evaluateAgentSessionOperationRow( rows: OperationRows, args: AgentSessionOperationAdmission ): EvaluatedOperationRows { + if (args.skipReceipt) { + return { rows, decision: evaluateAgentSessionOperation({ ...args, rows: new Map() }) } + } const pruned = pruneAgentSessionOperationRows(rows, args.now) return { rows: pruned, decision: evaluateAgentSessionOperation({ rows: pruned, ...args }) } } @@ -124,7 +130,7 @@ function placeAdmittedAgentSessionOperationRow( evaluated: EvaluatedOperationRows, args: AgentSessionOperationAdmission ): EvaluatedOperationRows { - if (evaluated.decision.decision === 'admit') { + if (evaluated.decision.decision === 'admit' && !args.skipReceipt) { evaluated.rows.set( agentSessionOperationKey(args.callerKey, args.operationId), evaluated.decision.row @@ -174,7 +180,8 @@ function mutationOperation( callerKey: args.callerKey, operationId: args.envelope.clientOperationId, fingerprint: args.hostFingerprint, - now: args.now + now: args.now, + ...(args.receiptPolicy === 'none' ? { skipReceipt: true as const } : {}) } } @@ -202,7 +209,16 @@ export function admitAgentSessionMutationOperation( ledger.rows.delete(agentSessionOperationKey(operation.callerKey, operation.operationId)) } state.operations = ledger.rows - return { admission, record } + const receiptRecorded = + ledger.decision.decision !== 'refused' && + ledger.rows.has( + agentSessionOperationKey(ledger.decision.row.callerKey, ledger.decision.row.operationId) + ) + return { + admission, + record, + ...(!receiptRecorded ? { receiptRecorded: false as const } : {}) + } } /** diff --git a/src/main/runtime/agent-session-operation-retention.test.ts b/src/main/runtime/agent-session-operation-retention.test.ts new file mode 100644 index 000000000000..4bf2ae7337b8 --- /dev/null +++ b/src/main/runtime/agent-session-operation-retention.test.ts @@ -0,0 +1,68 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, expect, it } from 'vitest' +import { + agentSessionOperationExpiry, + agentSessionOperationKey +} from '../../shared/agent-session-operation-ledger' +import { + openTestAgentSessionRecordStore, + readPersistedTestAgentSessionStore +} from './agent-session-record-store-test-harness' + +const NOW = 1_800_000_000_000 +let root: string | undefined +afterEach(async () => { + if (root) { + await rm(root, { recursive: true, force: true }) + } + root = undefined +}) + +it('keeps a settled receipt through restart, then deletes it without readmitting its retry', async () => { + root = await mkdtemp(join(tmpdir(), 'orca-operation-retention-')) + const operation = { + callerKey: 'client-1', + operationId: `${NOW}-${'a'.repeat(32)}`, + fingerprint: 'original-send', + now: NOW + } + const store = await openTestAgentSessionRecordStore(root) + const admitted = await store.admitGlobalOperation(operation) + if (admitted.decision !== 'admit') { + throw new Error('expected admitted operation') + } + await store.recordOperationOutcome({ + ...operation, + outcome: { status: 'succeeded', sessionId: 'session-1' } + }) + const reopened = await openTestAgentSessionRecordStore(root) + const expiresAt = agentSessionOperationExpiry(NOW, NOW) + expect( + await reopened.admitGlobalOperation({ + ...operation, + callerKey: 'reconnected-client', + now: expiresAt - 1 + }) + ).toMatchObject({ + decision: 'replay', + row: { callerKey: 'client-1', outcome: { status: 'succeeded' } } + }) + const key = agentSessionOperationKey(operation.callerKey, operation.operationId) + expect((await readPersistedTestAgentSessionStore(root)).operations[key]).toBeDefined() + + expect( + await reopened.admitGlobalOperation({ + ...operation, + operationId: `${expiresAt}-${'b'.repeat(32)}`, + now: expiresAt + }) + ).toMatchObject({ decision: 'admit' }) + expect((await readPersistedTestAgentSessionStore(root)).operations[key]).toBeUndefined() + expect(await reopened.admitGlobalOperation({ ...operation, now: expiresAt })).toMatchObject({ + decision: 'refused', + code: 'agent_session_operation_expired' + }) + expect(reopened.listOperationRows()).toHaveLength(1) +}) diff --git a/src/main/runtime/rpc/methods/agent-launch-restart-replay.test.ts b/src/main/runtime/rpc/methods/agent-launch-restart-replay.test.ts index 00ed70257c93..daccc05cfc36 100644 --- a/src/main/runtime/rpc/methods/agent-launch-restart-replay.test.ts +++ b/src/main/runtime/rpc/methods/agent-launch-restart-replay.test.ts @@ -517,6 +517,33 @@ describe('a caller cannot claim an identity', () => { }) describe('one row per launch, with no limit on how many a caller holds', () => { + it('allows a fresh desktop launch after 600 settled operations over a day', async () => { + const end = Date.now() + for (let index = 0; index < 600; index += 1) { + const now = end - (599 - index) * 138_000 + const operationId = `${now}-${index.toString(16).padStart(32, '0')}` + expect( + await store.admitOperation({ + callerKey: 'trusted-local:desktop', + operationId, + fingerprint: 'fp-history', + now + }) + ).toMatchObject({ decision: 'admit' }) + await store.recordOperationOutcome({ + callerKey: 'trusted-local:desktop', + operationId, + outcome: { status: 'succeeded', sessionId: '' } + }) + } + const host = hostRuntime() + await expect(launch(host, PROMPTED_LAUNCH, { ...DESKTOP_IPC })).resolves.toMatchObject({ + outcome: { kind: 'terminal' } + }) + expect(host.createTerminal).toHaveBeenCalledOnce() + expect(store.listOperationRows()).toHaveLength(601) + }) + it('keeps one row per launch, retained from admission, across both writes', async () => { let releasePaste: (pasted: boolean) => void = () => {} deliverTerminalPrompt.mockImplementationOnce( diff --git a/src/main/runtime/rpc/methods/structured-agent-session-close.ts b/src/main/runtime/rpc/methods/structured-agent-session-close.ts new file mode 100644 index 000000000000..fd93127579b3 --- /dev/null +++ b/src/main/runtime/rpc/methods/structured-agent-session-close.ts @@ -0,0 +1,31 @@ +import type { StructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-host' + +export async function closeStructuredAgentSessionForClient( + host: StructuredAgentSessionHost, + sessionId: string +): Promise<{ ok: true }> { + // Start shutdown before bookkeeping, so a failed or pending write cannot delay it. + const [closed, hidden] = await Promise.allSettled([ + host.close(sessionId, 'user-close'), + (async () => { + // Terminal-disposal closes use this RPC without the session-tabs retirement RPC. + if (typeof host.setSessionTabVisibility === 'function') { + await host.setSessionTabVisibility(sessionId, false) + } + })() + ]) + if (hidden.status === 'rejected') { + host.deps.logger.warn('hiding a closed chat tab failed', { + scope: 'tab-close-visibility', + sessionId, + error: hidden.reason + }) + } + if (closed.status === 'rejected') { + throw closed.reason + } + if (hidden.status === 'rejected') { + throw hidden.reason + } + return { ok: true } +} diff --git a/src/main/runtime/rpc/methods/structured-agent-session.test.ts b/src/main/runtime/rpc/methods/structured-agent-session.test.ts index f67dd8ed74fb..5dfebae0d8cc 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.test.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.test.ts @@ -136,18 +136,18 @@ describe('capability gating', () => { expect(response).toMatchObject({ ok: true, result: { ok: true } }) expect(hostCalls.close).toHaveBeenCalledWith(SESSION, 'user-close') expect(hostCalls.setSessionTabVisibility).toHaveBeenCalledWith(SESSION, false) - expect(hostCalls.setSessionTabVisibility.mock.invocationCallOrder[0]).toBeLessThan( - hostCalls.close.mock.invocationCallOrder[0]! + expect(hostCalls.close.mock.invocationCallOrder[0]).toBeLessThan( + hostCalls.setSessionTabVisibility.mock.invocationCallOrder[0]! ) }) - it('does not stop the provider when durable tab retirement fails', async () => { + it('stops the provider and reports failed durable tab retirement', async () => { hostCalls.setSessionTabVisibility.mockRejectedValueOnce(new Error('visibility write failed')) const response = await call('agentSession.close', { sessionId: SESSION }, STRUCTURED_CLIENT) expect(response).toMatchObject({ ok: false }) - expect(hostCalls.close).not.toHaveBeenCalled() + expect(hostCalls.close).toHaveBeenCalledExactlyOnceWith(SESSION, 'user-close') }) it('advertises the capability without bumping the protocol version', () => { diff --git a/src/main/runtime/rpc/methods/structured-agent-session.ts b/src/main/runtime/rpc/methods/structured-agent-session.ts index 37b9e62f960e..dcedf28c1093 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.ts @@ -58,6 +58,7 @@ import { UnsubscribeParams } from './structured-agent-session-schemas' import { sendStructuredAgentSessionForClient } from './structured-agent-session-send-compatibility' +import { closeStructuredAgentSessionForClient } from './structured-agent-session-close' export const STRUCTURED_AGENT_SESSION_METHODS = [ defineMethod({ @@ -130,16 +131,8 @@ export const STRUCTURED_AGENT_SESSION_METHODS = [ name: 'agentSession.close', permission: 'workspace', params: OptionsParams, - handler: async (params, ctx) => { - const host = requireStructuredCleanupHost(ctx) - // Terminal-disposal closes use this RPC without the session-tabs retirement RPC. - if (typeof host.setSessionTabVisibility === 'function') { - await host.setSessionTabVisibility(params.sessionId, false) - } - // Clients call this only when the user closes this chat's tab or cancels its launch. - await host.close(params.sessionId, 'user-close') - return { ok: true as const } - } + handler: (params, ctx) => + closeStructuredAgentSessionForClient(requireStructuredCleanupHost(ctx), params.sessionId) }), ...STRUCTURED_AGENT_SESSION_PROMPT_RESPONSE_METHODS, defineMethod({ diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-cancel.ts b/src/renderer/src/components/native-chat/use-structured-agent-session-cancel.ts new file mode 100644 index 000000000000..77e6900a0959 --- /dev/null +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-cancel.ts @@ -0,0 +1,55 @@ +import type { AgentSessionCancelResult } from '../../../../shared/agent-session-wire' +import type { RuntimeClientTarget } from '@/runtime/runtime-rpc-client' +import { supportsStructuredAgentSessionPromptCancel } from '@/runtime/structured-agent-session-client' +import { useStructuredAgentSessionHostCapability } from '@/runtime/structured-agent-session-host-capability' +import { AGENT_SESSION_TARGETED_STOP_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version' +import { agentSessionStopTarget } from '../../../../shared/agent-session-stop-target' +import { agentSessionBackgroundStopTarget } from '../../../../shared/agent-session-background-stop-target' +import type { StructuredAgentSessionMutate } from './use-structured-agent-session-mutate' +import type { useStructuredAgentSessionTransportState } from './use-structured-agent-session-transport-state' + +export function useStructuredAgentSessionCancel(input: { + target: RuntimeClientTarget + transportState: Pick< + ReturnType, + 'submissions' | 'fence' | 'backgroundTasks' + > + mutate: StructuredAgentSessionMutate +}) { + const { target, transportState, mutate } = input + const targetedStop = useStructuredAgentSessionHostCapability( + target, + AGENT_SESSION_TARGETED_STOP_RUNTIME_CAPABILITY + ) + return { + cancel: async ( + turnId: string | undefined, + prompt?: { itemId: string; expectedRevision: number } + ) => { + const stopTarget = agentSessionStopTarget( + turnId ?? null, + transportState.submissions, + transportState.fence + ) + const promptSupported = + prompt !== undefined && (await supportsStructuredAgentSessionPromptCancel(target)) + return mutate('agentSession.cancel', 'agentSession.cancel', { + ...(turnId ? { turnId } : {}), + ...(stopTarget && targetedStop ? { stopTarget } : {}), + ...(promptSupported ? { prompt } : {}) + }) + }, + stopBackgroundTask: (taskId?: string) => { + const stopTarget = agentSessionBackgroundStopTarget( + transportState.backgroundTasks.children, + taskId + ) + return mutate('agentSession.cancel', 'agentSession.cancel', { + turnId: 'background-tasks', + scope: 'background-tasks', + ...(taskId ? { taskId } : {}), + ...(targetedStop ? { stopTarget } : {}) + }) + } + } +} diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-stop.test.tsx b/src/renderer/src/components/native-chat/use-structured-agent-session-stop.test.tsx index 70f86c2bfed7..1fff0105fbb6 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session-stop.test.tsx +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-stop.test.tsx @@ -57,7 +57,8 @@ vi.mock('./use-structured-agent-session-sends', () => ({ import { AGENT_SESSION_ACCEPTED_SEND_RUNTIME_CAPABILITY, - AGENT_SESSION_CONVERSATION_STOP_RUNTIME_CAPABILITY + AGENT_SESSION_CONVERSATION_STOP_RUNTIME_CAPABILITY, + AGENT_SESSION_TARGETED_STOP_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version' import { setLocalRuntimeCapabilitiesForTests } from '@/runtime/local-runtime-capabilities' import { projectStructuredAgentSessionStatusSummary } from '../../../../shared/structured-agent-session-projection' @@ -153,7 +154,8 @@ describe('Stop against a host that stops the conversation', () => { beforeEach(() => { setLocalRuntimeCapabilitiesForTests([ AGENT_SESSION_ACCEPTED_SEND_RUNTIME_CAPABILITY, - AGENT_SESSION_CONVERSATION_STOP_RUNTIME_CAPABILITY + AGENT_SESSION_CONVERSATION_STOP_RUNTIME_CAPABILITY, + AGENT_SESSION_TARGETED_STOP_RUNTIME_CAPABILITY ]) }) @@ -412,6 +414,20 @@ describe('Stop against a host that stops the conversation', () => { expect(result.current.isWorking).toBe(false) expect(result.current.canStop).toBe(false) }) + it('sends the captured host target only when the host advertises targeted Stop', async () => { + setLocalRuntimeCapabilitiesForTests([ + AGENT_SESSION_CONVERSATION_STOP_RUNTIME_CAPABILITY, + AGENT_SESSION_TARGETED_STOP_RUNTIME_CAPABILITY + ]) + items = [RUNNING_TURN] + const { result } = render() + await act(async () => { + await result.current.stop() + }) + expect(cancels()).toEqual([ + expect.objectContaining({ stopTarget: { kind: 'turn', turnId: 'provider-turn' } }) + ]) + }) }) const EARLIER_TURN: AgentJournalRenderItem[] = [ @@ -481,7 +497,8 @@ describe('Stop against a host that stops the conversation, beside the sidebar', beforeEach(() => { setLocalRuntimeCapabilitiesForTests([ AGENT_SESSION_ACCEPTED_SEND_RUNTIME_CAPABILITY, - AGENT_SESSION_CONVERSATION_STOP_RUNTIME_CAPABILITY + AGENT_SESSION_CONVERSATION_STOP_RUNTIME_CAPABILITY, + AGENT_SESSION_TARGETED_STOP_RUNTIME_CAPABILITY ]) }) diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-stop.ts b/src/renderer/src/components/native-chat/use-structured-agent-session-stop.ts index f74e247d88ac..d733ae54512d 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session-stop.ts +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-stop.ts @@ -1,5 +1,10 @@ import type { RuntimeClientTarget } from '@/runtime/runtime-rpc-client' -import { useStructuredAgentSessionHostStopsConversation } from '@/runtime/structured-agent-session-host-capability' +import { AGENT_SESSION_TARGETED_STOP_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version' +import { agentSessionStopTarget } from '../../../../shared/agent-session-stop-target' +import { + useStructuredAgentSessionHostCapability, + useStructuredAgentSessionHostStopsConversation +} from '@/runtime/structured-agent-session-host-capability' import { agentStopDisplayStatus } from '../../../../shared/agent-stop-display-status' import type { StructuredAgentSessionMutate } from './use-structured-agent-session-mutate' import { useStructuredAgentSessionStopPress } from './use-structured-agent-session-stop-press' @@ -11,7 +16,7 @@ export function useStructuredAgentSessionStop(input: { target: RuntimeClientTarget transportState: Pick< ReturnType, - 'fence' | 'isWorking' + 'fence' | 'isWorking' | 'submissions' > /** The host says a person's Stop is still ending this session's work. */ hostStopping: boolean @@ -27,6 +32,10 @@ export function useStructuredAgentSessionStop(input: { } { const { sessionId, target, transportState, hostStopping, mutate } = input const press = useStructuredAgentSessionStopPress(sessionId) + const targetedStop = useStructuredAgentSessionHostCapability( + target, + AGENT_SESSION_TARGETED_STOP_RUNTIME_CAPABILITY + ) const stopsConversation = useStructuredAgentSessionHostStopsConversation(target) && transportState.fence !== null return { @@ -39,11 +48,22 @@ export function useStructuredAgentSessionStop(input: { pressed: press.pressed, stopsConversation, stop: (turnId, stopSends) => { + const stopTarget = agentSessionStopTarget( + turnId, + transportState.submissions, + transportState.fence + ) // The chat's own send goes no further if it has not gone out yet. Host-held cards are // never withdrawn by a Stop: the host pauses them, visible on every device, until acted on. if (stopsConversation) { stopSends() - return press.track(() => mutate('agentSession.cancel', 'agentSession.cancel', {})) + return press.track(() => + mutate( + 'agentSession.cancel', + 'agentSession.cancel', + stopTarget && targetedStop ? { stopTarget } : {} + ) + ) } if (!turnId) { return Promise.resolve(null) 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 95b025b2de38..3b616ed050f8 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 @@ -3,11 +3,9 @@ import { useStructuredAgentSessionSends } from './use-structured-agent-session-s import { useStructuredAgentSessionCommandWrite } from './use-structured-agent-session-command-write' import { structuredAgentSessionNewSendsQueue } from './structured-agent-session-queue-request' import type { AgentType } from '../../../../shared/agent-status-types' -import type { AgentSessionCancelResult } from '../../../../shared/agent-session-wire' import type { RuntimeClientTarget } from '@/runtime/runtime-rpc-client' import { structuredAgentLabel } from '@/lib/structured-agent-session-launch-label' import { useStructuredLegacyLaunchStop } from '@/lib/structured-agent-session-legacy-launch-stop' -import { supportsStructuredAgentSessionPromptCancel } from '@/runtime/structured-agent-session-client' import { useStructuredAgentSessionHostQueuesCommands, useStructuredAgentSessionHostEditsQueuedMessages, @@ -22,6 +20,7 @@ import { } from './structured-agent-session-message-projection' import { useStructuredAgentSessionMessages } from './use-structured-agent-session-messages' import { useStructuredAgentSessionTransportState } from './use-structured-agent-session-transport-state' +import { useStructuredAgentSessionCancel } from './use-structured-agent-session-cancel' import { useStructuredAgentSessionStop } from './use-structured-agent-session-stop' import { useStructuredAgentSessionTransport } from './use-structured-agent-session-transport' import { useStructuredAgentSessionOptions } from './use-structured-agent-session-options' @@ -43,8 +42,6 @@ import type { NativeChatRewindHost } from './use-native-chat-rewind' export type { StructuredPromptItem } from './structured-agent-session-message-projection' -type StructuredPromptCancelTarget = { itemId: string; expectedRevision: number } - export function useStructuredAgentSession(args: { sessionId: string target: RuntimeClientTarget @@ -126,6 +123,7 @@ export function useStructuredAgentSession(args: { () => (transportState.queuedMessages ?? []).map((message) => message.messageId), [transportState.queuedMessages] ) + const cancelActions = useStructuredAgentSessionCancel({ target, transportState, mutate }) const stopControl = useStructuredAgentSessionStop({ sessionId, target, @@ -296,22 +294,7 @@ export function useStructuredAgentSession(args: { /** A send made now while the agent works is held as a queued card: the host queues, and this * send asks it to (the setting is on and no pending prompt blocks the queue). */ sendsQueue: structuredAgentSessionNewSendsQueue(queue), - cancel: async (turnId: string | undefined, prompt?: StructuredPromptCancelTarget) => { - // Capability negotiation must complete before mutate fingerprints the payload: - // older hosts reject the strict prompt field. - const promptSupported = - prompt !== undefined && (await supportsStructuredAgentSessionPromptCancel(target)) - return mutate('agentSession.cancel', 'agentSession.cancel', { - ...(turnId ? { turnId } : {}), - ...(promptSupported ? { prompt } : {}) - }) - }, - stopBackgroundTask: (taskId?: string) => - mutate('agentSession.cancel', 'agentSession.cancel', { - turnId: 'background-tasks', - scope: 'background-tasks', - ...(taskId ? { taskId } : {}) - }), + ...cancelActions, respond: (item: StructuredPromptItem, response: AgentSessionPromptResponse) => respondToStructuredAgentSessionPrompt({ item, response, target, mutate }), optionSnapshot, diff --git a/src/shared/agent-child-work-invocation.ts b/src/shared/agent-child-work-invocation.ts new file mode 100644 index 000000000000..3ab5332b91f3 --- /dev/null +++ b/src/shared/agent-child-work-invocation.ts @@ -0,0 +1,11 @@ +export type AgentChildWorkInvocationFence = { + invocationId: string + generation: number +} + +export function agentChildWorkFencesEqual( + left: AgentChildWorkInvocationFence, + right: AgentChildWorkInvocationFence +): boolean { + return left.invocationId === right.invocationId && left.generation === right.generation +} diff --git a/src/shared/agent-session-background-stop-target.ts b/src/shared/agent-session-background-stop-target.ts new file mode 100644 index 000000000000..6ce4a7eaf644 --- /dev/null +++ b/src/shared/agent-session-background-stop-target.ts @@ -0,0 +1,44 @@ +import { agentChildWorkViewOffersStop } from './agent-child-work-stop-targets' +import { + agentChildWorkFencesEqual, + type AgentChildWorkInvocationFence +} from './agent-child-work-invocation' +import type { AgentChildWorkView } from './agent-status-child-work-view' + +export type AgentSessionBackgroundStopTarget = { + kind: 'background-tasks' + tasks: { id: string; invocation: AgentChildWorkInvocationFence }[] +} + +export function agentSessionBackgroundStopTarget( + views: readonly AgentChildWorkView[] | undefined, + taskId?: string +): AgentSessionBackgroundStopTarget { + return { + kind: 'background-tasks', + tasks: (views ?? []) + .filter( + (view) => + agentChildWorkViewOffersStop(view) && (taskId === undefined || view.providerId === taskId) + ) + .map((view) => ({ id: view.id, invocation: { ...view.invocation } })) + } +} + +/** Provider handles can be reused; the host's child identity and invocation must still match. */ +export function targetedAgentSessionBackgroundTaskIds( + views: readonly AgentChildWorkView[] | undefined, + target: AgentSessionBackgroundStopTarget, + taskId?: string +): string[] { + return (views ?? []).flatMap((view) => + agentChildWorkViewOffersStop(view) && + view.providerId !== undefined && + (taskId === undefined || view.providerId === taskId) && + target.tasks.some( + (task) => task.id === view.id && agentChildWorkFencesEqual(task.invocation, view.invocation) + ) + ? [view.providerId] + : [] + ) +} diff --git a/src/shared/agent-session-operation-ledger.ts b/src/shared/agent-session-operation-ledger.ts index 8c34edc7bcb0..31361291f1ee 100644 --- a/src/shared/agent-session-operation-ledger.ts +++ b/src/shared/agent-session-operation-ledger.ts @@ -271,6 +271,8 @@ export function evaluateAgentSessionOperation(args: { operationId: string fingerprint: string now: number + /** Targeted cleanup re-derives its effect from host state instead of replaying a receipt. */ + skipReceipt?: true }): AgentSessionOperationDecision { const { rows, callerKey, operationId, fingerprint, now } = args const operationTimestamp = parseAgentSessionOperationTimestamp(operationId) @@ -296,7 +298,7 @@ export function evaluateAgentSessionOperation(args: { details: { reason: 'operationIdReused' } } } - if (now - operationTimestamp > AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS) { + if (!args.skipReceipt && now - operationTimestamp > AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS) { // Why: once a tombstone could have expired, an unseen replay must never be reinterpreted as // permission to start another fresh agent. return { diff --git a/src/shared/agent-session-stop-capabilities.ts b/src/shared/agent-session-stop-capabilities.ts index c113161f96d8..7ef30d1ed4d0 100644 --- a/src/shared/agent-session-stop-capabilities.ts +++ b/src/shared/agent-session-stop-capabilities.ts @@ -14,9 +14,12 @@ export const AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY = // way itself there. Transitional: drop the client join once no supported host lacks this. export const AGENT_SESSION_REPEATED_STOP_RUNTIME_CAPABILITY = 'agent-session.repeated-stop.v1' as const +export const AGENT_SESSION_TARGETED_STOP_RUNTIME_CAPABILITY = + 'agent-session.targeted-stop.v1' as const export const AGENT_SESSION_STOP_RUNTIME_CAPABILITIES = [ AGENT_SESSION_CONVERSATION_STOP_RUNTIME_CAPABILITY, AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY, - AGENT_SESSION_REPEATED_STOP_RUNTIME_CAPABILITY + AGENT_SESSION_REPEATED_STOP_RUNTIME_CAPABILITY, + AGENT_SESSION_TARGETED_STOP_RUNTIME_CAPABILITY ] as const diff --git a/src/shared/agent-session-stop-target.test.ts b/src/shared/agent-session-stop-target.test.ts new file mode 100644 index 000000000000..d4ac9b988f41 --- /dev/null +++ b/src/shared/agent-session-stop-target.test.ts @@ -0,0 +1,102 @@ +import { expect, it } from 'vitest' +import { agentJournalItemKey } from './agent-session-journal-item-key' +import type { AgentJournalSubmission } from './agent-session-journal-types' +import type { AgentChildWorkView } from './agent-status-child-work-view' +import { agentSessionStopTarget, agentSessionStopTargetIsLive } from './agent-session-stop-target' +import { + agentSessionBackgroundStopTarget, + targetedAgentSessionBackgroundTaskIds +} from './agent-session-background-stop-target' + +const pending: AgentJournalSubmission = { + clientMessageId: 'send-1', + fence: 1, + payloadFingerprint: 'fp', + dispatchState: 'pending', + providerItemId: null, + reason: null, + submittedAt: 1, + resolvedAt: null +} + +it('follows an unanswered send into its own turn and excludes a later turn', () => { + const target = agentSessionStopTarget(null, [pending], 1) + if (!target) { + throw new Error('expected target') + } + expect(agentSessionStopTargetIsLive(target, null, [pending], 1)).toBe(true) + const accepted = { + ...pending, + dispatchState: 'accepted' as const, + providerItemId: agentJournalItemKey({ + provider: 'codex', + threadId: 'thread', + turnId: 'one', + ordinal: 0 + }) + } + expect(agentSessionStopTargetIsLive(target, 'one', [accepted], 1)).toBe(true) + expect(agentSessionStopTargetIsLive(target, 'two', [accepted], 1)).toBe(false) + expect( + agentSessionStopTargetIsLive(target, null, [{ ...pending, dispatchState: 'rejected' }], 1) + ).toBe(false) + expect(agentSessionStopTargetIsLive(target, 'one', [accepted], 2)).toBe(false) +}) + +it('never names a queued send, and a waiting send stays live whatever arrives after it', () => { + const handedOver = { ...pending, handoverRecorded: true as const, handedOverAt: 2 } + const queued = { ...pending, clientMessageId: 'send-2', handoverRecorded: true as const } + expect(agentSessionStopTarget(null, [handedOver, queued], 1)).toEqual({ + kind: 'submission', + clientMessageId: 'send-1' + }) + expect(agentSessionStopTarget(null, [queued], 1)).toBeUndefined() + const target = { kind: 'submission' as const, clientMessageId: 'send-1' } + const laterSent = { ...pending, clientMessageId: 'send-3' } + expect(agentSessionStopTargetIsLive(target, null, [handedOver, queued], 1)).toBe(true) + expect(agentSessionStopTargetIsLive(target, null, [handedOver, laterSent], 1)).toBe(true) +}) + +it('uses the host turn linkage for a provider whose item identity does not carry a turn id', () => { + const providerItemId = agentJournalItemKey({ + provider: 'legacy', + agent: 'pi', + sessionId: 'session', + recordId: 'user-1' + }) + expect( + agentSessionStopTargetIsLive( + { kind: 'submission', clientMessageId: 'send-1' }, + 'one', + [{ ...pending, providerItemId }], + 1, + providerItemId + ) + ).toBe(true) +}) + +it('cannot reuse an old background Stop against a new invocation with the same provider handle', () => { + const child: AgentChildWorkView = { + id: 'child-1', + providerId: 'task-1', + kind: 'agent', + state: 'working', + membership: 'live', + firstObservedAt: 1, + observedAt: 1, + stoppable: true, + invocation: { invocationId: 'invocation-1', generation: 1 } + } + const target = agentSessionBackgroundStopTarget([child]) + expect(targetedAgentSessionBackgroundTaskIds([child], target)).toEqual(['task-1']) + expect( + targetedAgentSessionBackgroundTaskIds([{ ...child, membership: 'settled' }], target) + ).toEqual([]) + expect( + targetedAgentSessionBackgroundTaskIds( + [{ ...child, invocation: { invocationId: 'invocation-1', generation: 2 } }], + target + ) + ).toEqual([]) + expect(targetedAgentSessionBackgroundTaskIds([{ ...child, id: 'child-2' }], target)).toEqual([]) +}) diff --git a/src/shared/agent-session-stop-target.ts b/src/shared/agent-session-stop-target.ts new file mode 100644 index 000000000000..6af05dd79ba0 --- /dev/null +++ b/src/shared/agent-session-stop-target.ts @@ -0,0 +1,66 @@ +import type { AgentJournalSubmission } from './agent-session-journal-types' +import { isUnansweredStructuredAgentSessionDispatch } from './structured-agent-session-unanswered-dispatch' +import { isQueuedAgentJournalSubmission } from './agent-session-queued-submission' +import { parseAgentJournalItemKey } from './agent-session-journal-item-key' +import type { AgentSessionBackgroundStopTarget } from './agent-session-background-stop-target' + +export type AgentSessionStopTarget = + | { kind: 'turn'; turnId: string } + | { kind: 'submission'; clientMessageId: string } + | AgentSessionBackgroundStopTarget + +/** Capture the host-published identity at the press, before capability discovery can await. */ +export function agentSessionStopTarget( + turnId: string | null, + submissions: readonly AgentJournalSubmission[], + fence: number | null +): AgentSessionStopTarget | undefined { + if (turnId) { + return { kind: 'turn', turnId } + } + // A queued send never started, so it is no Stop target: the unnamed Stop withdraws it. + const submission = submissions.findLast( + (entry) => + !isQueuedAgentJournalSubmission(entry) && + isUnansweredStructuredAgentSessionDispatch(entry, fence) + ) + return submission + ? { kind: 'submission', clientMessageId: submission.clientMessageId } + : undefined +} + +/** A submission target stops only work still awaiting its turn, never a later turn. Decided by the + * named send's own state: whatever was sent after it does not end it. */ +export function agentSessionStopTargetIsLive( + target: AgentSessionStopTarget, + activeTurnId: string | null, + submissions: readonly AgentJournalSubmission[], + fence: number, + activeUserItemId?: string +): boolean { + if (target.kind === 'turn') { + return target.turnId === activeTurnId + } + if (target.kind === 'background-tasks') { + return true + } + return submissions.some((entry) => { + if (entry.clientMessageId !== target.clientMessageId) { + return false + } + if (activeTurnId === null) { + return isUnansweredStructuredAgentSessionDispatch(entry, fence) + } + if (entry.providerItemId !== null && entry.providerItemId === activeUserItemId) { + return entry.fence === fence + } + const identity = entry.providerItemId ? parseAgentJournalItemKey(entry.providerItemId) : null + const turnId = + identity?.provider === 'codex' + ? identity.turnId + : identity?.provider === 'claude' + ? identity.uuid + : null + return turnId === activeTurnId && entry.fence === fence + }) +} diff --git a/src/shared/agent-status-child-work.ts b/src/shared/agent-status-child-work.ts index 1012d3e47a84..481eff9498d4 100644 --- a/src/shared/agent-status-child-work.ts +++ b/src/shared/agent-status-child-work.ts @@ -1,4 +1,9 @@ import { agentStatusSubjectsEqual, type AgentStatusSubject } from './agent-status-subject' +import type { AgentChildWorkInvocationFence } from './agent-child-work-invocation' +export { + agentChildWorkFencesEqual, + type AgentChildWorkInvocationFence +} from './agent-child-work-invocation' export const AGENT_CHILD_WORK_KINDS = [ 'agent', @@ -48,11 +53,6 @@ export type AgentChildWorkOperation = { observedAt: number } -export type AgentChildWorkInvocationFence = { - invocationId: string - generation: number -} - export type AgentChildWorkInvocationHistory = { fence: AgentChildWorkInvocationFence outcome?: AgentChildWorkOutcome @@ -109,13 +109,6 @@ export type AgentChildWorkRecord = AgentChildWorkInput & { revision: number } -export function agentChildWorkFencesEqual( - left: AgentChildWorkInvocationFence, - right: AgentChildWorkInvocationFence -): boolean { - return left.invocationId === right.invocationId && left.generation === right.generation -} - export function agentChildWorkBelongsTo( child: Pick, parent: AgentStatusSubject diff --git a/src/shared/rpc-contract/agent-session-stop-target-params.test.ts b/src/shared/rpc-contract/agent-session-stop-target-params.test.ts new file mode 100644 index 000000000000..74f7116e0c65 --- /dev/null +++ b/src/shared/rpc-contract/agent-session-stop-target-params.test.ts @@ -0,0 +1,62 @@ +import { expect, it } from 'vitest' +import { CancelParams } from './structured-agent-session-params' + +const envelope = { + sessionId: 'session-1', + clientOperationId: `1800000000000-${'a'.repeat(32)}`, + expectedRuntimeFence: 1, + payloadFingerprint: 'a'.repeat(64) +} + +it('keeps legacy Stop valid and accepts optional turn, submission, and task targets', () => { + expect(CancelParams.safeParse({ envelope }).success).toBe(true) + expect( + CancelParams.safeParse({ envelope, stopTarget: { kind: 'turn', turnId: 'one' } }).success + ).toBe(true) + expect( + CancelParams.safeParse({ + envelope, + stopTarget: { kind: 'submission', clientMessageId: 'send-1' } + }).success + ).toBe(true) + expect( + CancelParams.safeParse({ + envelope, + scope: 'background-tasks', + turnId: 'background-tasks', + stopTarget: { + kind: 'background-tasks', + tasks: [{ id: 'one', invocation: { invocationId: 'attempt', generation: 1 } }] + } + }).success + ).toBe(true) +}) + +it('refuses contradictory or oversized target payloads before execution', () => { + expect( + CancelParams.safeParse({ envelope, turnId: 'two', stopTarget: { kind: 'turn', turnId: 'one' } }) + .success + ).toBe(false) + expect( + CancelParams.safeParse({ + envelope, + scope: 'background-tasks', + turnId: 'one', + stopTarget: { kind: 'turn', turnId: 'one' } + }).success + ).toBe(false) + expect( + CancelParams.safeParse({ + envelope, + scope: 'background-tasks', + turnId: 'background-tasks', + stopTarget: { + kind: 'background-tasks', + tasks: Array.from({ length: 513 }, () => ({ + id: 'one', + invocation: { invocationId: 'attempt', generation: 1 } + })) + } + }).success + ).toBe(false) +}) diff --git a/src/shared/rpc-contract/agent-session-stop-target-params.ts b/src/shared/rpc-contract/agent-session-stop-target-params.ts new file mode 100644 index 000000000000..5c760ef36866 --- /dev/null +++ b/src/shared/rpc-contract/agent-session-stop-target-params.ts @@ -0,0 +1,49 @@ +import { z } from 'zod' +import { Identifier } from './structured-agent-session-identifiers' +import type { AgentSessionStopTarget } from '../agent-session-stop-target' + +export const StopTarget = z.discriminatedUnion('kind', [ + z.object({ kind: z.literal('turn'), turnId: Identifier('Invalid turn id') }).strict(), + z + .object({ kind: z.literal('submission'), clientMessageId: Identifier('Invalid message id') }) + .strict(), + z + .object({ + kind: z.literal('background-tasks'), + tasks: z + .array( + z + .object({ + id: Identifier('Invalid child id'), + invocation: z + .object({ + invocationId: Identifier('Invalid invocation id'), + generation: z.number().int().nonnegative() + }) + .strict() + }) + .strict() + ) + .max(512) + }) + .strict() +]) + +export function validateCancelStopTarget( + value: { stopTarget?: AgentSessionStopTarget; scope?: 'background-tasks'; turnId?: string }, + ctx: z.RefinementCtx +): void { + if ( + value.stopTarget && + (value.stopTarget.kind === 'background-tasks') !== (value.scope === 'background-tasks') + ) { + ctx.addIssue({ code: 'custom', message: 'Stop target must match its scope' }) + } + if ( + value.stopTarget?.kind === 'turn' && + value.turnId !== undefined && + value.turnId !== value.stopTarget.turnId + ) { + ctx.addIssue({ code: 'custom', message: 'Stop target must match its turn' }) + } +} diff --git a/src/shared/rpc-contract/structured-agent-session-params.ts b/src/shared/rpc-contract/structured-agent-session-params.ts index d9dd55c7b441..478fe5496e7f 100644 --- a/src/shared/rpc-contract/structured-agent-session-params.ts +++ b/src/shared/rpc-contract/structured-agent-session-params.ts @@ -1,3 +1,4 @@ +import { StopTarget, validateCancelStopTarget } from './agent-session-stop-target-params' import { SendBody, MAX_OPTION_LABEL, @@ -168,6 +169,7 @@ export const CancelParams = z turnId: Identifier('Invalid turn id').optional(), scope: z.literal('background-tasks').optional(), taskId: Identifier('Invalid task id').optional(), + stopTarget: StopTarget.optional(), prompt: z .object({ itemId: Identifier('Invalid item id'), @@ -178,6 +180,7 @@ export const CancelParams = z }) .strict() .superRefine((value, ctx) => { + validateCancelStopTarget(value, ctx) if (value.taskId !== undefined && value.scope !== 'background-tasks') { ctx.addIssue({ code: 'custom', message: 'A task id requires background-task scope' }) }