diff --git a/config/scripts/vitest-sqlite-database-runtime-files.mjs b/config/scripts/vitest-sqlite-database-runtime-files.mjs index 9098a6f566db..e79a96a65076 100644 --- a/config/scripts/vitest-sqlite-database-runtime-files.mjs +++ b/config/scripts/vitest-sqlite-database-runtime-files.mjs @@ -37,8 +37,15 @@ export const SQLITE_DATABASE_RUNTIME_INCLUDE = [ 'src/main/runtime/orchestration-cli-subprocess.test.ts', 'src/main/runtime/orchestration-dispatch-mailbox-delivery.test.ts', 'src/main/runtime/orchestration-mailbox-crash-recovery.test.ts', + 'src/main/runtime/orchestration-mailbox-cold-park-idle.test.ts', 'src/main/runtime/orchestration-mailbox-notification-consistency.test.ts', + 'src/main/runtime/orchestration-mailbox-transport-settlement.test.ts', + 'src/main/runtime/orchestration-mailbox-unacked-notice.test.ts', 'src/main/runtime/orchestration-message-delivery-identity.test.ts', + 'src/main/runtime/structured-chat-worker-unacked-notice.test.ts', + 'src/main/runtime/structured-chat-coordinator-mail-queue.test.ts', + 'src/main/runtime/structured-chat-coordinator-mail.test.ts', + 'src/main/runtime/structured-chat-mail-queue-recovery.test.ts', 'src/main/runtime/orchestration-messages-fake-parity.test.ts', 'src/main/runtime/orchestration/agent-facing-parity.test.ts', 'src/main/runtime/orchestration/coordinator-decision-gates.test.ts', @@ -68,6 +75,7 @@ export const SQLITE_DATABASE_RUNTIME_INCLUDE = [ 'src/main/runtime/orchestration/db/hot-path-statement-compilation.test.ts', 'src/main/runtime/orchestration/db/lifecycle-transition.test.ts', 'src/main/runtime/orchestration/db/messages/mailbox-consumer-lifecycle-fencing.test.ts', + 'src/main/runtime/orchestration/db/messages/replay-waiting-count.test.ts', 'src/main/runtime/orchestration/db/remaining-dispatch-statement-compilation.test.ts', 'src/main/runtime/orchestration/db/row-column-lists.test.ts', 'src/main/runtime/orchestration/db/runs/run-coordinator-orca-session-binding.test.ts', @@ -90,6 +98,7 @@ export const SQLITE_DATABASE_RUNTIME_INCLUDE = [ 'src/main/runtime/orchestration/mailbox-pointer-eligibility.test.ts', 'src/main/runtime/orchestration/mailbox-pointer-release-query.test.ts', 'src/main/runtime/orchestration/mailbox-pointer-stage.test.ts', + 'src/main/runtime/orchestration/mailbox-pointer-partial-fetch.test.ts', 'src/main/runtime/orchestration/mailbox-pointer-submit.test.ts', 'src/main/runtime/orchestration/message-batch-atomicity.test.ts', 'src/main/runtime/orchestration/mutation-receipt-capacity.test.ts', @@ -143,6 +152,10 @@ export const SQLITE_DATABASE_RUNTIME_INCLUDE = [ 'src/main/runtime/rpc/methods/orchestration/federation/federation-worker-brief-agent-in-front.test.ts', 'src/main/runtime/rpc/methods/orchestration/federation/federation.test.ts', 'src/main/runtime/rpc/methods/orchestration/messaging/check-worker-federated-attachment.test.ts', + 'src/main/runtime/rpc/methods/orchestration/messaging/check-replay-recovery.test.ts', + 'src/main/runtime/rpc/methods/orchestration/messaging/check-worker-consumer-fencing.test.ts', + 'src/main/runtime/rpc/methods/orchestration/messaging/run-bound-mailbox-interleavings.test.ts', + 'src/main/runtime/rpc/methods/orchestration/messaging/run-bound-mailbox-safety.test.ts', 'src/main/runtime/rpc/methods/orchestration/runs/migration-behavior.test.ts', 'src/main/runtime/rpc/methods/orchestration/worker/agent-status-producer-census.test.ts', 'src/main/runtime/rpc/methods/orchestration/worker/manual-dispatch-observation.test.ts', diff --git a/src/cli/handlers/orchestration-check-replay-recovery.test.ts b/src/cli/handlers/orchestration-check-replay-recovery.test.ts new file mode 100644 index 000000000000..02ad066cb219 --- /dev/null +++ b/src/cli/handlers/orchestration-check-replay-recovery.test.ts @@ -0,0 +1,131 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { HandlerContext } from '../dispatch' +import { RuntimeClient } from '../runtime-client' +import type { OrchestrationCheckOutput } from '../../shared/orchestration-check-output' + +const callMock = vi.hoisted(() => vi.fn()) +const printResultMock = vi.hoisted(() => vi.fn()) +vi.mock('../format', () => ({ printResult: printResultMock })) + +import { ORCHESTRATION_HANDLERS } from './orchestration' + +describe('CLI replay recovery across host versions', () => { + beforeEach(() => { + callMock.mockReset() + printResultMock.mockReset() + vi.stubEnv('ORCA_CLI_COMMAND', 'orca-ide') + }) + afterEach(() => { + vi.unstubAllEnvs() + vi.restoreAllMocks() + }) + + function invoke(flags: [string, string | boolean][], json: boolean) { + const client = new RuntimeClient('/folder-workspace/.test-orca', 60_000, null, null, 'orca-ide') + vi.spyOn(client, 'call').mockImplementation(callMock) + const context: HandlerContext = { + flags: new Map([['terminal', 'term_original'], ...flags]), + client, + cwd: '/folder-workspace', + json + } + return ORCHESTRATION_HANDLERS['orchestration check'](context) + } + function printed(): OrchestrationCheckOutput { + const response: unknown = printResultMock.mock.calls[0]?.[0] + if ( + typeof response === 'object' && + response && + 'result' in response && + typeof response.result === 'object' && + response.result && + 'messages' in response.result && + 'count' in response.result + ) { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: The mocked response is an orchestration check result and was prepared by the real handler. + return response.result as OrchestrationCheckOutput + } + throw new Error('Expected printed check receipt') + } + function oldHostResult() { + return { + runId: 'run_receipt_owner', + messages: [{ id: 'msg_old', from_handle: 'worker' }], + count: 1, + deliveryId: 'delivery_old', + replayed: true + } + } + + it.each([false, true])( + 'projects old-host recovery into actual JSON and text: json=%s', + async (json) => { + callMock.mockResolvedValue({ result: oldHostResult() }) + await invoke([], json) + const result = printed() + expect(result.replayRecovery).toMatchObject({ + ackCommand: 'orca-ide orchestration check --terminal term_original --ack delivery_old', + guidance: + 'Process every message in this batch before acknowledging, then process the next batch returned.' + }) + expect(result.replayRecovery).not.toHaveProperty('waitingCount') + expect(JSON.parse(JSON.stringify(result))).toHaveProperty('replayRecovery.ackCommand') + const formatter = printResultMock.mock.calls[0]?.[2] + expect(formatter?.(result)).toContain('Delivery delivery_old (replayed)') + expect(formatter?.(result)).not.toContain('0 unread messages') + expect(formatter?.(result)).not.toContain('--run run_receipt_owner') + expect(callMock).toHaveBeenCalledTimes(1) + } + ) + + it('preserves explicit Run selection while omitting unavailable backlog counts', async () => { + callMock.mockResolvedValue({ result: oldHostResult() }) + await invoke( + [ + ['run', 'run_selected'], + ['format', true] + ], + true + ) + expect(printed().replayRecovery?.ackCommand).toBe( + 'orca-ide orchestration check --terminal term_original --run run_selected --ack delivery_old' + ) + }) + + it('keeps new-host recovery count and guidance in JSON and formatted output', async () => { + callMock.mockResolvedValue({ + result: { + ...oldHostResult(), + formatted: 'Message body', + replayRecovery: { + waitingCount: 83, + ackCommand: 'orca-dev orchestration check --terminal term_original --ack delivery_old', + guidance: + 'Process every message in this batch before acknowledging, then process the next batch returned.' + } + } + }) + await invoke([['format', true]], true) + expect(printed().replayRecovery?.waitingCount).toBe(83) + expect(printResultMock.mock.calls[0]?.[2]?.(printed())).toContain( + '83 unread messages waiting behind this Delivery.' + ) + }) + + it.each([ + ['peek', true], + ['all', true] + ] as const)('omits fallback for inspection %s', async (flag, value) => { + callMock.mockResolvedValue({ result: oldHostResult() }) + await invoke([[flag, value]], true) + expect(printed()).not.toHaveProperty('replayRecovery') + }) + + it('omits fallback for legacy recovery reads', async () => { + callMock.mockResolvedValue({ + result: { ...oldHostResult(), legacyCompatibility: { readOnly: true, recovery: true } } + }) + await invoke([], true) + expect(printed()).not.toHaveProperty('replayRecovery') + }) +}) diff --git a/src/cli/handlers/orchestration/message-check-handler.ts b/src/cli/handlers/orchestration/message-check-handler.ts index 6ace43eac31a..346c7ef3e708 100644 --- a/src/cli/handlers/orchestration/message-check-handler.ts +++ b/src/cli/handlers/orchestration/message-check-handler.ts @@ -6,6 +6,7 @@ import type { RuntimeRpcSuccess } from '../../runtime-client' import { formatOrchestrationCheckText, prepareOrchestrationCheckOutput, + type OrchestrationReplayRecovery, type LegacyCompatibilityResult, type OrchestrationMessageSummary as MessageSummary } from '../../../shared/orchestration-check-output' @@ -20,6 +21,8 @@ type CheckResult = { count: number formatted?: string deliveryId?: string | null + replayed?: boolean + replayRecovery?: OrchestrationReplayRecovery runId?: string timedOut?: boolean cancelled?: boolean @@ -72,7 +75,12 @@ export const ORCHESTRATION_CHECK_HANDLER: Record = { } result = { ...result, - result: prepareOrchestrationCheckOutput(result.result, callerLabel, flags.has('format')) + result: prepareOrchestrationCheckOutput(result.result, callerLabel, flags.has('format'), { + cliCommand: resolveCompatibilityCliCommand(), + terminal, + run: getOptionalStringFlag(flags, 'run'), + consuming: !peek && !flags.has('all') + }) } printResult(result, json, (value) => formatOrchestrationCheckText(value, callerLabel)) const compatibilityAck = result.result.legacyCompatibility?.ackMessageIds diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index e0f193ba0b49..cbdf8593b37d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -331,6 +331,12 @@ export class StructuredAgentSessionHost { journalSnapshot = async (sessionId: string): Promise => (await this.lifetime.conversation(sessionId)).journal.snapshot() + /** Includes settled queue receipts, which the published card list omits. */ + mailboxPointerSnapshot = async (sessionId: string) => { + const { journal } = await this.lifetime.conversation(sessionId) + return { snapshot: journal.snapshot(), queuedMessages: journal.queuedMessages.list() } + } + subscribe = (input: AgentSessionSubscribeInput) => this.backgroundTasks.subscribe(input) settleLateDispatch = (input: Parameters[1]) => diff --git a/src/main/runtime/orchestration-dispatch-mailbox-delivery.test.ts b/src/main/runtime/orchestration-dispatch-mailbox-delivery.test.ts index 958a93392927..f7d3b8d96dd6 100644 --- a/src/main/runtime/orchestration-dispatch-mailbox-delivery.test.ts +++ b/src/main/runtime/orchestration-dispatch-mailbox-delivery.test.ts @@ -88,7 +88,7 @@ describe('Dispatch mailbox Delivery', () => { const restarted = createRuntime(restartedDb) await driveToLiveIdle(restarted.runtime) await vi.advanceTimersByTimeAsync(2_500) - expect(pointerCount(restarted.write)).toBe(0) + expect(pointerCount(restarted.write)).toBe(1) const replayed = await checkBoundMailbox(restarted.runtime) expect(replayed).toMatchObject({ diff --git a/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts b/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts index 3475ae70eb04..528d4daebf8e 100644 --- a/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts +++ b/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts @@ -724,7 +724,7 @@ describe('orchestration notification mailbox consistency', () => { }) ) const notificationQuery = vi.spyOn(db, 'getUndeliveredUnreadMessages') - const submitRevalidation = vi.spyOn(db, 'areUnreadMessages') + const submitRevalidation = vi.spyOn(db, 'getMailboxPointerAttentionIds') const prepare = vi.spyOn(sqliteFor(db), 'prepare') await driveToLiveIdle(harness.runtime) @@ -743,7 +743,7 @@ describe('orchestration notification mailbox consistency', () => { ) const revalidationSql = prepare.mock.calls .map(([sql]) => sql) - .find((sql) => sql.includes('SELECT COUNT(*)') && sql.includes('id IN')) + .find((sql) => sql.includes('SELECT id FROM messages') && sql.includes('id IN')) expect(revalidationSql).toContain('INDEXED BY idx_messages_id') const revalidationPlan = sqliteFor(db) .prepare(`EXPLAIN QUERY PLAN ${revalidationSql}`) @@ -833,7 +833,7 @@ describe('orchestration notification mailbox consistency', () => { db.close() }) - it('does not point newer Run mail behind an outstanding Delivery', async () => { + it('points newer Run mail while preserving an outstanding Delivery replay', async () => { vi.useFakeTimers() const db = createDatabase('orca-mailbox-outstanding-') const harness = createRuntime(db) @@ -855,7 +855,7 @@ describe('orchestration notification mailbox consistency', () => { await driveToLiveIdle(harness.runtime) const replayed = await checkBoundMailbox(harness.runtime) - expect(pointerCount(harness.write)).toBe(0) + expect(pointerCount(harness.write)).toBe(1) expect(replayed.messages).toEqual([expect.objectContaining({ id: firstMessage.id })]) expect(replayed.deliveryId).toBe(firstDelivery.deliveryId) diff --git a/src/main/runtime/orchestration-mailbox-unacked-notice.test.ts b/src/main/runtime/orchestration-mailbox-unacked-notice.test.ts new file mode 100644 index 000000000000..a1b9e4f29fed --- /dev/null +++ b/src/main/runtime/orchestration-mailbox-unacked-notice.test.ts @@ -0,0 +1,205 @@ +import './rpc/unused-default-rpc-methods.test-fixture' +import { tmpdir } from 'node:os' +import { rmSync } from 'node:fs' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { + checkBoundMailbox, + createDatabase, + createRuntime, + driveToLiveIdle, + PANE_KEY, + pointerCount, + PTY_ID, + temporaryDirectories, + TERMINAL_HANDLE +} from './orchestration-mailbox-notification-test-harness' +import { createRootDispatch } from './orchestration/db/root-dispatch-test-fixture' +import { + WRITE_ACCEPTED, + writeUnverifiable, + type WriteSettlement +} from '../../shared/pty-write-settlement' +import { + MAILBOX_POINTER_WRITE_ATTEMPTED, + MAILBOX_POINTER_ENTER_ATTEMPTED +} from './orchestration/db/messages/mailbox-pointer-enter-state' + +vi.mock('electron', () => ({ + app: { getPath: vi.fn(() => tmpdir()), isPackaged: false }, + BrowserWindow: { fromId: vi.fn(() => null) }, + ipcMain: { on: vi.fn(), removeListener: vi.fn() }, + webContents: { fromId: vi.fn(() => null) } +})) + +function workerMailbox() { + const db = createDatabase('orca-unacked-worker-notice-') + const harness = createRuntime(db) + const run = db.createRun({ + objective: 'Continue hearing coordinator guidance', + coordinatorHandle: 'term_coordinator', + coordinatorPaneKey: '33333333-3333-4333-8333-333333333333:44444444-4444-4444-8444-444444444444' + }) + const task = db.createTask({ spec: 'Read guidance', runId: run.id }) + const dispatch = createRootDispatch( + db, + task.id, + TERMINAL_HANDLE, + PANE_KEY, + undefined, + `${PTY_ID}:mailbox-incarnation` + ) + const mailbox = `dispatch:${dispatch.id}` + const receive = (subject: string) => { + const message = db.insertMessage({ + from: 'term_coordinator', + to: mailbox, + subject, + runId: run.id + }) + harness.runtime.notifyMessageArrived(mailbox, 'status') + return message + } + return { db, ...harness, mailbox, receive } +} + +describe('terminal worker notices after a check without acknowledgement', () => { + afterEach(() => { + vi.useRealTimers() + for (const directory of temporaryDirectories.splice(0)) { + rmSync(directory, { recursive: true, force: true }) + } + }) + + it('notices new mail on the only idle edge after an in-flight Enter loses settlement', async () => { + vi.useFakeTimers() + const { db, runtime, mailbox, receive } = workerMailbox() + const write = vi.fn((_ptyId: string, _data: string) => true) + let settleEnter: ((settlement: WriteSettlement) => void) | undefined + runtime.setPtyController({ + write, + writeWithSettlement: (ptyId, data) => { + write(ptyId, data) + if (data === '\r' && !settleEnter) { + return new Promise((resolve) => { + settleEnter = resolve + }) + } + return WRITE_ACCEPTED + }, + kill: vi.fn(), + getForegroundProcess: async () => null + }) + try { + await driveToLiveIdle(runtime) + const first = receive('Fetched guidance with lost Enter settlement') + await vi.advanceTimersByTimeAsync(500) + expect(pointerCount(write)).toBe(1) + expect(settleEnter).toEqual(expect.any(Function)) + await runtime.acceptPtyDataBounded(PTY_ID, '\x1b]0;Codex working\x07', 3).completion + settleEnter?.(writeUnverifiable('transport_settlement_lost', true)) + await vi.advanceTimersByTimeAsync(0) + expect(db.getMessageById(first.id)?.pointer_enter_pending).toBe( + MAILBOX_POINTER_ENTER_ATTEMPTED + ) + const checked = await checkBoundMailbox(runtime) + expect(checked.messages).toEqual([expect.objectContaining({ id: first.id })]) + const newer = receive('New guidance while the worker is busy') + await vi.advanceTimersByTimeAsync(500) + expect(pointerCount(write)).toBe(1) + await runtime.acceptPtyDataBounded(PTY_ID, '\x1b]0;Codex done\x07', 4).completion + await vi.advanceTimersByTimeAsync(500) + expect(pointerCount(write)).toBe(2) + expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(2) + expect(db.getMessageById(newer.id)?.delivered_at).toEqual(expect.any(String)) + expect(db.getMessageById(first.id)?.pointer_enter_pending).toBe(0) + expect((await checkBoundMailbox(runtime)).deliveryId).toBe(checked.deliveryId) + expect(db.hasOutstandingMailboxDelivery(mailbox)).toBe(true) + } finally { + db.close() + } + }) + + it.each([MAILBOX_POINTER_WRITE_ATTEMPTED, MAILBOX_POINTER_ENTER_ATTEMPTED])( + 'checks newer attention on the same idle edge that settles recovered phase %s', + async (phase) => { + vi.useFakeTimers() + const { db, runtime, receive } = workerMailbox() + try { + await driveToLiveIdle(runtime) + receive('Fetched guidance') + await vi.advanceTimersByTimeAsync(500) + const checked = await checkBoundMailbox(runtime) + await runtime.acceptPtyDataBounded(PTY_ID, '\x1b]0;Codex working\x07', 3).completion + const uncertain = receive('Notice handed off before restart') + const target = { ptyId: PTY_ID, processIncarnation: `${PTY_ID}:mailbox-incarnation` } + expect(db.stageMailboxPointerEnter([uncertain.id], target)).toBe(true) + expect(db.markMailboxPointerWriteAttempted([uncertain.id], target)).toBe(true) + if (phase === MAILBOX_POINTER_ENTER_ATTEMPTED) { + expect(db.markMailboxPointerEnterAttempted([uncertain.id], target)).toBe(true) + } + const newer = receive('Newer guidance before the only idle edge') + const restarted = createRuntime(db) + await restarted.runtime.listTerminals() + await restarted.runtime.acceptPtyDataBounded(PTY_ID, '\x1b]0;Codex done\x07', 1).completion + await vi.advanceTimersByTimeAsync(500) + expect(pointerCount(restarted.write)).toBe(1) + expect(restarted.write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(1) + expect(db.getMessageById(newer.id)?.delivered_at).toEqual(expect.any(String)) + expect(db.getMessageById(uncertain.id)).toMatchObject({ + pointer_enter_pending: 0, + delivered_at: expect.any(String) + }) + expect((await checkBoundMailbox(restarted.runtime)).deliveryId).toBe(checked.deliveryId) + } finally { + db.close() + } + } + ) + + it.each(['before new mail', 'after new mail'])( + 'notifies later guidance when the consumer goes idle %s', + async (idleTiming) => { + vi.useFakeTimers() + const { db, runtime, write, mailbox, receive } = workerMailbox() + try { + await driveToLiveIdle(runtime) + const first = receive('First guidance') + await vi.advanceTimersByTimeAsync(500) + expect(pointerCount(write)).toBe(1) + const checked = await checkBoundMailbox(runtime) + expect(checked.messages).toEqual([expect.objectContaining({ id: first.id })]) + expect(db.hasOutstandingMailboxDelivery(mailbox)).toBe(true) + + await runtime.acceptPtyDataBounded(PTY_ID, '\x1b]0;Codex working\x07', 3).completion + if (idleTiming === 'before new mail') { + await runtime.acceptPtyDataBounded(PTY_ID, '\x1b]0;Codex done\x07', 4).completion + } + const later = receive('Later guidance') + await vi.advanceTimersByTimeAsync(500) + if (idleTiming === 'after new mail') { + expect(pointerCount(write)).toBe(1) + await runtime.acceptPtyDataBounded(PTY_ID, '\x1b]0;Codex done\x07', 4).completion + await vi.advanceTimersByTimeAsync(500) + } + expect(pointerCount(write)).toBe(2) + expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(2) + expect(db.getMessageById(later.id)).toMatchObject({ + read: 0, + delivered_at: expect.any(String) + }) + + runtime.notifyMessageArrived(mailbox, 'status') + runtime.deliverPendingMessagesForHandle(mailbox) + await vi.advanceTimersByTimeAsync(500) + expect(pointerCount(write)).toBe(2) + const replayed = await checkBoundMailbox(runtime) + expect(replayed.deliveryId).toBe(checked.deliveryId) + expect(replayed.messages).toEqual([expect.objectContaining({ id: first.id })]) + const next = await checkBoundMailbox(runtime, { ack: checked.deliveryId ?? undefined }) + expect(next.messages).toEqual([expect.objectContaining({ id: later.id })]) + } finally { + db.close() + } + } + ) +}) diff --git a/src/main/runtime/orchestration/__snapshots__/preamble.test.ts.snap b/src/main/runtime/orchestration/__snapshots__/preamble.test.ts.snap index 6815b5ecc7f4..2b6f83f5b00c 100644 --- a/src/main/runtime/orchestration/__snapshots__/preamble.test.ts.snap +++ b/src/main/runtime/orchestration/__snapshots__/preamble.test.ts.snap @@ -62,6 +62,10 @@ Don't post to Slack, GitHub, or other channels during the run; report through th # start a new file and after a test run — and once more immediately before # you send worker_done, so a redirect lands before the task settles. orca orchestration check --terminal orca_session_id:WORKER --json + + # Process every message in the returned batch before acknowledging, + # then process the next batch returned. Use its deliveryId, not a message ID. + orca orchestration check --terminal orca_session_id:WORKER --ack "" --json \`\`\` === AFTER YOU SEND worker_done === @@ -146,6 +150,10 @@ Don't post to Slack, GitHub, or other channels during the run; report through th # start a new file and after a test run — and once more immediately before # you send worker_done, so a redirect lands before the task settles. orca orchestration check --terminal term_WORKER --json + + # Process every message in the returned batch before acknowledging, + # then process the next batch returned. Use its deliveryId, not a message ID. + orca orchestration check --terminal term_WORKER --ack "" --json \`\`\` === AFTER YOU SEND worker_done === diff --git a/src/main/runtime/orchestration/agent-facing-parity.test.ts b/src/main/runtime/orchestration/agent-facing-parity.test.ts index 627b841e09dc..5bdab2f17077 100644 --- a/src/main/runtime/orchestration/agent-facing-parity.test.ts +++ b/src/main/runtime/orchestration/agent-facing-parity.test.ts @@ -130,7 +130,7 @@ async function renderChatPointer(mailbox: string): Promise { getCliCommand: localOrchestrationCliCommand, senderName: () => null, host: { - readSessionFacts: async () => ({ submissions: [] }), + readSessionFacts: async () => ({ submissions: [], queuedSends: [] }), currentFence: () => 1, send: async (input) => { for (const block of input.body.blocks) { diff --git a/src/main/runtime/orchestration/db-messages.test.ts b/src/main/runtime/orchestration/db-messages.test.ts index e803aad36407..61fa10e40155 100644 --- a/src/main/runtime/orchestration/db-messages.test.ts +++ b/src/main/runtime/orchestration/db-messages.test.ts @@ -17,6 +17,34 @@ describe('OrchestrationDb', () => { } describe('messages', () => { + it('settles accepted notice members without changing prior timestamps or another mailbox', () => { + const d = createDb() + const pointed = d.insertMessage({ runId, from: 'a', to: 'b', subject: 'pointed' }) + const fetched = d.insertMessage({ runId, from: 'a', to: 'b', subject: 'fetched' }) + const foreign = d.insertMessage({ runId, from: 'a', to: 'c', subject: 'foreign' }) + const audit = d.insertMessage({ + runId, + from: 'a', + to: 'b', + subject: 'audit', + deliveryContract: 'audit_only' + }) + d.db + .prepare('UPDATE messages SET delivered_at = ? WHERE id = ?') + .run('2026-01-01 00:00:00', pointed.id) + d.markAsRead([fetched.id]) + + d.markUnpointedMailboxMessagesAsDelivered('b', [pointed.id, fetched.id, foreign.id, audit.id]) + + expect(d.getMessageById(pointed.id)?.delivered_at).toBe('2026-01-01T00:00:00Z') + expect(d.getMessageById(fetched.id)).toMatchObject({ + read: 1, + delivered_at: expect.any(String) + }) + expect(d.getMessageById(foreign.id)?.delivered_at).toBeNull() + expect(d.getMessageById(audit.id)?.delivered_at).toBeNull() + }) + it('inserts and retrieves a message', () => { const d = createDb() const msg = d.insertMessage({ @@ -88,6 +116,36 @@ describe('OrchestrationDb', () => { expect(unread).toHaveLength(2) }) + it('selects independent mail beyond the SQLite parameter limit of owned IDs', () => { + const d = createDb() + const owned = d.insertMessage({ runId, from: 'a', to: 'b', subject: 'owned' }) + const fresh = d.insertMessage({ runId, from: 'a', to: 'b', subject: 'fresh' }) + const excludeMessageIds = Array.from({ length: 250001 }, (_, index) => `owned-${index}`) + excludeMessageIds.push(owned.id) + + const selected = d.getUndeliveredUnreadMessages('b', undefined, { + excludeMessageIds, + limit: 50 + }) + + expect(selected.map((message) => message.id)).toEqual([fresh.id]) + }) + + it('excludes all owned members before applying the notice limit', () => { + const d = createDb() + const owned = Array.from({ length: 51 }, (_, index) => + d.insertMessage({ runId, from: 'a', to: 'b', subject: `owned-${index}` }) + ) + const fresh = d.insertMessage({ runId, from: 'a', to: 'b', subject: 'fresh' }) + + const selected = d.getUndeliveredUnreadMessages('b', undefined, { + excludeMessageIds: owned.map((message) => message.id), + limit: 50 + }) + + expect(selected.map((message) => message.id)).toEqual([fresh.id]) + }) + it('creates the undelivered inbox index used by push delivery', () => { const d = createDb() const sqlite = (d as unknown as { db: Database.Database }).db diff --git a/src/main/runtime/orchestration/db/messages/mailbox-pointer-attention.ts b/src/main/runtime/orchestration/db/messages/mailbox-pointer-attention.ts new file mode 100644 index 000000000000..ee9c48b88539 --- /dev/null +++ b/src/main/runtime/orchestration/db/messages/mailbox-pointer-attention.ts @@ -0,0 +1,78 @@ +import type { MessageType, MessageRow } from '../../types' +import type { OrchestrationDb } from '../orchestration-db' +import { exposeMessageListTimestamps } from '../utc-timestamp' + +const OUTSIDE_FETCHED_BATCH_SQL = `NOT EXISTS ( + SELECT 1 FROM outstanding_deliveries AS delivery, json_each(delivery.message_ids) AS member + WHERE delivery.mailbox_handle = messages.to_handle AND member.value = messages.id +)` + +// Why: delivered_at IS NULL filter — push-on-idle delivers each row at most once; read (set only by check) wouldn't prevent replay. +export function getUndeliveredUnreadMessages( + this: OrchestrationDb, + toHandle: string, + types?: MessageType[], + options?: { + excludeTypes?: readonly string[] + excludeFetched?: boolean + excludeMessageIds?: readonly string[] + limit?: number + } +): MessageRow[] { + const conditions = [ + 'to_handle = ?', + 'read = 0', + 'delivered_at IS NULL', + 'pointer_enter_pending = 0', + "delivery_contract = 'current_delivery'" + ] + const params: (string | number)[] = [toHandle] + if (options?.excludeFetched) { + conditions.push(OUTSIDE_FETCHED_BATCH_SQL) + } + if (options?.excludeMessageIds?.length) { + conditions.push('id NOT IN (SELECT value FROM json_each(?))') + params.push(JSON.stringify(options.excludeMessageIds)) + } + if (types?.length) { + conditions.push(`type IN (${types.map(() => '?').join(',')})`) + params.push(...types) + } + if (options?.excludeTypes?.length) { + conditions.push(`type NOT IN (${options.excludeTypes.map(() => '?').join(',')})`) + params.push(...options.excludeTypes) + } + const limitSql = options?.limit === undefined ? '' : ' LIMIT ?' + if (options?.limit !== undefined) { + params.push(Math.max(1, Math.floor(options.limit))) + } + return exposeMessageListTimestamps( + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: SELECT * reads the migrated messages schema represented by MessageRow. + this.db + .prepare( + `SELECT * FROM messages + WHERE ${conditions.join(' AND ')} + ORDER BY sequence${limitSql}` + ) + .all(...params) as MessageRow[] + ) +} + +export function getMailboxPointerAttentionIds( + this: OrchestrationDb, + mailboxHandle: string, + ids: readonly string[] +): string[] { + if (ids.length === 0) { + return [] + } + return this.db + .prepare( + `SELECT id FROM messages INDEXED BY idx_messages_id + WHERE to_handle = ? AND read = 0 AND delivered_at IS NULL + AND delivery_contract = 'current_delivery' + AND id IN (${ids.map(() => '?').join(',')}) AND ${OUTSIDE_FETCHED_BATCH_SQL}` + ) + .all(mailboxHandle, ...ids) + .flatMap((row) => (typeof row.id === 'string' ? [row.id] : [])) +} diff --git a/src/main/runtime/orchestration/db/messages/message-inbox.ts b/src/main/runtime/orchestration/db/messages/message-inbox.ts index e9b800d431e4..8b132a081f41 100644 --- a/src/main/runtime/orchestration/db/messages/message-inbox.ts +++ b/src/main/runtime/orchestration/db/messages/message-inbox.ts @@ -2,14 +2,19 @@ import type { MessageType, MessageRow } from '../../types' import { exposeMessageTimestamps, exposeMessageListTimestamps } from '../utc-timestamp' import { addLifecycleRejectionMarker } from '../lifecycle-rejection-marker' import type { OrchestrationDb } from '../orchestration-db' +import { + getUndeliveredUnreadMessages, + getMailboxPointerAttentionIds +} from './mailbox-pointer-attention' const MESSAGE_ID_UPDATE_BATCH_SIZE = 500 const MESSAGE_MUTATION_SAVEPOINT = 'message_id_mutation' function runBatchedMessageMutation( db: OrchestrationDb, - ids: string[], - sqlForPlaceholders: (placeholders: string) => string + ids: readonly string[], + sqlForPlaceholders: (placeholders: string) => string, + leadingParams: readonly string[] = [] ): void { if (ids.length === 0) { return @@ -19,7 +24,7 @@ function runBatchedMessageMutation( for (let offset = 0; offset < ids.length; offset += MESSAGE_ID_UPDATE_BATCH_SIZE) { const batch = ids.slice(offset, offset + MESSAGE_ID_UPDATE_BATCH_SIZE) const placeholders = batch.map(() => '?').join(',') - db.db.prepare(sqlForPlaceholders(placeholders)).run(...batch) + db.db.prepare(sqlForPlaceholders(placeholders)).run(...leadingParams, ...batch) } db.db.exec(`RELEASE ${MESSAGE_MUTATION_SAVEPOINT}`) } catch (error) { @@ -86,44 +91,6 @@ export function convertLifecycleMessageToRejection( return this.getMessageById(messageId) } -// Why: delivered_at IS NULL filter — push-on-idle delivers each row at most once; read (set only by check) wouldn't prevent replay. -export function getUndeliveredUnreadMessages( - this: OrchestrationDb, - toHandle: string, - types?: MessageType[], - options?: { excludeTypes?: readonly string[]; limit?: number } -): MessageRow[] { - const conditions = [ - 'to_handle = ?', - 'read = 0', - 'delivered_at IS NULL', - 'pointer_enter_pending = 0', - "delivery_contract = 'current_delivery'" - ] - const params: (string | number)[] = [toHandle] - if (types?.length) { - conditions.push(`type IN (${types.map(() => '?').join(',')})`) - params.push(...types) - } - if (options?.excludeTypes?.length) { - conditions.push(`type NOT IN (${options.excludeTypes.map(() => '?').join(',')})`) - params.push(...options.excludeTypes) - } - const limitSql = options?.limit === undefined ? '' : ' LIMIT ?' - if (options?.limit !== undefined) { - params.push(Math.max(1, Math.floor(options.limit))) - } - return exposeMessageListTimestamps( - this.db - .prepare( - `SELECT * FROM messages - WHERE ${conditions.join(' AND ')} - ORDER BY sequence${limitSql}` - ) - .all(...params) as MessageRow[] - ) -} - export function getUndeliveredUnreadMailboxHandles(this: OrchestrationDb): string[] { return ( this.db @@ -177,6 +144,23 @@ export function markAsDelivered(this: OrchestrationDb, ids: string[]): void { ) } +export function markUnpointedMailboxMessagesAsDelivered( + this: OrchestrationDb, + mailboxHandle: string, + ids: readonly string[] +): void { + runBatchedMessageMutation( + this, + ids, + (placeholders) => + `UPDATE messages SET delivered_at = datetime('now'), pointer_enter_pending = 0, + pointer_pty_id = NULL, pointer_process_incarnation = NULL + WHERE to_handle = ? AND delivered_at IS NULL + AND delivery_contract = 'current_delivery' AND id IN (${placeholders})`, + [mailboxHandle] + ) +} + export function markAsUndelivered(this: OrchestrationDb, ids: string[]): void { runBatchedMessageMutation( this, @@ -279,11 +263,13 @@ export type MessageInboxMethods = { getUnreadMessages: typeof getUnreadMessages convertLifecycleMessageToRejection: typeof convertLifecycleMessageToRejection getUndeliveredUnreadMessages: typeof getUndeliveredUnreadMessages + getMailboxPointerAttentionIds: typeof getMailboxPointerAttentionIds getUndeliveredUnreadMailboxHandles: typeof getUndeliveredUnreadMailboxHandles getAllMessages: typeof getAllMessages getMessageById: typeof getMessageById markAsRead: typeof markAsRead markAsDelivered: typeof markAsDelivered + markUnpointedMailboxMessagesAsDelivered: typeof markUnpointedMailboxMessagesAsDelivered markAsUndelivered: typeof markAsUndelivered areUnreadMessages: typeof areUnreadMessages markAsReadAndDelivered: typeof markAsReadAndDelivered @@ -297,11 +283,13 @@ export function attachMessageInbox(ctor: { prototype: object }): void { getUnreadMessages, convertLifecycleMessageToRejection, getUndeliveredUnreadMessages, + getMailboxPointerAttentionIds, getUndeliveredUnreadMailboxHandles, getAllMessages, getMessageById, markAsRead, markAsDelivered, + markUnpointedMailboxMessagesAsDelivered, markAsUndelivered, areUnreadMessages, markAsReadAndDelivered, diff --git a/src/main/runtime/orchestration/db/messages/replay-waiting-count.test.ts b/src/main/runtime/orchestration/db/messages/replay-waiting-count.test.ts new file mode 100644 index 000000000000..ce5730e57c5b --- /dev/null +++ b/src/main/runtime/orchestration/db/messages/replay-waiting-count.test.ts @@ -0,0 +1,68 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { OrchestrationDb } from '../../db' + +describe('unread mail outside a Delivery', () => { + let db: OrchestrationDb + beforeEach(() => { + db = new OrchestrationDb(':memory:') + }) + afterEach(() => db.close()) + + it('counts only unread current-contract mail in the selected Run and mailbox, with no limit', () => { + const run = db.createRun({ + objective: 'count', + coordinatorHandle: 'term_coord', + coordinatorPaneKey: 'tab_coord:pane' + }) + const mailboxHandle = `run:${run.id}` + const first = db.insertMessage({ + from: 'worker', + to: mailboxHandle, + runId: run.id, + subject: 'first' + }) + const delivery = db.getOrCreateRunDelivery({ + runId: run.id, + consumerGeneration: run.consumer_generation + }) + if (!delivery) { + throw new Error('Expected Delivery') + } + const later = Array.from({ length: 79 }, (_, i) => + db.insertMessage({ from: 'worker', to: mailboxHandle, runId: run.id, subject: `later ${i}` }) + ) + db.markAsDelivered(later.map((row) => row.id)) + const read = db.insertMessage({ + from: 'worker', + to: mailboxHandle, + runId: run.id, + subject: 'read' + }) + db.markAsRead([read.id]) + for (const deliveryContract of ['audit_only', 'legacy_direct'] as const) { + db.insertMessage({ + from: 'worker', + to: mailboxHandle, + runId: run.id, + subject: deliveryContract, + deliveryContract + }) + } + db.insertMessage({ from: 'worker', to: 'different', runId: run.id, subject: 'other address' }) + db.insertMessage({ + from: 'worker', + to: mailboxHandle, + runId: 'run_legacy_local', + subject: 'other Run' + }) + expect( + db.countUnreadMessagesOutsideDelivery({ + runId: run.id, + mailboxHandle, + deliveryId: delivery.delivery.id + }) + ).toBe(79) + expect(db.countUnreadMessagesOutsideDelivery({ runId: run.id, mailboxHandle })).toBe(80) + expect(db.getMessageById(first.id)?.read).toBe(0) + }) +}) diff --git a/src/main/runtime/orchestration/db/messages/role-mailbox-delivery.ts b/src/main/runtime/orchestration/db/messages/role-mailbox-delivery.ts index 3b5f5dad94f7..ae5cdbcae1c5 100644 --- a/src/main/runtime/orchestration/db/messages/role-mailbox-delivery.ts +++ b/src/main/runtime/orchestration/db/messages/role-mailbox-delivery.ts @@ -184,6 +184,27 @@ export function hasOutstandingMailboxDelivery( ) } +export function countUnreadMessagesOutsideDelivery( + this: OrchestrationDb, + params: { runId: string; mailboxHandle: string; deliveryId?: string } +): number { + const row = this.db + .prepare( + `SELECT COUNT(*) AS count FROM messages AS message + WHERE message.run_id = ? AND message.to_handle = ? AND message.read = 0 + AND message.delivery_contract = 'current_delivery' + AND NOT EXISTS ( + SELECT 1 FROM deliveries AS delivery, json_each(delivery.message_ids) AS member + WHERE delivery.id = ? AND member.value = message.id + )` + ) + .get(params.runId, params.mailboxHandle, params.deliveryId ?? null) + if (typeof row === 'object' && row !== null && 'count' in row && typeof row.count === 'number') { + return row.count + } + throw new Error('Unread message count was not returned.') +} + export function fenceUnacknowledgedMailboxDeliveries( this: OrchestrationDb, mailboxHandle: string @@ -201,6 +222,7 @@ export type RoleMailboxDeliveryMethods = { getOrCreateMailboxDelivery: typeof getOrCreateMailboxDelivery acknowledgeMailboxDelivery: typeof acknowledgeMailboxDelivery hasOutstandingMailboxDelivery: typeof hasOutstandingMailboxDelivery + countUnreadMessagesOutsideDelivery: typeof countUnreadMessagesOutsideDelivery fenceUnacknowledgedMailboxDeliveries: typeof fenceUnacknowledgedMailboxDeliveries } @@ -211,6 +233,7 @@ export function attachRoleMailboxDelivery(ctor: { prototype: object }): void { getOrCreateMailboxDelivery, acknowledgeMailboxDelivery, hasOutstandingMailboxDelivery, + countUnreadMessagesOutsideDelivery, fenceUnacknowledgedMailboxDeliveries }) } diff --git a/src/main/runtime/orchestration/mailbox-pointer-delivery.ts b/src/main/runtime/orchestration/mailbox-pointer-delivery.ts index b8c8ad7a0c45..1d7a6056876d 100644 --- a/src/main/runtime/orchestration/mailbox-pointer-delivery.ts +++ b/src/main/runtime/orchestration/mailbox-pointer-delivery.ts @@ -67,9 +67,6 @@ export class OrchestrationMailboxPointerDelivery 0 && + if (pending.length > 0) { resumePendingOrchestrationMailboxPointer({ deps: this.deps, state: this.state, @@ -115,8 +111,6 @@ export class OrchestrationMailboxPointerDelivery this.settle(ptyId, flight), redrive: (redriveMailbox, force) => this.redrive(redriveMailbox, force) }) - ) { - return } const unread = selectOrchestrationPointerBatch({ db, diff --git a/src/main/runtime/orchestration/mailbox-pointer-eligibility.test.ts b/src/main/runtime/orchestration/mailbox-pointer-eligibility.test.ts index 2276665fc1c9..c855fe0d2708 100644 --- a/src/main/runtime/orchestration/mailbox-pointer-eligibility.test.ts +++ b/src/main/runtime/orchestration/mailbox-pointer-eligibility.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from 'vitest' import { OrchestrationDb } from './db/orchestration-db' import { selectOrchestrationPointerBatch, + shouldReleaseOrchestrationPointer, type OrchestrationMessageWaiter } from './mailbox-pointer-eligibility' @@ -27,6 +28,51 @@ function waiters(...filters: (string[] | undefined)[]): ReadonlySet { + it('selects new mail behind a fetched batch before limiting the notice rows', () => { + const db = new OrchestrationDb(':memory:') + try { + const run = db.createRun({ + objective: 'Read a full batch', + coordinatorHandle: 'coordinator', + coordinatorPaneKey: 'tab:leaf' + }) + const mailbox = `run:${run.id}` + db.insertMessages( + Array.from({ length: 50 }, (_, index) => ({ + runId: run.id, + from: 'worker', + to: mailbox, + subject: `Fetched ${index}` + })) + ) + const fetched = db.getOrCreateRunDelivery({ + runId: run.id, + consumerGeneration: run.consumer_generation + }) + expect(fetched?.messages).toHaveLength(50) + const newer = db.insertMessage({ + runId: run.id, + from: 'worker', + to: mailbox, + subject: 'New mail' + }) + expect( + selectOrchestrationPointerBatch({ + db, + mailboxHandle: mailbox, + waiters: undefined, + reservedTypes: undefined + }).map((message) => message.id) + ).toEqual([newer.id]) + expect(shouldReleaseOrchestrationPointer(db, mailbox, [newer], undefined)).toBe(false) + expect( + shouldReleaseOrchestrationPointer(db, mailbox, fetched?.messages ?? [], undefined) + ).toBe(true) + } finally { + db.close() + } + }) + it('excludes a waiter-claimed type', () => { const db = seeded() try { diff --git a/src/main/runtime/orchestration/mailbox-pointer-eligibility.ts b/src/main/runtime/orchestration/mailbox-pointer-eligibility.ts index b095df340993..4239ac00e745 100644 --- a/src/main/runtime/orchestration/mailbox-pointer-eligibility.ts +++ b/src/main/runtime/orchestration/mailbox-pointer-eligibility.ts @@ -43,6 +43,7 @@ export function selectOrchestrationPointerBatch(input: { mailboxHandle: string waiters: ReadonlySet | undefined reservedTypes: ReadonlySet | undefined + excludeMessageIds?: readonly string[] }): MessageRow[] { if (hasUnfilteredOrchestrationWaiter(input.waiters)) { return [] @@ -55,26 +56,35 @@ export function selectOrchestrationPointerBatch(input: { } return input.db.getUndeliveredUnreadMessages(input.mailboxHandle, undefined, { excludeTypes: [...excludedTypes], + excludeFetched: true, + excludeMessageIds: input.excludeMessageIds, limit: ORCHESTRATION_DELIVERY_BATCH_LIMIT }) } +export function selectStagedOrchestrationPointerMessages( + db: OrchestrationDb | null, + mailboxHandle: string, + messages: readonly T[], + waiters: ReadonlySet | undefined +): T[] { + const eligibleIds = db?.getMailboxPointerAttentionIds?.( + mailboxHandle, + messages.map((message) => message.id) + ) + const eligible = eligibleIds ? new Set(eligibleIds) : null + return messages.filter( + (message) => + !messageTypeHasOrchestrationWaiter(waiters, message.type) && + (eligible?.has(message.id) ?? db?.areUnreadMessages?.(mailboxHandle, [message.id]) ?? true) + ) +} + export function shouldReleaseOrchestrationPointer( db: OrchestrationDb | null, mailboxHandle: string, messages: readonly { id: string; type: string }[], waiters: ReadonlySet | undefined ): boolean { - if (db?.hasOutstandingMailboxDelivery?.(mailboxHandle)) { - return true - } - if (messages.some((message) => messageTypeHasOrchestrationWaiter(waiters, message.type))) { - return true - } - return !( - db?.areUnreadMessages?.( - mailboxHandle, - messages.map((message) => message.id) - ) ?? true - ) + return selectStagedOrchestrationPointerMessages(db, mailboxHandle, messages, waiters).length === 0 } diff --git a/src/main/runtime/orchestration/mailbox-pointer-partial-fetch.test.ts b/src/main/runtime/orchestration/mailbox-pointer-partial-fetch.test.ts new file mode 100644 index 000000000000..b44ebfafbbbd --- /dev/null +++ b/src/main/runtime/orchestration/mailbox-pointer-partial-fetch.test.ts @@ -0,0 +1,171 @@ +import { describe, expect, it, vi } from 'vitest' +import { WRITE_ACCEPTED, type WriteSettlement } from '../../../shared/pty-write-settlement' +import { OrchestrationDb } from './db' +import { OrchestrationMailboxDeliveryTarget } from './mailbox-delivery-target' +import { OrchestrationMailboxOwner, type OrchestrationMailboxLeaf } from './mailbox-owner' +import { stageOrchestrationMailboxPointer } from './mailbox-pointer-stage' +import { OrchestrationMailboxPointerState } from './mailbox-pointer-state' + +describe('a terminal pointer partially fetched before Enter', () => { + it.each( + ['before staging', 'during text write', 'before Enter', 'during liveness probe'].flatMap( + (boundary) => [ + { boundary, ack: false, allFetched: false }, + { boundary, ack: true, allFetched: false }, + { boundary, ack: false, allFetched: true }, + { boundary, ack: true, allFetched: true } + ] + ) + )( + 'submits the unfetched tail when check runs $boundary (ack=$ack, all=$allFetched)', + async ({ boundary, ack, allFetched }) => { + vi.useFakeTimers() + const db = new OrchestrationDb(':memory:') + const state = new OrchestrationMailboxPointerState() + const leaf: OrchestrationMailboxLeaf = { + tabId: 'tab', + leafId: 'leaf', + ptyId: 'pty', + writable: true, + lastAgentStatus: 'idle', + lastAgentStatusObservedLive: true, + lastOscTitle: null + } + try { + const run = db.createRun({ + objective: 'Partial fetch', + coordinatorHandle: 'term', + coordinatorPaneKey: 'tab:leaf' + }) + const mailboxHandle = `run:${run.id}` + const earlier = db.insertMessages( + Array.from({ length: allFetched ? 48 : 49 }, (_, index) => ({ + runId: run.id, + from: 'worker', + to: mailboxHandle, + subject: `Earlier ${index}` + })) + ) + db.markAsDelivered(earlier.map((message) => message.id)) + const messages = db.insertMessages( + ['A', 'B'].map((subject) => ({ + runId: run.id, + from: 'worker', + to: mailboxHandle, + subject + })) + ) + const tail = messages[1]! + const checkAndAck = () => { + const receipt = db.getOrCreateRunDelivery({ + runId: run.id, + consumerGeneration: run.consumer_generation + }) + expect(receipt?.messages).toHaveLength(50) + expect(receipt?.messages.some((message) => message.id === tail.id)).toBe(allFetched) + if (!receipt) { + throw new Error('Missing consuming receipt') + } + if (ack) { + db.acknowledgeRunDelivery({ + runId: run.id, + consumerGeneration: run.consumer_generation, + deliveryId: receipt.delivery.id + }) + } + } + const mailboxOwner = new OrchestrationMailboxOwner({ + getDb: () => db, + getLeaf: () => leaf, + getLeafKey: () => 'tab:leaf', + getTerminalHandleForLeafKey: () => 'term', + getTerminalProcessIncarnation: () => 'inc', + onRoutedMessageTypes: vi.fn(), + onForeignMailboxRouted: vi.fn() + }) + const deliveryTarget = new OrchestrationMailboxDeliveryTarget({ + getDb: () => db, + getTerminalHandleForPaneKey: () => 'term', + hasTerminalHandle: () => true, + isStructuredSessionOwner: () => false, + canProbePtyLiveness: () => false, + controllerKnowsPtyIsLive: () => true, + isLeafPtyProvenAbsent: async () => false + }) + let settleText: ((settlement: WriteSettlement) => void) | undefined + const writePty = vi.fn( + (_ptyId: string, data: string): WriteSettlement | Promise => { + if (data !== '\r' && boundary === 'during text write') { + return new Promise((resolve) => { + settleText = resolve + }) + } + return WRITE_ACCEPTED + } + ) + if (boundary === 'before staging') { + checkAndAck() + } + stageOrchestrationMailboxPointer({ + deps: { + mailboxOwner, + deliveryTarget, + getDb: () => db, + getLeaf: () => leaf, + getLeafKey: () => 'tab:leaf', + getLiveLeafForHandle: () => leaf, + isAgentSettledForDelivery: () => true, + getMessageWaiters: () => undefined, + getTabTitle: () => null, + getCliCommand: () => 'orca', + getTerminalHandleForLeafKey: () => 'term', + resolveSubmitTarget: () => ({ + leaf, + terminalHandle: 'term', + processIncarnation: 'inc' + }), + isLeafPtyProvenAbsent: async () => { + if (boundary === 'during liveness probe') { + checkAndAck() + } + return false + }, + redriveMailbox: vi.fn(), + writePty + }, + state, + leaf, + mailboxHandle, + messages, + newestSequence: tail.sequence, + enterDelayMs: 5, + leafKey: 'tab:leaf', + settle: (ptyId, flight) => state.settleFlight(ptyId, flight), + redrive: vi.fn() + }) + if (boundary === 'during text write' || boundary === 'before Enter') { + checkAndAck() + } + settleText?.(WRITE_ACCEPTED) + await vi.advanceTimersByTimeAsync(10) + expect(writePty.mock.calls.filter(([, data]) => data === '\r')).toHaveLength( + allFetched ? 0 : 1 + ) + expect(db.getMessageById(tail.id)).toMatchObject({ + read: allFetched && ack ? 1 : 0, + pointer_enter_pending: 0 + }) + if (allFetched) { + expect(db.getMessageById(tail.id)?.delivered_at).toBeNull() + } else { + expect(db.getMessageById(tail.id)?.delivered_at).not.toBeNull() + } + expect(state.hasFlight('pty')).toBe(false) + } finally { + state.retirePty('pty') + db.close() + vi.useRealTimers() + } + } + ) +}) diff --git a/src/main/runtime/orchestration/mailbox-pointer-stage.ts b/src/main/runtime/orchestration/mailbox-pointer-stage.ts index 3c0bea2baadd..395b29f68853 100644 --- a/src/main/runtime/orchestration/mailbox-pointer-stage.ts +++ b/src/main/runtime/orchestration/mailbox-pointer-stage.ts @@ -6,6 +6,7 @@ import type { } from './mailbox-pointer-delivery-contract' import { shouldReleaseOrchestrationPointer, + selectStagedOrchestrationPointerMessages, type OrchestrationMessageWaiter } from './mailbox-pointer-eligibility' import type { OrchestrationMailboxLeaf } from './mailbox-owner' @@ -16,6 +17,7 @@ import type { import { submitOrchestrationMailboxPointer } from './mailbox-pointer-submit' import type { OrchestrationMailboxPointerSubmitTarget } from './mailbox-pointer-submit' import { isSettledWrite, type WriteSettlement } from '../../../shared/pty-write-settlement' +import { MAILBOX_POINTER_WRITE_ATTEMPTED } from './db/messages/mailbox-pointer-enter-state' type StagePointerArgs = { deps: PointerDeliveryDependencies @@ -46,17 +48,19 @@ export function stageOrchestrationMailboxPointer message.id) @@ -142,6 +146,14 @@ function finishPointerWriteAndStageEnter message.id) + let messageIds = input.messages.map((message) => message.id) const reservationTarget = { ptyId: input.ptyId, processIncarnation: input.expectedTarget.processIncarnation @@ -99,18 +100,24 @@ export function submitOrchestrationMailboxPointer message.id) + if (eligibleIds.length === 0) { releaseWithoutRedrive = true + coveredBeforeEnter = true } else { preserveAmbiguousDelivery = true - const db = deps.getDb() + const eligible = new Set(eligibleIds) + const coveredIds = messageIds.filter((id) => !eligible.has(id)) + if (coveredIds.length > 0) { + db?.releaseMailboxPointerEnter(coveredIds, reservationTarget, [expectedPhase]) + } + messageIds = eligibleIds if (!db?.markMailboxPointerEnterAttempted(messageIds, reservationTarget)) { return } @@ -151,7 +158,15 @@ export function submitOrchestrationMailboxPointer { .split('\n') .filter((line) => line.trimStart().startsWith('orca orchestration')) - expect(commandLines).toHaveLength(5) + expect(commandLines).toHaveLength(6) expect(result).not.toContain('\\\n') expect(commandLines.filter((line) => line.includes('--type worker_done'))).toHaveLength(1) expect(commandLines.filter((line) => line.includes('--type heartbeat'))).toHaveLength(1) @@ -186,6 +186,19 @@ describe('buildDispatchPreamble', () => { expect(cadence).toContain('immediately before\n # you send worker_done') }) + it.each(['orca', 'orca-ide', 'orca-dev'] as const)( + 'teaches one whole-batch ack with %s', + (cliCommand) => { + const commands = cliFence(buildDispatchPreamble(baseParams({ cliCommand }))) + expect(commands).toContain('Process every message in the returned batch before acknowledging') + expect(commands).toContain('then process the next batch returned') + expect(commands.split('--ack')).toHaveLength(2) + expect(commands).toContain( + `${cliCommand} orchestration check --terminal term_worker --ack "" --json` + ) + } + ) + it('renders worker_done and heartbeat recipes bound to the exact Dispatch', () => { const result = buildDispatchPreamble(baseParams()) diff --git a/src/main/runtime/orchestration/preamble.ts b/src/main/runtime/orchestration/preamble.ts index 8af6aa756bb8..ef15b136fa5d 100644 --- a/src/main/runtime/orchestration/preamble.ts +++ b/src/main/runtime/orchestration/preamble.ts @@ -143,6 +143,10 @@ Don't post to Slack, GitHub, or other channels during the run; report through th # start a new file and after a test run — and once more immediately before # you send worker_done, so a redirect lands before the task settles. ${cli} orchestration check --terminal ${params.workerHandle} --json + + # Process every message in the returned batch before acknowledging, + # then process the next batch returned. Use its deliveryId, not a message ID. + ${cli} orchestration check --terminal ${params.workerHandle} --ack "" --json \`\`\` ${postDoneInstructions}` diff --git a/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts b/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts index f8170c3dc301..c2cf8a35fa70 100644 --- a/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts +++ b/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts @@ -36,6 +36,9 @@ function harness(options: { const mailbox = options.mailbox ?? 'dispatch:d1' const dispatchId = options.dispatchId === undefined ? 'd1' : options.dispatchId let attached = options.attached ?? true + let runtimeFence = 4 + let targetSession = IDENTITY.sessionId + let dbAvailable = true // The session's recorded sends, as its journal reports them. let submissions: StructuredPointerSubmission[] = [] // The mailbox's unread mail; a pointed message is no longer selected for a pointer. @@ -43,6 +46,9 @@ function harness(options: { { id: 'm1', type: 'status', sequence: 3, from_handle: 'term_coord', run_id: 'run_1' } ] const pointed = new Set() + const fetched = new Set(options.outstandingOwnDelivery ? ['m1'] : []) + const read = new Set() + let waiters: ReadonlySet<{ typeFilter: string[] | undefined }> | undefined const markAsDelivered = vi.fn((ids: string[]) => { for (const id of ids) { pointed.add(id) @@ -55,28 +61,56 @@ function harness(options: { ) const sendMock = vi.mocked(send) const stored = new Map() + const readSessionFacts = vi.fn(async () => + attached ? { submissions, queuedSends: [] } : null + ) const db = { getDispatchContextById: () => ({ run_id: 'run_1' }), hasOutstandingMailboxDelivery: (handle: string) => ((options.outstandingRunDelivery ?? false) && handle.startsWith('run:')) || ((options.outstandingOwnDelivery ?? false) && !handle.startsWith('run:')), - getUndeliveredUnreadMessages: () => mail.filter((message) => !pointed.has(message.id)), + getUndeliveredUnreadMessages: ( + _handle: string, + _types: unknown, + selection?: { + excludeFetched?: boolean + excludeMessageIds?: readonly string[] + excludeTypes?: readonly string[] + } + ) => + mail.filter( + (message) => + !pointed.has(message.id) && + !read.has(message.id) && + !(selection?.excludeFetched && fetched.has(message.id)) && + !selection?.excludeMessageIds?.includes(message.id) && + !selection?.excludeTypes?.includes(message.type) + ), markAsDelivered, + markUnpointedMailboxMessagesAsDelivered: (handle: string, ids: readonly string[]) => { + const unpointed = ids.filter( + (id) => handle === mailbox && !pointed.has(id) && mail.some((message) => message.id === id) + ) + if (unpointed.length > 0) { + markAsDelivered(unpointed) + } + }, getStructuredPointerOperation: (key: string) => stored.get(key), putStructuredPointerOperation: (row: StructuredPointerOperationRow) => stored.set(row.mailbox_handle, row), deleteStructuredPointerOperation: (key: string) => stored.delete(key) } const delivery = new OrchestrationStructuredMailboxPointerDelivery({ - getDb: () => db as never, - getMessageWaiters: () => undefined, + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: The fixture implements every database operation this lane invokes. + getDb: () => (dbAvailable ? db : null) as never, + getMessageWaiters: () => waiters, resolveStructuredTarget: (mailboxHandle) => - mailboxHandle === mailbox ? { sessionId: IDENTITY.sessionId, dispatchId } : null, + mailboxHandle === mailbox ? { sessionId: targetSession, dispatchId } : null, getCliCommand: () => 'orca-dev', senderName: (party) => (party.address === 'term_coord' ? 'Coordinator' : null), host: { - readSessionFacts: async () => (attached ? { submissions } : null), - currentFence: () => 4, + readSessionFacts, + currentFence: () => runtimeFence, send } }) @@ -85,6 +119,21 @@ function harness(options: { markAsDelivered, send: sendMock, stored, + readSessionFacts, + fetch: (...ids: string[]) => ids.forEach((id) => fetched.add(id)), + acknowledge: (...ids: string[]) => ids.forEach((id) => read.add(id)), + setWaiters: (next: typeof waiters) => { + waiters = next + }, + setFence: (next: number) => { + runtimeFence = next + }, + setTargetSession: (next: string) => { + targetSession = next + }, + setDbAvailable: (next: boolean) => { + dbAvailable = next + }, setAttached: (next: boolean) => { attached = next }, @@ -202,6 +251,37 @@ describe('structured mailbox pointer delivery', () => { }) }) + it.each([ + { kind: 'sent', state: 'accepted' }, + { kind: 'queued' }, + { kind: 'sent', state: 'rejected' }, + { kind: 'sent', state: 'unknown' }, + { kind: 'unattached' } + ] as const)( + 'rederives arrived mail only after a successful $kind $state handoff', + async (outcome) => { + const { delivery, send, receive } = harness({}) + let finish: (() => void) | undefined + send.mockImplementationOnce(() => new Promise((resolve) => (finish = () => resolve(outcome)))) + delivery.deliverForHandle('dispatch:d1') + await flush() + expect(send).toHaveBeenCalledTimes(1) + receive('m2', 4) + delivery.deliverForHandle('dispatch:d1') + finish?.() + await flush() + await flush() + const handedOff = + outcome.kind === 'queued' || (outcome.kind === 'sent' && outcome.state === 'accepted') + expect(send).toHaveBeenCalledTimes(handedOff ? 2 : 1) + if (handedOff) { + expect(send.mock.calls[1]![0].body.from?.orchestration).toMatchObject({ + messages: [{ messageId: 'm2' }] + }) + } + } + ) + it('nudges the worker while its coordinator holds an unacked Run delivery', async () => { // The exact window in which a coordinator replies to its workers: it checked, is acting on the // batch, and has not acked yet. The gate is keyed on the handle being nudged, so the @@ -216,13 +296,178 @@ describe('structured mailbox pointer delivery', () => { expect(markAsDelivered).toHaveBeenCalledWith(['m1']) }) - it('does not re-nudge a mailbox still holding its own unacked batch', async () => { - // The other half of the same gate: the consumer already has this batch, so a second nudge - // spends a whole provider turn telling it something it was told. - const { delivery, send } = harness({ outstandingOwnDelivery: true }) + it('does not re-nudge exact members of its own unacked batch', async () => { + const { delivery, send, receive } = harness({ outstandingOwnDelivery: true }) delivery.deliverForHandle('dispatch:d1') await flush() expect(send).not.toHaveBeenCalled() + receive('m2', 4) + delivery.deliverForHandle('dispatch:d1') + await flush() + expect(send).toHaveBeenCalledTimes(1) + expect(send.mock.calls[0]![0].body.from?.orchestration).toMatchObject({ + messages: [{ messageId: 'm2' }] + }) + }) + + it.each(['pending', 'accepted'] as const)( + 'reconciles the original batch when a %s notice loses one member to check', + async (dispatchState) => { + const { delivery, send, stored, receive, fetch, setSubmissions, markAsDelivered } = harness({ + dispatchState: 'unknown' + }) + receive('m2', 4) + delivery.deliverForHandle('dispatch:d1') + await flush() + const operationId = send.mock.calls[0]![0].operationId + setSubmissions([ + { + clientMessageId: operationId, + dispatchState, + submittedAt: Date.now(), + mailNotice: { mailbox: 'dispatch:d1', messageIds: ['m1', 'm2'] } + } + ]) + fetch('m1') + delivery.onJournalActivity('session-1') + await flush() + expect(send).toHaveBeenCalledTimes(1) + if (dispatchState === 'pending') { + expect(markAsDelivered).not.toHaveBeenCalled() + expect(stored.get('dispatch:d1')?.operation_id).toBe(operationId) + } else { + expect(markAsDelivered).toHaveBeenCalledWith(['m1', 'm2']) + expect(stored.has('dispatch:d1')).toBe(false) + } + } + ) + + it('notifies independent mail while the journal owns an earlier pending batch', async () => { + const { delivery, send, receive, fetch, setSubmissions, markAsDelivered } = harness({ + dispatchState: 'unknown' + }) + receive('m2', 4) + delivery.deliverForHandle('dispatch:d1') + await flush() + const first = send.mock.calls[0]![0].operationId + setSubmissions([ + { + clientMessageId: first, + dispatchState: 'pending', + submittedAt: Date.now(), + mailNotice: { mailbox: 'dispatch:d1', messageIds: ['m1', 'm2'] } + } + ]) + fetch('m1') + receive('m3', 5) + delivery.onJournalActivity('session-1') + await flush() + expect(send).toHaveBeenCalledTimes(2) + expect(send.mock.calls[1]![0].operationId).not.toBe(first) + expect(send.mock.calls[1]![0].body.from?.orchestration).toMatchObject({ + messages: [{ messageId: 'm3' }] + }) + expect(markAsDelivered).not.toHaveBeenCalled() + }) + + it.each(['fetched', 'acknowledged', 'waiter'] as const)( + 'revalidates mail that becomes %s while reading the journal', + async (change) => { + const { delivery, send, stored, readSessionFacts, fetch, acknowledge, setWaiters } = harness( + {} + ) + let release: (() => void) | undefined + readSessionFacts.mockImplementationOnce( + () => + new Promise((resolve) => (release = () => resolve({ submissions: [], queuedSends: [] }))) + ) + delivery.deliverForHandle('dispatch:d1') + await flush() + if (change === 'fetched') { + fetch('m1') + } + if (change === 'acknowledged') { + acknowledge('m1') + } + if (change === 'waiter') { + setWaiters(new Set([{ typeFilter: ['status'] }])) + } + release?.() + await flush() + expect(send).not.toHaveBeenCalled() + expect(stored.size).toBe(0) + } + ) + + it('reconciles original acceptance after an independent accepted notice replaced its operation row', async () => { + const { delivery, send, receive, fetch, setSubmissions, markAsDelivered } = harness({ + dispatchState: 'unknown' + }) + receive('m2', 4) + delivery.deliverForHandle('dispatch:d1') + await flush() + const first = send.mock.calls[0]![0].operationId + const original: StructuredPointerSubmission = { + clientMessageId: first, + dispatchState: 'pending', + submittedAt: Date.now(), + mailNotice: { mailbox: 'dispatch:d1', messageIds: ['m1', 'm2'] } + } + setSubmissions([original]) + fetch('m1') + receive('m3', 5) + send.mockResolvedValueOnce({ kind: 'sent', state: 'accepted' }) + delivery.onJournalActivity('session-1') + await flush() + expect(send).toHaveBeenCalledTimes(2) + expect(markAsDelivered).toHaveBeenCalledWith(['m3']) + delivery.onJournalActivity('session-1') + await flush() + expect(send).toHaveBeenCalledTimes(2) + setSubmissions([{ ...original, dispatchState: 'accepted' }]) + delivery.onJournalActivity('session-1') + await flush() + expect(send).toHaveBeenCalledTimes(2) + expect(markAsDelivered).toHaveBeenCalledWith(['m1', 'm2']) + }) + + it.each(['lease', 'database'] as const)( + 'does not submit against %s authority that changed during the journal read', + async (change) => { + const { delivery, send, stored, readSessionFacts, setFence, setDbAvailable } = harness({}) + let release: (() => void) | undefined + readSessionFacts.mockImplementationOnce( + () => + new Promise((resolve) => (release = () => resolve({ submissions: [], queuedSends: [] }))) + ) + delivery.deliverForHandle('dispatch:d1') + await flush() + if (change === 'lease') { + setFence(5) + } else { + setDbAvailable(false) + } + release?.() + await flush() + expect(send).not.toHaveBeenCalled() + expect(stored.size).toBe(0) + } + ) + + it('reselects for the successor when clear moves the mailbox during its journal read', async () => { + const { delivery, send, readSessionFacts, setTargetSession } = harness({}) + let release: (() => void) | undefined + readSessionFacts.mockImplementationOnce( + () => + new Promise((resolve) => (release = () => resolve({ submissions: [], queuedSends: [] }))) + ) + delivery.deliverForHandle('dispatch:d1') + await flush() + setTargetSession('successor') + release?.() + await flush() + expect(send).toHaveBeenCalledTimes(1) + expect(send.mock.calls[0]![0].sessionId).toBe('successor') }) it('retries a rejected nudge on the next journal edge, under the same id', async () => { @@ -402,13 +647,23 @@ describe('forgetting one settled worker', () => { kind: 'sent' as const, state: 'accepted' as const })) + const pointed = new Set() const db = { getDispatchContextById: () => ({ run_id: 'run_1' }), hasOutstandingMailboxDelivery: () => false, - getUndeliveredUnreadMessages: () => [ - { id: 'm1', type: 'status', sequence: 3, from_handle: 'term_coord', run_id: 'run_1' } - ], - markAsDelivered: vi.fn(), + getUndeliveredUnreadMessages: (mailbox: string) => + pointed.has(mailbox) + ? [] + : [ + { + id: mailbox, + type: 'status', + sequence: 3, + from_handle: 'term_coord', + run_id: 'run_1' + } + ], + markAsDelivered: vi.fn((ids: string[]) => ids.forEach((id) => pointed.add(id))), getStructuredPointerOperation: () => undefined, putStructuredPointerOperation: () => {}, deleteStructuredPointerOperation: () => {} @@ -425,7 +680,7 @@ describe('forgetting one settled worker', () => { getCliCommand: () => 'orca', senderName: () => null, host: { - readSessionFacts: async () => (attached ? { submissions: [] } : null), + readSessionFacts: async () => (attached ? { submissions: [], queuedSends: [] } : null), currentFence: () => 4, send } @@ -523,7 +778,7 @@ describe('a mailbox a /clear moves while its nudge is in flight', () => { getCliCommand: () => 'orca-dev', senderName: () => null, host: { - readSessionFacts: async () => ({ submissions: [] }), + readSessionFacts: async () => ({ submissions: [], queuedSends: [] }), currentFence: () => 4, send } diff --git a/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.ts b/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.ts index d8a369ce4977..fe7df91705ba 100644 --- a/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.ts +++ b/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.ts @@ -13,7 +13,7 @@ */ import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' -import type { MessageRow, OrchestrationDb } from './db' +import type { OrchestrationDb } from './db' import { formatMessagePointer } from './formatter' import type { OrchestrationCliCommand } from './cli-command' import { @@ -22,7 +22,9 @@ import { } from './mailbox-pointer-eligibility' import { resolveStructuredPointerOperation, - type StructuredPointerSubmission + reconcileStructuredPointerOperations, + type StructuredPointerSubmission, + type StructuredPointerQueuedSend } from './structured-pointer-operation-id' import { structuredMailSource } from './structured-mail-source' import type { SenderNameResolver } from './agent-message-sender' @@ -57,6 +59,8 @@ export type StructuredPointerSendOutcome = export type StructuredPointerSessionFacts = { /** Every send the session recorded, oldest first: what the lane's own sends settled as. */ submissions: readonly StructuredPointerSubmission[] + /** Original queue acceptance, including cards later consumed, withdrawn or carried. */ + queuedSends: readonly StructuredPointerQueuedSend[] } export type StructuredMailboxPointerHost = { @@ -170,32 +174,27 @@ export class OrchestrationStructuredMailboxPointerDelivery< if (!db || this.inFlight.has(mailboxHandle)) { return } - // Don't re-nudge a mailbox whose consumer still holds an unacknowledged batch. The lookup is - // keyed on the exact handle being nudged, so a coordinator's own `run:` delivery is invisible - // to a worker's `dispatch:` gate and cannot suppress the nudges a coordinator sends its - // workers. Worth more here than in the PTY lane: a structured nudge costs a whole provider - // turn, not a line of text into a composer. - if (db.hasOutstandingMailboxDelivery?.(mailboxHandle)) { - return - } - const unread = selectOrchestrationPointerBatch({ - db, - mailboxHandle, - waiters: this.deps.getMessageWaiters(mailboxHandle), - reservedTypes - }) - if (unread.length === 0) { - return - } this.inFlight.add(mailboxHandle) + let handedOff = false try { - await this.attempt(db, mailboxHandle, target, unread, reservedTypes) + handedOff = await this.attempt(db, mailboxHandle, target, reservedTypes) } finally { this.inFlight.delete(mailboxHandle) // A thrown attempt follows too; its own failure is what still propagates. await this.followMovedTarget(mailboxHandle, target, reservedTypes, attemptedSessions).catch( () => undefined ) + const current = this.deps.resolveStructuredTarget(mailboxHandle) + if ( + handedOff && + current?.sessionId === target.sessionId && + current.dispatchId === target.dispatchId + ) { + // Mail arriving during handoff lost its in-flight edge; rederive after success only. + await this.deliver(mailboxHandle, current, reservedTypes, attemptedSessions).catch( + () => undefined + ) + } } } @@ -226,19 +225,57 @@ export class OrchestrationStructuredMailboxPointerDelivery< db: OrchestrationDb, mailboxHandle: string, target: StructuredPointerTarget, - unread: readonly MessageRow[], reservedTypes: ReadonlySet | undefined - ): Promise { + ): Promise { + if ( + !db.getStructuredPointerOperation(mailboxHandle) && + selectOrchestrationPointerBatch({ + db, + mailboxHandle, + waiters: this.deps.getMessageWaiters(mailboxHandle), + reservedTypes + }).length === 0 + ) { + return false + } const sessionId = target.sessionId + const fence = this.deps.host.currentFence(sessionId) const session = await this.deps.host.readSessionFacts(sessionId) - if (!session) { - this.retain(mailboxHandle, sessionId, 'session-not-attached', reservedTypes) - return + const current = this.deps.resolveStructuredTarget(mailboxHandle) + if ( + this.deps.getDb() !== db || + current?.sessionId !== sessionId || + current.dispatchId !== target.dispatchId + ) { + return false } - const fence = this.deps.host.currentFence(sessionId) - if (fence === null) { + if (!session || fence === null || this.deps.host.currentFence(sessionId) !== fence) { this.retain(mailboxHandle, sessionId, 'session-not-attached', reservedTypes) - return + return false + } + const ownership = reconcileStructuredPointerOperations({ + db, + mailboxHandle, + sessionId, + submissions: session.submissions, + queuedSends: session.queuedSends + }) + if (!db.getStructuredPointerOperation(mailboxHandle)) { + this.sentOperationIds.delete(mailboxHandle) + } + if (ownership.pendingMessageIds.length > 0) { + this.retain(mailboxHandle, sessionId, 'turn-unsettled', reservedTypes) + } + // A check or waiter may have acquired mail while the journal was being read. + const unread = selectOrchestrationPointerBatch({ + db, + mailboxHandle, + waiters: this.deps.getMessageWaiters(mailboxHandle), + reservedTypes, + excludeMessageIds: ownership.ownedMessageIds + }) + if (unread.length === 0) { + return false } const body: AgentJournalMessageItem = { kind: 'message', @@ -264,18 +301,19 @@ export class OrchestrationStructuredMailboxPointerDelivery< sessionId, messageIds: staged, submissions: session.submissions, + queuedSends: session.queuedSends, sentByThisProcess: this.sentOperationIds.get(mailboxHandle) }) if (operation.kind === 'stamp') { // A send this lane gave up waiting on ran after all. - db.markAsDelivered(staged) + db.markAsDelivered([...operation.messageIds]) db.deleteStructuredPointerOperation(mailboxHandle) this.sentOperationIds.delete(mailboxHandle) - return + return true } if (operation.kind === 'park') { this.retain(mailboxHandle, sessionId, 'turn-unsettled', reservedTypes) - return + return false } this.sentOperationIds.set(mailboxHandle, operation.operationId) const outcome = await this.deps.host.send({ @@ -287,19 +325,20 @@ export class OrchestrationStructuredMailboxPointerDelivery< }) if (outcome.kind === 'unattached') { this.retain(mailboxHandle, sessionId, 'session-not-attached', reservedTypes) - return + return false } // A queued pointer is the chat's queue's to send, as a person's queued message is. if (outcome.kind === 'sent' && !structuredDispatchDelivered(outcome.state)) { // The row stays: resending under its id replays this verdict and starts nothing. this.retain(mailboxHandle, sessionId, retainReasonForDispatch(outcome.state), reservedTypes) - return + return false } db.markAsDelivered(staged) // The nudge landed as its own turn, so the next settle edge is the natural retry point for // anything that arrives while it runs. db.deleteStructuredPointerOperation(mailboxHandle) this.sentOperationIds.delete(mailboxHandle) + return true } /** diff --git a/src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts b/src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts index fb0b8313cfd5..7773a296f386 100644 --- a/src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts +++ b/src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts @@ -1,6 +1,11 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import type { AgentJournalRenderItem } from '../../../shared/agent-session-journal-types' import type { AgentMessageSource } from '../../../shared/agent-session-message-source' +import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' +import { + resolveStructuredPointerOperation, + structuredPointerBatchFingerprint +} from './structured-pointer-operation-id' const hostRef: { current: unknown } = { current: null } @@ -35,6 +40,10 @@ function transcript(count: number): AgentJournalRenderItem[] { ) } +function pointerSnapshot(snapshot: { items: readonly unknown[]; submissions: readonly unknown[] }) { + return { mailboxPointerSnapshot: () => ({ snapshot, queuedMessages: [] }) } +} + const NOTICE_SOURCE: AgentMessageSource = { kind: 'agent', senders: [], @@ -61,23 +70,299 @@ describe('structured mailbox pointer host', () => { it("reads what the session's sends settled as", async () => { const submissions = [{ clientMessageId: 'op1', dispatchState: 'unknown' }] - hostRef.current = { journalSnapshot: () => ({ items: [], submissions }) } + hostRef.current = pointerSnapshot({ items: [], submissions }) expect(await createStructuredMailboxPointerHost().readSessionFacts('s1')).toEqual({ + queuedSends: [], submissions }) }) + it.each(['pending', 'accepted'] as const)( + 'projects original mailbox members from the %s submission body', + async (dispatchState) => { + const submissions = [{ clientMessageId: 'op1', dispatchState, submittedAt: 1 }] + const item: AgentJournalRenderItem = { + itemId: agentJournalSubmissionKey('op1'), + revision: 1, + sequence: 1, + observedAt: 1, + body: { + kind: 'message', + role: 'user', + blocks: [], + from: { + ...NOTICE_SOURCE, + orchestration: { + message: 'mail-notice', + mailbox: 'dispatch:d1', + dispatchId: 'd1', + messages: [ + { messageId: 'm1', runId: 'r1', from: 'term_coord' }, + { messageId: 'm2', runId: 'r1', from: 'term_coord' } + ] + } + } + } + } + hostRef.current = pointerSnapshot({ items: [item], submissions }) + expect(await createStructuredMailboxPointerHost().readSessionFacts('s1')).toEqual({ + queuedSends: [], + submissions: [ + { + ...submissions[0], + mailNotice: { mailbox: 'dispatch:d1', messageIds: ['m1', 'm2'] } + } + ] + }) + } + ) + + it.each(['completed', 'interrupted', 'running', 'unverifiable'] as const)( + 'derives legacy operation settlement from its own %s turn', + async (state) => { + const submission = { clientMessageId: 'op1', dispatchState: 'accepted', submittedAt: 1 } + const turn: AgentJournalRenderItem = { + itemId: 'turn-item-1', + revision: 1, + sequence: 2, + observedAt: 2, + body: { + kind: 'turn', + turnId: 'turn-1', + state, + userItemId: agentJournalSubmissionKey('op1') + } + } + hostRef.current = pointerSnapshot({ items: [turn], submissions: [submission] }) + expect(await createStructuredMailboxPointerHost().readSessionFacts('s1')).toEqual({ + queuedSends: [], + submissions: [ + { + ...submission, + ...(['completed', 'interrupted'].includes(state) ? { turnSettled: true } : {}) + } + ] + }) + } + ) + + it('does not treat an unrelated settled turn as the original operation settling', async () => { + const submission = { clientMessageId: 'op1', dispatchState: 'accepted', submittedAt: 1 } + const turn: AgentJournalRenderItem = { + itemId: 'unrelated-turn', + revision: 1, + sequence: 2, + observedAt: 2, + body: { kind: 'turn', turnId: 'turn-2', state: 'completed', userItemId: 'another-send' } + } + hostRef.current = pointerSnapshot({ items: [turn], submissions: [submission] }) + expect(await createStructuredMailboxPointerHost().readSessionFacts('s1')).toEqual({ + queuedSends: [], + submissions: [submission] + }) + }) + + it('reads legacy settlement through a send`s explicit turn scope', async () => { + const submission = { clientMessageId: 'op1', dispatchState: 'accepted', submittedAt: 1 } + const items: AgentJournalRenderItem[] = [ + { + itemId: agentJournalSubmissionKey('op1'), + revision: 1, + sequence: 1, + observedAt: 1, + turnScope: { kind: 'turn', turnItemId: 'turn-item-1' }, + body: { kind: 'message', role: 'user', blocks: [] } + }, + { + itemId: 'turn-item-1', + revision: 1, + sequence: 2, + observedAt: 2, + body: { kind: 'turn', turnId: 'turn-1', state: 'completed' } + } + ] + hostRef.current = pointerSnapshot({ items, submissions: [submission] }) + expect(await createStructuredMailboxPointerHost().readSessionFacts('s1')).toEqual({ + queuedSends: [], + submissions: [{ ...submission, turnSettled: true }] + }) + }) + it('answers null rather than nothing recorded when the session cannot be read', async () => { // Null retains the pointer; an empty answer would send into a session this runtime cannot see. expect(await createStructuredMailboxPointerHost().readSessionFacts('s1')).toBeNull() hostRef.current = { - journalSnapshot: () => { + mailboxPointerSnapshot: () => { throw new Error('agent_session_ownership_unknown') } } expect(await createStructuredMailboxPointerHost().readSessionFacts('s1')).toBeNull() }) + it('retains unreadable queue ownership even when journal-only facts are readable', async () => { + hostRef.current = { + journalSnapshot: () => ({ items: [], submissions: [] }), + mailboxPointerSnapshot: () => { + throw new Error('queue read failed') + } + } + expect(await createStructuredMailboxPointerHost().readSessionFacts('s1')).toBeNull() + }) + + it.each(['waiting', 'returned', 'dispatched', 'withdrawn'] as const)( + 'projects historical acceptance of a %s queue row under its original identity', + async (state) => { + const body = { + kind: 'message', + role: 'user', + blocks: [], + from: { + ...NOTICE_SOURCE, + orchestration: { + message: 'mail-notice', + mailbox: 'dispatch:d1', + dispatchId: 'd1', + messages: [{ messageId: 'm1', runId: 'r1', from: 'term_coord' }] + } + } + } + hostRef.current = { + mailboxPointerSnapshot: () => ({ + snapshot: { items: [], submissions: [] }, + queuedMessages: [{ messageId: 'original', state, body, consumedAs: 'fresh' }] + }) + } + expect(await createStructuredMailboxPointerHost().readSessionFacts('s1')).toEqual({ + submissions: [], + queuedSends: [ + { + operationId: 'original', + mailNotice: { mailbox: 'dispatch:d1', messageIds: ['m1'] }, + ...(state === 'withdrawn' ? { settled: true } : {}) + } + ] + }) + } + ) + + it.each(['withdrawn', 'completed', 'interrupted', 'running', 'unverifiable', 'waiting'] as const)( + 'ends missing-provenance queue ownership only with its own proven %s lifecycle', + async (state) => { + const turn: AgentJournalRenderItem = { + itemId: 'own-turn', + revision: 1, + sequence: 2, + observedAt: 2, + body: { + kind: 'turn', + turnId: 'turn-1', + state: state === 'withdrawn' || state === 'waiting' ? 'completed' : state, + userItemId: agentJournalSubmissionKey('fresh-handoff') + } + } + hostRef.current = { + mailboxPointerSnapshot: () => ({ + snapshot: { + items: [turn], + submissions: [ + { + clientMessageId: 'fresh-handoff', + queuedMessageId: 'original', + dispatchState: 'accepted', + submittedAt: 1, + providerItemId: null + } + ] + }, + queuedMessages: [ + { + messageId: 'original', + state: state === 'withdrawn' || state === 'waiting' ? state : 'dispatched', + consumedAs: 'fresh-handoff', + body: { kind: 'message', role: 'user', blocks: [] } + } + ] + }) + } + const facts = await createStructuredMailboxPointerHost().readSessionFacts('s1') + if (!facts) { + throw new Error('fixture facts were unreadable') + } + const put = vi.fn() + const outcome = resolveStructuredPointerOperation({ + db: { + getStructuredPointerOperation: () => ({ + mailbox_handle: 'dispatch:d1', + session_id: 's1', + operation_id: 'original', + batch_fingerprint: structuredPointerBatchFingerprint('s1', ['old']), + minted_at_ms: 1 + }), + putStructuredPointerOperation: put + }, + mailboxHandle: 'dispatch:d1', + sessionId: 's1', + messageIds: ['new'], + submissions: facts.submissions, + queuedSends: facts.queuedSends, + sentByThisProcess: undefined + }) + const ended = state === 'withdrawn' || state === 'completed' || state === 'interrupted' + expect(outcome.kind).toBe(ended ? 'send' : 'park') + expect(put).toHaveBeenCalledTimes(ended ? 1 : 0) + } + ) + + it('does not let an older ended handoff retire a newer handoff across a clock step', async () => { + hostRef.current = { + mailboxPointerSnapshot: () => ({ + snapshot: { + items: [ + { + itemId: 'old-turn', + revision: 1, + sequence: 2, + observedAt: 2, + body: { + kind: 'turn', + turnId: 'old', + state: 'completed', + userItemId: agentJournalSubmissionKey('old-handoff') + } + } + ], + submissions: [ + { + clientMessageId: 'new-handoff', + queuedMessageId: 'original', + submittedSequence: 3, + submittedAt: 1, + dispatchState: 'pending' + }, + { + clientMessageId: 'old-handoff', + queuedMessageId: 'original', + submittedSequence: 1, + submittedAt: 10, + dispatchState: 'accepted' + } + ] + }, + queuedMessages: [ + { + messageId: 'original', + state: 'dispatched', + consumedAs: 'new-handoff', + body: { kind: 'message', role: 'user', blocks: [] } + } + ] + }) + } + expect( + (await createStructuredMailboxPointerHost().readSessionFacts('s1'))?.queuedSends + ).toEqual([{ operationId: 'original' }]) + }) + it('reports an unattached host rather than a rejection when nothing can be sent', async () => { await expect( createStructuredMailboxPointerHost().send({ diff --git a/src/main/runtime/orchestration/structured-mailbox-pointer-host.ts b/src/main/runtime/orchestration/structured-mailbox-pointer-host.ts index f81ef2c229f3..f0addf15ec2a 100644 --- a/src/main/runtime/orchestration/structured-mailbox-pointer-host.ts +++ b/src/main/runtime/orchestration/structured-mailbox-pointer-host.ts @@ -8,11 +8,22 @@ import { AGENT_SESSION_NOT_ATTACHED } from '../../native-chat/agent-session-wire/structured-agent-session-mutation-admission' import { getStructuredAgentSessionHost } from '../../native-chat/agent-session-wire/structured-agent-session-registry' +import type { StructuredAgentSessionHost } from '../../native-chat/agent-session-wire/structured-agent-session-host' +import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' +import { isRootAgentJournalItem } from '../../../shared/agent-session-journal-producer' +import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' import type { StructuredMailboxPointerHost, StructuredPointerSessionFacts } from './structured-mailbox-pointer-delivery' -import type { AgentJournalSnapshot } from '../../../shared/agent-session-journal-types' +import type { + AgentJournalItemBody, + AgentJournalSnapshot +} from '../../../shared/agent-session-journal-types' +import type { + StructuredPointerQueuedSend, + StructuredPointerSubmission +} from './structured-pointer-operation-id' import { structuredSessionGateFacts, type StructuredSessionGateFacts @@ -54,21 +65,107 @@ export async function readStructuredSessionGateFacts( async function readPointerSessionFacts( sessionId: string ): Promise { - const snapshot = await readSessionJournal(sessionId) - return snapshot ? { submissions: snapshot.submissions } : null + const state = await readSession(sessionId, (host) => host.mailboxPointerSnapshot(sessionId)) + if (!state) { + return null + } + const { snapshot, queuedMessages } = state + const items = new Map(snapshot.items.map((item) => [item.itemId, item])) + const queuedSends = new Map() + const queuedRows = new Map(queuedMessages.map((row) => [row.messageId, row])) + const handoffSequences = new Map() + for (const row of queuedMessages) { + const mailNotice = readMailNotice(row.body) + queuedSends.set(row.messageId, { + operationId: row.messageId, + ...(mailNotice ? { mailNotice } : {}), + ...(row.state === 'withdrawn' ? { settled: true as const } : {}) + }) + } + const settledTurns = new Set() + const settledOpeners = new Set() + for (const item of snapshot.items) { + const turn = readAgentJournalTurn(item.body) + if ( + isRootAgentJournalItem(item) && + (turn?.state === 'completed' || turn?.state === 'interrupted') + ) { + settledTurns.add(item.itemId) + if (turn.userItemId) { + settledOpeners.add(turn.userItemId) + } + } + } + return { + submissions: snapshot.submissions.map((submission) => { + const submissionKey = agentJournalSubmissionKey(submission.clientMessageId) + const item = items.get(submissionKey) + const mailNotice = readMailNotice(item?.body) + const turnSettled = + settledOpeners.has(submissionKey) || + (submission.providerItemId !== null && settledOpeners.has(submission.providerItemId)) || + (item?.turnScope?.kind === 'turn' && settledTurns.has(item.turnScope.turnItemId)) + if (submission.queuedMessageId) { + const operationId = submission.queuedMessageId + const previous = queuedSends.get(operationId) + const notice = previous?.mailNotice ?? mailNotice + const draft = queuedRows.get(operationId) + const sequence = submission.submittedSequence ?? 0 + const latest = sequence >= (handoffSequences.get(operationId) ?? -1) + const settled = latest + ? draft?.state === 'withdrawn' || + (!draft && submission.dispatchState === 'rejected') || + ((!draft || + (draft.state === 'dispatched' && draft.consumedAs === submission.clientMessageId)) && + turnSettled) + : previous?.settled + if (latest) { + handoffSequences.set(operationId, sequence) + } + // A fresh handoff still names its original card after retention prunes the row. + queuedSends.set(operationId, { + operationId, + ...(notice ? { mailNotice: notice } : {}), + ...(settled ? { settled: true as const } : {}) + }) + } + return { + ...submission, + ...(turnSettled ? { turnSettled: true as const } : {}), + ...(mailNotice ? { mailNotice } : {}) + } + }), + queuedSends: [...queuedSends.values()] + } +} + +function readMailNotice( + body: AgentJournalItemBody | undefined +): StructuredPointerSubmission['mailNotice'] { + const notice = body?.kind === 'message' && body.role === 'user' ? body.from?.orchestration : null + return notice?.message === 'mail-notice' + ? { mailbox: notice.mailbox, messageIds: notice.messages.map((message) => message.messageId) } + : undefined +} + +function readSessionJournal(sessionId: string): Promise { + return readSession(sessionId, (host) => host.journalSnapshot(sessionId)) } -async function readSessionJournal(sessionId: string): Promise { +async function readSession( + sessionId: string, + read: (host: StructuredAgentSessionHost) => Promise +): Promise { const host = getStructuredAgentSessionHost() if (!host) { return null } try { // Opens a conversation the idle sweep closed; that starts no agent. - return await host.journalSnapshot(sessionId) + return await read(host) } catch (error) { // Not attached is a retain reason, not a failure; anything else is still unreadable. - if ((error as Error)?.message !== AGENT_SESSION_NOT_ATTACHED.code) { + if (!(error instanceof Error && error.message === AGENT_SESSION_NOT_ATTACHED.code)) { console.warn('[orchestration] structured journal unreadable', sessionId, error) } return null diff --git a/src/main/runtime/orchestration/structured-pointer-operation-id.test.ts b/src/main/runtime/orchestration/structured-pointer-operation-id.test.ts index f828409aeffe..3c06154fe987 100644 --- a/src/main/runtime/orchestration/structured-pointer-operation-id.test.ts +++ b/src/main/runtime/orchestration/structured-pointer-operation-id.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from 'vitest' +import type { StructuredPointerOperationRow } from './db/messages/structured-pointer-operation-store' import { AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS, AGENT_SESSION_MAX_OPERATION_REPLAY_AGE_MS @@ -7,6 +8,8 @@ import { decideStructuredPointerAttempt, mintAgentSessionOperationId, resolveStructuredPointerOperation, + structuredPointerBatchFingerprint, + type StructuredPointerOperationStore, type StructuredPointerSubmission } from './structured-pointer-operation-id' @@ -32,14 +35,13 @@ function resolveId( return resolved } -function fakeDb() { - const rows = new Map() +function fakeDb(): StructuredPointerOperationStore { + const rows = new Map() return { - rows, getStructuredPointerOperation: (handle: string) => rows.get(handle), - putStructuredPointerOperation: (row: { mailbox_handle: string; operation_id: string }) => + putStructuredPointerOperation: (row: StructuredPointerOperationRow) => rows.set(row.mailbox_handle, row) - } as never + } } describe('structured pointer operation id', () => { @@ -169,6 +171,154 @@ describe('structured pointer operation id', () => { }) expect(moved.operationId).not.toBe(first.operationId) }) + + it.each(['pending', 'accepted'] as const)( + 'retains a changed batch when a legacy %s submission cannot identify original members', + (dispatchState) => { + const db = fakeDb() + const input = { db, mailboxHandle: 'dispatch:d1', sessionId: 's1', now: 1_000 } + const first = resolveId({ ...input, messageIds: ['m1', 'm2'] }) + expect( + resolveStructuredPointerOperation({ + ...input, + messageIds: ['m2'], + submissions: [{ clientMessageId: first.operationId, dispatchState, submittedAt: 1_000 }], + sentByThisProcess: first.operationId + }) + ).toEqual({ kind: 'park' }) + expect(db.getStructuredPointerOperation('dispatch:d1')?.operation_id).toBe(first.operationId) + } + ) + + it('stamps the original proven batch rather than an eligible subset', () => { + const db = fakeDb() + const input = { db, mailboxHandle: 'dispatch:d1', sessionId: 's1', now: 1_000 } + const first = resolveId({ ...input, messageIds: ['m1', 'm2'] }) + expect( + resolveStructuredPointerOperation({ + ...input, + messageIds: ['m2'], + submissions: [ + { + clientMessageId: first.operationId, + dispatchState: 'accepted', + submittedAt: 1_000, + mailNotice: { mailbox: 'dispatch:d1', messageIds: ['m1', 'm2'] } + } + ], + sentByThisProcess: first.operationId + }) + ).toEqual({ kind: 'stamp', messageIds: ['m1', 'm2'] }) + }) + + it('settles queue acceptance across actor and session changes using the original fingerprint', () => { + const db = fakeDb() + const input = { db, mailboxHandle: 'dispatch:d1', sessionId: 's1', now: 1_000 } + const first = resolveId({ ...input, messageIds: ['m1', 'm2'] }) + expect( + resolveStructuredPointerOperation({ + ...input, + sessionId: 'successor', + messageIds: ['m2'], + submissions: [], + queuedSends: [ + { + operationId: first.operationId, + mailNotice: { mailbox: 'dispatch:d1', messageIds: ['m1', 'm2'] } + } + ], + sentByThisProcess: undefined + }) + ).toEqual({ kind: 'stamp', messageIds: ['m1', 'm2'] }) + }) + + it('retains queue acceptance with unverifiable original provenance instead of minting a duplicate', () => { + const db = fakeDb() + const input = { db, mailboxHandle: 'dispatch:d1', sessionId: 's1', now: 1_000 } + const first = resolveId({ ...input, messageIds: ['m1', 'm2'] }) + expect( + resolveStructuredPointerOperation({ + ...input, + messageIds: ['m2'], + submissions: [], + queuedSends: [{ operationId: first.operationId }], + sentByThisProcess: undefined + }) + ).toEqual({ kind: 'park' }) + expect(db.getStructuredPointerOperation('dispatch:d1')?.operation_id).toBe(first.operationId) + }) + + it('admits independent messages without replacing an overlapping pending notice', () => { + const db = fakeDb() + const input = { db, mailboxHandle: 'dispatch:d1', sessionId: 's1', now: 1_000 } + const first = resolveId({ ...input, messageIds: ['m1', 'm2'] }) + const submissions: StructuredPointerSubmission[] = [ + { + clientMessageId: first.operationId, + dispatchState: 'pending', + submittedAt: 1_000, + mailNotice: { mailbox: 'dispatch:d1', messageIds: ['m1', 'm2'] } + } + ] + expect( + resolveStructuredPointerOperation({ + ...input, + messageIds: ['m2', 'm3'], + submissions, + sentByThisProcess: first.operationId + }) + ).toEqual({ kind: 'park' }) + const independent = resolveId({ ...input, messageIds: ['m3'], submissions }) + expect(independent.operationId).not.toBe(first.operationId) + }) + + it('requires mailbox and fingerprint agreement before trusting projected original IDs', () => { + const db = fakeDb() + const input = { db, mailboxHandle: 'dispatch:d1', sessionId: 's1', now: 1_000 } + const first = resolveId({ ...input, messageIds: ['m1', 'm2'] }) + for (const mailNotice of [ + { mailbox: 'dispatch:other', messageIds: ['m1', 'm2'] }, + { mailbox: 'dispatch:d1', messageIds: ['m2'] } + ]) { + expect( + resolveStructuredPointerOperation({ + ...input, + messageIds: ['m3'], + submissions: [ + { + clientMessageId: first.operationId, + dispatchState: 'accepted', + submittedAt: 1_000, + mailNotice + } + ], + sentByThisProcess: first.operationId + }) + ).toEqual({ kind: 'park' }) + } + }) + + it('releases legacy acceptance uncertainty once its owning journal turn settles', () => { + const db = fakeDb() + const input = { db, mailboxHandle: 'dispatch:d1', sessionId: 's1', now: 1_000 } + const first = resolveId({ ...input, messageIds: ['m1', 'm2'] }) + const next = resolveId({ + ...input, + messageIds: ['m2'], + submissions: [ + { + clientMessageId: first.operationId, + dispatchState: 'accepted', + submittedAt: 1_000, + turnSettled: true + } + ] + }) + expect(next.operationId).not.toBe(first.operationId) + expect(db.getStructuredPointerOperation('dispatch:d1')?.batch_fingerprint).toBe( + structuredPointerBatchFingerprint('s1', ['m2']) + ) + }) }) describe('what a pointer attempt does with its operation row', () => { @@ -236,6 +386,16 @@ describe('what a pointer attempt does with its operation row', () => { expect(decide([sent('pending')])).toBe('park') }) + it.each([ + ['pending', 'park'], + ['accepted', 'stamp'] + ] as const)( + 'reconciles a %s operation before comparing changed membership', + (state, expected) => { + expect(decide([sent(state)], { batchFingerprint: 'batch-2' })).toBe(expected) + } + ) + it.each([ ['in doubt', sent('unknown')], ['refused', sent('rejected')] diff --git a/src/main/runtime/orchestration/structured-pointer-operation-id.ts b/src/main/runtime/orchestration/structured-pointer-operation-id.ts index 26dc951edcbf..5b7594193edb 100644 --- a/src/main/runtime/orchestration/structured-pointer-operation-id.ts +++ b/src/main/runtime/orchestration/structured-pointer-operation-id.ts @@ -25,8 +25,21 @@ import type { StructuredPointerOperationRow } from './db/messages/structured-poi /** What the pointer lane reads off a session's journal for its own sends. */ export type StructuredPointerSubmission = Pick< AgentJournalSubmission, - 'clientMessageId' | 'dispatchState' | 'submittedAt' -> + 'clientMessageId' | 'dispatchState' | 'submittedAt' | 'queuedMessageId' +> & { + /** Projected from the submitted message's existing provenance, never persisted separately. */ + mailNotice?: { mailbox: string; messageIds: readonly string[] } + /** Its journal-owned turn ended; absent when that ownership or end cannot be proven. */ + turnSettled?: true +} + +/** Host queue acceptance survives withdrawal, pause, carry and a fresh submission's verdict. */ +export type StructuredPointerQueuedSend = { + operationId: string + mailNotice?: StructuredPointerSubmission['mailNotice'] + /** No retained card can send: withdrawn, or its current handoff's own turn conclusively ended. */ + settled?: true +} /** * What to do with a mailbox's pointer, given its operation row and the session's recorded sends. @@ -48,16 +61,19 @@ export function decideStructuredPointerAttempt(input: { batchFingerprint: string /** The session's recorded sends; a rewind may have dropped the row's. */ submissions: readonly StructuredPointerSubmission[] + queuedSends?: readonly StructuredPointerQueuedSend[] /** Whether this process minted the row's id. */ mintedByThisProcess: boolean now: number }): StructuredPointerAttempt { const { row, submissions } = input - if ( - !row || - row.session_id !== input.sessionId || - row.batch_fingerprint !== input.batchFingerprint - ) { + if (!row) { + return 'mint' + } + if (input.queuedSends?.some((entry) => entry.operationId === row.operation_id)) { + return 'stamp' + } + if (row.session_id !== input.sessionId) { return 'mint' } const sent = submissions.find((entry) => entry.clientMessageId === row.operation_id) @@ -67,6 +83,9 @@ export function decideStructuredPointerAttempt(input: { if (sent?.dispatchState === 'pending') { return 'park' } + if (row.batch_fingerprint !== input.batchFingerprint) { + return 'mint' + } const ranSince = submissions.some( (entry) => entry.dispatchState === 'accepted' && entry.submittedAt > row.minted_at_ms ) @@ -91,16 +110,98 @@ export function structuredPointerBatchFingerprint( export type StructuredPointerOperation = | { kind: 'send'; operationId: string } - | { kind: 'stamp' } + | { kind: 'stamp'; messageIds: readonly string[] } | { kind: 'park' } -export function resolveStructuredPointerOperation(args: { +export type StructuredPointerOperationStore = { + getStructuredPointerOperation: OmitThisParameter + putStructuredPointerOperation: OmitThisParameter +} + +function operationMessageIds( + row: StructuredPointerOperationRow, + submission: Pick | undefined +): readonly string[] | undefined { + const notice = submission?.mailNotice + return notice?.mailbox === row.mailbox_handle && + structuredPointerBatchFingerprint(row.session_id, notice.messageIds) === row.batch_fingerprint + ? notice.messageIds + : undefined +} + +/** Settle original work before a fresh selection can replace its mailbox operation row. */ +export function reconcileStructuredPointerOperations(args: { db: OrchestrationDb mailboxHandle: string sessionId: string + submissions: readonly StructuredPointerSubmission[] + queuedSends: readonly StructuredPointerQueuedSend[] +}): { pendingMessageIds: readonly string[]; ownedMessageIds: readonly string[] } { + const stored = args.db.getStructuredPointerOperation(args.mailboxHandle) + const pending = new Set() + const accepted = new Set() + let storedAccepted = false + for (const queued of args.queuedSends) { + const notice = queued.mailNotice + if (notice?.mailbox !== args.mailboxHandle) { + continue + } + const ownsStored = stored?.operation_id === queued.operationId + if (ownsStored && !operationMessageIds(stored, queued)) { + continue + } + for (const id of notice.messageIds) { + accepted.add(id) + } + storedAccepted ||= ownsStored + } + for (const submission of args.submissions) { + const notice = submission.mailNotice + if (submission.queuedMessageId || notice?.mailbox !== args.mailboxHandle) { + continue + } + const ownsStored = + stored?.session_id === args.sessionId && stored.operation_id === submission.clientMessageId + if (ownsStored && !operationMessageIds(stored, submission)) { + continue + } + if (submission.dispatchState === 'pending') { + for (const id of notice.messageIds) { + pending.add(id) + } + } else if (submission.dispatchState === 'accepted') { + for (const id of notice.messageIds) { + accepted.add(id) + } + if (ownsStored) { + storedAccepted = true + } + } + } + try { + if (accepted.size > 0) { + args.db.markUnpointedMailboxMessagesAsDelivered(args.mailboxHandle, [...accepted]) + } + if (storedAccepted) { + args.db.deleteStructuredPointerOperation(args.mailboxHandle) + } + } catch (error) { + console.warn('[orchestration] structured pointer bookkeeping failed', args.mailboxHandle, error) + } + return { + pendingMessageIds: [...pending], + ownedMessageIds: [...new Set([...pending, ...accepted])] + } +} + +export function resolveStructuredPointerOperation(args: { + db: StructuredPointerOperationStore + mailboxHandle: string + sessionId: string /** The rows this nudge stands for; batch identity, not the body, decides reuse. */ messageIds: readonly string[] submissions: readonly StructuredPointerSubmission[] + queuedSends?: readonly StructuredPointerQueuedSend[] /** The operation id this process last sent for this mailbox, if any. */ sentByThisProcess: string | undefined now?: number @@ -108,16 +209,44 @@ export function resolveStructuredPointerOperation(args: { const now = args.now ?? Date.now() const batchFingerprint = structuredPointerBatchFingerprint(args.sessionId, args.messageIds) const stored = args.db.getStructuredPointerOperation(args.mailboxHandle) - const attempt = decideStructuredPointerAttempt({ + let attempt = decideStructuredPointerAttempt({ row: stored, sessionId: args.sessionId, batchFingerprint, submissions: args.submissions, + queuedSends: args.queuedSends, mintedByThisProcess: stored?.operation_id === args.sentByThisProcess, now }) - if (attempt === 'stamp' || attempt === 'park') { - return { kind: attempt } + const queued = args.queuedSends?.find((entry) => entry.operationId === stored?.operation_id) + if (stored && (stored.session_id === args.sessionId || queued)) { + const sent = args.submissions.find((entry) => entry.clientMessageId === stored.operation_id) + const originalIds = operationMessageIds(stored, queued ?? sent) + if (attempt === 'park' && originalIds) { + // Pending owns only its original members; independent mail remains a new send. + if (args.messageIds.some((id) => originalIds.includes(id))) { + return { kind: 'park' } + } + attempt = 'mint' + } else if (attempt === 'stamp') { + if (originalIds && args.messageIds.every((id) => !originalIds.includes(id))) { + // Proven independent work must not wait on the accepted owner's bookkeeping. + attempt = 'mint' + } else if ( + originalIds || + (stored.session_id === args.sessionId && stored.batch_fingerprint === batchFingerprint) + ) { + return { kind: 'stamp', messageIds: originalIds ?? args.messageIds } + } else if (queued ? !queued.settled : !sent?.turnSettled) { + return { kind: 'park' } + } else { + // A settled legacy turn may be replaced without guessing its original members. + attempt = 'mint' + } + } + } + if (attempt === 'park') { + return { kind: 'park' } } if (attempt === 'reuse' && stored) { return { kind: 'send', operationId: stored.operation_id } diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/check-delivery-result.ts b/src/main/runtime/rpc/methods/orchestration/messaging/check-delivery-result.ts new file mode 100644 index 000000000000..d4b05f1eccac --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/messaging/check-delivery-result.ts @@ -0,0 +1,20 @@ +import type { OrchestrationDb } from '../../../../orchestration/db' +import { exposeMessages } from './mailbox-message-receipt' +import type { CheckParams } from '../schemas' +import type { z } from 'zod' +import { checkReplayRecovery, type ReplayRecoveryMailbox } from './check-replay-recovery' + +export function checkDeliveryResult( + db: OrchestrationDb, + params: z.infer, + current: ReturnType, + additionalMailboxes?: ReplayRecoveryMailbox[] | (() => ReplayRecoveryMailbox[]) +) { + return { + deliveryId: current?.delivery.id ?? null, + messages: exposeMessages(current?.messages ?? []), + count: current?.messages.length ?? 0, + replayed: current?.replayed ?? false, + ...checkReplayRecovery({ db, params, current, additionalMailboxes }) + } +} diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/check-replay-recovery.test.ts b/src/main/runtime/rpc/methods/orchestration/messaging/check-replay-recovery.test.ts new file mode 100644 index 000000000000..2bbc7d73c3d3 --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/messaging/check-replay-recovery.test.ts @@ -0,0 +1,264 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + createRootDispatch, + reattachDispatchConsumer +} from '../../../../orchestration/db/root-dispatch-test-fixture' +import { createOrchestrationRpcHarness } from '../rpc-test-harness' + +const LEAD = 'tab_lead:22222222-2222-4222-9222-222222222222' +const OTHER = 'tab_other:33333333-3333-4333-8333-333333333333' + +function deliveryId(result: unknown): string { + if ( + typeof result === 'object' && + result && + 'deliveryId' in result && + typeof result.deliveryId === 'string' + ) { + return result.deliveryId + } + throw new Error('Expected a Delivery') +} + +describe('consuming check replay recovery', () => { + const h = createOrchestrationRpcHarness() + let state: ReturnType + let dispatch: ReturnType + + beforeEach(() => { + state = h.setup() + vi.mocked(state.runtime.getTerminalPaneKey).mockImplementation((handle) => + handle === 'term_lead' ? LEAD : handle === 'term_coord' ? h.coordinatorPaneKey : OTHER + ) + dispatch = createRootDispatch( + state.db, + state.db.createTask({ spec: 'lead' }).id, + 'term_lead', + LEAD + ) + }) + afterEach(() => { + h.cleanup() + vi.restoreAllMocks() + }) + + function check(terminal: string, params: Record = {}) { + return h.call( + 'orchestration.check', + { terminal, compatibilityCliCommand: 'orca', ...params }, + state.ctx + ) + } + function mail(runId: string, address: string, subject: string) { + return state.db.insertMessage({ from: 'term_worker', to: address, runId, subject }) + } + function bind() { + return state.db.createRun({ + objective: 'child', + coordinatorHandle: 'term_lead', + coordinatorPaneKey: LEAD + }) + } + + it.each(['run', 'dispatch'] as const)( + 'returns an exact unbounded count for %s replay, ignoring wake filters and pointer stamps', + async (kind) => { + const runId = kind === 'run' ? state.activeRunId : dispatch.run_id + if (!runId) { + throw new Error('Expected Run') + } + const address = kind === 'run' ? `run:${runId}` : `dispatch:${dispatch.id}` + const terminal = kind === 'run' ? 'term_coord' : 'term_lead' + const firstMessage = mail(runId, address, 'first') + const first = await check(terminal) + expect(first).not.toHaveProperty('replayRecovery') + const waiting = Array.from({ length: 73 }, (_, i) => mail(runId, address, `later ${i}`)) + state.db.markAsDelivered(waiting.map((message) => message.id)) + const read = mail(runId, address, 'already read') + state.db.markAsRead([read.id]) + state.db.insertMessage({ + from: 'term_worker', + to: address, + runId, + subject: 'audit', + deliveryContract: 'audit_only' + }) + state.db.insertMessage({ + from: 'term_worker', + to: address, + runId, + subject: 'legacy', + deliveryContract: 'legacy_direct' + }) + mail(runId, 'unrelated', 'different mailbox') + const replay = await check(terminal, { + wait: true, + types: 'question', + compatibilityCliCommand: 'orca-ide' + }) + expect(replay).toMatchObject({ + deliveryId: deliveryId(first), + count: 1, + replayed: true, + messages: [{ id: firstMessage.id }], + replayRecovery: { + waitingCount: 73, + ackCommand: `orca-ide orchestration check --terminal ${terminal} --ack ${deliveryId(first)}`, + guidance: + 'Process every message in this batch before acknowledging, then process the next batch returned.' + } + }) + expect(JSON.parse(JSON.stringify(replay))).toHaveProperty('replayRecovery.waitingCount', 73) + const next = await check(terminal, { ack: deliveryId(first) }) + expect(next).toMatchObject({ acknowledged: deliveryId(first), count: 50, replayed: false }) + expect(next).not.toHaveProperty('replayRecovery') + expect(await check(terminal, { ack: deliveryId(first) })).toMatchObject({ + deliveryId: deliveryId(next), + replayed: true, + replayRecovery: { waitingCount: 23 } + }) + } + ) + + it('reports zero waiting rows without modifying the replayed batch', async () => { + mail(dispatch.run_id, `dispatch:${dispatch.id}`, 'only') + const first = await check('term_lead') + expect(await check('term_lead')).toMatchObject({ + deliveryId: deliveryId(first), + replayed: true, + replayRecovery: { waitingCount: 0 } + }) + expect(state.db.getDeliveryRaw(deliveryId(first))?.status).toBe('outstanding') + }) + + it('counts valid residual Dispatch mail deferred by Run replay without changing FIFO ownership', async () => { + const run = bind() + mail(run.id, `run:${run.id}`, 'Run first') + const first = await check('term_lead', { run: run.id }) + mail(run.id, `run:${run.id}`, 'Run later') + const residual = Array.from({ length: 61 }, (_, i) => + mail(dispatch.run_id, `dispatch:${dispatch.id}`, `residual ${i}`) + ) + state.db.markAsDelivered(residual.map((message) => message.id)) + const rawResidual = residual.at(-1) + if (!rawResidual) { + throw new Error('Expected residual mail') + } + state.db.db + .prepare('UPDATE messages SET to_handle = ? WHERE id = ?') + .run('term_lead', rawResidual.id) + expect(await check('term_lead')).toMatchObject({ + deliveryId: deliveryId(first), + replayed: true, + replayRecovery: { + waitingCount: 62, + ackCommand: `orca orchestration check --terminal term_lead --ack ${deliveryId(first)}` + } + }) + expect(await check('term_lead', { run: run.id })).toMatchObject({ + replayRecovery: { + waitingCount: 1, + ackCommand: `orca orchestration check --terminal term_lead --run ${run.id} --ack ${deliveryId(first)}` + } + }) + const tail = await check('term_lead', { ack: deliveryId(first) }) + expect(tail).toMatchObject({ + runId: dispatch.run_id, + messages: residual.slice(0, 50).map((message) => ({ id: message.id })) + }) + const replay = await check('term_lead') + expect(replay).toMatchObject({ + deliveryId: deliveryId(tail), + replayRecovery: { + ackCommand: `orca orchestration check --terminal term_lead --ack ${deliveryId(tail)}` + } + }) + const residualNext = await check('term_lead', { ack: deliveryId(tail) }) + expect(residualNext).toMatchObject({ runId: dispatch.run_id, count: 11 }) + expect(await check('term_lead', { ack: deliveryId(residualNext) })).toMatchObject({ + runId: run.id, + messages: [{ subject: 'Run later' }] + }) + }) + + it('does not count an old residual Dispatch after its process identity was replaced', async () => { + const run = bind() + mail(run.id, `run:${run.id}`, 'Run batch') + const first = await check('term_lead', { run: run.id }) + mail(dispatch.run_id, `dispatch:${dispatch.id}`, 'private old mail') + reattachDispatchConsumer(state.db, { + dispatchId: dispatch.id, + paneKey: LEAD, + processIncarnation: 'old:pty:1' + }) + expect(await check('term_lead')).toMatchObject({ + deliveryId: deliveryId(first), + replayRecovery: { waitingCount: 0 } + }) + }) + + it.each([{ peek: true }, { all: true }, { unread: false }])( + 'omits recovery for inspection %j', + async (mode) => { + mail(dispatch.run_id, `dispatch:${dispatch.id}`, 'first') + await check('term_lead') + mail(dispatch.run_id, `dispatch:${dispatch.id}`, 'later') + expect(await check('term_lead', mode)).not.toHaveProperty('replayRecovery') + } + ) + + it('refuses a fenced consumer before returning recovery', async () => { + mail(dispatch.run_id, `dispatch:${dispatch.id}`, 'first') + await check('term_lead') + reattachDispatchConsumer(state.db, { + dispatchId: dispatch.id, + paneKey: OTHER, + processIncarnation: 'replacement:1' + }) + await expect(check('term_lead')).rejects.toMatchObject({ code: 'consumer_fenced' }) + }) + + it('keeps the batch and ack guidance available if deriving the waiting count fails', async () => { + mail(dispatch.run_id, `dispatch:${dispatch.id}`, 'first') + const first = await check('term_lead') + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined) + vi.spyOn(state.db, 'countUnreadMessagesOutsideDelivery').mockImplementation(() => { + throw new Error('count unavailable') + }) + const replay = await check('term_lead') + expect(replay).toMatchObject({ + deliveryId: deliveryId(first), + count: 1, + replayRecovery: { + ackCommand: `orca orchestration check --terminal term_lead --ack ${deliveryId(first)}` + } + }) + expect(replay).not.toHaveProperty('replayRecovery.waitingCount') + expect(warn).toHaveBeenCalled() + }) + + it('omits the entire waiting count if deferred residual ownership cannot be projected', async () => { + const run = bind() + mail(run.id, `run:${run.id}`, 'Run batch') + const first = await check('term_lead', { run: run.id }) + mail(dispatch.run_id, `dispatch:${dispatch.id}`, 'residual') + const lookup = state.db.getActiveDispatchForIdentity.bind(state.db) + vi.spyOn(state.db, 'getActiveDispatchForIdentity') + .mockImplementationOnce(lookup) + .mockImplementationOnce(() => { + throw new Error('residual projection unavailable') + }) + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined) + const replay = await check('term_lead') + expect(replay).toMatchObject({ + deliveryId: deliveryId(first), + count: 1, + messages: [{ subject: 'Run batch' }], + replayRecovery: { + ackCommand: `orca orchestration check --terminal term_lead --ack ${deliveryId(first)}` + } + }) + expect(replay).not.toHaveProperty('replayRecovery.waitingCount') + expect(warn).toHaveBeenCalled() + }) +}) diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/check-replay-recovery.ts b/src/main/runtime/rpc/methods/orchestration/messaging/check-replay-recovery.ts new file mode 100644 index 000000000000..67e74c9cba5b --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/messaging/check-replay-recovery.ts @@ -0,0 +1,65 @@ +import type { DeliveryRow, OrchestrationDb } from '../../../../orchestration/db' +import { localOrchestrationCliCommand } from '../../../../orchestration/cli-command' +import { + createOrchestrationReplayRecovery, + type OrchestrationReplayRecovery +} from '../../../../../../shared/orchestration-check-output' +import type { CheckParams } from '../schemas' +import type { z } from 'zod' + +export type ReplayRecoveryMailbox = { runId: string; mailboxHandle: string } + +export function checkReplayRecovery(args: { + db: OrchestrationDb + params: z.infer + current: { delivery: DeliveryRow; replayed: boolean } | undefined + additionalMailboxes?: ReplayRecoveryMailbox[] | (() => ReplayRecoveryMailbox[]) +}): { replayRecovery?: OrchestrationReplayRecovery } { + const { db, params, current } = args + if ( + !current?.replayed || + !current.delivery.mailbox_handle || + params.peek || + params.all || + params.unread === false + ) { + return {} + } + let waitingCount: number | undefined + try { + if (db.getRun(current.delivery.run_id)?.legacy === 1) { + return {} + } + const additionalMailboxes = + typeof args.additionalMailboxes === 'function' + ? args.additionalMailboxes() + : args.additionalMailboxes + const mailboxes: ReplayRecoveryMailbox[] = [ + { runId: current.delivery.run_id, mailboxHandle: current.delivery.mailbox_handle }, + ...(additionalMailboxes ?? []) + ] + waitingCount = mailboxes.reduce( + (count, mailbox) => + count + + db.countUnreadMessagesOutsideDelivery({ + ...mailbox, + deliveryId: current.delivery.id + }), + 0 + ) + } catch (error) { + console.warn('[orchestration] Could not count mail waiting behind a replayed Delivery.', error) + } + return { + replayRecovery: createOrchestrationReplayRecovery( + current.delivery.id, + { + cliCommand: params.compatibilityCliCommand ?? localOrchestrationCliCommand(), + terminal: params.terminal, + run: params.run, + consuming: true + }, + waitingCount + ) + } +} diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/check-run-pending-mail.ts b/src/main/runtime/rpc/methods/orchestration/messaging/check-run-pending-mail.ts index 5d453616f355..6735fdf5d302 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/check-run-pending-mail.ts +++ b/src/main/runtime/rpc/methods/orchestration/messaging/check-run-pending-mail.ts @@ -8,6 +8,7 @@ import { orchestrationCallerIdentity } from '../runs/run-scope' import type { OrchestrationSessionCaller } from '../../../../orchestration/orchestration-caller-identity' import type { CheckParams } from '../schemas' import type { z } from 'zod' +import type { ReplayRecoveryMailbox } from './check-replay-recovery' export async function checkRunPendingMail(args: { params: z.infer @@ -21,7 +22,11 @@ export async function checkRunPendingMail(args: { signal: AbortSignal | undefined revalidateConsumer: () => void recordMutationReceipt: ((receipt: unknown) => void) | undefined -}): Promise<{ acknowledged?: string; result?: unknown }> { +}): Promise<{ + acknowledged?: string + result?: unknown + readResidualMailboxes?: () => ReplayRecoveryMailbox[] +}> { const { params, runtime, @@ -68,15 +73,14 @@ export async function checkRunPendingMail(args: { } } recordAcknowledged() - if ( - residual && - residual.run_id !== run.id && - (residual.assignee_orca_session_id === null || - residual.assignee_orca_session_id === caller.orcaSessionId) && - callerHoldsDispatchPane(residual, paneKey) && - currentDispatchAssigneeRun(runtime, db, residual)?.id === run.id && - (residualAck || !db.hasOutstandingMailboxDelivery(address)) - ) { + const canReadResidual = (candidate: typeof residual): candidate is NonNullable => + candidate !== undefined && + candidate.run_id !== run.id && + (candidate.assignee_orca_session_id === null || + candidate.assignee_orca_session_id === caller.orcaSessionId) && + callerHoldsDispatchPane(candidate, paneKey) && + currentDispatchAssigneeRun(runtime, db, candidate)?.id === run.id + if (canReadResidual(residual) && (residualAck || !db.hasOutstandingMailboxDelivery(address))) { const result = await checkWorkerMailbox({ params: { ...params, ack: residualAck, wait: false }, runtime, @@ -90,6 +94,7 @@ export async function checkRunPendingMail(args: { wakeTypes: params.wait ? typeFilter : undefined, // Accept the original owner's ack without creating a batch ahead of Run replay. deferDelivery: () => db.hasOutstandingMailboxDelivery(address), + recoveryMailboxes: [{ runId: run.id, mailboxHandle: address }], revalidateConsumer: () => { revalidateConsumer() const current = db.getActiveDispatchForIdentity(handle, paneKey) @@ -136,5 +141,19 @@ export async function checkRunPendingMail(args: { if (params.ack && !acknowledged) { acknowledgeRun() } - return { acknowledged: acknowledged?.delivery.id } + return { + acknowledged: acknowledged?.delivery.id, + readResidualMailboxes: () => { + const latest = !params.run ? db.getActiveDispatchForIdentity(handle, paneKey) : undefined + if (latest?.id !== residual?.id || !canReadResidual(latest)) { + return [] + } + // Direct aliases stay under the original Dispatch Run until its next consuming check. + const addresses = new Set([`dispatch:${latest.id}`, handle]) + if (latest.assignee_handle) { + addresses.add(latest.assignee_handle) + } + return [...addresses].map((mailboxHandle) => ({ runId: latest.run_id, mailboxHandle })) + } + } } diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/check-run.ts b/src/main/runtime/rpc/methods/orchestration/messaging/check-run.ts index c48141cf7ddb..d9fea2540503 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/check-run.ts +++ b/src/main/runtime/rpc/methods/orchestration/messaging/check-run.ts @@ -11,6 +11,7 @@ import type { CheckParams } from '../schemas' import type { z } from 'zod' import type { OrchestrationSessionCaller } from '../../../../orchestration/orchestration-caller-identity' import { checkRunPendingMail } from './check-run-pending-mail' +import { checkDeliveryResult } from './check-delivery-result' type CheckParamsInput = z.infer @@ -105,6 +106,8 @@ export async function checkRunMailbox(args: { return pending.result } const acknowledged = pending.acknowledged + const deliveryResult = (current: Parameters[2]) => + checkDeliveryResult(db, params, current, pending.readResidualMailboxes) if (params.all || (params.unread === false && !params.peek)) { const messages = db.getRunMailboxHistory(run.id, 100, typeFilter) const result = { @@ -142,10 +145,7 @@ export async function checkRunMailbox(args: { if (current) { return { runId: run.id, - deliveryId: current.delivery.id, - messages: exposeMessages(current.messages), - count: current.messages.length, - replayed: current.replayed, + ...deliveryResult(current), acknowledged: acknowledged ?? null, timedOut: false, cancelled: false, @@ -246,10 +246,7 @@ export async function checkRunMailbox(args: { current = readDelivery(typeFilter) return { runId: run.id, - deliveryId: current?.delivery.id ?? null, - messages: exposeMessages(current?.messages ?? []), - count: current?.messages.length ?? 0, - replayed: current?.replayed ?? false, + ...deliveryResult(current), acknowledged: acknowledged ?? null, timedOut: false, cancelled: false, diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/check-worker.ts b/src/main/runtime/rpc/methods/orchestration/messaging/check-worker.ts index 7616f3e5cc10..8a9a19c097e4 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/check-worker.ts +++ b/src/main/runtime/rpc/methods/orchestration/messaging/check-worker.ts @@ -9,6 +9,8 @@ import { interruptedAcknowledgedCheck } from '../routing' import { currentDispatchAssigneeRun } from './recipient-routing' import type { CheckParams } from '../schemas' import type { z } from 'zod' +import type { ReplayRecoveryMailbox } from './check-replay-recovery' +import { checkDeliveryResult } from './check-delivery-result' type CheckParamsInput = z.infer type ActiveDispatch = NonNullable> @@ -29,6 +31,7 @@ export async function checkWorkerMailbox(args: { wakeTypes?: MessageType[] revalidateConsumer?: () => void deferDelivery?: () => boolean + recoveryMailboxes?: ReplayRecoveryMailbox[] recordMutationReceipt?: (receipt: unknown) => void }) { const { @@ -240,10 +243,7 @@ export async function checkWorkerMailbox(args: { if (current || !params.wait) { return { ...mailboxIdentity, - deliveryId: current?.delivery.id ?? null, - messages: exposeMessages(current?.messages ?? []), - count: current?.messages.length ?? 0, - replayed: current?.replayed ?? false, + ...checkDeliveryResult(db, params, current, args.recoveryMailboxes), acknowledged: acknowledged?.delivery.id ?? null, timedOut: false, cancelled: false, @@ -293,10 +293,7 @@ export async function checkWorkerMailbox(args: { const arrived = readDelivery(typeFilter) return { ...mailboxIdentity, - deliveryId: arrived?.delivery.id ?? null, - messages: exposeMessages(arrived?.messages ?? []), - count: arrived?.messages.length ?? 0, - replayed: arrived?.replayed ?? false, + ...checkDeliveryResult(db, params, arrived, args.recoveryMailboxes), acknowledged: acknowledged?.delivery.id ?? null, ...(params.format || params.inject ? { formatted: arrived?.messages.map(formatMessageBanner).join('\n\n') ?? '' } diff --git a/src/main/runtime/structured-chat-coordinator-mail.test.ts b/src/main/runtime/structured-chat-coordinator-mail.test.ts index 5eb0f57209cb..321d5706e989 100644 --- a/src/main/runtime/structured-chat-coordinator-mail.test.ts +++ b/src/main/runtime/structured-chat-coordinator-mail.test.ts @@ -88,7 +88,7 @@ describe('a worker result reaches the structured chat that coordinates it', () = }) /** Fires both edges and waits until every gate read they started has answered. */ - const edgesAnswered = (): Promise => observationClock.edgesAnswered(runtime, WAIT) + const edgesAnswered = (): Promise => observationClock.edgesAnswered(runtime) /** The operation ids the coordinator's journal recorded for its pointer turns. */ async function pointerSends(): Promise { diff --git a/src/main/runtime/structured-chat-coordinator-observation-clock.test-fixture.ts b/src/main/runtime/structured-chat-coordinator-observation-clock.test-fixture.ts index cced136ce8b6..c6d56282afd6 100644 --- a/src/main/runtime/structured-chat-coordinator-observation-clock.test-fixture.ts +++ b/src/main/runtime/structured-chat-coordinator-observation-clock.test-fixture.ts @@ -24,14 +24,13 @@ export function createCoordinatorMailObservationClock( await new Promise((resolve) => setImmediate(resolve)) }, async edgesAnswered( - runtime: Pick, - wait: { timeout: number } + runtime: Pick ): Promise { - const reads = vi.spyOn(getHost(), 'journalSnapshot') + const reads = vi.spyOn(getHost(), 'mailboxPointerSnapshot') runtime.onStructuredSessionStatusForMail({ sessionId, status: null }) runtime.onStructuredSessionStatusForMail({ sessionId, status: 'idle' }) - await vi.waitFor(() => expect(reads).toHaveBeenCalled(), wait) await Promise.all(reads.mock.results.map((read) => read.value)) + await getHost().flushStreamedEvents(sessionId) await new Promise((resolve) => setImmediate(resolve)) reads.mockRestore() }, diff --git a/src/main/runtime/structured-chat-mail-queue-recovery.test.ts b/src/main/runtime/structured-chat-mail-queue-recovery.test.ts new file mode 100644 index 000000000000..e60cebdd31dd --- /dev/null +++ b/src/main/runtime/structured-chat-mail-queue-recovery.test.ts @@ -0,0 +1,368 @@ +import './rpc/unused-default-rpc-methods.test-fixture' +import { beforeEach, describe, expect, it, vi } from 'vitest' +import type { AgentJournalMessageItem } from '../../shared/agent-session-journal-types' +import { QUEUED_MESSAGE_PAUSED_SEND_FAILED } from '../../shared/agent-session-wire' +import { QUEUED_MESSAGE_REPLAY_WINDOW_MS } from '../native-chat/agent-session-journal/journal-queued-messages' +import { + OrchestrationStructuredMailboxPointerDelivery, + type StructuredMailboxPointerHost +} from './orchestration/structured-mailbox-pointer-delivery' +import { createStructuredMailboxPointerHost } from './orchestration/structured-mailbox-pointer-host' +import { operationId } from './structured-chat-coordinator-fake-codex-fixture' +import { + COORDINATOR, + WAIT, + call, + clearChat, + connectionFor, + coordinatorRunAndTask, + db, + host, + openChat, + runtime, + sendUserMessage, + settleTurn, + startSuccessor +} from './structured-chat-coordinator-mail-rig.test-fixture' + +function journal(sessionId = COORDINATOR) { + const opened = host.collaboratorsForTests().sessions.get(sessionId) + if (!opened) { + throw new Error('the test conversation is not open') + } + return opened.journal +} + +function noticeMessages(body: AgentJournalMessageItem | undefined) { + const notice = body?.from?.orchestration + if (notice?.message !== 'mail-notice') { + throw new Error('expected an orchestration mail notice') + } + return notice.messages +} + +function pointerLane(mailbox: string, sessionId: () => string = () => COORDINATOR) { + const adapter = createStructuredMailboxPointerHost() + const send = vi.fn((input) => adapter.send(input)) + const readSessionFacts = vi.fn((id) => + adapter.readSessionFacts(id) + ) + const delivery = new OrchestrationStructuredMailboxPointerDelivery({ + getDb: () => db, + getMessageWaiters: () => undefined, + resolveStructuredTarget: (handle) => + handle === mailbox ? { sessionId: sessionId(), dispatchId: null } : null, + getCliCommand: () => 'orca', + senderName: () => null, + host: { ...adapter, send, readSessionFacts } + }) + return { delivery, send, readSessionFacts } +} + +function insertMail(runId: string, subject: string) { + return db.insertMessage({ from: 'term_worker', to: `run:${runId}`, runId, subject }) +} + +async function busyMailbox() { + const chat = await openChat(COORDINATOR) + const { runId } = await coordinatorRunAndTask() + expect(await sendUserMessage(COORDINATOR, 'working')).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(chat.turns).toHaveLength(1), WAIT) + chat.handlers.onNotification?.('turn/started', { turn: { id: 'turn-1' } }) + await host.flushStreamedEvents(COORDINATOR) + return { chat, runId, mailbox: `run:${runId}` } +} + +async function loseQueueStamp(mailbox: string, lane: ReturnType) { + const stamp = vi.spyOn(db, 'markAsDelivered').mockImplementationOnce(() => { + throw new Error('mailbox stamp failed after queue commit') + }) + lane.delivery.deliverForHandle(mailbox) + await vi.waitFor(() => expect(stamp).toHaveBeenCalledTimes(1), WAIT) + expect(journal().queuedMessages.list()).toHaveLength(1) + const original = journal().queuedMessages.list()[0]! + expect(db.getStructuredPointerOperation(mailbox)?.operation_id).toBe(original.messageId) + expect(db.getUndeliveredUnreadMessages(mailbox).length).toBeGreaterThan(0) + return original +} + +beforeEach(() => { + // These tests drive the delivery actor explicitly; host queue/drain and database remain real. + vi.spyOn(runtime, 'onStructuredSessionStatusForMail').mockImplementation(() => {}) +}) + +describe('queued orchestration notice recovery', () => { + it('skips the journal when there is no candidate attention and no stored operation', async () => { + await openChat(COORDINATOR) + const { runId } = await coordinatorRunAndTask() + const lane = pointerLane(`run:${runId}`) + lane.delivery.deliverForHandle(`run:${runId}`) + await host.flushStreamedEvents(COORDINATOR) + expect(lane.readSessionFacts).not.toHaveBeenCalled() + expect(lane.send).not.toHaveBeenCalled() + }) + + it('does not read the journal again after successful handoff leaves no candidate or operation', async () => { + await openChat(COORDINATOR) + const { runId } = await coordinatorRunAndTask() + const mailbox = `run:${runId}` + insertMail(runId, 'A') + const lane = pointerLane(mailbox) + lane.delivery.deliverForHandle(mailbox) + await vi.waitFor(() => expect(connectionFor(COORDINATOR).turns).toHaveLength(1), WAIT) + await settleTurn(COORDINATOR, 0) + await vi.waitFor(() => expect(db.getUndeliveredUnreadMessages(mailbox)).toEqual([]), WAIT) + await host.flushStreamedEvents(COORDINATOR) + expect(lane.send).toHaveBeenCalledTimes(1) + expect(lane.readSessionFacts).toHaveBeenCalledTimes(1) + }) + + it('reconciles a stored queue operation even when all its members are already fetched', async () => { + const { runId, mailbox } = await busyMailbox() + insertMail(runId, 'A') + await loseQueueStamp(mailbox, pointerLane(mailbox)) + expect(await call('orchestration.check', {}, { sessionId: COORDINATOR })).toMatchObject({ + count: 1 + }) + const recovered = pointerLane(mailbox) + recovered.delivery.deliverForHandle(mailbox) + await vi.waitFor(() => expect(db.getStructuredPointerOperation(mailbox)).toBeUndefined(), WAIT) + expect(recovered.readSessionFacts).toHaveBeenCalledTimes(1) + expect(recovered.send).not.toHaveBeenCalled() + expect(db.getUndeliveredUnreadMessages(mailbox)).toEqual([]) + }) + + it('retains the original operation when the host cannot read its queue', async () => { + const { runId, mailbox } = await busyMailbox() + const a = insertMail(runId, 'A') + const original = await loseQueueStamp(mailbox, pointerLane(mailbox)) + const queueRead = vi.spyOn(journal().queuedMessages, 'list').mockImplementationOnce(() => { + throw new Error('queue facts are unavailable') + }) + const recovered = pointerLane(mailbox) + recovered.delivery.deliverForHandle(mailbox) + await vi.waitFor(() => expect(recovered.readSessionFacts).toHaveBeenCalledTimes(1), WAIT) + await expect(recovered.readSessionFacts.mock.results[0]?.value).resolves.toBeNull() + expect(queueRead).toHaveBeenCalled() + expect(recovered.send).not.toHaveBeenCalled() + expect(db.getStructuredPointerOperation(mailbox)?.operation_id).toBe(original.messageId) + expect(db.getMessageById(a.id)?.delivered_at).toBeNull() + }) + + it('keeps original queue ownership without gating independent mail on repeated stamp failure', async () => { + const { runId, mailbox } = await busyMailbox() + const a = insertMail(runId, 'A') + const original = await loseQueueStamp(mailbox, pointerLane(mailbox)) + vi.spyOn(db, 'markUnpointedMailboxMessagesAsDelivered').mockImplementation(() => { + throw new Error('queue acceptance stamp still unavailable') + }) + const c = insertMail(runId, 'C') + const recovered = pointerLane(mailbox) + recovered.delivery.deliverForHandle(mailbox) + await vi.waitFor(() => expect(recovered.send).toHaveBeenCalledTimes(1), WAIT) + expect(noticeMessages(recovered.send.mock.calls[0]?.[0].body)).toMatchObject([ + { messageId: c.id } + ]) + expect( + journal() + .queuedMessages.list() + .map((card) => card.messageId) + ).toEqual([original.messageId, recovered.send.mock.calls[0]?.[0].operationId]) + expect(db.getMessageById(a.id)?.delivered_at).toBeNull() + expect(await call('orchestration.check', {}, { sessionId: COORDINATOR })).toMatchObject({ + count: 2, + messages: [{ id: a.id }, { id: c.id }] + }) + }) + + it('selects fresh mail after the preflight candidate is fetched during the journal await', async () => { + const { runId, mailbox } = await busyMailbox() + const a = insertMail(runId, 'A') + const lane = pointerLane(mailbox) + const adapter = createStructuredMailboxPointerHost() + let release: (() => void) | undefined + lane.readSessionFacts.mockImplementationOnce(async (id) => { + await new Promise((resolve) => { + release = resolve + }) + return adapter.readSessionFacts(id) + }) + lane.delivery.deliverForHandle(mailbox) + await vi.waitFor(() => expect(release).toBeDefined(), WAIT) + expect(await call('orchestration.check', {}, { sessionId: COORDINATOR })).toMatchObject({ + count: 1, + messages: [{ id: a.id }] + }) + const b = insertMail(runId, 'B') + release?.() + await vi.waitFor(() => expect(lane.send).toHaveBeenCalledTimes(1), WAIT) + expect(noticeMessages(lane.send.mock.calls[0]?.[0].body)).toMatchObject([{ messageId: b.id }]) + }) + + it.each(['unchanged', 'partially fetched'] as const)( + 'a new actor reconciles an accepted %s batch without another card or provider turn', + async (selection) => { + const { chat, runId, mailbox } = await busyMailbox() + if (selection === 'partially fetched') { + const prefix = Array.from({ length: 49 }, (_, index) => + insertMail(runId, `prefix ${index}`) + ) + db.markAsDelivered(prefix.map((mail) => mail.id)) + } + const a = insertMail(runId, 'A') + const b = insertMail(runId, 'B') + const original = await loseQueueStamp(mailbox, pointerLane(mailbox)) + expect(noticeMessages(original.body).map((mail) => mail.messageId)).toEqual([a.id, b.id]) + if (selection === 'partially fetched') { + const fetched = await call('orchestration.check', {}, { sessionId: COORDINATOR }) + expect(fetched).toMatchObject({ count: 50 }) + expect( + db.getUndeliveredUnreadMessages(mailbox, undefined, { excludeFetched: true }) + ).toEqual([expect.objectContaining({ id: b.id })]) + } + const recovered = pointerLane(mailbox) + recovered.delivery.deliverForHandle(mailbox) + await vi.waitFor(() => expect(db.getUndeliveredUnreadMessages(mailbox)).toEqual([]), WAIT) + expect(recovered.send).not.toHaveBeenCalled() + expect( + journal() + .queuedMessages.list() + .map((card) => card.messageId) + ).toEqual([original.messageId]) + expect(db.getStructuredPointerOperation(mailbox)).toBeUndefined() + + const c = insertMail(runId, 'C') + recovered.delivery.deliverForHandle(mailbox) + await vi.waitFor(() => expect(db.getUndeliveredUnreadMessages(mailbox)).toEqual([]), WAIT) + expect(recovered.send).toHaveBeenCalledTimes(1) + const cards = journal().queuedMessages.list() + expect(cards).toHaveLength(2) + expect(noticeMessages(cards[1]?.body).map((mail) => mail.messageId)).toEqual([c.id]) + await settleTurn(COORDINATOR, 0) + await vi.waitFor(() => expect(chat.turns).toHaveLength(2), WAIT) + await settleTurn(COORDINATOR, 1) + await vi.waitFor(() => expect(chat.turns).toHaveLength(3), WAIT) + await settleTurn(COORDINATOR, 2) + recovered.delivery.deliverForHandle(mailbox) + await host.flushStreamedEvents(COORDINATOR) + expect(chat.turns).toHaveLength(3) + const notices = (await host.journalSnapshot(COORDINATOR)).submissions.filter( + (submission) => submission.queuedMessageId + ) + expect(notices.map((submission) => submission.queuedMessageId)).toEqual( + cards.map((card) => card.messageId) + ) + expect(new Set(notices.map((submission) => submission.clientMessageId)).size).toBe(2) + } + ) + + it('respects withdrawal of an accepted notice and still queues independent mail', async () => { + const { chat, runId, mailbox } = await busyMailbox() + insertMail(runId, 'A') + const original = await loseQueueStamp(mailbox, pointerLane(mailbox)) + await journal().queuedMessages.withdraw({ + messageIds: [original.messageId], + settledByOp: 'test-withdrawal' + }) + const recovered = pointerLane(mailbox) + recovered.delivery.deliverForHandle(mailbox) + await vi.waitFor(() => expect(db.getUndeliveredUnreadMessages(mailbox)).toEqual([]), WAIT) + expect(recovered.send).not.toHaveBeenCalled() + insertMail(runId, 'C') + recovered.delivery.deliverForHandle(mailbox) + await vi.waitFor(() => expect(recovered.send).toHaveBeenCalledTimes(1), WAIT) + await settleTurn(COORDINATOR, 0) + await vi.waitFor(() => expect(chat.turns).toHaveLength(2), WAIT) + await settleTurn(COORDINATOR, 1) + expect( + journal() + .submissions() + .some((submission) => submission.queuedMessageId === original.messageId) + ).toBe(false) + }) + + it('recognizes the original accepted card carried into a successor after clear', async () => { + const { runId, mailbox } = await busyMailbox() + insertMail(runId, 'A') + const original = await loseQueueStamp(mailbox, pointerLane(mailbox)) + await journal().queuedMessages.hold({ + messageIds: [original.messageId], + reason: QUEUED_MESSAGE_PAUSED_SEND_FAILED + }) + await settleTurn(COORDINATOR, 0) + const successor = await clearChat(COORDINATOR) + const carried = journal(successor).queuedMessages.get(original.messageId) + expect(carried).toMatchObject({ carriedFrom: COORDINATOR, state: 'waiting' }) + const recovered = pointerLane(mailbox, () => successor) + recovered.delivery.deliverForHandle(mailbox) + await vi.waitFor(() => expect(db.getUndeliveredUnreadMessages(mailbox)).toEqual([]), WAIT) + expect(recovered.send).not.toHaveBeenCalled() + expect(journal(successor).queuedMessages.list()).toHaveLength(1) + expect(db.getStructuredPointerOperation(mailbox)).toBeUndefined() + await startSuccessor(successor) + await vi.waitFor(() => expect(connectionFor(successor).turns).toHaveLength(2), WAIT) + await settleTurn(successor, 1) + expect( + journal(successor) + .submissions() + .filter((submission) => submission.queuedMessageId) + ).toMatchObject([{ queuedMessageId: original.messageId }]) + }) + + it.each(['pending', 'accepted'] as const)( + 'reconciles a fresh %s handoff as ownership by the original queue operation', + async (state) => { + const { runId, mailbox } = await busyMailbox() + insertMail(runId, 'A') + const original = await loseQueueStamp(mailbox, pointerLane(mailbox)) + const handoffId = operationId() + const conversation = journal() + const fence = host.deps.store.getRecord(COORDINATOR)!.lease.runtimeFence + await conversation.appendSubmission( + { + clientMessageId: handoffId, + payloadFingerprint: original.fingerprint, + body: original.body, + fence, + handoverRecorded: true, + origin: 'host' + }, + { + messageId: original.messageId, + expect: 'waiting', + settledByOp: null, + hostInstance: original.hostInstance + } + ) + if (state === 'accepted') { + await conversation.resolveDispatch({ + clientMessageId: handoffId, + state, + fence, + providerIdentity: null + }) + const later = Date.now() + QUEUED_MESSAGE_REPLAY_WINDOW_MS + 1 + const clock = vi.spyOn(Date, 'now').mockReturnValue(later) + try { + await conversation.queuedMessages.repairAndPrune() + } finally { + clock.mockRestore() + } + expect(conversation.queuedMessages.list()).toEqual([]) + } + expect(conversation.submission(handoffId)).toMatchObject({ + queuedMessageId: original.messageId, + dispatchState: state + }) + const recovered = pointerLane(mailbox) + recovered.delivery.deliverForHandle(mailbox) + await vi.waitFor( + () => expect(db.getStructuredPointerOperation(mailbox)).toBeUndefined(), + WAIT + ) + expect(db.getUndeliveredUnreadMessages(mailbox)).toEqual([]) + expect(recovered.send).not.toHaveBeenCalled() + expect(conversation.queuedMessages.list()).toHaveLength(state === 'accepted' ? 0 : 1) + } + ) +}) diff --git a/src/main/runtime/structured-chat-worker-unacked-notice.test.ts b/src/main/runtime/structured-chat-worker-unacked-notice.test.ts new file mode 100644 index 000000000000..ea9aa2b4de4a --- /dev/null +++ b/src/main/runtime/structured-chat-worker-unacked-notice.test.ts @@ -0,0 +1,100 @@ +import './rpc/unused-default-rpc-methods.test-fixture' +import { describe, expect, it, vi } from 'vitest' +import { formatOrcaSessionAddress } from '../../shared/orca-session-address' +import { testOrcaSessionId } from '../../shared/orca-session-address-test-fixture' +import { idOf } from './rpc/orchestration-session-caller-test-fixture' +import { + COORDINATOR, + PEER_CHAT, + WAIT, + call, + coordinatorRunAndTask, + db, + host, + openChat, + queuedCardTexts, + runtime, + settleTurn, + turnText +} from './structured-chat-coordinator-mail-rig.test-fixture' + +async function workerMailbox() { + await openChat(COORDINATOR) + const chat = await openChat(PEER_CHAT) + const { taskId } = await coordinatorRunAndTask() + const { dispatch } = await call( + 'orchestration.dispatch', + { task: taskId, to: formatOrcaSessionAddress(testOrcaSessionId(PEER_CHAT)) }, + { sessionId: COORDINATOR } + ) + const mailbox = `dispatch:${idOf(dispatch)}` + const receive = async (subject: string) => { + const result = await call( + 'orchestration.send', + { to: mailbox, subject }, + { sessionId: COORDINATOR } + ) + return idOf(result.message) + } + const check = (ack?: string) => call('orchestration.check', { ack }, { sessionId: PEER_CHAT }) + return { chat, mailbox, receive, check } +} + +describe('structured worker notices after a check without acknowledgement', () => { + it('starts one later notice turn after the consumer settles without acknowledging', async () => { + const { chat, mailbox, receive, check } = await workerMailbox() + const first = await receive('First guidance') + await vi.waitFor(() => expect(chat.turns).toHaveLength(1), WAIT) + expect(turnText(chat.turns[0]!)).toContain('You have 1 orchestration message.') + const checked = await check() + expect(checked.messages).toEqual([expect.objectContaining({ id: first })]) + expect(db.hasOutstandingMailboxDelivery(mailbox)).toBe(true) + await settleTurn(PEER_CHAT, 0) + + const later = await receive('Later guidance') + await vi.waitFor(() => expect(chat.turns).toHaveLength(2), WAIT) + expect(turnText(chat.turns[1]!)).toContain('You have 1 orchestration message.') + const replayed = await check() + expect(replayed.deliveryId).toBe(checked.deliveryId) + expect(replayed.messages).toEqual([expect.objectContaining({ id: first })]) + if (typeof checked.deliveryId !== 'string') { + throw new Error('Expected a consuming delivery') + } + const next = await check(checked.deliveryId) + expect(next.messages).toEqual([expect.objectContaining({ id: later })]) + await settleTurn(PEER_CHAT, 1) + runtime.deliverPendingMessagesForHandle(mailbox) + runtime.onStructuredSessionStatusForMail({ sessionId: PEER_CHAT, status: 'idle' }) + expect(chat.turns).toHaveLength(2) + }) + + it('queues later mail once while the checked batch is active, without starting a second provider turn', async () => { + const { chat, mailbox, receive, check } = await workerMailbox() + await receive('First guidance') + await vi.waitFor(() => expect(chat.turns).toHaveLength(1), WAIT) + chat.handlers.onNotification?.('turn/started', { turn: { id: 'turn-1' } }) + chat.handlers.onNotification?.('item/completed', { + item: { + type: 'userMessage', + id: `echo-${chat.turns[0]!.clientUserMessageId}`, + clientId: chat.turns[0]!.clientUserMessageId, + content: [{ type: 'text', text: turnText(chat.turns[0]!) }] + } + }) + await host.flushStreamedEvents(PEER_CHAT) + await check() + expect(db.hasOutstandingMailboxDelivery(mailbox)).toBe(true) + await receive('Later guidance while working') + await vi.waitFor(async () => expect(await queuedCardTexts(PEER_CHAT)).toHaveLength(1), WAIT) + runtime.deliverPendingMessagesForHandle(mailbox) + runtime.deliverPendingMessagesForHandle(mailbox) + expect(chat.turns).toHaveLength(1) + expect(await queuedCardTexts(PEER_CHAT)).toHaveLength(1) + + await settleTurn(PEER_CHAT, 0) + await vi.waitFor(() => expect(chat.turns).toHaveLength(2), WAIT) + await settleTurn(PEER_CHAT, 1) + expect(await queuedCardTexts(PEER_CHAT)).toHaveLength(0) + expect(chat.turns).toHaveLength(2) + }) +}) diff --git a/src/shared/orchestration-check-output.test.ts b/src/shared/orchestration-check-output.test.ts index df3248bd2fdb..f4ae75122910 100644 --- a/src/shared/orchestration-check-output.test.ts +++ b/src/shared/orchestration-check-output.test.ts @@ -1,7 +1,8 @@ import { describe, expect, it } from 'vitest' import { formatOrchestrationCheckText, - prepareOrchestrationCheckOutput + prepareOrchestrationCheckOutput, + type OrchestrationCheckOutput } from './orchestration-check-output' describe('prepareOrchestrationCheckOutput', () => { @@ -44,6 +45,48 @@ describe('prepareOrchestrationCheckOutput', () => { }) }) +describe('replayed Delivery output', () => { + const recovery = { + waitingCount: 71, + ackCommand: 'orca-ide orchestration check --terminal term_worker --ack delivery_old', + guidance: + 'Process every message in this batch before acknowledging, then process the next batch returned.' + } + it.each([undefined, 'Message body'])( + 'renders recovery in plain and formatted output: %s', + (formatted) => { + const result: OrchestrationCheckOutput = { + messages: [{ id: 'msg_one', from_handle: 'worker', subject: 'old' }], + count: 1, + deliveryId: 'delivery_old', + replayed: true, + formatted, + replayRecovery: recovery + } + const output = formatOrchestrationCheckText(result, 'term_worker') + expect(output).toContain('Delivery delivery_old (replayed)') + expect(output).toContain('71 unread messages waiting behind this Delivery.') + expect(output).toContain(recovery.ackCommand) + expect(output).toContain(recovery.guidance) + expect(output.split(recovery.ackCommand)).toHaveLength(2) + expect(output).toContain(formatted ?? 'msg_one [status]') + } + ) + + it('retains the optional recovery object in the actual JSON result', () => { + const result: OrchestrationCheckOutput = { + messages: [], + count: 0, + deliveryId: 'delivery_old', + replayed: true, + replayRecovery: recovery + } + expect( + JSON.parse(JSON.stringify(prepareOrchestrationCheckOutput(result, 'term_worker', false))) + ).toHaveProperty('replayRecovery', recovery) + }) +}) + describe('formatted delivery acknowledgment', () => { it('retains the delivery ID above formatted message bodies', () => { expect( diff --git a/src/shared/orchestration-check-output.ts b/src/shared/orchestration-check-output.ts index 5429c5423410..9c809cc43e03 100644 --- a/src/shared/orchestration-check-output.ts +++ b/src/shared/orchestration-check-output.ts @@ -36,12 +36,42 @@ export type OrchestrationCheckOutput = { count: number formatted?: string deliveryId?: string | null + replayed?: boolean + replayRecovery?: OrchestrationReplayRecovery timedOut?: boolean cancelled?: boolean connectionLost?: boolean legacyCompatibility?: LegacyCompatibilityResult } +export type OrchestrationReplayRecovery = { + waitingCount?: number + ackCommand: string + guidance: string +} + +export type OrchestrationCheckInvocation = { + cliCommand: 'orca' | 'orca-ide' | 'orca-dev' + terminal?: string + run?: string + consuming: boolean +} + +export function createOrchestrationReplayRecovery( + deliveryId: string, + invocation: OrchestrationCheckInvocation, + waitingCount?: number +): OrchestrationReplayRecovery { + const terminal = invocation.terminal ? ` --terminal ${invocation.terminal}` : '' + const run = invocation.run ? ` --run ${invocation.run}` : '' + return { + ...(waitingCount === undefined ? {} : { waitingCount }), + ackCommand: `${invocation.cliCommand} orchestration check${terminal}${run} --ack ${deliveryId}`, + guidance: + 'Process every message in this batch before acknowledging, then process the next batch returned.' + } +} + export function formatMessageReadOnlyTag( message: OrchestrationMessageSummary, legacyCompatibilityActive = false @@ -80,8 +110,12 @@ export function formatOrchestrationCheckText( : '[LEGACY COMPATIBILITY]\n' : '' const deliveryNotice = formatCurrentDeliveryNotice(prepared.legacyCompatibility?.currentDelivery) + const replayNotice = formatReplayRecoveryNotice(prepared.replayRecovery) + const deliveryHeader = prepared.deliveryId + ? `Delivery ${prepared.deliveryId}${prepared.replayRecovery ? ' (replayed)' : ''}\n` + : '' if (prepared.formatted) { - return `${legacyHeader}${prepared.deliveryId ? `Delivery ${prepared.deliveryId}\n` : ''}${prepared.formatted}${deliveryNotice}` + return `${legacyHeader}${deliveryHeader}${prepared.formatted}${replayNotice}${deliveryNotice}` } if (prepared.count === 0) { if (prepared.timedOut) { @@ -104,30 +138,55 @@ export function formatOrchestrationCheckText( )} [${message.type ?? 'status'}] from=${message.from_handle} "${message.subject}"` ) .join('\n') - const output = prepared.deliveryId ? `Delivery ${prepared.deliveryId}\n${rendered}` : rendered - return `${legacyHeader}${output}${deliveryNotice}` + return `${legacyHeader}${deliveryHeader}${rendered}${replayNotice}${deliveryNotice}` } export function prepareOrchestrationCheckOutput( result: T, checkedTerminal: string, - formattedRequested: boolean + formattedRequested: boolean, + invocation?: OrchestrationCheckInvocation ): T { const compatibilityActive = Boolean( result.legacyCompatibility && !result.legacyCompatibility.readOnly ) + const prepared = + invocation?.consuming && + result.replayed && + result.deliveryId && + !result.legacyCompatibility && + !result.messages.some((message) => isLegacyReadOnlyMessage(message)) + ? { + ...result, + replayRecovery: + result.replayRecovery ?? + createOrchestrationReplayRecovery(result.deliveryId, invocation) + } + : result if ( !formattedRequested || !result.messages.some((message) => isLegacyReadOnlyMessage(message, compatibilityActive)) ) { - return result + return prepared } return { - ...result, + ...prepared, formatted: formatLegacyAwareCheckMessages(result.messages, checkedTerminal, compatibilityActive) } } +function formatReplayRecoveryNotice(recovery: OrchestrationReplayRecovery | undefined): string { + if (!recovery) { + return '' + } + const count = recovery.waitingCount + const waiting = + count === undefined + ? '' + : `${count} unread ${count === 1 ? 'message' : 'messages'} waiting behind this Delivery.\n` + return `\n${waiting}${recovery.guidance}\nAcknowledge: ${recovery.ackCommand}` +} + function formatMessagePriorityTag(message: OrchestrationMessageSummary): string { return message.priority === 'urgent' ? ' [URGENT]' : message.priority === 'high' ? ' [HIGH]' : '' }