fix(orchestration): bound mailbox reconciliation

This commit is contained in:
Jinwoo-H
2026-08-14 16:35:55 -04:00
committed by Jinwoo-H
parent db767a82c3
commit ddac8cbe7c
17 changed files with 526 additions and 130 deletions
+13 -7
View File
@@ -8804,10 +8804,10 @@
"providers": ["local", "daemon", "ssh", "wsl", "remote-runtime"],
"coveredPlatforms": ["macos"],
"coveredProviders": ["local"],
"coverageNotes": "A deterministic in-process runtime/SQLite/RPC harness fixes one terminal handle and pane across Run and Dispatch ownership changes, captures mocked PTY-controller writes, executes the production check, reopens the durable database under a fresh runtime, and verifies Run Delivery acknowledgment plus worker-check consumption. The host-side authority decision is provider-independent and adds no wire field, renderer state, polling, or PTY-provider call. Daemon, paired remote-runtime, SSH, WSL, Linux, Windows, and headed restart evidence remain gaps.",
"coverageNotes": "A deterministic in-process runtime/SQLite/RPC harness covers two remint-equivalent panes across Run and Dispatch ownership changes, captures mocked PTY-controller writes, executes the production check, reopens a durable production-shape mismatch under a fresh runtime, and verifies Run Delivery acknowledgment plus worker-check consumption. A 151-row backlog proves reconciliation stays within one 50-row page and one coalesced mailbox wake per event-loop turn. The authority decision adds no wire field, renderer state, polling, or provider call. Daemon, paired remote-runtime, SSH, WSL, Linux, Windows, and headed restart evidence remain gaps.",
"motivatingLinks": ["internal incident evidence: improve-vps-setup, 2026-08-10"],
"invariant": "Every injected orchestration pointer must select the same mailbox as the exact check from that pane and expose at least the notified unread messages through that mailbox's acknowledgment semantics. Rebinding cannot let foreign direct mail wake a new owner, newer Run mail cannot be announced behind an outstanding fixed Delivery, and restart cannot replay a successfully submitted pointer. Notification records only a durable pointer cursor; it never marks mail read or globally clears it.",
"oracle": "Create Run A with unread current-delivery mail addressed directly to its coordinator handle, bind the same pane to Run B, drive a controlled working-to-idle event, and require zero PTY pointers while Run B check returns zero and Run A mail stays unread. Reopen the database and repeat. For same-Run mail, require one submitted pointer, a bound Delivery containing that row, explicit acknowledgment, and no replay after a fresh runtime. Require an active worker's direct mail to route through its Dispatch check, suppress newer Run pointers behind an outstanding Delivery, normalize mail that arrived while its Run was displaced, avoid full mailbox materialization while working, and let bound waiters consume without a duplicate PTY pointer.",
"oracle": "Create Run A with unread current-delivery mail addressed directly to its coordinator handle, bind the same pane to Run B, drive a controlled working-to-idle event, and require zero PTY pointers while Run B check returns zero and Run A mail stays unread. Reopen a database already containing that mismatch and require live-idle reconciliation to make the row actionable to Run A without pointing Run B. For same-Run mail, require one Run-pinned pointer, a bound Delivery containing that row, explicit acknowledgment, and no replay after a fresh runtime. Require reminted active-worker mail to retain Dispatch ownership, suppress newer Run pointers behind an outstanding Delivery, preserve waiter reservations, reject absent local PTYs, and reconcile a 151-row mismatch in 50-row pages with one mailbox wake per turn.",
"commands": [
"pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration-mailbox-notification-consistency.test.ts src/main/runtime/orchestration/orchestration-run-delivery-db.test.ts src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts",
"pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/terminal-send-stale-leaf-liveness.test.ts src/main/runtime/rpc/methods/orchestration.test.ts"
@@ -8826,13 +8826,18 @@
"assertions": [
"Run A direct mail cannot point after the pane binds Run B",
"a fresh runtime cannot replay the unactionable pointer",
"a fresh runtime reconciles a database that already contains the production mismatch",
"large persisted mismatches route in 50-row pages with coalesced per-mailbox wakes",
"Run pointers pin the existing mixed-version-compatible --run identity",
"a delayed pointer submit cannot press Enter after the agent becomes working",
"an explicit check releases its staged pointer reservation without redrive",
"a provider-refused Enter remains undelivered and retryable",
"a staged pointer follows its Run to a new live pane without cross-PTY duplication",
"equivalent panes serialize one mailbox while newer mail arrives during Enter delay",
"same-Run direct mail is delivered and acknowledged through the bound Run",
"a submitted actionable pointer is durably deduplicated across restart without marking mail read",
"newer Run mail is not pointed behind an outstanding fixed Delivery",
"active worker direct mail is exposed by its Dispatch check",
"active worker direct mail is exposed by its Dispatch check after handle remint",
"late direct mail is normalized when its displaced Run rebinds",
"a bound waiter wakes without a duplicate PTY pointer",
"working-agent arrival uses bounded ownership lookup without a full mailbox scan"
@@ -8852,8 +8857,8 @@
"platform": "macos",
"command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration-mailbox-notification-consistency.test.ts src/main/runtime/orchestration/orchestration-run-delivery-db.test.ts src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts",
"result": "passed",
"durationSeconds": 3.1,
"summary": "31 tests passed across mailbox identity, cross-PTY serialization, live ownership transfer, durable restart, submit races, bounded batches, outstanding Deliveries, Dispatch ownership, acknowledgment, waiters, Run Delivery, and remote pane remints. The original 10-scenario mailbox oracle failed 10/10 on origin/main plus the test commit and again with the implementation commit explicitly reverted, including pointerCount > 0 with bound check count 0."
"durationSeconds": 5.1,
"summary": "37 tests passed across mailbox identity, persisted-upgrade reconciliation, 50-row paging and coalesced wakes, cross-PTY serialization, live ownership transfer, durable restart, submit races, outstanding Deliveries, reminted Dispatch ownership, acknowledgment, waiters, Run Delivery, and remote pane remints. The original 10-scenario mailbox oracle failed 10/10 on origin/main plus the test commit and again with the implementation commit explicitly reverted, including pointerCount > 0 with bound check count 0."
},
{
"date": "2026-08-10",
@@ -8875,11 +8880,11 @@
},
"redGreenEvidence": {
"status": "complete",
"evidence": "The identical 10-scenario oracle failed 10/10 on origin/main plus the oracle commit, passed 10/10 on the candidate, failed 10/10 after explicitly reverting the implementation commit, and passed again after restoration. Live topology coverage remains a separate promotion gap."
"evidence": "The identical original 10-scenario oracle failed 10/10 on origin/main plus the oracle commit, passed 10/10 on the candidate, failed 10/10 after explicitly reverting the implementation commit, and passed again after restoration. Added restart-upgrade, reminted-Dispatch, pointer-pinning, refusal, and bounded-backlog scenarios are candidate regressions; live topology coverage remains a separate promotion gap."
},
"performanceBudget": {
"required": true,
"evidence": "Arrival ownership uses indexed identity and SELECT 1 LIMIT 1 probes. Pointer selection is capped to 50 rows inside SQLite, submit-time row revalidation is forced through the unique message-ID index, durable cursor writes update only the submitted batch in 500-ID chunks, and outstanding-Delivery suppression is a bounded existence query. No polling, new timers, subprocesses, provider calls, listeners, or renderer work were added."
"evidence": "Arrival ownership uses indexed identity and SELECT 1 LIMIT 1 probes. Pointer selection and direct-mail reconciliation are capped to 50 rows inside SQLite; a single setImmediate continuation yields between pages and coalesces each mailbox/type wake. Submit-time row revalidation uses the unique message-ID index, durable cursor writes update only the submitted batch in 500-ID chunks, and outstanding-Delivery suppression is a bounded existence query. No polling, subprocesses, provider calls, persistent timers, listeners, or renderer work were added."
},
"promotionCriteria": [
"Add isolated live restart and SSH or paired-runtime evidence.",
@@ -8888,6 +8893,7 @@
"knownGaps": [
"No headed Electron restart journey was required to prove the runtime/PTY/check boundary.",
"No daemon, paired remote-runtime, live SSH, WSL, Linux, or Windows journey has been collected.",
"Daemon and SSH PTY writes remain transport-accepted rather than owner-acknowledged; capability-negotiated owner acknowledgments are separate remote-protocol work.",
"Passive renderer-native unread UI and opt-in coordinator wake policy remain separate product work."
],
"demotionRule": "Demote or quarantine if the deterministic oracle flakes without a product or harness bug owned by orchestration."
+2 -1
View File
@@ -1064,7 +1064,7 @@ export class DaemonPtyAdapter implements IPtyProvider {
}
}
write(id: string, data: string): void {
write(id: string, data: string): boolean {
this.markSessionDirty(id)
// Why recoverable and not just active: rejecting a write asks the pane to remount,
// which only helps if this endpoint can come back. A legacy adapter has no respawn,
@@ -1086,6 +1086,7 @@ export class DaemonPtyAdapter implements IPtyProvider {
this.reconnectAfterWriteFailure()
throw new PtyWriteUnavailableError(`Daemon PTY "${id}" is awaiting recovery`)
}
return delivered
}
resize(id: string, cols: number, rows: number): void {
+2 -2
View File
@@ -89,8 +89,8 @@ export class DaemonPtyRouter implements IPtyProvider {
return await this.ownerResolver.probe(id)
}
write(id: string, data: string): void {
this.adapterFor(id).write(id, data)
write(id: string, data: string): boolean {
return this.adapterFor(id).write(id, data)
}
resize(id: string, cols: number, rows: number): void {
@@ -108,8 +108,8 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
this.sessionProviders.get(ptyId) ?? this.findProviderForExistingSession(ptyId)
)?.providesAgentSessionOwnerListings?.(ptyId) === true
write(id: string, data: string): void {
this.providerFor(id).write(id, data)
write(id: string, data: string): boolean | void {
return this.providerFor(id).write(id, data)
}
resize(id: string, cols: number, rows: number): void {
+1 -2
View File
@@ -5430,8 +5430,7 @@ export function registerPtyHandlers(
},
write: (ptyId, data) => {
try {
getProviderForPty(ptyId).write(ptyId, data)
return true
return getProviderForPty(ptyId).write(ptyId, data) !== false
} catch {
return false
}
@@ -1454,12 +1454,12 @@ describe('LocalPtyProvider', () => {
describe('write', () => {
it('writes data to the PTY process', async () => {
const { id } = await provider.spawn({ cols: 80, rows: 24 })
provider.write(id, 'hello')
expect(provider.write(id, 'hello')).toBe(true)
expect(mockProc.write).toHaveBeenCalledWith('hello')
})
it('is a no-op for unknown PTY ids', () => {
provider.write('nonexistent', 'hello')
expect(provider.write('nonexistent', 'hello')).toBe(false)
expect(mockProc.write).not.toHaveBeenCalled()
})
})
+8 -3
View File
@@ -1150,15 +1150,20 @@ export class LocalPtyProvider implements IPtyProvider {
hasPty(id: string): boolean {
return ptyProcesses.has(id)
}
write(id: string, data: string): void {
write(id: string, data: string): boolean {
// Cooked PTYs echo private DSR/OSC replies; CPR/DA remain immediate (#13137, #7329).
if (extractOnlyCookedEchoSafeQueryReplies(data)) {
const ingress = startupIngressByPty.get(id)
if (ingress?.answerLiveQueryReply(data)) {
return
return true
}
}
ptyProcesses.get(id)?.write(data)
const proc = ptyProcesses.get(id)
if (!proc) {
return false
}
proc.write(data)
return true
}
resize(id: string, cols: number, rows: number): void {
ptyProcesses.get(id)?.resize(cols, rows)
+1 -1
View File
@@ -126,7 +126,7 @@ export type IPtyProvider = {
hasPty?: (id: string) => boolean
/** Exact provider readback: false only when the provider answered that the PTY is absent. */
probePtyLiveness?: (id: string) => Promise<boolean | null>
write(id: string, data: string): void
write(id: string, data: string): boolean | void
resize(id: string, cols: number, rows: number): void
/**
* Producer-side flow control: stop/restart reading the underlying PTY so a
+2 -2
View File
@@ -34642,7 +34642,7 @@ describe('OrcaRuntimeService', () => {
runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
expect(write).toHaveBeenCalledWith(
'pty-1',
'\nYou have 1 orchestration message. Run `orca orchestration check`.\n'
'\nYou have 1 orchestration message. Run `orca orchestration check --run run_mailbox`.\n'
)
expect(write).not.toHaveBeenCalledWith(
'pty-1',
@@ -34990,7 +34990,7 @@ describe('OrcaRuntimeService', () => {
await vi.waitFor(() => {
expect(write).toHaveBeenCalledWith(
'pty-1',
'\nYou have 1 orchestration message. Run `orca orchestration check`.\n'
'\nYou have 1 orchestration message. Run `orca orchestration check --run run_codex_native_title`.\n'
)
})
db.close()
+94 -49
View File
@@ -2915,6 +2915,7 @@ export class OrcaRuntimeService {
string,
ReadonlySet<string> | null
>()
private readonly pendingForeignDirectReconciliationsByLeafKey = new Set<string>()
private syntheticTerminalHandles = new Set<string>()
private detachedPreAllocatedLeaves = new Map<string, RuntimeLeafRecord>()
private graphSyncCallbacks: (() => void)[] = []
@@ -32420,12 +32421,76 @@ export class OrcaRuntimeService {
}
private deliverPendingMessagesForLeaf(leaf: RuntimeLeafRecord): void {
this.notifyForwardedOrchestrationMailboxes(this.routeForeignDirectMessagesForLeaf(leaf))
const mailboxHandle = this.resolveActionableMailboxForLeaf(leaf)
if (mailboxHandle) {
this.deliverPendingMessages(leaf, { mailboxHandle })
}
}
private scheduleForeignDirectReconciliation(leaf: RuntimeLeafRecord): void {
const leafKey = this.getLeafKey(leaf.tabId, leaf.leafId)
if (this.pendingForeignDirectReconciliationsByLeafKey.has(leafKey)) {
return
}
this.pendingForeignDirectReconciliationsByLeafKey.add(leafKey)
setImmediate(() => {
this.pendingForeignDirectReconciliationsByLeafKey.delete(leafKey)
const currentLeaf = this.leaves.get(leafKey)
if (!currentLeaf || currentLeaf.ptyId !== leaf.ptyId) {
return
}
this.notifyForwardedOrchestrationMailboxes(
this.routeForeignDirectMessagesForLeaf(currentLeaf)
)
})
}
private routeForeignDirectMessagesForLeaf(
leaf: RuntimeLeafRecord
): { mailboxHandle: string; types: string[] }[] {
if (!this._orchestrationDb) {
return []
}
const leafKey = this.getLeafKey(leaf.tabId, leaf.leafId)
const handle = this.handleByLeafKey.get(leafKey)
if (!handle) {
return []
}
const paneKey = `${leaf.tabId}:${leaf.leafId}`
const ownerMailbox = this.resolveActionableMailboxForLeaf(leaf, undefined, {
routeDirectMail: false
})
const ownerRunId = ownerMailbox?.startsWith('run:')
? ownerMailbox.slice('run:'.length)
: ownerMailbox?.startsWith('dispatch:')
? this._orchestrationDb.getDispatchContextById?.(ownerMailbox.slice('dispatch:'.length))
?.run_id
: undefined
if (!ownerRunId) {
return []
}
const routed = this._orchestrationDb.routeForeignDirectMessagesToOwnedMailboxes?.(
handle,
ownerRunId,
paneKey
)
if (routed?.hasMore) {
this.scheduleForeignDirectReconciliation(leaf)
}
return routed?.mailboxes ?? []
}
private notifyForwardedOrchestrationMailboxes(
forwarded: readonly { mailboxHandle: string; types: string[] }[]
): void {
for (const routed of forwarded) {
for (const routedType of routed.types) {
this.notifyMessageArrived(routed.mailboxHandle, routedType)
}
}
}
private resolveActionableMailboxForLeaf(
leaf: RuntimeLeafRecord,
requestedMailbox?: string,
@@ -32515,31 +32580,18 @@ export class OrcaRuntimeService {
private resolveArrivedMessageMailboxes(handle: string): {
mailboxHandle: string | null
forwarded: { runId: string; types: string[] }[]
forwarded: { mailboxHandle: string; types: string[] }[]
} {
if (!this.handles.has(handle) || !this._orchestrationDb) {
return { mailboxHandle: handle, forwarded: [] }
}
try {
const { leaf } = this.getLiveLeafForHandle(handle)
const ownerMailbox = this.resolveActionableMailboxForLeaf(leaf, undefined, {
routeDirectMail: false
})
const ownerRunId = ownerMailbox?.startsWith('run:')
? ownerMailbox.slice('run:'.length)
: ownerMailbox?.startsWith('dispatch:')
? this._orchestrationDb.getDispatchContextById?.(ownerMailbox.slice('dispatch:'.length))
?.run_id
: undefined
const forwarded = ownerRunId
? (this._orchestrationDb.routeForeignDirectMessagesToRunMailboxes?.(handle, ownerRunId) ??
[])
: []
return {
mailboxHandle: this.resolveActionableMailboxForLeaf(leaf, handle, {
requireRequestedMail: true
}),
forwarded
forwarded: this.routeForeignDirectMessagesForLeaf(leaf)
}
} catch {
return { mailboxHandle: null, forwarded: [] }
@@ -32552,11 +32604,7 @@ export class OrcaRuntimeService {
this.mailPointerRepointScheduler.schedule(handle)
}
const { mailboxHandle, forwarded } = this.resolveArrivedMessageMailboxes(handle)
for (const routed of forwarded) {
for (const routedType of routed.types) {
this.notifyMessageArrived(`run:${routed.runId}`, routedType)
}
}
this.notifyForwardedOrchestrationMailboxes(forwarded)
if (!mailboxHandle) {
return
}
@@ -33243,7 +33291,6 @@ export class OrcaRuntimeService {
this.pointedMessageMailboxHandlesByPtyId.delete(prior.ptyId)
}
}
this.lastPointedMessageSequenceByHandle.set(mailboxHandle, sequence)
this.pointedMessageWatermarkOwnerByHandle.set(mailboxHandle, {
ptyId,
sequence,
@@ -33265,9 +33312,6 @@ export class OrcaRuntimeService {
return false
}
this.pointedMessageWatermarkOwnerByHandle.delete(mailboxHandle)
if (this.lastPointedMessageSequenceByHandle.get(mailboxHandle) === sequence) {
this.lastPointedMessageSequenceByHandle.delete(mailboxHandle)
}
const mailboxes = this.pointedMessageMailboxHandlesByPtyId.get(ptyId)
mailboxes?.delete(mailboxHandle)
if (mailboxes?.size === 0) {
@@ -33294,13 +33338,13 @@ export class OrcaRuntimeService {
reservedTypes: ReadonlySet<string> | undefined
): void {
const prior = this.parkedMessageRedeliveryTypesByMailboxHandle.get(mailboxHandle)
if (prior === null || reservedTypes === undefined) {
if (prior === null || (prior === undefined && reservedTypes === undefined)) {
this.parkedMessageRedeliveryTypesByMailboxHandle.set(mailboxHandle, null)
return
}
this.parkedMessageRedeliveryTypesByMailboxHandle.set(
mailboxHandle,
new Set([...(prior ?? []), ...reservedTypes])
new Set([...(prior ?? []), ...(reservedTypes ?? [])])
)
}
@@ -33354,30 +33398,30 @@ export class OrcaRuntimeService {
this.pointedMessageMailboxHandlesByPtyId.delete(ptyId)
}
private canSubmitMessagePointer(
private getMessagePointerSubmitDecision(
leaf: RuntimeLeafRecord,
mailboxHandle: string,
messages: readonly { id: string; type: string }[]
): boolean {
): 'submit' | 'release' {
if (this.resolveActionableMailboxForLeaf(leaf) !== mailboxHandle) {
return false
return 'release'
}
if (
mailboxHandle.startsWith('run:') &&
this._orchestrationDb?.hasOutstandingRunDelivery?.(mailboxHandle.slice('run:'.length))
) {
return false
return 'release'
}
const waiters = this.messageWaitersByHandle.get(mailboxHandle)
if (messages.some((message) => messageTypeHasLiveWaiter(waiters, message.type))) {
return false
return 'release'
}
return (
this._orchestrationDb?.areUndeliveredUnreadMessages?.(
mailboxHandle,
messages.map((message) => message.id)
) ?? true
)
return (this._orchestrationDb?.areUndeliveredUnreadMessages?.(
mailboxHandle,
messages.map((message) => message.id)
) ?? true)
? 'submit'
: 'release'
}
// Why: push-on-idle delivery is event-driven (no polling) because the runtime owns both the message store and terminal status detection.
@@ -33485,7 +33529,8 @@ export class OrcaRuntimeService {
if (newestSequence === undefined) {
return
}
const pointedSequence = this.lastPointedMessageSequenceByHandle.get(mailboxHandle) ?? -1
const pointedSequence =
this.pointedMessageWatermarkOwnerByHandle.get(mailboxHandle)?.sequence ?? -1
if (newestSequence <= pointedSequence) {
const owner = this.pointedMessageWatermarkOwnerByHandle.get(mailboxHandle)
const leafKey = this.getLeafKey(leaf.tabId, leaf.leafId)
@@ -33556,12 +33601,9 @@ export class OrcaRuntimeService {
const deliveryPtyId = leaf.ptyId
const flight: { enterTimer: ReturnType<typeof setTimeout> | null } = { enterTimer: null }
this.messageDeliveryFlightsByPtyId.set(deliveryPtyId, flight)
// Why: every sync outcome — failed write, Cursor branch, or a throw —
// must end the flight here, or a leaked flag parks this pty's deliveries
// forever. Only an armed Enter hands settling to its own callback.
let settlesInEnterCallback = false
try {
const payload = formatMessagePointer(unread.length)
const payload = formatMessagePointer(unread.length, mailboxHandle)
const wrote = this.ptyController?.write(deliveryPtyId, payload) ?? false
if (!wrote) {
return
@@ -33585,7 +33627,6 @@ export class OrcaRuntimeService {
const tabTitle = this.tabs.get(leaf.tabId)?.title
if (isCursorAgentOrchestrationTarget(leaf, tabTitle)) {
// Why: Cursor Agent treats injected PTY text as editable prompt input, so submitting must stay under user control.
let delivered = false
try {
this._orchestrationDb.markAsDelivered(unread.map((message) => message.id))
@@ -33605,11 +33646,10 @@ export class OrcaRuntimeService {
flight.enterTimer = setTimeout(() => {
let clearAndRedrive = false
let delivered = false
let releaseWithoutRedrive = false
let finalizeReservation = true
void this.isLeafPtyProvenAbsent(deliveryPtyId)
.then((absent) => {
// Why current state, not the closure: ownership, graph records, or the
// process can change during the Enter delay and liveness probe.
if (absent) {
clearAndRedrive = true
return
@@ -33629,11 +33669,16 @@ export class OrcaRuntimeService {
}
if (
currentLeaf.lastAgentStatus !== 'idle' ||
!currentLeaf.lastAgentStatusObservedLive ||
!this.canSubmitMessagePointer(currentLeaf, mailboxHandle, unread)
!currentLeaf.lastAgentStatusObservedLive
) {
return
}
if (
this.getMessagePointerSubmitDecision(currentLeaf, mailboxHandle, unread) === 'release'
) {
releaseWithoutRedrive = true
return
}
const submitted = this.ptyController?.write(deliveryPtyId, '\r') ?? false
if (submitted) {
this._orchestrationDb?.markAsDelivered(unread.map((message) => message.id))
@@ -33647,7 +33692,7 @@ export class OrcaRuntimeService {
let released = false
if (finalizeReservation) {
released =
delivered || clearAndRedrive
delivered || clearAndRedrive || releaseWithoutRedrive
? this.clearPointedMessageWatermark(mailboxHandle, newestSequence, deliveryPtyId)
: this.deactivatePointedMessageWatermark(
mailboxHandle,
@@ -33656,7 +33701,7 @@ export class OrcaRuntimeService {
)
}
this.settlePendingMessageDelivery(deliveryPtyId, flight)
if (released) {
if (released && !releaseWithoutRedrive) {
this.redrivePendingMessageMailbox(mailboxHandle, clearAndRedrive)
}
})
@@ -256,6 +256,88 @@ describe('orchestration notification mailbox consistency', () => {
db.close()
})
it('reconciles the persisted production mismatch after runtime restart', async () => {
vi.useFakeTimers()
const directory = mkdtempSync(join(tmpdir(), 'orca-mailbox-upgrade-restart-'))
temporaryDirectories.push(directory)
const dbPath = join(directory, 'orchestration.db')
const oldDb = new OrchestrationDb(dbPath)
const runA = createBoundRun(oldDb, 'Persisted Run A')
const message = insertDirectRunMessage(oldDb, runA.id, 'Persisted Run A completion')
createBoundRun(oldDb, 'Persisted Run B')
sqliteFor(oldDb)
.prepare('UPDATE messages SET to_handle = ? WHERE id = ?')
.run(TERMINAL_HANDLE, message.id)
oldDb.close()
const restartedDb = new OrchestrationDb(dbPath)
const restarted = createRuntime(restartedDb)
await driveToLiveIdle(restarted.runtime)
expect(pointerCount(restarted.write)).toBe(0)
expect(restartedDb.getMessageById(message.id)).toMatchObject({
to_handle: `run:${runA.id}`,
read: 0,
delivered_at: null
})
registerSecondPane(restarted.runtime)
restartedDb.bindRun({
runId: runA.id,
coordinatorHandle: SECOND_TERMINAL_HANDLE,
coordinatorPaneKey: SECOND_PANE_KEY
})
restarted.runtime.onPtyData(SECOND_PTY_ID, '\x1b]0;Codex working\x07', 1)
restarted.runtime.onPtyData(SECOND_PTY_ID, '\x1b]0;Codex done\x07', 2)
const checked = await checkBoundMailbox(restarted.runtime, {
terminal: SECOND_TERMINAL_HANDLE,
paneKey: SECOND_PANE_KEY,
launchToken: SECOND_LAUNCH_TOKEN
})
expect(checked).toMatchObject({ runId: runA.id, count: 1 })
expect(checked.messages).toEqual([expect.objectContaining({ id: message.id })])
restartedDb.close()
})
it('pages a large persisted mismatch and coalesces each mailbox wake', async () => {
vi.useFakeTimers()
const db = createDatabase('orca-mailbox-paged-reconciliation-')
const harness = createRuntime(db)
const runA = createBoundRun(db, 'Backlog Run A')
const messages = Array.from({ length: 151 }, (_, index) =>
insertDirectRunMessage(db, runA.id, `Backlog message ${index}`)
)
createBoundRun(db, 'Backlog Run B')
const placeholders = messages.map(() => '?').join(',')
sqliteFor(db)
.prepare(`UPDATE messages SET to_handle = ? WHERE id IN (${placeholders})`)
.run(TERMINAL_HANDLE, ...messages.map((message) => message.id))
const notify = vi.spyOn(harness.runtime, 'notifyMessageArrived')
const directRemaining = (): number =>
(
sqliteFor(db)
.prepare('SELECT COUNT(*) AS count FROM messages WHERE to_handle = ?')
.get(TERMINAL_HANDLE) as { count: number }
).count
await driveToLiveIdle(harness.runtime)
expect(directRemaining()).toBe(101)
expect(notify).toHaveBeenCalledTimes(1)
await vi.advanceTimersToNextTimerAsync()
expect(directRemaining()).toBe(51)
expect(notify).toHaveBeenCalledTimes(2)
await vi.advanceTimersToNextTimerAsync()
expect(directRemaining()).toBe(1)
expect(notify).toHaveBeenCalledTimes(3)
await vi.advanceTimersToNextTimerAsync()
expect(directRemaining()).toBe(0)
expect(notify).toHaveBeenCalledTimes(4)
expect(pointerCount(harness.write)).toBe(0)
db.close()
})
it('does not submit or replay a pointer after ownership changes during Enter delay', async () => {
vi.useFakeTimers()
const directory = mkdtempSync(join(tmpdir(), 'orca-mailbox-restart-'))
@@ -289,6 +371,52 @@ describe('orchestration notification mailbox consistency', () => {
restartedDb.close()
})
it('fences the pointed mailbox instead of checking a rebound empty Run', async () => {
vi.useFakeTimers()
const db = createDatabase('orca-mailbox-post-submit-rebind-')
const harness = createRuntime(db)
const runA = createBoundRun(db, 'Run A')
insertDirectRunMessage(db, runA.id, 'Run A completion')
await driveToLiveIdle(harness.runtime)
await vi.advanceTimersByTimeAsync(500)
expect(harness.write).toHaveBeenCalledWith(
PTY_ID,
expect.stringContaining(`orchestration check --run ${runA.id}`)
)
db.bindRun({
runId: runA.id,
coordinatorHandle: 'term_new_coordinator',
coordinatorPaneKey:
'55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666'
})
const runB = createBoundRun(db, 'Run B')
const response = await new RpcDispatcher({
runtime: harness.runtime,
methods: ORCHESTRATION_METHODS
}).dispatch({
id: 'req-pointed-mailbox-fenced',
authToken: 'test-auth-token',
method: 'orchestration.check',
orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION,
orchestrationCompatibilityEvidence: {
terminalHandle: TERMINAL_HANDLE,
paneKey: PANE_KEY,
launchToken: LAUNCH_TOKEN
},
params: { terminal: TERMINAL_HANDLE, run: runA.id }
})
expect(response).toMatchObject({
ok: false,
error: { code: 'consumer_fenced' }
})
const generic = await checkBoundMailbox(harness.runtime)
expect(generic).toMatchObject({ runId: runB.id, count: 0 })
db.close()
})
it('releases a staged pointer when its Run moves to another live pane', async () => {
vi.useFakeTimers()
const db = createDatabase('orca-mailbox-live-rebind-')
@@ -393,6 +521,35 @@ describe('orchestration notification mailbox consistency', () => {
db.close()
})
it('releases staged pointer state when an explicit check owns the batch', async () => {
vi.useFakeTimers()
const db = createDatabase('orca-mailbox-explicit-check-')
const harness = createRuntime(db)
const run = createBoundRun(db, 'Explicit-check Run')
insertDirectRunMessage(db, run.id, 'Checked before Enter')
await driveToLiveIdle(harness.runtime)
expect(pointerCount(harness.write)).toBe(1)
const checked = await checkBoundMailbox(harness.runtime)
expect(checked).toMatchObject({ runId: run.id, count: 1 })
await vi.advanceTimersByTimeAsync(500)
const internals = harness.runtime as unknown as {
pointedMessageWatermarkOwnerByHandle: Map<string, unknown>
pointedMessageMailboxHandlesByPtyId: Map<string, Set<string>>
parkedMessageRedeliveryTypesByMailboxHandle: Map<string, unknown>
}
expect(internals.pointedMessageWatermarkOwnerByHandle.size).toBe(0)
expect(internals.pointedMessageMailboxHandlesByPtyId.size).toBe(0)
expect(internals.parkedMessageRedeliveryTypesByMailboxHandle.size).toBe(0)
const redrive = vi.spyOn(harness.runtime, 'deliverPendingMessagesForHandle')
harness.runtime.onPtyExit(PTY_ID, 0)
await Promise.resolve()
expect(redrive).not.toHaveBeenCalled()
db.close()
})
it('routes same-Run direct mail through the bound check and acknowledgment path', async () => {
vi.useFakeTimers()
const db = createDatabase('orca-mailbox-current-run-')
@@ -485,6 +642,34 @@ describe('orchestration notification mailbox consistency', () => {
restartedDb.close()
})
it('keeps mail retryable when the provider rejects Enter', async () => {
vi.useFakeTimers()
const db = createDatabase('orca-mailbox-provider-refusal-')
const first = createRuntime(db)
const recordWrite = first.write as unknown as (id: string, payload: string) => unknown
const write = vi.fn((ptyId: string, data: string) => {
recordWrite(ptyId, data)
return data !== '\r'
})
first.runtime.setPtyController({
write,
kill: vi.fn(),
getForegroundProcess: async () => null
})
const run = createBoundRun(db, 'Provider-refusal Run')
const message = insertDirectRunMessage(db, run.id, 'Retry after refusal')
await driveToLiveIdle(first.runtime)
await vi.advanceTimersByTimeAsync(500)
expect(pointerCount(first.write)).toBe(1)
expect(db.getMessageById(message.id)?.delivered_at).toBeNull()
const restarted = createRuntime(db)
await driveToLiveIdle(restarted.runtime)
expect(pointerCount(restarted.write)).toBe(1)
db.close()
})
it('does not point newer Run mail behind an outstanding Delivery', async () => {
vi.useFakeTimers()
const db = createDatabase('orca-mailbox-outstanding-')
@@ -551,11 +736,9 @@ describe('orchestration notification mailbox consistency', () => {
delivered_at: expect.any(String)
})
const internals = harness.runtime as unknown as {
lastPointedMessageSequenceByHandle: Map<string, number>
pointedMessageWatermarkOwnerByHandle: Map<string, unknown>
pointedMessageMailboxHandlesByPtyId: Map<string, Set<string>>
}
expect(internals.lastPointedMessageSequenceByHandle.has(`dispatch:${dispatch.id}`)).toBe(false)
expect(internals.pointedMessageWatermarkOwnerByHandle.has(`dispatch:${dispatch.id}`)).toBe(
false
)
@@ -599,6 +782,51 @@ describe('orchestration notification mailbox consistency', () => {
db.close()
})
it('preserves an active Dispatch as owner of displaced direct mail', async () => {
vi.useFakeTimers()
const db = createDatabase('orca-mailbox-displaced-dispatch-')
const harness = createRuntime(db)
const workerRun = db.createRun({
objective: 'Worker Run',
coordinatorHandle: 'term_worker_coordinator',
coordinatorPaneKey:
'55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666'
})
const task = db.createTask({ spec: 'Worker task', runId: workerRun.id })
const dispatch = db.createDispatchContext(
task.id,
'term_mailbox_before_remint',
`99999999-9999-4999-8999-999999999999:${LEAF_ID}`
)
createBoundRun(db, 'Current coordinator Run')
const waiting = harness.runtime.waitForMessage(`dispatch:${dispatch.id}`, {
timeoutMs: 5_000
})
const message = db.insertMessage({
from: 'term_worker_coordinator',
to: TERMINAL_HANDLE,
subject: 'Late worker instruction',
runId: workerRun.id
})
harness.runtime.notifyMessageArrived(TERMINAL_HANDLE, message.type)
await expect(waiting).resolves.toBe('notified')
expect(pointerCount(harness.write)).toBe(0)
expect(db.getMessageById(message.id)?.to_handle).toBe(`dispatch:${dispatch.id}`)
const currentRun = db.getCurrentRunForPane(PANE_KEY)
db.bindRun({
runId: currentRun!.id,
coordinatorHandle: 'term_rebound_coordinator',
coordinatorPaneKey:
'77777777-7777-4777-8777-777777777777:88888888-8888-4888-8888-888888888888'
})
const checked = await checkBoundMailbox(harness.runtime)
expect(checked).toMatchObject({ runId: workerRun.id, dispatchId: dispatch.id, count: 1 })
expect(checked.messages).toEqual([expect.objectContaining({ id: message.id })])
db.close()
})
it('uses bounded ownership lookup for direct arrivals while the agent is working', async () => {
vi.useFakeTimers()
const db = createDatabase('orca-mailbox-bounded-')
+129 -50
View File
@@ -90,6 +90,14 @@ function parseWorkerTerminalPriorOwnerIds(value: string): string[] | null {
const MESSAGE_ID_UPDATE_BATCH_SIZE = 500
export const ORCHESTRATION_DELIVERY_BATCH_LIMIT = 50
export type DirectMailboxRoutingPage = {
routedCount: number
hasMore: boolean
}
export type ForeignDirectMailboxRoutingPage = DirectMailboxRoutingPage & {
mailboxes: { mailboxHandle: string; types: MessageType[] }[]
}
// Why: indexable pre-filter for isEquivalentPaneKey — equal strings and equal leaves both share the
// text after the first ':', so this narrows candidates without deciding equivalence itself.
const RUN_PANE_KEY_MATCH_SUFFIX_SQL =
@@ -2710,7 +2718,10 @@ export class OrchestrationDb {
limit?: number
wakeTypes?: MessageType[]
}): { delivery: DeliveryRow; messages: MessageRow[]; replayed: boolean } | undefined {
const limit = Math.min(Math.max(params.limit ?? 50, 1), 50)
const limit = Math.min(
Math.max(params.limit ?? ORCHESTRATION_DELIVERY_BATCH_LIMIT, 1),
ORCHESTRATION_DELIVERY_BATCH_LIMIT
)
this.db.exec('BEGIN IMMEDIATE')
try {
this.requireCurrentConsumer(params.runId, params.consumerGeneration)
@@ -3586,65 +3597,137 @@ export class OrchestrationDb {
}
// Why: change mailbox ownership without changing unread or acknowledgment state.
routeUnreadDirectMessagesToRunMailbox(runId: string, directHandle: string): number {
const result = this.db
private routeDirectMessagePage(
mailboxHandle: string,
runId: string,
directHandle: string
): DirectMailboxRoutingPage {
const rows = this.db
.prepare(
`UPDATE messages
SET to_handle = ?
`SELECT id FROM messages
WHERE run_id = ? AND to_handle = ? AND read = 0
AND delivery_contract = 'current_delivery'`
AND delivery_contract = 'current_delivery'
ORDER BY sequence LIMIT ?`
)
.run(`run:${runId}`, runId, directHandle)
return Number(result.changes)
.all(runId, directHandle, ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1) as { id: string }[]
const page = rows.slice(0, ORCHESTRATION_DELIVERY_BATCH_LIMIT)
if (page.length === 0) {
return { routedCount: 0, hasMore: false }
}
const placeholders = page.map(() => '?').join(',')
const result = this.db
.prepare(`UPDATE messages SET to_handle = ? WHERE id IN (${placeholders})`)
.run(mailboxHandle, ...page.map((row) => row.id))
return {
routedCount: Number(result.changes),
hasMore: rows.length > ORCHESTRATION_DELIVERY_BATCH_LIMIT
}
}
routeUnreadDirectMessagesToRunMailbox(
runId: string,
directHandle: string
): DirectMailboxRoutingPage {
return this.routeDirectMessagePage(`run:${runId}`, runId, directHandle)
}
routeUnreadDirectMessagesToDispatchMailbox(
dispatchId: string,
runId: string,
directHandle: string
): number {
const result = this.db
.prepare(
`UPDATE messages
SET to_handle = ?
WHERE run_id = ? AND to_handle = ? AND read = 0
AND delivery_contract = 'current_delivery'`
)
.run(`dispatch:${dispatchId}`, runId, directHandle)
return Number(result.changes)
): DirectMailboxRoutingPage {
return this.routeDirectMessagePage(`dispatch:${dispatchId}`, runId, directHandle)
}
routeForeignDirectMessagesToRunMailboxes(
private findActiveDispatchForDirectMessageOwner(
runId: string,
directHandle: string,
currentRunId: string
): { runId: string; types: MessageType[] }[] {
const routedRows = this.db
paneKey?: string
): DispatchContextRow | undefined {
const exact = this.db
.prepare(
`SELECT DISTINCT messages.run_id, messages.type FROM messages
WHERE to_handle = ? AND run_id <> ? AND read = 0 AND delivered_at IS NULL
AND delivery_contract = 'current_delivery'
AND EXISTS (SELECT 1 FROM runs WHERE runs.id = messages.run_id AND runs.legacy = 0)`
`SELECT * FROM dispatch_contexts
WHERE run_id = ? AND assignee_handle = ? AND status IN ('pending', 'dispatched')
ORDER BY rowid DESC LIMIT 1`
)
.all(directHandle, currentRunId) as { run_id: string; type: MessageType }[]
if (routedRows.length === 0) {
return []
.get(runId, directHandle) as DispatchContextRow | undefined
if (exact || !paneKey || !parsePaneKey(paneKey)) {
return exact
}
this.db
return this.db
.prepare(
`UPDATE messages
SET to_handle = 'run:' || run_id
`SELECT * FROM dispatch_contexts
WHERE run_id = ? AND assignee_pane_key IS NOT NULL
AND status IN ('pending', 'dispatched') AND instr(assignee_pane_key, ':') > 1
AND ${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL} = ?
ORDER BY rowid DESC LIMIT 1`
)
.get(runId, paneKeyMatchSuffix(paneKey)) as DispatchContextRow | undefined
}
routeForeignDirectMessagesToOwnedMailboxes(
directHandle: string,
currentRunId: string,
paneKey?: string
): ForeignDirectMailboxRoutingPage {
const rows = this.db
.prepare(
`SELECT id, run_id, type FROM messages
WHERE to_handle = ? AND run_id <> ? AND read = 0 AND delivered_at IS NULL
AND delivery_contract = 'current_delivery'
AND EXISTS (SELECT 1 FROM runs WHERE runs.id = messages.run_id AND runs.legacy = 0)`
AND EXISTS (SELECT 1 FROM runs WHERE runs.id = messages.run_id AND runs.legacy = 0)
ORDER BY sequence LIMIT ?`
)
.run(directHandle, currentRunId)
const byRun = new Map<string, Set<MessageType>>()
for (const row of routedRows) {
const types = byRun.get(row.run_id) ?? new Set<MessageType>()
.all(directHandle, currentRunId, ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1) as {
id: string
run_id: string
type: MessageType
}[]
const page = rows.slice(0, ORCHESTRATION_DELIVERY_BATCH_LIMIT)
if (page.length === 0) {
return { routedCount: 0, hasMore: false, mailboxes: [] }
}
const runIds = [...new Set(page.map((row) => row.run_id))]
const dispatchByRun = new Map<string, DispatchContextRow>()
for (const runId of runIds) {
const dispatch = this.findActiveDispatchForDirectMessageOwner(runId, directHandle, paneKey)
if (dispatch) {
dispatchByRun.set(runId, dispatch)
}
}
const idsByMailbox = new Map<string, string[]>()
const byMailbox = new Map<string, Set<MessageType>>()
for (const row of page) {
const dispatch = dispatchByRun.get(row.run_id)
const mailboxHandle = dispatch ? `dispatch:${dispatch.id}` : `run:${row.run_id}`
const ids = idsByMailbox.get(mailboxHandle) ?? []
ids.push(row.id)
idsByMailbox.set(mailboxHandle, ids)
const types = byMailbox.get(mailboxHandle) ?? new Set<MessageType>()
types.add(row.type)
byRun.set(row.run_id, types)
byMailbox.set(mailboxHandle, types)
}
this.db.exec('BEGIN IMMEDIATE')
try {
for (const [mailboxHandle, ids] of idsByMailbox) {
const placeholders = ids.map(() => '?').join(',')
this.db
.prepare(`UPDATE messages SET to_handle = ? WHERE id IN (${placeholders})`)
.run(mailboxHandle, ...ids)
}
this.db.exec('COMMIT')
} catch (error) {
this.db.exec('ROLLBACK')
throw error
}
return {
routedCount: page.length,
hasMore: rows.length > ORCHESTRATION_DELIVERY_BATCH_LIMIT,
mailboxes: [...byMailbox].map(([mailboxHandle, types]) => ({
mailboxHandle,
types: [...types]
}))
}
return [...byRun].map(([runId, types]) => ({ runId, types: [...types] }))
}
areUndeliveredUnreadMessages(toHandle: string, ids: string[]): boolean {
@@ -5114,18 +5197,14 @@ export class OrchestrationDb {
}
findActiveRemoteAttachmentForPane(paneKey: string): RemoteDispatchAttachmentRow | undefined {
const exact = this.db
.prepare(
`SELECT * FROM remote_dispatch_attachments
WHERE state IN ('starting', 'ready') AND pane_key = ?
ORDER BY rowid DESC LIMIT 1`
)
.get(paneKey) as RemoteDispatchAttachmentRow | undefined
if (exact) {
return exact
}
if (!parsePaneKey(paneKey)) {
return undefined
return this.db
.prepare(
`SELECT * FROM remote_dispatch_attachments
WHERE state IN ('starting', 'ready') AND pane_key = ?
ORDER BY rowid DESC LIMIT 1`
)
.get(paneKey) as RemoteDispatchAttachmentRow | undefined
}
return this.db
.prepare(
@@ -186,8 +186,8 @@ describe('formatMessagesForInjection', () => {
describe('formatMessagePointer', () => {
it('formats a singular pointer without message content', () => {
expect(formatMessagePointer(1)).toBe(
'\nYou have 1 orchestration message. Run `orca orchestration check`.\n'
expect(formatMessagePointer(1, 'run:run_1')).toBe(
'\nYou have 1 orchestration message. Run `orca orchestration check --run run_1`.\n'
)
})
+5 -2
View File
@@ -108,7 +108,10 @@ export function formatMessagesForInjection(messages: MessageRow[]): string {
return `\n--- Orchestration Messages (${messages.length}) ---\n${banners}\n---\n`
}
export function formatMessagePointer(count: number): string {
export function formatMessagePointer(count: number, mailboxHandle?: string): string {
const noun = count === 1 ? 'message' : 'messages'
return `\nYou have ${count} orchestration ${noun}. Run \`orca orchestration check\`.\n`
const runFlag = mailboxHandle?.startsWith('run:')
? ` --run ${mailboxHandle.slice('run:'.length)}`
: ''
return `\nYou have ${count} orchestration ${noun}. Run \`orca orchestration check${runFlag}\`.\n`
}
@@ -328,6 +328,9 @@ describe('OrchestrationDb worker Dispatch state', () => {
expect(d.findActiveRemoteAttachmentForPane(`tab_reminted:${leafId}`)?.dispatch_id).toBe(
'ctx_valid_new'
)
expect(d.findActiveRemoteAttachmentForPane(`tab_old:${leafId}`)?.dispatch_id).toBe(
'ctx_valid_new'
)
})
it('returns already-settled when completion wins before stop', () => {
@@ -741,6 +741,14 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
const generation = run.consumer_generation
const address = `run:${run.id}`
runtime.ensureOrchestrationFederationRelay(run.id)
const routedDirect = db.routeUnreadDirectMessagesToRunMailbox(run.id, handle)
if (
routedDirect.routedCount === 0 &&
run.coordinator_handle &&
run.coordinator_handle !== handle
) {
db.routeUnreadDirectMessagesToRunMailbox(run.id, run.coordinator_handle)
}
const acknowledged = params.ack
? db.acknowledgeRunDelivery({
@@ -909,6 +917,24 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
: undefined
if (workerMailbox) {
const address = `dispatch:${workerMailbox.dispatchId}`
if (activeDispatch) {
const routedDirect = db.routeUnreadDirectMessagesToDispatchMailbox(
activeDispatch.id,
activeDispatch.run_id,
handle
)
if (
routedDirect.routedCount === 0 &&
activeDispatch.assignee_handle &&
activeDispatch.assignee_handle !== handle
) {
db.routeUnreadDirectMessagesToDispatchMailbox(
activeDispatch.id,
activeDispatch.run_id,
activeDispatch.assignee_handle
)
}
}
const showAll = params.all === true || (params.unread === false && params.peek !== true)
const messages = showAll
? db.getAllMessagesForHandle(address, 100, typeFilter)
@@ -607,20 +607,21 @@ describe('push-on-idle orchestration delivery absence gate', () => {
const internals = runtime as unknown as {
messageDeliveryFlightsByPtyId: Map<string, unknown>
parkedMessageRedeliveriesByPtyId: Map<string, unknown>
lastPointedMessageSequenceByHandle: Map<string, unknown>
pointedMessageWatermarkOwnerByHandle: Map<string, unknown>
}
stub.insert('first')
runtime.deliverPendingMessagesForHandle(handle)
stub.insert('second')
runtime.deliverPendingMessagesForHandle(handle)
await Promise.resolve()
expect(internals.messageDeliveryFlightsByPtyId.size).toBe(1)
expect(internals.parkedMessageRedeliveriesByPtyId.size).toBe(1)
expect(internals.lastPointedMessageSequenceByHandle.size).toBe(1)
expect(internals.pointedMessageWatermarkOwnerByHandle.size).toBe(1)
runtime.onPtyExit(STALE_PTY_ID, 0)
expect(internals.messageDeliveryFlightsByPtyId.size).toBe(0)
expect(internals.parkedMessageRedeliveriesByPtyId.size).toBe(0)
expect(internals.lastPointedMessageSequenceByHandle.size).toBe(0)
expect(internals.pointedMessageWatermarkOwnerByHandle.size).toBe(0)
await vi.advanceTimersByTimeAsync(500)
expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(0)