diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index 81b1cf789f1..72c331d08aa 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -9747,7 +9747,7 @@ "invariant": "A public CLI send containing non-empty text plus Enter and no interrupt identifies agent-prompt intent. The provider PTY-input owner grants one generation-aware prompt transaction across renderer, RPC/mobile, orchestration, startup, and internal writers. Admitted user input has priority: it safely closes an open prompt or automation paste, invalidates the unacknowledged prompt, and proceeds without declaring the provider unavailable; unversioned large renderer input cannot stale-invalidate the owner. Complete query replies are deferred only until the paste frame closes, with overflow preempting while preserving reply order. The runtime revalidates the foreground target before at most one Enter, and only fresh lifecycle evidence acknowledges delivery. Any unacknowledged post-byte failure is operation_unknown, and orchestration retains the exact terminal in start_unknown rather than transferring or retrying it.", "oracle": "Withhold OpenCode readiness beyond the old 500 ms deadline, emit a stale pre-attempt marker and then one split fresh marker, and require zero premature Enter plus one lifecycle-acknowledged Enter. A swallowed Enter followed by autonomous redraw output must remain operation_unknown. Missing readiness emits no Enter but remains operation_unknown because paste bytes reached the composer. Current foreground identity must override stale launch metadata, a switch to shell before submit must emit no Enter, and unavailable legacy inspection must retain bounded metadata fallback. Manual input must close an open paste, cancel prompt authority, and proceed without WriteUnavailable recovery; query replies must be delivered after the paste frame. Partial-paste, post-paste, and post-Enter generation reset must remain unknown; an unknown attempt must prevent already-queued prompt delivery. Local and federated receipts persist operation_unknown as start_unknown, and the exact local terminal is not transferable.", "commands": [ - "pnpm exec vitest run --config config/vitest.config.ts src/cli/handlers/terminal.test.ts src/main/pty-input-transaction.test.ts src/main/ipc/pty-write-ipc-validation.test.ts src/main/runtime/rpc/terminal-agent-prompt-send.test.ts src/main/runtime/rpc/terminal-send.test.ts src/main/runtime/rpc/terminal-multiplex-input-write-rejection.test.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/agent-prompt-delivery-state-machine.test.ts src/main/runtime/agent-prompt-delivery-outcome.test.ts src/main/runtime/agent-prompt-submission-runtime.test.ts src/main/runtime/agent-prompt-submission-verification.test.ts src/main/runtime/terminal-send-stale-leaf-liveness.test.ts src/main/runtime/orchestration-mailbox-notification-consistency.test.ts src/main/runtime/orchestration-mailbox-transport-settlement.test.ts src/main/runtime/rpc/methods/orchestration-composed-workers.test.ts src/main/runtime/rpc/methods/orchestration-federation-start-receipt.test.ts src/shared/agent-prompt-injection.test.ts", + "pnpm exec vitest run --config config/vitest.config.ts src/cli/handlers/terminal.test.ts src/main/pty-input-transaction.test.ts src/main/ipc/pty-write-ipc-validation.test.ts src/main/runtime/rpc/terminal-agent-prompt-send.test.ts src/main/runtime/rpc/terminal-send.test.ts src/main/runtime/rpc/terminal-multiplex-input-write-rejection.test.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/agent-prompt-delivery-state-machine.test.ts src/main/runtime/agent-prompt-delivery-outcome.test.ts src/main/runtime/agent-prompt-submission-runtime.test.ts src/main/runtime/agent-prompt-submission-verification.test.ts src/main/runtime/terminal-send-stale-leaf-liveness.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-pointer-state.test.ts src/main/runtime/rpc/methods/orchestration-composed-workers.test.ts src/main/runtime/rpc/methods/orchestration-federation-start-receipt.test.ts src/shared/agent-prompt-injection.test.ts", "pnpm exec vitest run --config config/vitest.config.ts src/cli/handlers/terminal.test.ts src/main/runtime/rpc/terminal-agent-prompt-send.test.ts src/main/runtime/rpc/terminal-send.test.ts src/main/runtime/orca-runtime.test.ts src/shared/agent-prompt-injection.test.ts", "pnpm run build:cli && pnpm exec playwright test tests/e2e/terminal-send-agent-prompt-submit.spec.ts --config tests/playwright.config.ts --project electron-headless --workers=1", "pnpm.cmd exec playwright test tests/e2e/terminal-send-agent-prompt-submit.spec.ts --config tests/playwright.config.ts --project electron-headless --workers=1", @@ -9950,6 +9950,7 @@ "commands": [ "pnpm exec vitest run --config config/vitest.config.ts src/shared/agent-prompt-injection.test.ts src/main/pty-input-transaction.test.ts src/main/ipc/pty-write-ipc-validation.test.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/agent-prompt-delivery-state-machine.test.ts src/main/runtime/agent-prompt-delivery-outcome.test.ts src/main/runtime/agent-prompt-submission-runtime.test.ts src/main/runtime/agent-prompt-submission-verification.test.ts src/main/runtime/terminal-send-stale-leaf-liveness.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-pointer-state.test.ts src/main/runtime/rpc/terminal-multiplex-input-write-rejection.test.ts src/main/runtime/rpc/methods/orchestration-composed-workers.test.ts src/main/runtime/rpc/methods/orchestration-federation-start-receipt.test.ts src/main/runtime/rpc/methods/orchestration-tasks-dispatch.test.ts src/main/runtime/orchestration/coordinator.test.ts", "pnpm exec vitest run --config config/vitest.config.ts src/shared/agent-prompt-injection.test.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/rpc/methods/orchestration-tasks-dispatch.test.ts src/main/runtime/orchestration/coordinator.test.ts", + "pnpm exec vitest run --config config/vitest.config.ts src/cli/handlers/terminal.test.ts src/main/pty-input-transaction.test.ts src/main/ipc/pty-write-ipc-validation.test.ts src/main/runtime/rpc/terminal-agent-prompt-send.test.ts src/main/runtime/rpc/terminal-send.test.ts src/main/runtime/rpc/terminal-multiplex-input-write-rejection.test.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/agent-prompt-delivery-state-machine.test.ts src/main/runtime/agent-prompt-delivery-outcome.test.ts src/main/runtime/agent-prompt-submission-runtime.test.ts src/main/runtime/agent-prompt-submission-verification.test.ts src/main/runtime/terminal-send-stale-leaf-liveness.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-pointer-state.test.ts src/main/runtime/rpc/methods/orchestration-composed-workers.test.ts src/main/runtime/rpc/methods/orchestration-federation-start-receipt.test.ts src/shared/agent-prompt-injection.test.ts", "node tests/tools/repro-orchestration-long-prompt.mjs --cli out/bin/orca-dev --mode codex-like --size-kb 32 --timeout-ms 20000" ], "testFiles": [ diff --git a/src/main/pty-input-transaction-lifecycle.ts b/src/main/pty-input-transaction-lifecycle.ts index a4ddbdf5f50..4ce8b97e607 100644 --- a/src/main/pty-input-transaction-lifecycle.ts +++ b/src/main/pty-input-transaction-lifecycle.ts @@ -80,6 +80,13 @@ export function releaseOwnerToken( ownerByPtyId.delete(ptyId) } } + +export function releaseDormantOwner( + owner: OwnedTokenState, + flush: (owner: OwnedTokenState) => boolean +): boolean { + return owner.tokens.size === 0 && !owner.pasteOpen && !owner.pasteMutationPending && flush(owner) +} export function invalidateOwner( ownerByPtyId: Map>, ptyId: string, @@ -99,6 +106,17 @@ export function invalidateOwner( export function createTransactionToken(generation: number | null): TransactionToken { return { generation, active: true, reason: null, invalidate: () => undefined } } +export function invalidateTransactionToken( + token: TransactionToken, + reason: PtyInputInvalidationReason +): void { + if (!token.active) { + return + } + token.active = false + token.reason = reason + token.invalidate(reason) +} export function transactionForToken( token: TransactionToken, operations: { write: (data: string) => PtyInputWriteResult; release: () => void } diff --git a/src/main/pty-input-transaction-state.ts b/src/main/pty-input-transaction-state.ts index 31ad3bf47f9..8273e9540b4 100644 --- a/src/main/pty-input-transaction-state.ts +++ b/src/main/pty-input-transaction-state.ts @@ -40,6 +40,25 @@ export function flushQueryReplies( state.pendingQueryReplyCodeUnits = 0 return true } +export function writeQueuedQueryReply( + state: PendingQueryReplyState, + data: string, + write: (data: string) => boolean +): boolean { + if (state.pendingQueryReplies.length > 0 && !flushQueryReplies(state, write)) { + if (canQueueQueryReply(state, data)) { + enqueueQueryReply(state, data) + return true + } + return false + } + const wrote = write(data) + if (!wrote && canQueueQueryReply(state, data)) { + enqueueQueryReply(state, data) + return true + } + return wrote +} export function nextPasteState(initial: boolean, data: string): boolean { let open = initial let cursor = 0 diff --git a/src/main/pty-input-transaction.test.ts b/src/main/pty-input-transaction.test.ts index a1ee6293c37..ad7fff96bee 100644 --- a/src/main/pty-input-transaction.test.ts +++ b/src/main/pty-input-transaction.test.ts @@ -212,6 +212,78 @@ describe('PTY input transaction owner', () => { expect(prompt.active).toBe(true) }) + it('rejects a concurrent direct client while a renderer transaction owns the PTY', () => { + const writer = vi.fn(() => true) + const owner = new PtyInputTransactionOwner(writer) + const renderer = owner.beginUnversionedInteractive('pty-1') + + expect(renderer.write('first chunk')).toBe(true) + expect(owner.write('pty-1', 'concurrent client')).toBe(false) + expect(writer).toHaveBeenCalledOnce() + }) + + it('queues terminal query replies behind an interactive paste transaction', () => { + const writes: string[] = [] + const owner = new PtyInputTransactionOwner((_ptyId, data) => { + writes.push(data) + return true + }) + const renderer = owner.beginUnversionedInteractive('pty-1') + + expect(renderer.write(`${START}chunk`)).toBe(true) + expect(owner.write('pty-1', '\x1b[6n')).toBe(true) + expect(owner.write('pty-1', 'concurrent client')).toBe(false) + expect(renderer.write(END)).toBe(true) + renderer.release() + + expect(writes).toEqual([`${START}chunk`, END, '\x1b[6n']) + }) + + it('retries a deferred query reply before admitting a replacement transaction', () => { + let failReply = true + const writer = vi.fn((_ptyId: string, data: string) => { + if (data === '\x1b[6n' && failReply) { + return false + } + return true + }) + const owner = new PtyInputTransactionOwner(writer) + const prompt = owner.begin('pty-1', 1, 'agent-prompt')! + expect(prompt.write(`${START}prompt`)).toBe(true) + expect(owner.write('pty-1', '\x1b[6n')).toBe(true) + expect(prompt.write(END)).toBe(true) + prompt.release() + + failReply = false + const replacement = owner.begin('pty-1', 1, 'agent-prompt') + expect(replacement).not.toBeNull() + expect(writer.mock.calls).toEqual([ + ['pty-1', `${START}prompt`], + ['pty-1', END], + ['pty-1', '\x1b[6n'], + ['pty-1', '\x1b[6n'], + ['pty-1', '\x1b[6n'] + ]) + }) + + it('keeps a newly arrived query reply behind a deferred reply', () => { + let failReply = true + const writes: string[] = [] + const owner = new PtyInputTransactionOwner((_ptyId, data) => { + writes.push(data) + return !failReply || data !== '\x1b[6n' + }) + const prompt = owner.begin('pty-1', 1, 'agent-prompt')! + prompt.write(`${START}prompt`) + owner.write('pty-1', '\x1b[6n') + prompt.write(END) + prompt.release() + expect(owner.write('pty-1', '\x1b[5n')).toBe(true) + failReply = false + expect(owner.begin('pty-1', 1, 'agent-prompt')).not.toBeNull() + expect(writes.slice(-2)).toEqual(['\x1b[6n', '\x1b[5n']) + }) + it('joins interactive transactions only when each attempts a write', () => { const owner = new PtyInputTransactionOwner(() => true) const first = owner.begin('pty-1', 1, 'interactive')! diff --git a/src/main/pty-input-transaction.ts b/src/main/pty-input-transaction.ts index e3e9e6066af..c4b367ce561 100644 --- a/src/main/pty-input-transaction.ts +++ b/src/main/pty-input-transaction.ts @@ -3,6 +3,7 @@ import { canQueueQueryReply, enqueueQueryReply, flushQueryReplies, + writeQueuedQueryReply, nextPasteState, isPtyInputTransactionQueryReply } from './pty-input-transaction-state' @@ -14,7 +15,9 @@ import { pruneLazyTokens, releaseLazyToken, releaseOwnerToken, - transactionForToken + releaseDormantOwner, + transactionForToken, + invalidateTransactionToken } from './pty-input-transaction-lifecycle' import type { TransactionToken } from './pty-input-transaction-lifecycle' export type PtyInputTransactionKind = 'agent-prompt' | 'automation' | 'interactive' @@ -29,11 +32,10 @@ export type PtyInputTransaction = Readonly<{ }> type PtyInputWriter = (ptyId: string, data: string) => boolean type SettledPtyInputWriter = (ptyId: string, data: string) => PtyInputWriteResult -type InputToken = TransactionToken type InputOwner = { generation: number | null kind: PtyInputTransactionKind - tokens: Set + tokens: Set pasteOpen: boolean pasteMutationPending: boolean pendingQueryReplies: string[] @@ -41,25 +43,28 @@ type InputOwner = { } export class PtyInputTransactionOwner { private readonly ownerByPtyId = new Map() - private readonly lazyInteractiveByPtyId = new Map>() + private readonly lazyInteractiveByPtyId = new Map>() constructor( private readonly writeInput: PtyInputWriter, private readonly writeInputWithSettlement: SettledPtyInputWriter = writeInput ) {} write(ptyId: string, data: string): boolean { const owner = this.ownerByPtyId.get(ptyId) - if (!owner || owner.kind === 'interactive') { + if (!owner) { return this.writeInput(ptyId, data) } if (isPtyInputTransactionQueryReply(data)) { if (!owner.pasteOpen && !owner.pasteMutationPending) { - return this.writeInput(ptyId, data) + return writeQueuedQueryReply(owner, data, (reply) => this.writeInput(ptyId, reply)) } if (canQueueQueryReply(owner, data)) { enqueueQueryReply(owner, data) return true } } + if (owner.kind === 'interactive') { + return false + } if (!this.interruptOwner(ptyId, owner)) { return false } @@ -77,12 +82,21 @@ export class PtyInputTransactionOwner { ptyId, generation, undefined, - (token, reason) => this.invalidateToken(token, reason), + invalidateTransactionToken, 'terminal_handle_stale' ) } return this.beginLazyInteractive(ptyId, generation) } + const existing = this.ownerByPtyId.get(ptyId) + if ( + existing && + releaseDormantOwner(existing, (state) => + flushQueryReplies(state, (reply) => this.writeInput(ptyId, reply)) + ) + ) { + this.ownerByPtyId.delete(ptyId) + } if (this.ownerByPtyId.has(ptyId)) { return null } @@ -91,7 +105,7 @@ export class PtyInputTransactionOwner { ptyId, generation, undefined, - (token, reason) => this.invalidateToken(token, reason), + invalidateTransactionToken, 'terminal_handle_stale' ) const owner: InputOwner = { @@ -113,19 +127,19 @@ export class PtyInputTransactionOwner { const owner = this.ownerByPtyId.get(ptyId) if (owner && (owner.generation === null || owner.generation !== generation)) { invalidateOwner(this.ownerByPtyId, ptyId, owner, 'terminal_handle_stale', (token, reason) => - this.invalidateToken(token, reason) + invalidateTransactionToken(token, reason) ) } for (const token of this.lazyInteractiveByPtyId.get(ptyId) ?? []) { if (token.active && (token.generation === null || token.generation !== generation)) { - this.invalidateToken(token, 'terminal_handle_stale') + invalidateTransactionToken(token, 'terminal_handle_stale') } } pruneLazyTokens(this.lazyInteractiveByPtyId, ptyId) } private beginLazyInteractive(ptyId: string, generation: number | null): PtyInputTransaction { const token = createTransactionToken(generation) - const lazy = this.lazyInteractiveByPtyId.get(ptyId) ?? new Set() + const lazy = this.lazyInteractiveByPtyId.get(ptyId) ?? new Set() lazy.add(token) this.lazyInteractiveByPtyId.set(ptyId, lazy) let joinedOwner: InputOwner | null = null @@ -154,7 +168,7 @@ export class PtyInputTransactionOwner { private claimInteractive( ptyId: string, generation: number | null, - lazyToken: InputToken + lazyToken: TransactionToken ): InputOwner | null { const current = this.ownerByPtyId.get(ptyId) if ( @@ -164,7 +178,7 @@ export class PtyInputTransactionOwner { current.generation !== generation ) { releaseLazyToken(this.lazyInteractiveByPtyId, ptyId, lazyToken, false) - this.invalidateToken(lazyToken, 'terminal_handle_stale') + invalidateTransactionToken(lazyToken, 'terminal_handle_stale') return null } if (current && current.kind !== 'interactive' && !this.interruptOwner(ptyId, current)) { @@ -188,7 +202,7 @@ export class PtyInputTransactionOwner { ptyId, generation, lazyToken, - (token, reason) => this.invalidateToken(token, reason), + invalidateTransactionToken, 'terminal_handle_stale' ) } @@ -214,7 +228,7 @@ export class PtyInputTransactionOwner { private writeOwned( ptyId: string, owner: InputOwner, - token: InputToken, + token: TransactionToken, data: string ): PtyInputWriteResult { if (!token.active || this.ownerByPtyId.get(ptyId) !== owner || !owner.tokens.has(token)) { @@ -275,21 +289,12 @@ export class PtyInputTransactionOwner { return false } invalidateOwner(this.ownerByPtyId, ptyId, owner, 'terminal_input_superseded', (token, reason) => - this.invalidateToken(token, reason) + invalidateTransactionToken(token, reason) ) return true } - private releaseOwnerToken(ptyId: string, owner: InputOwner, token: InputToken): void { + private releaseOwnerToken = (ptyId: string, owner: InputOwner, token: TransactionToken): void => releaseOwnerToken(this.ownerByPtyId, ptyId, owner, token, (current) => flushQueryReplies(current, (reply) => this.writeInput(ptyId, reply)) ) - } - private invalidateToken(token: InputToken, reason: PtyInputInvalidationReason): void { - if (!token.active) { - return - } - token.active = false - token.reason = reason - token.invalidate(reason) - } } diff --git a/src/main/runtime/agent-prompt-delivery-state-machine.test.ts b/src/main/runtime/agent-prompt-delivery-state-machine.test.ts index 35c9bd29f87..f52966aa646 100644 --- a/src/main/runtime/agent-prompt-delivery-state-machine.test.ts +++ b/src/main/runtime/agent-prompt-delivery-state-machine.test.ts @@ -218,10 +218,19 @@ describe('agent prompt delivery state machine', () => { expect(writes).toEqual([]) }) - it('rejects an authoritative missing foreground before writing', async () => { + it('rejects a null foreground observation instead of trusting launch metadata', async () => { + vi.useFakeTimers() const { runtime, handle, writes } = await createPromptRuntime({ launchAgent: 'opencode', - inspectProcess: () => ({ foregroundProcess: null, hasChildProcesses: false }) + inspectProcess: () => ({ foregroundProcess: null, hasChildProcesses: false }), + onWrite: (runtime, data) => { + if (data.includes(AGENT_PROMPT_BRACKETED_PASTE_END)) { + runtime.onPtyData('pty-prompt', '\x1b[?25h', Date.now()) + } + if (data === '\r') { + runtime.onPtyData('pty-prompt', '\x1b]0;OpenCode working\x07', Date.now()) + } + } }) await expect(runtime.sendTerminalAgentPrompt(handle, 'review this')).rejects.toThrow( @@ -230,6 +239,31 @@ describe('agent prompt delivery state machine', () => { expect(writes).toEqual([]) }) + it('uses launch metadata only when inspection is explicitly unavailable', async () => { + vi.useFakeTimers() + const { runtime, handle, writes } = await createPromptRuntime({ + launchAgent: 'opencode', + inspectProcess: () => ({ + foregroundProcess: null, + hasChildProcesses: true, + unavailable: true + }), + onWrite: (runtime, data) => { + if (data.includes(AGENT_PROMPT_BRACKETED_PASTE_END)) { + runtime.onPtyData('pty-prompt', '\x1b[?25h', Date.now()) + } + if (data === '\r') { + runtime.onPtyData('pty-prompt', '\x1b]0;OpenCode working\x07', Date.now()) + } + } + }) + + const submission = runtime.sendTerminalAgentPrompt(handle, 'review this') + await vi.runAllTimersAsync() + await expect(submission).resolves.toMatchObject({ accepted: true }) + expect(writes).toContain('\r') + }) + it('does not submit after the agent returns to a shell', async () => { vi.useFakeTimers() let foregroundProcess = 'opencode' diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 1d9eebe57a0..1e2e48ef6c2 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -3237,19 +3237,26 @@ export class OrcaRuntimeService { getLiveLeafForHandle: (handle) => this.getLiveLeafForHandle(handle).leaf, getMessageWaiters: (mailboxHandle) => this.messageWaitersByHandle.get(mailboxHandle), getTabTitle: (tabId) => this.tabs.get(tabId)?.title, - getTerminalHandleForLeaf: (leaf) => this.issueHandle(leaf as RuntimeLeafRecord), + getTerminalHandleForLeaf: (leaf) => + this.handleByLeafKey.get(this.getLeafKey(leaf.tabId, leaf.leafId)), redriveMailbox: (mailboxHandle, reservedTypes) => this.deliverPendingMessagesForHandle(mailboxHandle, reservedTypes), sendPrompt: async (handle, prompt, options) => { + let providerAttempted = false try { await this.sendTerminalAgentPrompt(handle, prompt, { signal: options.signal, beforeWrite: options.beforeWrite, - beforeAttempt: options.beforeAttempt + beforeAttempt: () => { + options.beforeAttempt() + providerAttempted = true + } }) return 'delivered' } catch (error) { - return isAgentPromptDeliveryUnknownError(error) ? 'unknown' : 'rejected' + return providerAttempted || isAgentPromptDeliveryUnknownError(error) + ? 'unknown' + : 'rejected' } } }) @@ -6209,7 +6216,10 @@ export class OrcaRuntimeService { // Why: CLI terminal writes must go through the main-owned PTY registry // instead of tunneling back through renderer IPC, or live handles could // drift from the process they are supposed to control during reloads. - if (controller && !controller.beginInputTransaction) { + if ( + controller && + (!controller.beginInputTransaction || !controller.invalidateInputTransactions) + ) { const write = controller.write.bind(controller) const writeWithSettlement = controller.writeWithSettlement?.bind(controller) const inputOwner = new PtyInputTransactionOwner( @@ -19807,6 +19817,7 @@ export class OrcaRuntimeService { try { const inspection = await controller.inspectProcess(ptyId) return { + // Null is a valid idle-shell observation; only the explicit unavailable bit permits fallback. available: inspection.unavailable !== true, process: inspection.foregroundProcess } diff --git a/src/main/runtime/orchestration-mailbox-transport-settlement.test.ts b/src/main/runtime/orchestration-mailbox-transport-settlement.test.ts index 4d9d1abf58e..3cac267f6e4 100644 --- a/src/main/runtime/orchestration-mailbox-transport-settlement.test.ts +++ b/src/main/runtime/orchestration-mailbox-transport-settlement.test.ts @@ -88,6 +88,7 @@ describe('orchestration mailbox prompt settlement', () => { } expect(db.getMessageById(message.id)?.delivered_at).toEqual(expect.any(String)) + expect(db.getMessageById(message.id)?.delivery_state).toBe('unknown') const restarted = createRuntime(db) await driveToLiveIdle(restarted.runtime) expect(pointerCount(restarted.write)).toBe(0) @@ -106,7 +107,7 @@ describe('orchestration mailbox prompt settlement', () => { }) const run = createBoundRun(db, 'Staging failure Run') const message = insertDirectRunMessage(db, run.id, 'Must remain explicit') - vi.spyOn(db, 'markAsDelivered').mockImplementation(() => { + vi.spyOn(db, 'markAsDeliveryStaged').mockImplementation(() => { throw new Error('disk unavailable') }) diff --git a/src/main/runtime/orchestration/db/contract-constants.ts b/src/main/runtime/orchestration/db/contract-constants.ts index 6d175498bc4..2a5aa877055 100644 --- a/src/main/runtime/orchestration/db/contract-constants.ts +++ b/src/main/runtime/orchestration/db/contract-constants.ts @@ -6,5 +6,5 @@ export const LEGACY_RUN_ID = ORCHESTRATION_LEGACY_RUN_ID export const LEGACY_CONTRACT_VERSION = 0 export const CURRENT_CONTRACT_VERSION = ORCHESTRATION_CONTRACT_VERSION -// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 legacy scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity, v27 durable federation acknowledgments, v28 durable local mutation caller identity. -export const SCHEMA_VERSION = 29 +// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 durable scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity, v27 durable federation acknowledgments, v28 durable local mutation caller identity, v30 durable mailbox delivery attempts. +export const SCHEMA_VERSION = 30 diff --git a/src/main/runtime/orchestration/db/messages/message-inbox.ts b/src/main/runtime/orchestration/db/messages/message-inbox.ts index af94b2e7c16..9efe341a0a3 100644 --- a/src/main/runtime/orchestration/db/messages/message-inbox.ts +++ b/src/main/runtime/orchestration/db/messages/message-inbox.ts @@ -96,7 +96,7 @@ export function getUndeliveredUnreadMessages( const conditions = [ 'to_handle = ?', 'read = 0', - 'delivered_at IS NULL', + "COALESCE(delivery_state, CASE WHEN delivered_at IS NULL THEN 'none' ELSE 'delivered' END) = 'none'", "delivery_contract = 'current_delivery'" ] const params: (string | number)[] = [toHandle] @@ -128,7 +128,7 @@ export function getUndeliveredUnreadMailboxHandles(this: OrchestrationDb): strin this.db .prepare( `SELECT DISTINCT to_handle FROM messages - WHERE read = 0 AND delivered_at IS NULL + WHERE read = 0 AND COALESCE(delivery_state, CASE WHEN delivered_at IS NULL THEN 'none' ELSE 'delivered' END) = 'none' AND delivery_contract = 'current_delivery'` ) .all() as { to_handle: string }[] @@ -164,7 +164,7 @@ export function markAsDelivered(this: OrchestrationDb, ids: string[]): void { this, ids, (placeholders) => - `UPDATE messages SET delivered_at = datetime('now') WHERE id IN (${placeholders})` + `UPDATE messages SET delivered_at = COALESCE(delivered_at, datetime('now')), delivery_state = 'delivered' WHERE id IN (${placeholders})` ) } @@ -173,11 +173,29 @@ export function markAsUndelivered(this: OrchestrationDb, ids: string[]): void { this, ids, (placeholders) => - `UPDATE messages SET delivered_at = NULL + `UPDATE messages SET delivered_at = NULL, delivery_state = 'none' WHERE read = 0 AND id IN (${placeholders})` ) } +export function markAsDeliveryStaged(this: OrchestrationDb, ids: string[]): void { + runBatchedMessageMutation( + this, + ids, + (placeholders) => + `UPDATE messages SET delivered_at = COALESCE(delivered_at, datetime('now')), delivery_state = 'staged' WHERE read = 0 AND id IN (${placeholders})` + ) +} + +export function markAsDeliveryUnknown(this: OrchestrationDb, ids: string[]): void { + runBatchedMessageMutation( + this, + ids, + (placeholders) => + `UPDATE messages SET delivered_at = COALESCE(delivered_at, datetime('now')), delivery_state = 'unknown' WHERE read = 0 AND id IN (${placeholders})` + ) +} + export function areUnreadMessages(this: OrchestrationDb, toHandle: string, ids: string[]): boolean { let matched = 0 for (let offset = 0; offset < ids.length; offset += MESSAGE_ID_UPDATE_BATCH_SIZE) { @@ -201,7 +219,7 @@ export function markAsReadAndDelivered(this: OrchestrationDb, ids: string[]): vo this, ids, (placeholders) => - `UPDATE messages SET read = 1, delivered_at = COALESCE(delivered_at, datetime('now')) WHERE id IN (${placeholders})` + `UPDATE messages SET read = 1, delivered_at = COALESCE(delivered_at, datetime('now')), delivery_state = 'delivered' WHERE id IN (${placeholders})` ) } @@ -270,6 +288,8 @@ export type MessageInboxMethods = { markAsRead: typeof markAsRead markAsDelivered: typeof markAsDelivered markAsUndelivered: typeof markAsUndelivered + markAsDeliveryStaged: typeof markAsDeliveryStaged + markAsDeliveryUnknown: typeof markAsDeliveryUnknown areUnreadMessages: typeof areUnreadMessages markAsReadAndDelivered: typeof markAsReadAndDelivered getInbox: typeof getInbox @@ -288,6 +308,8 @@ export function attachMessageInbox(ctor: { prototype: object }): void { markAsRead, markAsDelivered, markAsUndelivered, + markAsDeliveryStaged, + markAsDeliveryUnknown, areUnreadMessages, markAsReadAndDelivered, getInbox, diff --git a/src/main/runtime/orchestration/db/schema/create-core-tables-sql.ts b/src/main/runtime/orchestration/db/schema/create-core-tables-sql.ts index 1b3172edf31..ee6e796f06d 100644 --- a/src/main/runtime/orchestration/db/schema/create-core-tables-sql.ts +++ b/src/main/runtime/orchestration/db/schema/create-core-tables-sql.ts @@ -36,6 +36,8 @@ CREATE TABLE IF NOT EXISTS messages ( sequence INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL DEFAULT (datetime('now')), delivered_at TEXT, + delivery_state TEXT NOT NULL DEFAULT 'none' + CHECK(delivery_state IN ('none', 'staged', 'delivered', 'unknown')), sender_pane_key TEXT ); diff --git a/src/main/runtime/orchestration/db/schema/migrate.ts b/src/main/runtime/orchestration/db/schema/migrate.ts index 3c33b5c8ce5..1a730cb8140 100644 --- a/src/main/runtime/orchestration/db/schema/migrate.ts +++ b/src/main/runtime/orchestration/db/schema/migrate.ts @@ -16,6 +16,23 @@ export function migrate(this: OrchestrationDb): void { try { applySchemaMigrationsV2ToV12.call(this, current) applySchemaMigrationsV13ToV29.call(this, current) + if (current < 30) { + if (!this.hasColumn('messages', 'delivery_state')) { + this.db.exec(`ALTER TABLE messages ADD COLUMN delivery_state TEXT NOT NULL DEFAULT 'none'`) + } + this.db.exec(` + UPDATE messages + SET delivery_state = CASE + WHEN delivered_at IS NULL THEN 'none' + ELSE 'delivered' + END + WHERE delivery_state = 'none' AND delivered_at IS NOT NULL + `) + this.db.exec(` + CREATE INDEX IF NOT EXISTS idx_messages_delivery_state + ON messages(to_handle, read, delivery_state, sequence) + `) + } this.db.pragma(`user_version = ${SCHEMA_VERSION}`) this.db.exec('COMMIT') } catch (err) { diff --git a/src/main/runtime/orchestration/mailbox-pointer-delivery-contract.ts b/src/main/runtime/orchestration/mailbox-pointer-delivery-contract.ts index f4827dd8554..4998f2eeedc 100644 --- a/src/main/runtime/orchestration/mailbox-pointer-delivery-contract.ts +++ b/src/main/runtime/orchestration/mailbox-pointer-delivery-contract.ts @@ -12,7 +12,7 @@ export type PointerDeliveryDependencies OrchestrationMailboxLeaf getMessageWaiters: (mailboxHandle: string) => ReadonlySet | undefined getTabTitle: (tabId: string) => string | null | undefined - getTerminalHandleForLeaf: (leaf: OrchestrationMailboxLeaf) => string + getTerminalHandleForLeaf: (leaf: OrchestrationMailboxLeaf) => string | undefined redriveMailbox: (mailboxHandle: string, reservedTypes?: ReadonlySet) => void sendPrompt: ( handle: string, diff --git a/src/main/runtime/orchestration/mailbox-pointer-delivery-validation.ts b/src/main/runtime/orchestration/mailbox-pointer-delivery-validation.ts new file mode 100644 index 00000000000..897b8a43684 --- /dev/null +++ b/src/main/runtime/orchestration/mailbox-pointer-delivery-validation.ts @@ -0,0 +1,50 @@ +import type { PointerDeliveryDependencies } from './mailbox-pointer-delivery-contract' +import { shouldReleaseOrchestrationPointer } from './mailbox-pointer-eligibility' +import type { OrchestrationMailboxLeaf } from './mailbox-owner' +import type { + OrchestrationMailboxDeliveryFlight, + OrchestrationMailboxPointerState +} from './mailbox-pointer-state' +export function markMailboxDeliveryDelivered( + db: { markAsDelivered: (messageIds: string[]) => void }, + messageIds: readonly string[], + usedLegacyDeliveredTransition: boolean +): void { + if (messageIds.length > 0 && !usedLegacyDeliveredTransition) { + db.markAsDelivered([...messageIds]) + } +} + +export function assertMailboxPointerDeliveryCurrent< + TWaiter extends { typeFilter: string[] | undefined } +>( + deps: PointerDeliveryDependencies, + state: OrchestrationMailboxPointerState, + leaf: OrchestrationMailboxLeaf, + mailboxHandle: string, + unread: readonly { id: string; type: string; sequence: number }[], + ptyId: string, + flight: OrchestrationMailboxDeliveryFlight +): void { + const currentLeaf = deps.getLeaf(deps.getLeafKey(leaf.tabId, leaf.leafId)) + const ownerChanged = + !state.isCurrentFlight(ptyId, flight) || + currentLeaf?.ptyId !== ptyId || + !currentLeaf.writable || + deps.mailboxOwner.resolve(currentLeaf) !== mailboxHandle + const agentNoLongerIdle = + currentLeaf?.lastAgentStatus !== 'idle' || !currentLeaf.lastAgentStatusObservedLive + const messagesReleased = shouldReleaseOrchestrationPointer( + deps.getDb(), + mailboxHandle, + unread, + deps.getMessageWaiters(mailboxHandle) + ) + if (!ownerChanged && !agentNoLongerIdle && !messagesReleased) { + return + } + if (!flight.mutated && ownerChanged) { + flight.redriveMailbox = mailboxHandle + } + throw new Error('orchestration_pointer_released') +} diff --git a/src/main/runtime/orchestration/mailbox-pointer-delivery.ts b/src/main/runtime/orchestration/mailbox-pointer-delivery.ts index 58d4ef2b062..12a72f13bc3 100644 --- a/src/main/runtime/orchestration/mailbox-pointer-delivery.ts +++ b/src/main/runtime/orchestration/mailbox-pointer-delivery.ts @@ -5,7 +5,6 @@ import type { PointerDeliveryDependencies } from './mailbox-pointer-delivery-con import { hasUnfilteredOrchestrationWaiter, messageTypeHasOrchestrationWaiter, - shouldReleaseOrchestrationPointer, type OrchestrationMessageWaiter } from './mailbox-pointer-eligibility' import type { OrchestrationMailboxLeaf } from './mailbox-owner' @@ -13,6 +12,11 @@ import { OrchestrationMailboxPointerState, type OrchestrationMailboxDeliveryFlight } from './mailbox-pointer-state' +import { redriveMailboxPointer } from './mailbox-pointer-redrive' +import { + assertMailboxPointerDeliveryCurrent, + markMailboxDeliveryDelivered +} from './mailbox-pointer-delivery-validation' export type { OrchestrationMessageWaiter } from './mailbox-pointer-eligibility' export class OrchestrationMailboxPointerDelivery { private readonly state = new OrchestrationMailboxPointerState() @@ -44,7 +48,7 @@ export class OrchestrationMailboxPointerDelivery): void { - this.state.parkRedelivery(mailboxHandle, reservedTypes) - } retirePty(ptyId: string): void { const { flight, releasedMailboxes } = this.state.retirePty(ptyId) flight?.abortController.abort() @@ -117,7 +125,7 @@ export class OrchestrationMailboxPointerDelivery - this.assertPromptDeliveryCurrent(leaf, mailboxHandle, unread, ptyId, flight), + assertMailboxPointerDeliveryCurrent( + this.deps, + this.state, + leaf, + mailboxHandle, + unread, + ptyId, + flight + ), beforeAttempt: () => this.stagePromptDeliveryAttempt( leaf, @@ -170,9 +186,16 @@ export class OrchestrationMailboxPointerDelivery 0) { + const db = this.deps.getDb() + if (db?.markAsDelivered && !flight.usedLegacyDeliveredTransition) { + markMailboxDeliveryDelivered(db, messageIds, flight.usedLegacyDeliveredTransition) + } + } if ( outcome === 'delivered' && [leaf.lastOscTitle, leaf.paneTitle, this.deps.getTabTitle(leaf.tabId)].some( @@ -183,11 +206,19 @@ export class OrchestrationMailboxPointerDelivery 0) { + const db = this.deps.getDb() + if (outcome === 'rejected') { + db?.markAsUndelivered?.(messageIds) + } else { + ;(db?.markAsDeliveryUnknown ?? db?.markAsDelivered)?.call(db, messageIds) + } + } const released = - outcome === 'delivered' - ? this.state.clearWatermark(mailboxHandle, newestSequence, ptyId) - : this.state.deactivateWatermark(mailboxHandle, newestSequence, ptyId) - if (released && outcome === 'delivered') { + outcome === 'unknown' + ? this.state.deactivateWatermark(mailboxHandle, newestSequence, ptyId) + : this.state.clearWatermark(mailboxHandle, newestSequence, ptyId) + if (released && outcome !== 'unknown') { this.redrive(mailboxHandle) } } finally { @@ -197,35 +228,6 @@ export class OrchestrationMailboxPointerDelivery message.id)) - this.state.setWatermark(mailboxHandle, newestSequence, ptyId, this.leafKey(leaf)) + const stage = db.markAsDeliveryStaged ?? db.markAsDelivered + flight.usedLegacyDeliveredTransition = !db.markAsDeliveryStaged + stage.call( + db, + unread.map((message) => message.id) + ) + flight.messageIds = unread.map((message) => message.id) + this.state.setWatermark( + mailboxHandle, + newestSequence, + ptyId, + this.deps.getLeafKey(leaf.tabId, leaf.leafId) + ) flight.mutated = true if (!this.state.isCurrentFlight(ptyId, flight)) { this.state.deactivateWatermark(mailboxHandle, newestSequence, ptyId) } - this.assertPromptDeliveryCurrent(leaf, mailboxHandle, unread, ptyId, flight) + assertMailboxPointerDeliveryCurrent( + this.deps, + this.state, + leaf, + mailboxHandle, + unread, + ptyId, + flight + ) if ( [leaf.lastOscTitle, leaf.paneTitle, this.deps.getTabTitle(leaf.tabId)].some( isCursorAgentTitle @@ -259,12 +280,14 @@ export class OrchestrationMailboxPointerDelivery { - try { - this.deps.redriveMailbox(mailboxHandle, parkedTypes ?? undefined) - } catch {} - }) - } - private leafKey(leaf: OrchestrationMailboxLeaf): string { - return this.deps.getLeafKey(leaf.tabId, leaf.leafId) + redriveMailboxPointer(this.state, this.deps, mailboxHandle, force) } } diff --git a/src/main/runtime/orchestration/mailbox-pointer-redrive.ts b/src/main/runtime/orchestration/mailbox-pointer-redrive.ts new file mode 100644 index 00000000000..667466a788f --- /dev/null +++ b/src/main/runtime/orchestration/mailbox-pointer-redrive.ts @@ -0,0 +1,24 @@ +import type { PointerDeliveryDependencies } from './mailbox-pointer-delivery-contract' +import type { OrchestrationMessageWaiter } from './mailbox-pointer-eligibility' +import type { OrchestrationMailboxPointerState } from './mailbox-pointer-state' + +export function redriveMailboxPointer( + state: OrchestrationMailboxPointerState, + deps: PointerDeliveryDependencies, + mailboxHandle: string, + force = false +): void { + const parkedTypes = state.takeRedelivery(mailboxHandle, force) + if (parkedTypes === undefined) { + return + } + queueMicrotask(() => { + try { + const db = deps.getDb() + if (db && db.getUndeliveredUnreadMessages(mailboxHandle).length === 0) { + return + } + deps.redriveMailbox(mailboxHandle, parkedTypes ?? undefined) + } catch {} + }) +} diff --git a/src/main/runtime/orchestration/mailbox-pointer-state.test.ts b/src/main/runtime/orchestration/mailbox-pointer-state.test.ts index 3508e2ecc01..11a17d0e870 100644 --- a/src/main/runtime/orchestration/mailbox-pointer-state.test.ts +++ b/src/main/runtime/orchestration/mailbox-pointer-state.test.ts @@ -1,11 +1,27 @@ -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import { OrchestrationMailboxPointerState } from './mailbox-pointer-state' +import { markMailboxDeliveryDelivered } from './mailbox-pointer-delivery-validation' describe('OrchestrationMailboxPointerState', () => { - it('does not admit newer mail after an unknown prompt is deactivated', () => { + it('persists delivery before a Cursor fast-path redrive', () => { + const markAsDelivered = vi.fn() + markMailboxDeliveryDelivered({ markAsDelivered }, ['message-1'], false) + expect(markAsDelivered).toHaveBeenCalledWith(['message-1']) + }) + it('keeps an unknown attempt fenced until its PTY incarnation retires', () => { const state = new OrchestrationMailboxPointerState() state.setWatermark('run:1', 1, 'pty-1', 'pane-1') expect(state.deactivateWatermark('run:1', 1, 'pty-1')).toBe(true) - expect(state.releaseSupersededWatermark('run:1', 2)).toBe(false) + expect(state.releaseSupersededWatermark('run:1', 2, 'pty-1', 'pane-1')).toBe(false) + expect(state.releaseSupersededWatermark('run:1', 2, 'pty-2', 'pane-2')).toBe(true) + }) + + it('releases an unknown watermark with its retired PTY', () => { + const state = new OrchestrationMailboxPointerState() + state.setWatermark('run:1', 1, 'pty-1', 'pane-1') + state.deactivateWatermark('run:1', 1, 'pty-1') + + expect(state.retirePty('pty-1').releasedMailboxes).toEqual(['run:1']) + expect(state.releaseSupersededWatermark('run:1', 2, 'pty-2', 'pane-2')).toBe(true) }) }) diff --git a/src/main/runtime/orchestration/mailbox-pointer-state.ts b/src/main/runtime/orchestration/mailbox-pointer-state.ts index 626ec67189f..138bbac5dbe 100644 --- a/src/main/runtime/orchestration/mailbox-pointer-state.ts +++ b/src/main/runtime/orchestration/mailbox-pointer-state.ts @@ -3,6 +3,8 @@ import type { OrchestrationMailboxLeaf } from './mailbox-owner' export type OrchestrationMailboxDeliveryFlight = { abortController: AbortController mutated: boolean + messageIds: string[] + usedLegacyDeliveredTransition: boolean redriveMailbox: string | null } @@ -29,7 +31,13 @@ export class OrchestrationMailboxPointerState { } beginFlight(ptyId: string): OrchestrationMailboxDeliveryFlight { - const flight = { abortController: new AbortController(), mutated: false, redriveMailbox: null } + const flight = { + abortController: new AbortController(), + mutated: false, + messageIds: [], + usedLegacyDeliveredTransition: false, + redriveMailbox: null + } this.flightsByPtyId.set(ptyId, flight) return flight } @@ -75,15 +83,20 @@ export class OrchestrationMailboxPointerState { return this.watermarkByMailbox.get(mailboxHandle)?.active === true } - releaseSupersededWatermark(mailboxHandle: string, newestSequence: number): boolean { + releaseSupersededWatermark( + mailboxHandle: string, + newestSequence: number, + ptyId: string | undefined, + leafKey: string + ): boolean { const owner = this.watermarkByMailbox.get(mailboxHandle) - if (owner && !owner.active) { - return false - } - if (newestSequence > (owner?.sequence ?? -1)) { + if (!owner) { return true } - return false + if (newestSequence <= owner.sequence) { + return false + } + return owner.active || owner.ptyId !== ptyId || owner.leafKey !== leafKey } setWatermark(mailbox: string, sequence: number, ptyId: string, leafKey: string): void { @@ -153,17 +166,11 @@ export class OrchestrationMailboxPointerState { if (!owner || owner.ptyId !== ptyId) { continue } - if (flight?.mutated || !owner.active) { - owner.active = false - continue - } if (this.clearWatermark(mailboxHandle, owner.sequence, ptyId)) { releasedMailboxes.push(mailboxHandle) } } - if (!flight?.mutated) { - this.watermarkMailboxesByPtyId.delete(ptyId) - } + this.watermarkMailboxesByPtyId.delete(ptyId) return { flight, releasedMailboxes } } diff --git a/src/main/runtime/orchestration/types.ts b/src/main/runtime/orchestration/types.ts index 7b73f227a0a..7ae8dbbf213 100644 --- a/src/main/runtime/orchestration/types.ts +++ b/src/main/runtime/orchestration/types.ts @@ -17,6 +17,8 @@ export type MessagePriority = 'normal' | 'high' | 'urgent' export type MessageDeliveryContract = 'legacy_direct' | 'current_delivery' | 'audit_only' +export type MessageDeliveryState = 'none' | 'staged' | 'delivered' | 'unknown' + export type TaskStatus = 'pending' | 'ready' | 'dispatched' | 'completed' | 'failed' | 'blocked' export type DispatchStatus = 'pending' | 'dispatched' | 'completed' | 'failed' | 'circuit_broken' @@ -240,6 +242,7 @@ export type MessageRow = { sequence: number created_at: string delivered_at: string | null + delivery_state?: MessageDeliveryState sender_pane_key: string | null }