diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 502bd97e826..7983ebacb40 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -31166,7 +31166,7 @@ export class OrcaRuntimeService { if (!this._orchestrationDb) { return } - const run = this._orchestrationDb.getCurrentRunForPane(`${leaf.tabId}:${leaf.leafId}`) + const run = this._orchestrationDb.getCurrentRunForPane?.(`${leaf.tabId}:${leaf.leafId}`) if (run) { this.deliverPendingMessages(leaf, { mailboxHandle: `run:${run.id}` }) } @@ -31832,7 +31832,7 @@ export class OrcaRuntimeService { } } - // Why: an Enter armed for a dead session must not fire into a same-id cold restore. + // Why: a dead session's Enter or watermark must not affect a same-id cold restore. private retirePendingMessageDeliveryForPty(ptyId: string): void { const flight = this.messageDeliveryFlightsByPtyId.get(ptyId) if (flight?.enterTimer != null) { @@ -31840,6 +31840,16 @@ export class OrcaRuntimeService { } this.messageDeliveryFlightsByPtyId.delete(ptyId) this.parkedMessageRedeliveriesByPtyId.delete(ptyId) + for (const leaf of this.getLeavesForPty(ptyId)) { + const handle = this.handleByLeafKey.get(this.getLeafKey(leaf.tabId, leaf.leafId)) + if (handle) { + this.lastPointedMessageSequenceByHandle.delete(handle) + } + const run = this._orchestrationDb?.getCurrentRunForPane?.(`${leaf.tabId}:${leaf.leafId}`) + if (run) { + this.lastPointedMessageSequenceByHandle.delete(`run:${run.id}`) + } + } } // Why: push-on-idle delivery is event-driven (no polling) because the runtime owns both the message store and terminal status detection. diff --git a/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts b/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts index 823c04e5a6d..41e272ecd29 100644 --- a/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts +++ b/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts @@ -267,6 +267,7 @@ function makeOrchestrationDbStub(toHandle: () => string) { getUndeliveredUnreadMessages: (handle: string) => rows.filter((row) => row.to_handle === handle && row.read === 0 && !row.delivered_at), getActiveCoordinatorRun: () => null, + getCurrentRunForPane: () => undefined, // Consulted by onPtyExit's dispatch-failure path. getActiveDispatchForTerminal: () => null, markAsDelivered, @@ -327,7 +328,7 @@ describe('push-on-idle orchestration delivery absence gate', () => { expect(write).toHaveBeenCalledWith( STALE_PTY_ID, - expect.stringContaining('Subject: for the old session') + expect.stringContaining('You have 1 orchestration message') ) }) @@ -369,7 +370,10 @@ describe('push-on-idle orchestration delivery absence gate', () => { const payloads = write.mock.calls .map(([, data]) => data) .filter((data): data is string => typeof data === 'string') - expect(payloads.some((data) => data.includes('Subject: unclaimed status'))).toBe(true) + const pointers = payloads.filter((data) => data.includes('orca orchestration check')) + expect(pointers).toHaveLength(1) + expect(pointers[0]).toContain('You have 1 orchestration message') + expect(payloads.some((data) => data.includes('unclaimed status'))).toBe(false) expect(payloads.some((data) => data.includes('Subject: worker completion'))).toBe(false) }) @@ -411,8 +415,11 @@ describe('push-on-idle orchestration delivery absence gate', () => { const payloads = write.mock.calls .map(([, data]) => data) .filter((data): data is string => typeof data === 'string') - expect(payloads.some((data) => data.includes('Subject: unclaimed status'))).toBe(true) - expect(payloads.some((data) => data.includes('Subject: late completion'))).toBe(true) + const pointers = payloads.filter((data) => data.includes('orca orchestration check')) + expect(pointers).toHaveLength(1) + expect(pointers[0]).toContain('You have 2 orchestration messages') + expect(payloads.some((data) => data.includes('unclaimed status'))).toBe(false) + expect(payloads.some((data) => data.includes('late completion'))).toBe(false) }) it('keeps messages queued instead of marking a proven-absent pty delivered', async () => { @@ -440,12 +447,14 @@ describe('push-on-idle orchestration delivery absence gate', () => { // Why twice: the probe continuation yields a turn before delivering. await new Promise((resolve) => setTimeout(resolve, 0)) - expect(write).toHaveBeenCalledWith(STALE_PTY_ID, expect.stringContaining('Subject: hello')) + expect(write).toHaveBeenCalledWith( + STALE_PTY_ID, + expect.stringContaining('You have 1 orchestration message') + ) }) - // Why: delivered_at stamps only in the delayed-Enter callback, so the whole - // write→settle span — not just the probe — must be single-flight; a trigger - // landing inside the 500ms window would re-read the same un-stamped rows. + // Why: the whole pointer→Enter span must be single-flight; a trigger inside + // the 500ms window must park until the sequence watermark advances. it('delivers once across concurrent probe triggers and an in-window re-trigger, then flushes parked rows', async () => { vi.useFakeTimers() try { @@ -464,36 +473,33 @@ describe('push-on-idle orchestration delivery absence gate', () => { resolveProbe(null) await vi.advanceTimersByTimeAsync(0) - const firstSubjectWrites = () => + const pointerWrites = () => write.mock.calls.filter( - ([, data]) => typeof data === 'string' && data.includes('Subject: exactly once') + ([, data]) => typeof data === 'string' && data.includes('orca orchestration check') ) - expect(firstSubjectWrites()).toHaveLength(1) + expect(pointerWrites()).toHaveLength(1) + expect(pointerWrites()[0]?.[1]).toContain('You have 1 orchestration message') - // Re-trigger INSIDE the 500ms Enter window: the first batch is written but - // not yet stamped, so a fresh probe cycle would re-inject it. (On the - // fixed code no new probe is armed — the trigger parks; resolveProbe then - // re-resolves the settled first probe, a no-op.) + // Re-trigger inside the Enter window parks; resolving the settled first + // probe again is a no-op. stub.insert('second message') runtime.deliverPendingMessagesForHandle(handle) resolveProbe(null) await vi.advanceTimersByTimeAsync(0) - expect(firstSubjectWrites()).toHaveLength(1) + expect(pointerWrites()).toHaveLength(1) - // Enter fires, delivered_at stamps, the flight settles, and the parked - // trigger re-runs on its own — arming a fresh probe for the new row. + // Enter settles the flight and re-runs the parked trigger, arming a fresh + // probe for the newer sequence. await vi.advanceTimersByTimeAsync(500) resolveProbe(null) await vi.advanceTimersByTimeAsync(0) - const secondSubjectWrites = write.mock.calls.filter( - ([, data]) => typeof data === 'string' && data.includes('Subject: second message') - ) - expect(secondSubjectWrites).toHaveLength(1) - expect(firstSubjectWrites()).toHaveLength(1) + expect(pointerWrites()).toHaveLength(2) + expect(pointerWrites()[1]?.[1]).toContain('You have 2 orchestration messages') await vi.advanceTimersByTimeAsync(500) - expect(stub.rows.every((row) => row.delivered_at !== null)).toBe(true) + expect(stub.markAsDelivered).not.toHaveBeenCalled() + expect(stub.rows.every((row) => row.delivered_at === null)).toBe(true) } finally { vi.useRealTimers() } @@ -514,35 +520,30 @@ describe('push-on-idle orchestration delivery absence gate', () => { stub.insert('second') runtime.deliverPendingMessagesForHandle(handle) - const firstSubjectWrites = () => + const pointerWrites = () => write.mock.calls.filter( - ([, data]) => typeof data === 'string' && data.includes('Subject: first') + ([, data]) => typeof data === 'string' && data.includes('orca orchestration check') ) - expect(firstSubjectWrites()).toHaveLength(1) + expect(pointerWrites()).toHaveLength(1) + expect(pointerWrites()[0]?.[1]).toContain('You have 1 orchestration message') expect(probe).not.toHaveBeenCalled() - // Settle flushes the parked trigger; the second row delivers alone — - // its batch must not re-contain the already-stamped first row. + // Settle flushes the parked trigger; both still-pending rows are counted, + // while the newer sequence authorizes exactly one fresh pointer. await vi.advanceTimersByTimeAsync(500) - const secondOnlyWrites = write.mock.calls.filter( - ([, data]) => - typeof data === 'string' && - data.includes('Subject: second') && - !data.includes('Subject: first') - ) - expect(secondOnlyWrites).toHaveLength(1) - expect(firstSubjectWrites()).toHaveLength(1) + expect(pointerWrites()).toHaveLength(2) + expect(pointerWrites()[1]?.[1]).toContain('You have 2 orchestration messages') await vi.advanceTimersByTimeAsync(500) - expect(stub.rows.every((row) => row.delivered_at !== null)).toBe(true) + expect(stub.markAsDelivered).not.toHaveBeenCalled() + expect(stub.rows.every((row) => row.delivered_at === null)).toBe(true) } finally { vi.useRealTimers() } }) // Why: cold restore respawns under the SAME session id. An Enter armed for - // the dead incarnation must not fire into the replacement — it would inject - // \r and stamp rows the new session never received. + // the dead incarnation must not submit stale input into the replacement. it('retires an armed Enter when the pty exits and respawns under the same id inside the window', async () => { vi.useFakeTimers() try { @@ -569,18 +570,19 @@ describe('push-on-idle orchestration delivery absence gate', () => { runtime.deliverPendingMessagesForHandle(handle) expect( write.mock.calls.filter( - ([, data]) => typeof data === 'string' && data.includes('Subject: for the old session') + ([, data]) => typeof data === 'string' && data.includes('orca orchestration check') ) ).toHaveLength(1) runtime.onPtyData(STALE_PTY_ID, '\x1b]0;Codex working\x07', 200) runtime.onPtyData(STALE_PTY_ID, '\x1b]0;Codex done\x07', 201) const payloadWrites = write.mock.calls.filter( - ([, data]) => typeof data === 'string' && data.includes('Subject: for the old session') + ([, data]) => typeof data === 'string' && data.includes('orca orchestration check') ) expect(payloadWrites).toHaveLength(2) await vi.advanceTimersByTimeAsync(500) expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(1) - expect(stub.rows[0].delivered_at).not.toBeNull() + expect(stub.markAsDelivered).not.toHaveBeenCalled() + expect(stub.rows[0].delivered_at).toBeNull() } finally { vi.useRealTimers() } @@ -595,18 +597,21 @@ describe('push-on-idle orchestration delivery absence gate', () => { }) const internals = runtime as unknown as { messageDeliveryFlightsByPtyId: Map - parkedMessageRedeliveryLeavesByPtyId: Map + parkedMessageRedeliveriesByPtyId: Map + lastPointedMessageSequenceByHandle: Map } stub.insert('first') runtime.deliverPendingMessagesForHandle(handle) stub.insert('second') runtime.deliverPendingMessagesForHandle(handle) expect(internals.messageDeliveryFlightsByPtyId.size).toBe(1) - expect(internals.parkedMessageRedeliveryLeavesByPtyId.size).toBe(1) + expect(internals.parkedMessageRedeliveriesByPtyId.size).toBe(1) + expect(internals.lastPointedMessageSequenceByHandle.size).toBe(1) runtime.onPtyExit(STALE_PTY_ID, 0) expect(internals.messageDeliveryFlightsByPtyId.size).toBe(0) - expect(internals.parkedMessageRedeliveryLeavesByPtyId.size).toBe(0) + expect(internals.parkedMessageRedeliveriesByPtyId.size).toBe(0) + expect(internals.lastPointedMessageSequenceByHandle.size).toBe(0) await vi.advanceTimersByTimeAsync(500) expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(0)