From 5deafe9883ee205c073b4b1ed2444d848d554fa7 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Tue, 29 Sep 2026 01:46:58 -0700 Subject: [PATCH] fix(orchestration): point a structured chat's mail by the same rule as a terminal's The chat lane pointed newer mail past a batch its reader had checked and not acknowledged, while the terminal lane skips a mailbox until that batch is acked. A chat now waits for the ack the same way a terminal does, and the lookup that let the chat lane filter the held batch out is removed. --- .../db/messages/role-mailbox-delivery.ts | 27 ------------ ...tructured-mailbox-pointer-delivery.test.ts | 42 ++++--------------- .../structured-mailbox-pointer-delivery.ts | 14 ++++--- .../structured-chat-coordinator-mail.test.ts | 37 ---------------- 4 files changed, 16 insertions(+), 104 deletions(-) diff --git a/src/main/runtime/orchestration/db/messages/role-mailbox-delivery.ts b/src/main/runtime/orchestration/db/messages/role-mailbox-delivery.ts index 7bb8f6656a3..3b5f5dad94f 100644 --- a/src/main/runtime/orchestration/db/messages/role-mailbox-delivery.ts +++ b/src/main/runtime/orchestration/db/messages/role-mailbox-delivery.ts @@ -184,31 +184,6 @@ export function hasOutstandingMailboxDelivery( ) } -/** The batch a consumer has read and not yet acknowledged on this mailbox, if any. */ -export function getOutstandingMailboxDelivery( - this: OrchestrationDb, - mailboxHandle: string -): { id: string; messageIds: ReadonlySet } | undefined { - const row: unknown = this.db - .prepare('SELECT id, message_ids FROM outstanding_deliveries WHERE mailbox_handle = ? LIMIT 1') - .get(mailboxHandle) - if ( - !row || - typeof row !== 'object' || - !('id' in row) || - typeof row.id !== 'string' || - !('message_ids' in row) || - typeof row.message_ids !== 'string' - ) { - return undefined - } - const ids: unknown = JSON.parse(row.message_ids) - return { - id: row.id, - messageIds: new Set(Array.isArray(ids) ? ids.filter((id) => typeof id === 'string') : []) - } -} - export function fenceUnacknowledgedMailboxDeliveries( this: OrchestrationDb, mailboxHandle: string @@ -226,7 +201,6 @@ export type RoleMailboxDeliveryMethods = { getOrCreateMailboxDelivery: typeof getOrCreateMailboxDelivery acknowledgeMailboxDelivery: typeof acknowledgeMailboxDelivery hasOutstandingMailboxDelivery: typeof hasOutstandingMailboxDelivery - getOutstandingMailboxDelivery: typeof getOutstandingMailboxDelivery fenceUnacknowledgedMailboxDeliveries: typeof fenceUnacknowledgedMailboxDeliveries } @@ -237,7 +211,6 @@ export function attachRoleMailboxDelivery(ctor: { prototype: object }): void { getOrCreateMailboxDelivery, acknowledgeMailboxDelivery, hasOutstandingMailboxDelivery, - getOutstandingMailboxDelivery, fenceUnacknowledgedMailboxDeliveries }) } diff --git a/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts b/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts index 25b0f4e5947..e0b91843b66 100644 --- a/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts +++ b/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts @@ -4,7 +4,6 @@ import { OrchestrationStructuredMailboxPointerDelivery, type StructuredMailboxPointerHost } from './structured-mailbox-pointer-delivery' -import { formatMessagePointer } from './formatter' import type { StructuredPointerOperationRow } from './db/messages/structured-pointer-operation-store' import { structuredPointerBatchFingerprint, @@ -83,8 +82,6 @@ function harness(options: { /** The coordinator of this worker's Run is mid-batch: it checked and has not acked yet. */ outstandingRunDelivery?: boolean outstandingOwnDelivery?: boolean - /** Undelivered unread rows on the mailbox, oldest first. */ - unreadIds?: string[] /** The mailbox this worker owns; its own handle for direct peer mail outside a dispatch. */ mailbox?: string dispatchId?: string | null @@ -103,18 +100,10 @@ function harness(options: { const stored = new Map() const db = { getDispatchContextById: () => ({ run_id: 'run_1' }), - // The reader holds `m1` unacknowledged, on whichever mailbox the option names. - getOutstandingMailboxDelivery: (handle: string) => + hasOutstandingMailboxDelivery: (handle: string) => ((options.outstandingRunDelivery ?? false) && handle.startsWith('run:')) || - ((options.outstandingOwnDelivery ?? false) && !handle.startsWith('run:')) - ? { id: 'delivery_held', messageIds: new Set(['m1']) } - : undefined, - getUndeliveredUnreadMessages: () => - (options.unreadIds ?? ['m1']).map((id, index) => ({ - id, - type: 'status', - sequence: index + 3 - })), + ((options.outstandingOwnDelivery ?? false) && !handle.startsWith('run:')), + getUndeliveredUnreadMessages: () => [{ id: 'm1', type: 'status', sequence: 3 }], markAsDelivered, getStructuredPointerOperation: (key: string) => stored.get(key), putStructuredPointerOperation: (row: StructuredPointerOperationRow) => @@ -275,32 +264,15 @@ describe('structured mailbox pointer delivery', () => { expect(markAsDelivered).toHaveBeenCalledWith(['m1']) }) - it('does not re-point mail the reader already holds unacknowledged', async () => { - // It already has this batch, so a second pointer spends a whole provider turn telling it - // something it was told. + it('does not re-nudge a mailbox still holding its own unacked batch', async () => { + // The other half of the same gate: the consumer already has this batch, so a second nudge + // spends a whole provider turn telling it something it was told. const { delivery, send } = harness({ journal: idleJournal(), outstandingOwnDelivery: true }) delivery.deliverForHandle('dispatch:d1') await flush() expect(send).not.toHaveBeenCalled() }) - it('points newer mail while the reader holds an unacknowledged batch, in the PTY pointer text', async () => { - // The strand this pins: a chat reads a result, ends its turn without acking, and the gate on - // "an unacknowledged batch exists" silenced every later result. The text stays the PTY lane's. - const { delivery, send, markAsDelivered } = harness({ - journal: idleJournal(), - outstandingOwnDelivery: true, - unreadIds: ['m1', 'm2'] - }) - delivery.deliverForHandle('dispatch:d1') - await flush() - expect(send).toHaveBeenCalledTimes(1) - expect(send.mock.calls[0]![0].body.blocks[0]).toMatchObject({ - text: formatMessagePointer(1, 'dispatch:d1', 'orca-dev').trim() - }) - expect(markAsDelivered).toHaveBeenCalledWith(['m2']) - }) - it('retries a rejected nudge on the next journal edge, under the same id', async () => { // A rejection consumes no mail and nothing else redrives this mailbox, so leaving it unparked // stranded the worker until unrelated mail happened to arrive. The retry keeps the id: the host @@ -419,7 +391,7 @@ describe('forgetting one settled worker', () => { })) const db = { getDispatchContextById: () => ({ run_id: 'run_1' }), - getOutstandingMailboxDelivery: () => undefined, + hasOutstandingMailboxDelivery: () => false, getUndeliveredUnreadMessages: () => [{ id: 'm1', type: 'status', sequence: 3 }], markAsDelivered: vi.fn(), getStructuredPointerOperation: () => undefined, diff --git a/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.ts b/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.ts index 283d3b0feaf..c35065923cc 100644 --- a/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.ts +++ b/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.ts @@ -163,16 +163,20 @@ export class OrchestrationStructuredMailboxPointerDelivery< if (!db || this.inFlight.has(mailboxHandle)) { return } - // Eligibility is "not yet pointed" (`delivered_at`), never "has the consumer acked": a chat that - // reads a batch and ends its turn without acking must still be pointed at the NEXT result. The - // batch it holds is excluded; its own `check` replays that batch and names its ack. - const outstanding = db.getOutstandingMailboxDelivery?.(mailboxHandle) + // Don't re-nudge a mailbox whose consumer still holds an unacknowledged batch. The lookup is + // keyed on the exact handle being nudged, so a coordinator's own `run:` delivery is invisible + // to a worker's `dispatch:` gate and cannot suppress the nudges a coordinator sends its + // workers. Worth more here than in the PTY lane: a structured nudge costs a whole provider + // turn, not a line of text into a composer. + if (db.hasOutstandingMailboxDelivery?.(mailboxHandle)) { + return + } const unread = selectOrchestrationPointerBatch({ db, mailboxHandle, waiters: this.deps.getMessageWaiters(mailboxHandle), reservedTypes - }).filter((message) => !outstanding?.messageIds.has(message.id)) + }) if (unread.length === 0) { return } diff --git a/src/main/runtime/structured-chat-coordinator-mail.test.ts b/src/main/runtime/structured-chat-coordinator-mail.test.ts index e6dfb89063c..ea4a243cc84 100644 --- a/src/main/runtime/structured-chat-coordinator-mail.test.ts +++ b/src/main/runtime/structured-chat-coordinator-mail.test.ts @@ -338,43 +338,6 @@ describe('a worker result reaches the structured chat that coordinates it', () = expect(await userTexts(COORDINATOR)).toHaveLength(1) }) - it('points the next result at a coordinator that read the last one without acking', async () => { - // The strand this pins: a flagless `check` opens a delivery that `check` replays until acked, - // and a lane gated on "an unacknowledged batch exists" never pointed the chat at a later result. - const chat = await openChat(COORDINATOR) - const { runId, taskId } = await coordinatorRunAndTask() - await finishWorker(taskId) - await vi.waitFor(() => expect(chat.turns).toHaveLength(1), WAIT) - await settleTurn(COORDINATOR, 0) - const first = await call('orchestration.check', {}, { sessionId: COORDINATOR }) - const heldDelivery = String(first.deliveryId) - expect(first).toMatchObject({ count: 1, messages: [{ type: 'worker_done' }] }) - - const second = await call( - 'orchestration.taskCreate', - { spec: 'more' }, - { sessionId: COORDINATOR } - ) - await finishWorker(idOf(second.task), { handle: 'term_worker_2', paneKey: WORKER_2_PANE }) - await vi.waitFor(() => expect(chat.turns).toHaveLength(2), WAIT) - // The PTY lane's text: `check` itself replays the held batch and names its ack. - expect(turnText(chat.turns[1]!)).toBe(ptyPointer(`run:${runId}`)) - await settleTurn(COORDINATOR, 1) - - // Exactly once per new message: a retry and the idle edge point nothing further. - runtime.deliverPendingMessagesForHandle(`run:${runId}`) - await new Promise((resolve) => setTimeout(resolve, 20)) - expect(chat.turns).toHaveLength(2) - - const acked = await call( - 'orchestration.check', - { ack: heldDelivery }, - { sessionId: COORDINATOR } - ) - expect(acked).toMatchObject({ acknowledged: heldDelivery, count: 1 }) - expect(acked.messages).not.toEqual(first.messages) - }) - /** Fires both edges and waits until every gate read they started has answered. */ async function edgesAnswered(): Promise { const reads = vi.spyOn(host, 'journalSnapshot')