fix(orchestration): reset mail pointer state on pty exit

This commit is contained in:
Jinwoo-H
2026-08-07 01:17:47 -04:00
parent 8bc659f83e
commit 6b4bc00625
2 changed files with 63 additions and 48 deletions
+12 -2
View File
@@ -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.
@@ -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<string, unknown>
parkedMessageRedeliveryLeavesByPtyId: Map<string, unknown>
parkedMessageRedeliveriesByPtyId: Map<string, unknown>
lastPointedMessageSequenceByHandle: Map<string, unknown>
}
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)