diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-codex-stop-row.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-codex-stop-row.test.ts index 1dbbac5282d8..eb00daafd545 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-codex-stop-row.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-codex-stop-row.test.ts @@ -8,6 +8,12 @@ import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it, vi, type MockInstance } from 'vitest' import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' +import { + createStructuredAgentSessionOutboxEntry, + structuredAgentSessionSendMutation +} from '../../../shared/structured-agent-session-outbox' +import { reconcileStructuredAgentSessionOutbox } from '../../../shared/structured-agent-session-outbox-reconcile' +import { admitStructuredAgentSessionOutboxEntry } from '../../../shared/structured-agent-session-outbox-admission' import { CodexAppServerRequestError } from '../../codex/codex-app-server-connection' import { CodexAppServerTimeoutError } from '../../codex/codex-app-server-session' import { @@ -534,3 +540,166 @@ describe('a message after a Codex Stop whose exit was unproven', () => { expect(host['sessions'].get(SESSION)?.lastEndedChild).toMatchObject({ rootGone: true }) }) }) + +// A follow-up steered into an open Codex turn and not yet echoed, when a Stop's interrupt fails. +// The desktop's outbox is fed the host's own rows, as its reconcile and drain read them: nothing +// the host keeps in doubt may hold the next message, which goes to a fresh Codex when the Stop +// ended the child, or to the same one when it did not. +describe('a follow-up Codex never echoed, when a Stop whose interrupt failed', () => { + /** A running turn, then a follow-up Codex took as a steer and has not echoed. */ + async function steeredFollowUp(): Promise { + await runningTurn() + codex.routes['turn/steer'] = turns.routes['turn/steer'] + const followUp = await send('and also this') + if (!followUp.ok) { + throw new Error(JSON.stringify(followUp.refusal)) + } + await vi.waitFor(() => + expect(codex.connections.at(-1)!.calls.some((call) => call.method === 'turn/steer')).toBe( + true + ) + ) + return followUp.value.clientMessageId + } + + function queued(clientMessageId: string, text: string) { + return createStructuredAgentSessionOutboxEntry({ + clientMessageId, + sessionId: SESSION, + text, + attachments: [], + queuedAt: NOW + }) + } + + async function followUpRow(followUpId: string) { + const { submissions } = await host.journalSnapshot(SESSION) + return submissions.find((row) => row.clientMessageId === followUpId) + } + + /** The desktop's queue as it held them, the follow-up on its way, read through the host's rows: + * the next message is admitted and reaches Codex as a turn of its own. Resolves with the queue + * as it stands once that message is on its way. */ + async function nextMessageGoesOut(followUpId: string, child: number) { + const { submissions } = await host.journalSnapshot(SESSION) + const next = queued(hostTestOperationId(), 'carry on') + const outbox = reconcileStructuredAgentSessionOutbox( + [{ ...queued(followUpId, 'and also this'), state: 'dispatching', lastAttemptAt: NOW }, next], + submissions, + [] + ) + expect(admitStructuredAgentSessionOutboxEntry(outbox)).toEqual({ + state: 'dispatch', + entry: next + }) + launch.resumeThreadId = THREAD + expect(await host.send(CALLER, structuredAgentSessionSendMutation(next, 1))).toMatchObject({ + ok: true + }) + await vi.waitFor(() => expect(codex.connections).toHaveLength(child + 1)) + await vi.waitFor(() => + expect( + codex.connections[child]!.calls.some( + (call) => call.method === 'turn/start' && JSON.stringify(call.params).includes('carry on') + ) + ).toBe(true) + ) + return outbox.map((entry) => + entry === next ? { ...entry, state: 'dispatching' as const, lastAttemptAt: NOW } : entry + ) + } + + /** Read through the host's rows as they stand now, the queue admits what is sent after. */ + async function queueStaysOpen(outbox: ReturnType) { + const { submissions } = await host.journalSnapshot(SESSION) + const later = queued(hostTestOperationId(), 'and then') + expect( + admitStructuredAgentSessionOutboxEntry( + reconcileStructuredAgentSessionOutbox([...outbox, later], submissions, []) + ) + ).toEqual({ state: 'dispatch', entry: later }) + } + + /** As the real connection: once a close begins it refuses every request, proven or not. The + * first close does not prove the exit. */ + function firstCloseUnproven(): void { + const old = codex.connections.at(-1)! + const close = old.close + const request = old.request + let unproven = 1 + old.close = async () => { + old.closed = true + if (unproven === 0) { + return close() + } + unproven -= 1 + return false + } + old.request = (method, params) => + old.closed + ? Promise.reject(new Error('codex app-server is closing')) + : request(method, params) + } + + it.each(['internal error', 'unanswered'] as const)( + 'stays in doubt when the Stop ends the child (%s), and the next message goes to a fresh Codex', + async (failure) => { + const followUpId = await steeredFollowUp() + codex.routes['turn/interrupt'] = () => { + throw interruptFailure(failure) + } + + await stop() + await host.flushStreamedEvents(SESSION) + expect(childEndedByStop()).toBe(true) + expect(await followUpRow(followUpId)).toMatchObject({ + dispatchState: 'unknown', + recovered: true + }) + await queueStaysOpen(await nextMessageGoesOut(followUpId, 1)) + } + ) + + // Codex marks the turn ended before it writes the refusal, so the turn's end follows it. + it('keeps the child when Codex had no turn to interrupt, and the next message goes to it', async () => { + const followUpId = await steeredFollowUp() + codex.routes['turn/interrupt'] = () => { + setTimeout(() => turns.end('completed'), 2) + throw refused('no active turn to interrupt') + } + + await stop() + await vi.waitFor(async () => expect((await journalRows()).turns).toEqual(['completed'])) + await host.flushStreamedEvents(SESSION) + expect(childEndedByStop()).toBe(false) + // Nothing settled it, so the desktop's entry stays on its way, which holds nothing up. + expect(await followUpRow(followUpId)).toMatchObject({ dispatchState: 'pending' }) + await queueStaysOpen(await nextMessageGoesOut(followUpId, 0)) + }) + + // The stop's unproven close is joined, and proven, by the next send, which settles the follow-up + // in doubt only then. + it.each(['internal error', 'unanswered'] as const)( + 'stays in doubt once a Stop that could not prove the exit (%s) is proven, and the queue never stops', + async (failure) => { + const followUpId = await steeredFollowUp() + codex.routes['turn/interrupt'] = () => { + throw interruptFailure(failure) + } + firstCloseUnproven() + + await stop() + await host.flushStreamedEvents(SESSION) + expect(host['sessions'].get(SESSION)?.child?.close).toBeDefined() + expect(await followUpRow(followUpId)).toMatchObject({ dispatchState: 'pending' }) + + const outbox = await nextMessageGoesOut(followUpId, 1) + expect(host['sessions'].get(SESSION)?.child?.close).toBeUndefined() + expect(await followUpRow(followUpId)).toMatchObject({ + dispatchState: 'unknown', + recovered: true + }) + await queueStaysOpen(outbox) + } + ) +}) diff --git a/src/renderer/src/components/native-chat/NativeChatStructuredSessionDelivery.probe.test.tsx b/src/renderer/src/components/native-chat/NativeChatStructuredSessionDelivery.probe.test.tsx index 4a89074d23fb..a60206a7bc01 100644 --- a/src/renderer/src/components/native-chat/NativeChatStructuredSessionDelivery.probe.test.tsx +++ b/src/renderer/src/components/native-chat/NativeChatStructuredSessionDelivery.probe.test.tsx @@ -119,7 +119,7 @@ describe('NativeChatStructuredSession delivery probe', () => { ) }, 20000) - it('parks a host-confirmed unknown instead of probing it', async () => { + it('never probes a host-confirmed unknown, and sends what follows it', async () => { useProbeClock() mocks.mode = 'outbox' mocks.call.mockRejectedValueOnce(new Error('socket closed')).mockResolvedValue({ @@ -147,7 +147,7 @@ describe('NativeChatStructuredSession delivery probe', () => { expect(mocks.call).toHaveBeenCalledOnce() const sent = mocks.call.mock.calls[0]?.[2] as { envelope: { clientOperationId: string } } - // The host now reports an unresolved unknown: another replay is the user's call. + // The host now reports an unresolved unknown: it never sends it again, so nothing waits on it. mocks.submissions = [ { clientMessageId: sent.envelope.clientOperationId, @@ -160,15 +160,16 @@ describe('NativeChatStructuredSession delivery probe', () => { resolvedAt: null } ] - // Queue a second message purely to re-render so the effect observes the - // new submissions; it must stay wedged behind the parked head. + // The second message also re-renders, so the effects observe the new submissions. await act(async () => { send?.('second', []) }) await advanceProbeClock(3000) - // From the row on, only the user's Retry moves it, so it says so. - expect(screen.getByText('Message delivery is unconfirmed.')).toBeTruthy() - expect(mocks.call).toHaveBeenCalledOnce() + const texts = mocks.call.mock.calls.map( + (call) => (call[2] as { body?: { blocks?: { text?: string }[] } })?.body?.blocks?.[0]?.text + ) + expect(texts).toEqual(['first', 'second']) + expect(screen.queryByText('Message delivery is unconfirmed.')).toBeNull() }, 20000) it('still probes while streaming batches rebuild the submissions array', async () => { diff --git a/src/renderer/src/components/native-chat/NativeChatStructuredSessionDelivery.test.tsx b/src/renderer/src/components/native-chat/NativeChatStructuredSessionDelivery.test.tsx index 791889c3370b..6a6e3835237e 100644 --- a/src/renderer/src/components/native-chat/NativeChatStructuredSessionDelivery.test.tsx +++ b/src/renderer/src/components/native-chat/NativeChatStructuredSessionDelivery.test.tsx @@ -313,7 +313,8 @@ describe('NativeChatStructuredSession delivery', () => { ['a recovered unknown', { recovered: true }], ["an older host's recovered unknown", { reason: 'host_restarted_before_acknowledgement' }] ])( - 'says a send reopened mid-send is unconfirmed, with its Retry, once the journal holds %s', + // Recorded in doubt, the host never sends it again: its row draws it, and nothing waits on it. + 'says nothing of a send reopened mid-send once the journal holds %s, and never resends it', async (label, patch) => { mocks.mode = 'outbox' const sessionId = `session-reopened-${label.replace(/\W+/g, '-')}` @@ -334,12 +335,12 @@ describe('NativeChatStructuredSession delivery', () => { renderSession(sessionId) - await waitFor(() => expect(screen.getByText('Message delivery is unconfirmed.')).toBeTruthy()) - expect(screen.getByRole('button', { name: /Retry/ })).toBeTruthy() - expect(screen.queryByText('Sending…')).toBeNull() await act(async () => { await new Promise((resolve) => setTimeout(resolve, 1500)) }) + expect(screen.queryByText('Message delivery is unconfirmed.')).toBeNull() + expect(screen.queryByRole('button', { name: /Retry/ })).toBeNull() + expect(screen.queryByText('Sending…')).toBeNull() expect(mocks.call).not.toHaveBeenCalled() }, 10000 diff --git a/src/renderer/src/components/native-chat/structured-agent-session-delivery-notices.rejected-retry.test.ts b/src/renderer/src/components/native-chat/structured-agent-session-delivery-notices.rejected-retry.test.ts new file mode 100644 index 000000000000..34e8ec6df350 --- /dev/null +++ b/src/renderer/src/components/native-chat/structured-agent-session-delivery-notices.rejected-retry.test.ts @@ -0,0 +1,87 @@ +// Which message the host recorded and then rejected has a Retry: only one no agent ever took, on a +// host that can queue that same message again. Every other one is drawn where it was rejected, in +// the host's words, and sending it again is a new message. + +import { describe, expect, it, vi } from 'vitest' +import { agentSessionFailureWords } from '../../../../shared/agent-session-failure-words' +import type { SubmissionRejectionFact } from '../../../../shared/agent-session-failure' +import { agentJournalSubmissionKey } from '../../../../shared/agent-session-journal-item-key' +import type { + AgentJournalRenderItem, + AgentJournalSubmission +} from '../../../../shared/agent-session-journal-types' +import { structuredAgentSessionDeliveryNotices } from './structured-agent-session-delivery-notices' +import { projectStructuredAgentSessionMessages } from './structured-agent-session-message-projection' + +const ID = 'op-message' +const KEY = agentJournalSubmissionKey(ID) +const ITEM: AgentJournalRenderItem = { + itemId: KEY, + revision: 0, + sequence: 5, + observedAt: 5, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hello' }] } +} + +function rejected( + fact: SubmissionRejectionFact, + patch: Partial = {} +): AgentJournalSubmission { + return { + clientMessageId: ID, + fence: 1, + payloadFingerprint: 'fingerprint', + dispatchState: 'rejected', + providerItemId: null, + submittedAt: 4, + resolvedAt: 7, + handoverRecorded: true, + ...agentSessionFailureWords(fact, { surface: 'rejection', agentName: 'Codex' }), + ...patch + } +} + +function noticeOn(submission: AgentJournalSubmission, retriesInPlace: boolean) { + const retry = vi.fn() + const notice = structuredAgentSessionDeliveryNotices( + [], + 'Codex', + retry, + [submission], + [], + new Set(), + [], + new Set(), + [], + retriesInPlace + ).get(KEY) + return { notice, retry } +} + +describe('a message the host recorded and then rejected', () => { + it("is drawn in place with the host's words and no Retry, even on a host that retries in place", () => { + // Handed to the agent, which never got it: the agent may have seen it, so no Retry. + const undelivered = rejected({ kind: 'notDelivered' }, { handedOverAt: 5 }) + + expect(projectStructuredAgentSessionMessages([ITEM], [], [undelivered], [])).toEqual([ + expect.objectContaining({ id: KEY, unsent: true }) + ]) + const { notice } = noticeOn(undelivered, true) + expect(notice).toEqual({ text: 'This message was not delivered. Send it again to continue.' }) + }) +}) + +describe('a failed start no agent took', () => { + const failedStart = rejected({ kind: 'providerStartFailed' }) + + it('gets Retry in place when the host advertises retry-message', () => { + const { notice, retry } = noticeOn(failedStart, true) + expect(notice?.text).toBe('Codex stopped before it finished starting.') + notice?.onRetry?.() + expect(retry).toHaveBeenCalledExactlyOnceWith(ID) + }) + + it('gets no Retry on a host that does not advertise retry-message', () => { + expect(noticeOn(failedStart, false).notice).not.toHaveProperty('onRetry') + }) +}) diff --git a/src/renderer/src/components/native-chat/structured-agent-session-delivery-notices.test.ts b/src/renderer/src/components/native-chat/structured-agent-session-delivery-notices.test.ts index 85fdf9de5225..6c696d216ccb 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-delivery-notices.test.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-delivery-notices.test.ts @@ -519,9 +519,8 @@ describe('the notice on each message that did not go through', () => { ] ])('says it is sending while your Retry is %s', (_label, patch, rowPatch) => { const doubt = row('m', 'unknown', rowPatch) - expect(texts([entry('m', { state: 'unconfirmed', lastAttemptAt: 1 })], [doubt])).toEqual({ - [agentJournalSubmissionKey('m')]: 'Message delivery is unconfirmed.' - }) + // Recorded in doubt, it leaves the outbox: its row draws it and nothing resends it. + expect(texts([entry('m', { state: 'unconfirmed', lastAttemptAt: 1 })], [doubt])).toEqual({}) const retried = entry('m', { lastAttemptAt: 1, retryAfterUnknownSubmittedAt: 7, ...patch }) expect(texts([retried], [doubt])).toEqual({ [agentJournalSubmissionKey('m')]: SENDING }) }) @@ -529,14 +528,14 @@ describe('the notice on each message that did not go through', () => { it.each([ ['a live unknown', {}], ['a recovered unknown', { recovered: true as const }] - ])('says the second of two in doubt is sending when each row holds %s', (_label, patch) => { - const a = entry('a', { state: 'unconfirmed' }) - const b = entry('b', { state: 'unconfirmed' }) - expect(texts([a, b], [row('a', 'unknown', patch), row('b', 'unknown', patch)])).toEqual({ - [agentJournalSubmissionKey('a')]: 'Message delivery is unconfirmed.', - [agentJournalSubmissionKey('b')]: SENDING - }) - }) + ])( + 'says nothing of two in doubt once each row holds %s: the rows draw them', + (_label, patch) => { + const a = entry('a', { state: 'unconfirmed' }) + const b = entry('b', { state: 'unconfirmed' }) + expect(texts([a, b], [row('a', 'unknown', patch), row('b', 'unknown', patch)])).toEqual({}) + } + ) // The rejection makes it the host's: not sent, in the host's words, whether its outbox copy or // its loaded row draws it. @@ -575,22 +574,6 @@ describe('the notice on each message that did not go through', () => { it.each([ ['rejected', entry('m', { state: 'rejected' }), []], ['held for its Retry', entry('m', { lastFailure: { kind: 'failed' } }), []], - [ - 'in doubt with a row the host holds', - entry('m', { state: 'unconfirmed', lastAttemptAt: 1 }), - [ - { - clientMessageId: 'm', - fence: 1, - payloadFingerprint: 'fp', - dispatchState: 'unknown', - providerItemId: null, - reason: null, - submittedAt: 1, - resolvedAt: null - } satisfies AgentJournalSubmission - ] - ], [ 'retried by the user', entry('m', { state: 'unconfirmed', retryAfterUnknownSubmittedAt: 1 }), @@ -653,22 +636,20 @@ describe('the notice on each message that did not go through', () => { "an older host's recovered unknown", row({ reason: 'host_restarted_before_acknowledgement' }) ] - ])('says it is unconfirmed, with its Retry, once the journal holds %s', (_label, answer) => { - const retry = vi.fn() - const notices = structuredAgentSessionDeliveryNotices( - [doubt()], - 'Claude', - retry, - [answer], - [], - NOT_FAILED_HERE - ) - expect(Object.fromEntries([...notices].map(([id, notice]) => [id, notice.text]))).toEqual( - UNCONFIRMED - ) - notices.get(agentJournalSubmissionKey('doubt'))?.onRetry?.() - expect(retry).toHaveBeenCalledExactlyOnceWith('doubt') - }) + ])( + 'says nothing once the journal holds %s: its row draws it, and no one resends it', + (_label, answer) => { + const notices = structuredAgentSessionDeliveryNotices( + [doubt()], + 'Claude', + vi.fn(), + [answer], + [], + NOT_FAILED_HERE + ) + expect(notices.size).toBe(0) + } + ) it.each([ ['the user already retried', { retryAfterUnknownSubmittedAt: 1 }], diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-outbox-admission.test.tsx b/src/renderer/src/components/native-chat/use-structured-agent-session-outbox-admission.test.tsx index a810a331a2cd..01ca844cd77e 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session-outbox-admission.test.tsx +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-outbox-admission.test.tsx @@ -305,7 +305,7 @@ describe('structured agent session outbox admission', () => { expect(heldIds(result.current.outbox)).toEqual([heldId]) }) - it('keeps a refused message held when an unconfirmed message ahead of it is retried', async () => { + it('keeps a refused message held when the message in doubt ahead of it leaves', async () => { const none: readonly AgentJournalSubmission[] = [] mocks.call.mockImplementation((_target, _method, params) => { const id = requestId(params) @@ -337,12 +337,10 @@ describe('structured agent session outbox admission', () => { dispatchState: 'unknown' } rerender({ submissions: [unknown] }) - await waitFor(() => expect(result.current.outbox[0]?.state).toBe('unconfirmed')) - - act(() => result.current.retry(firstId)) + // The host recorded it and never sends it again, so its entry goes; the refused one stays held. await waitFor(() => expect(result.current.outbox).toHaveLength(1)) await settleTimers(50) - expect(sentTexts()).toEqual(['first', 'held', 'first']) + expect(sentTexts()).toEqual(['first', 'held']) expect(heldIds(result.current.outbox)).toEqual([heldId]) }) }) diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-outbox-recorded-doubt.test.tsx b/src/renderer/src/components/native-chat/use-structured-agent-session-outbox-recorded-doubt.test.tsx new file mode 100644 index 000000000000..6d30995e0193 --- /dev/null +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-outbox-recorded-doubt.test.tsx @@ -0,0 +1,307 @@ +// @vitest-environment happy-dom + +// A message the journal records in doubt (its child ended before answering it) leaves the outbox, +// so nothing waits on it, now or after a reopen whose loaded rows no longer reach it. A queue an +// earlier build saved stuck behind one is freed once the chat opens again, and the held message is +// never sent again. + +import { act, cleanup, renderHook, waitFor } from '@testing-library/react' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +import { AGENT_SESSION_ACCEPTED_SEND_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version' +import type { + AgentJournalRenderItem, + AgentJournalSubmission +} from '../../../../shared/agent-session-journal-types' + +const mocks = vi.hoisted(() => ({ call: vi.fn() })) + +vi.mock('@/runtime/structured-agent-session-client', () => ({ + callStructuredAgentSession: mocks.call +})) + +import { setLocalRuntimeCapabilitiesForTests } from '@/runtime/local-runtime-capabilities' +import { useStructuredAgentSessionOutbox } from './use-structured-agent-session-outbox' +import { + createStructuredAgentSessionOutboxEntry, + type StructuredAgentSessionOutboxEntry +} from '../../../../shared/structured-agent-session-outbox' +import { writeOutbox } from './structured-agent-session-outbox-storage' + +const SESSION = 'session-1' +const LOCAL_TARGET = { kind: 'local' } as const +const NO_JOURNAL_ITEMS: readonly AgentJournalRenderItem[] = [] + +type SendRequest = { envelope?: { clientOperationId?: string } } +type SentText = { body?: { blocks?: { text?: string }[] } } + +function requestId(params: SendRequest | undefined): string { + return String(params?.envelope?.clientOperationId) +} + +function sentIds(): string[] { + return mocks.call.mock.calls.map((call) => requestId(call[2])) +} + +function saved( + clientMessageId: string, + patch: Partial = {} +): StructuredAgentSessionOutboxEntry { + return { + ...createStructuredAgentSessionOutboxEntry({ + clientMessageId, + sessionId: SESSION, + text: clientMessageId, + attachments: [], + queuedAt: 1 + }), + ...patch + } +} + +/** What the host wrote when the child ended with the follow-up unanswered. */ +const IN_DOUBT: AgentJournalSubmission = { + clientMessageId: 'op-follow-up', + fence: 1, + payloadFingerprint: 'fingerprint', + dispatchState: 'unknown', + providerItemId: null, + reason: 'provider_closed_before_acknowledgement', + submittedAt: 2, + resolvedAt: 3, + recovered: true +} + +/** The host's answer to a send it recorded in doubt, first try or replay. */ +function inDoubt(clientMessageId: string) { + return { + ok: true, + replayed: clientMessageId === IN_DOUBT.clientMessageId, + fence: 1, + cursor: { epoch: 'epoch-1', sequence: 4 }, + value: { clientMessageId, submission: { ...IN_DOUBT, clientMessageId } } + } +} + +function accepted(clientMessageId: string) { + return { + ok: true, + replayed: false, + fence: 1, + cursor: { epoch: 'epoch-1', sequence: 4 }, + value: { + clientMessageId, + submission: { + clientMessageId, + fence: 1, + payloadFingerprint: 'fingerprint', + dispatchState: 'accepted', + providerItemId: `provider-${clientMessageId}`, + reason: null, + submittedAt: 4, + resolvedAt: 4 + } + } + } +} + +describe('a queue saved behind a message the host holds in doubt', () => { + afterEach(() => { + cleanup() + setLocalRuntimeCapabilitiesForTests(null) + }) + + beforeEach(() => { + vi.clearAllMocks() + localStorage.clear() + setLocalRuntimeCapabilitiesForTests([AGENT_SESSION_ACCEPTED_SEND_RUNTIME_CAPABILITY]) + // The host answers a replay of the recorded follow-up with its doubt, and takes anything else. + mocks.call.mockImplementation((_target, _method, params: SendRequest) => + Promise.resolve( + requestId(params) === IN_DOUBT.clientMessageId + ? inDoubt(requestId(params)) + : accepted(requestId(params)) + ) + ) + }) + + it.each([ + ['held as unconfirmed', 'unconfirmed'], + ['left on its way out', 'dispatching'] + ] as const)( + 'sends the message behind it when the chat opens again, the held one %s', + async (_case, state) => { + writeOutbox(SESSION, [saved('op-follow-up', { state, lastAttemptAt: 2 }), saved('op-next')]) + + const { result } = renderHook(() => + useStructuredAgentSessionOutbox({ + journalItems: NO_JOURNAL_ITEMS, + sessionId: SESSION, + target: LOCAL_TARGET, + fence: 1, + submissions: [IN_DOUBT] + }) + ) + + await waitFor(() => expect(sentIds()).toEqual(['op-next'])) + // Past the unconfirmed probe's first delay: the held message is never sent again. + await act(() => new Promise((resolve) => setTimeout(resolve, 1_200))) + expect(sentIds()).toEqual(['op-next']) + expect(result.current.outbox).toEqual([]) + } + ) + + // Its rows can age out of what a reopen loads; the entry left with the first reconcile. + it('never holds a send after a reopen whose loaded rows no longer reach it', async () => { + writeOutbox(SESSION, [saved('op-follow-up', { state: 'dispatching', lastAttemptAt: 2 })]) + const before = renderHook(() => + useStructuredAgentSessionOutbox({ + journalItems: NO_JOURNAL_ITEMS, + sessionId: SESSION, + target: LOCAL_TARGET, + fence: 1, + submissions: [IN_DOUBT] + }) + ) + await waitFor(() => expect(before.result.current.outbox).toEqual([])) + before.unmount() + + const after = renderHook(() => + useStructuredAgentSessionOutbox({ + journalItems: NO_JOURNAL_ITEMS, + sessionId: SESSION, + target: LOCAL_TARGET, + fence: 1, + submissions: [] + }) + ) + act(() => expect(after.result.current.send('next')).toBe(true)) + await waitFor(() => expect(after.result.current.outbox).toEqual([])) + expect(sentIds()).toHaveLength(1) + expect(sentIds()).not.toContain(IN_DOUBT.clientMessageId) + }) + + // A client that never loaded the row: the probe's replay is the host's answer that it has it. + it('frees a queue whose message in doubt was recorded out of sight, by the replay', async () => { + writeOutbox(SESSION, [ + saved('op-follow-up', { state: 'unconfirmed', lastAttemptAt: 2 }), + saved('op-next') + ]) + const { result } = renderHook(() => + useStructuredAgentSessionOutbox({ + journalItems: NO_JOURNAL_ITEMS, + sessionId: SESSION, + target: LOCAL_TARGET, + fence: 1, + submissions: [] + }) + ) + + await waitFor(() => expect(sentIds()).toEqual(['op-follow-up', 'op-next']), { + timeout: 5_000 + }) + await waitFor(() => expect(result.current.outbox).toEqual([])) + expect(result.current.error).toBeNull() + }) + + // The one the host holds leaves; the probe resends the one it may not have. + it('probes a message the host may never have received, saved behind one it holds', async () => { + writeOutbox(SESSION, [ + saved('op-follow-up', { state: 'unconfirmed', lastAttemptAt: 2 }), + saved('op-lost', { state: 'unconfirmed', lastAttemptAt: 3 }), + saved('op-next') + ]) + + renderHook(() => + useStructuredAgentSessionOutbox({ + journalItems: NO_JOURNAL_ITEMS, + sessionId: SESSION, + target: LOCAL_TARGET, + fence: 1, + submissions: [IN_DOUBT] + }) + ) + + await waitFor(() => expect(sentIds()).toEqual(['op-lost', 'op-next']), { timeout: 5_000 }) + }) + + it('holds the queue until the journal shows the host has it, then sends what waits', async () => { + // Retried once already, so the probe leaves it alone and only the journal can free the queue. + writeOutbox(SESSION, [ + saved('op-follow-up', { + state: 'unconfirmed', + lastAttemptAt: 2, + retryAfterUnknownSubmittedAt: -1 + }), + saved('op-next') + ]) + const { rerender } = renderHook( + ({ submissions }: { submissions: readonly AgentJournalSubmission[] }) => + useStructuredAgentSessionOutbox({ + journalItems: NO_JOURNAL_ITEMS, + sessionId: SESSION, + target: LOCAL_TARGET, + fence: 1, + submissions + }), + { initialProps: { submissions: [] as readonly AgentJournalSubmission[] } } + ) + await act(() => new Promise((resolve) => setTimeout(resolve, 100))) + expect(sentIds()).toEqual([]) + + rerender({ submissions: [IN_DOUBT] }) + await waitFor(() => expect(sentIds()).toEqual(['op-next'])) + }) + + it('drops a send the host answers in doubt, and sends what follows it at once', async () => { + mocks.call.mockImplementation((_target, _method, params: SendRequest & SentText) => + Promise.resolve( + params.body?.blocks?.[0]?.text === 'first' + ? inDoubt(requestId(params)) + : accepted(requestId(params)) + ) + ) + const { result } = renderHook(() => + useStructuredAgentSessionOutbox({ + journalItems: NO_JOURNAL_ITEMS, + sessionId: SESSION, + target: LOCAL_TARGET, + fence: 1, + submissions: [] + }) + ) + + act(() => expect(result.current.send('first')).toBe(true)) + await waitFor(() => expect(result.current.outbox).toEqual([])) + act(() => expect(result.current.send('second')).toBe(true)) + await waitFor(() => expect(result.current.outbox).toEqual([])) + expect(sentIds()).toHaveLength(2) + expect(result.current.error).toBeNull() + }) + + // A Retry an earlier build saved on this very submission waits for its answer: the host refuses + // to send it again, and the user is told to check the chat before sending it again. + it('drops a saved Retry of a recorded doubt once the host only replays it', async () => { + writeOutbox(SESSION, [ + saved(IN_DOUBT.clientMessageId, { + lastAttemptAt: 2, + retryAfterUnknownSubmittedAt: IN_DOUBT.submittedAt + }) + ]) + const { result } = renderHook(() => + useStructuredAgentSessionOutbox({ + journalItems: NO_JOURNAL_ITEMS, + sessionId: SESSION, + target: LOCAL_TARGET, + fence: 1, + submissions: [IN_DOUBT] + }) + ) + + await waitFor(() => expect(result.current.outbox).toEqual([])) + expect(sentIds()).toEqual([IN_DOUBT.clientMessageId]) + expect(result.current.error).toBe( + "Orca couldn't confirm your message reached the agent. Check the chat, then send it again if needed." + ) + }) +}) diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-outbox.test.tsx b/src/renderer/src/components/native-chat/use-structured-agent-session-outbox.test.tsx index f9957ed41cc9..d58f6e15d163 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session-outbox.test.tsx +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-outbox.test.tsx @@ -91,28 +91,6 @@ function acceptedResultFor(clientMessageId: string, fence: number) { } } -function unknownResultFor(clientMessageId: string, submittedAt: number) { - return { - ok: true, - replayed: false, - fence: 1, - cursor: { epoch: 'epoch-1', sequence: submittedAt }, - value: { - clientMessageId, - submission: { - clientMessageId, - fence: 1, - payloadFingerprint: 'fingerprint', - dispatchState: 'unknown' as const, - providerItemId: null, - reason: 'socket closed', - submittedAt, - resolvedAt: submittedAt - } - } - } -} - function pendingResultFor(clientMessageId: string, submittedAt: number) { return { ok: true, @@ -480,60 +458,6 @@ describe('useStructuredAgentSessionOutbox', () => { expect(mocks.call).toHaveBeenCalledOnce() }) - it('drains a head the host refuses to redeliver so the queue behind it advances', async () => { - // The guard refuses a retry it cannot prove is a first delivery. That must - // not leave a Retry that does nothing in front of a wedged queue: the entry - // leaves the outbox, the user is told to check the chat before sending it again, and the - // message queued behind it goes out. - // The second send never settles, so the refusal's error is still on screen - // when the queue behind it advances. - mocks.call.mockImplementation(async (_target, _method, params) => { - const request = params as { - envelope: { clientOperationId: string } - body: { blocks: { text?: string }[] } - } - if (request.body.blocks[0]?.text === 'second') { - return new Promise(() => {}) - } - return unknownResultFor(request.envelope.clientOperationId, 10) - }) - const { result, rerender } = renderHook( - ({ submissions }: { submissions: readonly AgentJournalSubmission[] }) => - useStructuredAgentSessionOutbox({ - journalItems: NO_JOURNAL_ITEMS, - sessionId: 'session-1', - target: LOCAL_TARGET, - fence: 1, - submissions - }), - { initialProps: { submissions: [] as readonly AgentJournalSubmission[] } } - ) - - act(() => expect(result.current.send('first')).toBe(true)) - await waitFor(() => expect(result.current.outbox[0]?.state).toBe('unconfirmed')) - const firstId = result.current.outbox[0]!.clientMessageId - rerender({ submissions: [unknownResultFor(firstId, 10).value.submission] }) - - act(() => expect(result.current.send('second')).toBe(true)) - expect(result.current.outbox).toHaveLength(2) - - act(() => result.current.retry(firstId)) - await waitFor(() => - expect(result.current.outbox.some((entry) => entry.clientMessageId === firstId)).toBe(false) - ) - await act(async () => { - await new Promise((resolve) => setTimeout(resolve, 200)) - }) - - const sent = mocks.call.mock.calls.map( - (call) => (call[2] as { body?: { blocks?: { text?: string }[] } })?.body?.blocks?.[0]?.text - ) - expect(sent).toContain('second') - expect(result.current.error).toBe( - "Orca couldn't confirm your message reached the agent. Check the chat, then send it again if needed." - ) - }) - it('retains a send operation after a pending-admission refusal', async () => { mocks.call .mockResolvedValueOnce(refusedResult('agent_session_checkpoint_stale')) @@ -632,27 +556,18 @@ describe('useStructuredAgentSessionOutbox', () => { }) }) - it('retries an unknown head and advances a queued tail', async () => { + it('drops a head in transport doubt once the journal records it, and sends the tail', async () => { // oxlint-disable-next-line no-restricted-properties -- stubbing the global the generator reads, to pin ids in this test vi.mocked(globalThis.crypto.randomUUID) .mockReturnValueOnce('aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa') .mockReturnValueOnce('bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb') mocks.call - .mockImplementationOnce(async (_target, _method, params) => { - const clientMessageId = (params as { envelope: { clientOperationId: string } }).envelope - .clientOperationId - return unknownResultFor(clientMessageId, 10) - }) + .mockRejectedValueOnce(new Error('socket closed')) .mockImplementationOnce(async (_target, _method, params) => { const clientMessageId = (params as { envelope: { clientOperationId: string } }).envelope .clientOperationId return acceptedResultFor(clientMessageId, 11) }) - .mockImplementationOnce(async (_target, _method, params) => { - const clientMessageId = (params as { envelope: { clientOperationId: string } }).envelope - .clientOperationId - return acceptedResultFor(clientMessageId, 12) - }) const { result, rerender } = renderHook( ({ submissions }: { submissions: readonly AgentJournalSubmission[] }) => useStructuredAgentSessionOutbox({ @@ -670,6 +585,12 @@ describe('useStructuredAgentSessionOutbox', () => { }) await waitFor(() => expect(result.current.outbox[0]?.state).toBe('unconfirmed')) const firstId = result.current.outbox[0]!.clientMessageId + act(() => { + expect(result.current.send('second')).toBe(true) + }) + // The host may never have received it, so the tail waits. + expect(result.current.outbox).toHaveLength(2) + expect(mocks.call).toHaveBeenCalledOnce() rerender({ submissions: [ { @@ -684,25 +605,16 @@ describe('useStructuredAgentSessionOutbox', () => { } ] }) - act(() => { - expect(result.current.send('second')).toBe(true) - }) - expect(result.current.outbox).toHaveLength(2) - act(() => result.current.retry(firstId)) - await waitFor(() => expect(mocks.call).toHaveBeenCalledTimes(3)) + // The host recorded it and never sends it again: it goes, and the tail goes out unasked. + await waitFor(() => expect(mocks.call).toHaveBeenCalledTimes(2)) await waitFor(() => expect(result.current.outbox).toHaveLength(0)) - const retryParams = mocks.call.mock.calls[1]?.[2] as { retryUnknown?: true } | undefined - expect(retryParams?.retryUnknown).toBeUndefined() + expect(mocks.call.mock.calls[1]?.[2]?.body?.blocks?.[0]?.text).toBe('second') }) it('lets a history-rejected unknown head go to the journal so the queued tail advances', async () => { mocks.call - .mockImplementationOnce(async (_target, _method, params) => { - const clientMessageId = (params as { envelope: { clientOperationId: string } }).envelope - .clientOperationId - return unknownResultFor(clientMessageId, 10) - }) + .mockRejectedValueOnce(new Error('socket closed')) .mockImplementationOnce(async (_target, _method, params) => { const clientMessageId = (params as { envelope: { clientOperationId: string } }).envelope .clientOperationId @@ -751,11 +663,7 @@ describe('useStructuredAgentSessionOutbox', () => { }) it('loads the new session outbox when a pane switches sessions', async () => { - mocks.call.mockImplementationOnce(async (_target, _method, params) => { - const clientMessageId = (params as { envelope: { clientOperationId: string } }).envelope - .clientOperationId - return unknownResultFor(clientMessageId, 10) - }) + mocks.call.mockRejectedValueOnce(new Error('socket closed')) const { result, rerender } = renderHook( ({ sessionId }: { sessionId: string }) => diff --git a/src/shared/structured-agent-session-outbox-admission.ts b/src/shared/structured-agent-session-outbox-admission.ts index ec12333fe07a..edc3945c8de9 100644 --- a/src/shared/structured-agent-session-outbox-admission.ts +++ b/src/shared/structured-agent-session-outbox-admission.ts @@ -23,9 +23,11 @@ export type StructuredAgentSessionOutboxAdmission = * per-session serialize chain before dispatching, so it keeps its place, and waiting for its echo * costs delivery of everything queued behind it. One whose start was refused before it ran is the * exception the host makes: it waits for its next try while the messages behind it go first. An - * `unconfirmed` entry is a barrier — sending past it would reorder around a message that may yet - * land. One the user was told did not go through is not: it lands only by its own Retry, so what - * the user sends after it goes out as they send it. + * `unconfirmed` entry is one the host may never have received, and a barrier — sending past it + * would reorder around a message that may yet land. Once the journal records it in doubt it leaves + * the outbox (the reconcile): its place is fixed and the host never sends it again. One the user + * was told did not go through is not a barrier either: it lands only by its own Retry, so what the + * user sends after it goes out as they send it. */ export function admitStructuredAgentSessionOutboxEntry( entries: readonly StructuredAgentSessionOutboxEntry[] diff --git a/src/shared/structured-agent-session-outbox-reconcile.test.ts b/src/shared/structured-agent-session-outbox-reconcile.test.ts index 0336bd649e40..ef89345328ea 100644 --- a/src/shared/structured-agent-session-outbox-reconcile.test.ts +++ b/src/shared/structured-agent-session-outbox-reconcile.test.ts @@ -83,7 +83,7 @@ it('returns every kept entry, and the list, as themselves when read again', () = ['queued', 'queued'], ['landing', 'dispatching'], ['sending', 'dispatching'], - ['doubt', 'unconfirmed'], + // Recorded in doubt: its row draws it and nothing resends it, so it leaves. ['rejected-unloaded', 'rejected'], ['refused', 'rejected'] ]) @@ -93,7 +93,11 @@ it('returns every kept entry, and the list, as themselves when read again', () = }) it('returns the list it was given when nothing changes', () => { - const entries = [entry('queued'), entry('doubt', { state: 'unconfirmed' })] + // The user's Retry of this very submission in doubt waits for its answer, unchanged. + const entries = [ + entry('queued'), + entry('doubt', { state: 'unconfirmed', retryAfterUnknownSubmittedAt: 5 }) + ] expect(reconcileStructuredAgentSessionOutbox(entries, [submission('doubt', 'unknown')], [])).toBe( entries ) diff --git a/src/shared/structured-agent-session-outbox-reconcile.ts b/src/shared/structured-agent-session-outbox-reconcile.ts index 9fa699eece1f..d370d005c91e 100644 --- a/src/shared/structured-agent-session-outbox-reconcile.ts +++ b/src/shared/structured-agent-session-outbox-reconcile.ts @@ -49,17 +49,14 @@ export function reconcileStructuredAgentSessionOutbox( const { lastFailure: _landed, ...landed } = entry return [{ ...landed, state: 'dispatching' as const }] } + // Recorded, but whether the agent got it is unknown: its journal row draws it in place and the + // host never sends it again, so the entry goes. Only a Retry the user asked of this very + // submission stays, for its answer. if ( submission?.dispatchState === 'unknown' && - entry.retryAfterUnknownSubmittedAt !== -1 && entry.retryAfterUnknownSubmittedAt !== submission.submittedAt ) { - // In doubt now, not failed: the probe's resend decides it, as for any unconfirmed send. - if (entry.state === 'unconfirmed' && entry.lastFailure === undefined) { - return [entry] - } - const { lastFailure: _superseded, ...inDoubt } = entry - return [{ ...inDoubt, state: 'unconfirmed' as const }] + return [] } return [entry] }) diff --git a/src/shared/structured-agent-session-outbox-recorded-doubt.test.ts b/src/shared/structured-agent-session-outbox-recorded-doubt.test.ts new file mode 100644 index 000000000000..58fdec366f77 --- /dev/null +++ b/src/shared/structured-agent-session-outbox-recorded-doubt.test.ts @@ -0,0 +1,141 @@ +// A message whose delivery is in doubt holds the queue only while the host may never have received +// it. Once the host records it in doubt, its place is fixed and the host never sends it again, so +// its entry leaves the outbox and what follows it goes out. + +import { describe, expect, it } from 'vitest' +import type { AgentJournalSubmission } from './agent-session-journal-types' +import { + createStructuredAgentSessionOutboxEntry, + type StructuredAgentSessionOutboxEntry +} from './structured-agent-session-outbox' +import { reconcileStructuredAgentSessionOutbox } from './structured-agent-session-outbox-reconcile' +import { admitStructuredAgentSessionOutboxEntry } from './structured-agent-session-outbox-admission' +import { disposeStructuredAgentSessionSendResult } from './structured-agent-session-send-disposition' + +function entry( + clientMessageId: string, + patch: Partial = {} +): StructuredAgentSessionOutboxEntry { + return { + ...createStructuredAgentSessionOutboxEntry({ + clientMessageId, + sessionId: 'session-1', + text: clientMessageId, + attachments: [], + queuedAt: 1 + }), + ...patch + } +} + +function unknown( + clientMessageId: string, + patch: Partial = {} +): AgentJournalSubmission { + return { + clientMessageId, + fence: 1, + payloadFingerprint: 'fingerprint', + dispatchState: 'unknown', + providerItemId: null, + reason: 'provider_closed_before_acknowledgement', + submittedAt: 5, + resolvedAt: 6, + recovered: true, + ...patch + } +} + +function answered( + submission: AgentJournalSubmission, + entries: readonly StructuredAgentSessionOutboxEntry[] +) { + return disposeStructuredAgentSessionSendResult({ + entries, + entry: entries[0]!, + result: { + ok: true, + replayed: true, + fence: 1, + cursor: { epoch: 'epoch-1', sequence: 7 }, + value: { clientMessageId: submission.clientMessageId, submission } + }, + createOperationId: () => 'unused' + }) +} + +describe('a message the host records in doubt', () => { + it.each<[string, Partial, Partial]>([ + ['on its way, the doubt outliving its writer', { state: 'dispatching' }, {}], + [ + 'already unconfirmed, the doubt still outstanding', + { state: 'unconfirmed' }, + { recovered: undefined } + ], + [ + 'retried before the record was seen', + { state: 'unconfirmed', retryAfterUnknownSubmittedAt: -1 }, + {} + ] + ])('leaves the outbox when the journal records it, %s', (_case, entryPatch, recorded) => { + const outbox = reconcileStructuredAgentSessionOutbox( + [entry('doubt', { lastAttemptAt: 2, ...entryPatch }), entry('next')], + [unknown('doubt', recorded)], + [] + ) + expect(outbox.map((candidate) => candidate.clientMessageId)).toEqual(['next']) + expect(admitStructuredAgentSessionOutboxEntry(outbox)).toEqual({ + state: 'dispatch', + entry: outbox[0] + }) + }) + + it('leaves the outbox when the host answers a send or its replay in doubt', () => { + for (const recovered of [true, undefined] as const) { + const { entries, error } = answered(unknown('doubt', { recovered }), [ + entry('doubt', { state: 'dispatching', lastAttemptAt: 2 }), + entry('next') + ]) + expect(entries.map((candidate) => candidate.clientMessageId)).toEqual(['next']) + expect(error).toBeNull() + } + }) + + // Only this entry shows a message whose row the host lost, so it stays, as before. + it('stays when the host lost its row', () => { + const { entries } = answered(unknown('doubt', { reason: 'durable_send_submission_missing' }), [ + entry('doubt', { state: 'dispatching', lastAttemptAt: 2 }) + ]) + expect(entries).toMatchObject([{ state: 'unconfirmed', retryAfterUnknownSubmittedAt: -1 }]) + }) + + // A Retry asked of this very submission waits for the host's answer to it. + it('stays while the user retries that same submission', () => { + const retried = entry('doubt', { lastAttemptAt: 2, retryAfterUnknownSubmittedAt: 5 }) + expect(reconcileStructuredAgentSessionOutbox([retried], [unknown('doubt')], [])).toEqual([ + retried + ]) + }) +}) + +describe('a message the host may never have received', () => { + it('holds what follows it until the host answers it', () => { + const outbox = reconcileStructuredAgentSessionOutbox( + [entry('doubt', { state: 'unconfirmed', lastAttemptAt: 2 }), entry('next')], + // Another message's row, or the window of rows loaded after this one's aged out. + [unknown('other')], + [] + ) + expect(admitStructuredAgentSessionOutboxEntry(outbox)).toEqual({ + state: 'blocked', + entry: outbox[0] + }) + // The probe's replay of it is the host's answer: it recorded it, and what follows goes. + const { entries } = answered(unknown('doubt'), outbox) + expect(admitStructuredAgentSessionOutboxEntry(entries)).toEqual({ + state: 'dispatch', + entry: entries[0] + }) + expect(entries.map((candidate) => candidate.clientMessageId)).toEqual(['next']) + }) +}) diff --git a/src/shared/structured-agent-session-outbox-retry-hold.test.ts b/src/shared/structured-agent-session-outbox-retry-hold.test.ts index cd2bb570b57a..2a44fef2d67f 100644 --- a/src/shared/structured-agent-session-outbox-retry-hold.test.ts +++ b/src/shared/structured-agent-session-outbox-retry-hold.test.ts @@ -79,7 +79,7 @@ describe('a message held for its Retry', () => { }) // Any fresh word on where a message stands supersedes an earlier attempt's failure. - it('is in doubt, and holds the queue, once the host says it cannot tell whether it landed', () => { + it('leaves the outbox once the host records that it cannot tell whether it landed', () => { const submission: AgentJournalSubmission = { clientMessageId: 'held', fence: 1, @@ -95,11 +95,11 @@ describe('a message held for its Retry', () => { [submission], [] ) - expect(reconciled[0]).toMatchObject({ state: 'unconfirmed' }) - expect(reconciled[0]?.lastFailure).toBeUndefined() + // Its journal row draws it, and the host never sends it again, so nothing waits on it. + expect(reconciled.map((entry) => entry.clientMessageId)).toEqual(['next']) expect(admitStructuredAgentSessionOutboxEntry(reconciled)).toMatchObject({ - state: 'blocked', - entry: { clientMessageId: 'held' } + state: 'dispatch', + entry: { clientMessageId: 'next' } }) }) }) diff --git a/src/shared/structured-agent-session-outbox-withdrawn-send.test.ts b/src/shared/structured-agent-session-outbox-withdrawn-send.test.ts index 214ed97872aa..f8e1d0102efa 100644 --- a/src/shared/structured-agent-session-outbox-withdrawn-send.test.ts +++ b/src/shared/structured-agent-session-outbox-withdrawn-send.test.ts @@ -1,5 +1,6 @@ // A send the host withdrew (a Stop took it back before it ran) leaves the outbox with no notice, so -// the next queued send goes out. One the host left in doubt holds the queue until its Retry. +// the next queued send goes out. One the host recorded in doubt leaves it too: its row draws it, and +// the host never sends it again. import { describe, expect, it } from 'vitest' import { agentSessionFailureFact } from './agent-session-failure' @@ -61,7 +62,7 @@ describe('the send a Stop ended before its turn opened', () => { }) }) - it('holds the queue when left in doubt', () => { + it('leaves the outbox when recorded in doubt, and the next send goes out', () => { const inDoubt = submission({ dispatchState: 'unknown', recovered: true, @@ -70,9 +71,10 @@ describe('the send a Stop ended before its turn opened', () => { const reconciled = reconcileStructuredAgentSessionOutbox(OUTBOX, [inDoubt], []) + expect(reconciled.map((each) => each.clientMessageId)).toEqual(['next']) expect(admitStructuredAgentSessionOutboxEntry(reconciled)).toMatchObject({ - state: 'blocked', - entry: { clientMessageId: 'stopped' } + state: 'dispatch', + entry: { clientMessageId: 'next' } }) }) }) diff --git a/src/shared/structured-agent-session-send-disposition.ts b/src/shared/structured-agent-session-send-disposition.ts index 5358a20cf585..b4a4612b10e9 100644 --- a/src/shared/structured-agent-session-send-disposition.ts +++ b/src/shared/structured-agent-session-send-disposition.ts @@ -68,6 +68,9 @@ function withLastFailure( return lastFailure === undefined ? rest : { ...rest, lastFailure } } +/** The host's replay of a send whose operation it recorded but whose submission row it lost. */ +const SUBMISSION_MISSING = 'durable_send_submission_missing' + function dropEntry(input: SendDispositionInput): StructuredAgentSessionOutboxEntry[] { return input.entries.filter( (candidate) => candidate.clientMessageId !== input.entry.clientMessageId @@ -287,15 +290,19 @@ export function disposeStructuredAgentSessionSendResult( error: null } } - if (submission.dispatchState === 'unknown' && submission.recovered) { - return { - entries: input.entries.map((candidate) => - candidate.clientMessageId === input.entry.clientMessageId - ? { ...candidate, state: 'unconfirmed', retryAfterUnknownSubmittedAt: -1 } - : candidate - ), - error: null - } + if (submission.dispatchState === 'unknown') { + // The host recorded it and never sends it again, so nothing waits on it: its row draws it, as + // the reconcile drops it. One whose row the host lost has only this entry to show it. + return submission.reason === SUBMISSION_MISSING + ? { + entries: input.entries.map((candidate) => + candidate.clientMessageId === input.entry.clientMessageId + ? { ...candidate, state: 'unconfirmed', retryAfterUnknownSubmittedAt: -1 } + : candidate + ), + error: null + } + : { entries: dropEntry(input), error: null } } // `pending` is the host saying the message was written and is awaiting the // provider's acknowledgement, which cannot arrive until the turn ahead of it @@ -304,10 +311,7 @@ export function disposeStructuredAgentSessionSendResult( // because a `pending` can still settle `rejected` or `unknown`, and only the // entry carries the retry state that answer needs. return { - entries: replaceEntryState( - input, - submission.dispatchState === 'unknown' ? 'unconfirmed' : 'dispatching' - ), + entries: replaceEntryState(input, 'dispatching'), error: null } }