diff --git a/src/main/acp/acp-structured-mail-admission.test.ts b/src/main/acp/acp-structured-mail-admission.test.ts new file mode 100644 index 000000000000..ce3de90bd9e2 --- /dev/null +++ b/src/main/acp/acp-structured-mail-admission.test.ts @@ -0,0 +1,54 @@ +import { afterEach, expect, it } from 'vitest' +import { openAttachedHostRig, send, hello, framesOf } from './acp-structured-host.test-support' +import { PROVIDER_SESSION, waitFor } from './acp-structured-adapter.test-support' +import { closeProviderTimelineRigs } from '../native-chat/agent-session-timeline/provider-timeline-assembler-test-support' +import { HOST_TEST_SESSION as SESSION } from '../native-chat/agent-session-wire/structured-agent-session-host-test-data' +import { setStructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-registry' +import { commandReceiptScope } from '../native-chat/agent-session-journal/command-receipt-schema' +import { createStructuredMailboxPointerHost } from '../runtime/orchestration/structured-mailbox-pointer-host' +import { mintAgentSessionOperationId } from '../runtime/orchestration/structured-pointer-operation-id' +import type { AgentJournalMessageItem } from '../../shared/agent-session-journal-types' + +afterEach(async () => { + setStructuredAgentSessionHost(null) + await closeProviderTimelineRigs() +}) + +it('holds mail without interrupting an answerable ACP question', async () => { + const { host, rig } = await openAttachedHostRig() + setStructuredAgentSessionHost(host) + await send(host, 'work') + await rig.frame('session/prompt') + void rig.child().agent.request(601, 'x.ai/ask_user_question', { + sessionId: PROVIDER_SESSION, + toolCallId: 'question-1', + questions: [{ question: 'Which file?', options: [{ label: 'a.ts' }] }] + }) + const pointer = createStructuredMailboxPointerHost() + await waitFor(() => expect(pointer.hasActionablePromptNow(SESSION)).toBe(true)) + const body: AgentJournalMessageItem = { + ...hello, + from: { + kind: 'agent', + senders: [], + orchestration: { message: 'mail-notice', mailbox: 'run:r1', dispatchId: null, messages: [] } + } + } + const operationId = mintAgentSessionOperationId(Date.now()) + const expectedRuntimeFence = host.deps.store.getRecord(SESSION)!.lease.runtimeFence + expect( + await pointer.send({ + sessionId: SESSION, + dispatchId: null, + operationId, + expectedRuntimeFence, + body + }) + ).toEqual({ kind: 'held' }) + expect((await host.journalSnapshot(SESSION)).submissions).toHaveLength(1) + expect( + host.deps.store.readCommandReceipt(commandReceiptScope('any', 'global'), operationId) + ).toMatchObject({ verdict: 'absent' }) + expect(framesOf(rig.child(), 'session/cancel')).toHaveLength(0) + await host.close(SESSION, 'user-close') +}) diff --git a/src/main/runtime/orca-runtime-structured-status-sink-wiring.test.ts b/src/main/runtime/orca-runtime-structured-status-sink-wiring.test.ts index afb73faa2f0e..6c68bed3ee07 100644 --- a/src/main/runtime/orca-runtime-structured-status-sink-wiring.test.ts +++ b/src/main/runtime/orca-runtime-structured-status-sink-wiring.test.ts @@ -109,7 +109,8 @@ describe('mail dispatch through the current runtime', () => { }) target.onStructuredSessionStatusForMail({ sessionId: 'session-mailbox', status: 'working' }) - expect(events).toEqual([]) + expect(events).toEqual(['journal:session-mailbox']) + events.length = 0 target.onStructuredSessionStatusForMail({ sessionId: 'session-mailbox', status: 'idle' }) expect(events).toEqual(['journal:session-mailbox', 'delivery:term_mailbox']) }) diff --git a/src/main/runtime/orchestration/agent-facing-parity.test.ts b/src/main/runtime/orchestration/agent-facing-parity.test.ts index d860128fa4a7..4836203e17ae 100644 --- a/src/main/runtime/orchestration/agent-facing-parity.test.ts +++ b/src/main/runtime/orchestration/agent-facing-parity.test.ts @@ -129,6 +129,7 @@ async function renderChatPointer(mailbox: string): Promise { senderName: () => null, host: { readSessionFacts: async () => ({ submissions: [] }), + hasActionablePromptNow: () => false, currentFence: () => 1, currentContextClearOperationId: () => undefined, send: async (input) => { diff --git a/src/main/runtime/orchestration/send-agent-turn.ts b/src/main/runtime/orchestration/send-agent-turn.ts index b7a24c4c4365..c9f30abb2e04 100644 --- a/src/main/runtime/orchestration/send-agent-turn.ts +++ b/src/main/runtime/orchestration/send-agent-turn.ts @@ -41,6 +41,8 @@ export type StructuredSessionTurn = { operationId: string expectedRuntimeFence: number delivery: AgentTurnDelivery + /** Host-local admission for an automated send. */ + beforeRun?: () => void } export type StructuredSessionTurnSend = { @@ -134,7 +136,10 @@ async function sendStructuredSessionTurn( body: turn.body, delivery: turn.delivery === 'queue' ? 'queue-if-active' : undefined }) - const result = await send.host.send({ callerKey: send.callerKey }, message) + const result = await send.host.send( + { callerKey: send.callerKey }, + turn.beforeRun ? { ...message, beforeRun: turn.beforeRun } : message + ) if (!result.ok) { return { kind: 'refused', refusal: result.refusal } } 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 76c2c964f611..c2f1ac9fb062 100644 --- a/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts +++ b/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts @@ -9,6 +9,7 @@ import { type StructuredPointerSubmission } from './structured-pointer-operation-id' import type { StructuredWorkerIdentity } from '../structured-worker-identity' +import { RuntimeMessageWaiters } from '../runtime-message-waiters' const IDENTITY: StructuredWorkerIdentity = { handle: 'structworker_1', @@ -32,10 +33,12 @@ function harness(options: { /** The mailbox this worker owns; its own handle for direct peer mail outside a dispatch. */ mailbox?: string dispatchId?: string | null + waiters?: RuntimeMessageWaiters }) { const mailbox = options.mailbox ?? 'dispatch:d1' const dispatchId = options.dispatchId === undefined ? 'd1' : options.dispatchId let attached = options.attached ?? true + let awaitingHuman = false // The session's recorded sends, as its journal reports them. let submissions: StructuredPointerSubmission[] = [] // The mailbox's unread mail; a pointed message is no longer selected for a pointer. @@ -49,9 +52,11 @@ function harness(options: { } }) const send: StructuredMailboxPointerHost['send'] = vi.fn(async () => - options.queued - ? { kind: 'queued' as const } - : { kind: 'sent' as const, state: options.dispatchState ?? ('accepted' as const) } + awaitingHuman + ? { kind: 'held' as const } + : options.queued + ? { kind: 'queued' as const } + : { kind: 'sent' as const, state: options.dispatchState ?? ('accepted' as const) } ) const sendMock = vi.mocked(send) const readContext = vi.fn( @@ -63,7 +68,14 @@ function harness(options: { hasOutstandingMailboxDelivery: (handle: string) => ((options.outstandingRunDelivery ?? false) && handle.startsWith('run:')) || ((options.outstandingOwnDelivery ?? false) && !handle.startsWith('run:')), - getUndeliveredUnreadMessages: () => mail.filter((message) => !pointed.has(message.id)), + getUndeliveredUnreadMessages: ( + _handle: string, + _filter: unknown, + input: { excludeTypes: readonly string[] } + ) => + mail.filter( + (message) => !pointed.has(message.id) && !input.excludeTypes.includes(message.type) + ), markAsDelivered, getStructuredPointerOperation: (key: string) => stored.get(key), putStructuredPointerOperation: (row: StructuredPointerOperationRow) => @@ -72,13 +84,14 @@ function harness(options: { } const delivery = new OrchestrationStructuredMailboxPointerDelivery({ getDb: () => db as never, - getMessageWaiters: () => undefined, + getMessageWaiters: (handle) => options.waiters?.get(handle), resolveStructuredTarget: (mailboxHandle) => mailboxHandle === mailbox ? { sessionId: IDENTITY.sessionId, dispatchId } : null, getCliCommand: () => 'orca-dev', senderName: (party) => (party.address === 'term_coord' ? 'Coordinator' : null), host: { readSessionFacts: async () => (attached ? { submissions } : null), + hasActionablePromptNow: () => awaitingHuman, currentFence: () => 4, currentContextClearOperationId: readContext, send @@ -93,11 +106,14 @@ function harness(options: { setAttached: (next: boolean) => { attached = next }, - receive: (id: string, sequence: number) => { - mail.push({ id, type: 'status', sequence, from_handle: 'term_coord', run_id: 'run_1' }) + receive: (id: string, sequence: number, type = 'status') => { + mail.push({ id, type, sequence, from_handle: 'term_coord', run_id: 'run_1' }) }, setSubmissions: (next: StructuredPointerSubmission[]) => { submissions = next + }, + setAwaitingHuman: (next: boolean) => { + awaitingHuman = next } } } @@ -211,6 +227,20 @@ describe('structured mailbox pointer delivery', () => { expect(markAsDelivered).toHaveBeenCalledWith(['m1']) }) + it('rechecks a held notice when the answer edge lands during its host call', async () => { + const { delivery, send, markAsDelivered, setAwaitingHuman } = harness({}) + setAwaitingHuman(true) + send.mockImplementationOnce(async () => { + setAwaitingHuman(false) + delivery.onJournalActivity('session-1') + return { kind: 'held' } + }) + delivery.deliverForHandle('dispatch:d1') + await flush() + expect(send).toHaveBeenCalledTimes(2) + expect(markAsDelivered).toHaveBeenCalledWith(['m1']) + }) + it('sends the pointer through the chat, with who it is from, and counts it pointed once queued', async () => { const { delivery, send, markAsDelivered, stored } = harness({ queued: true }) delivery.deliverForHandle('dispatch:d1') @@ -250,6 +280,38 @@ describe('structured mailbox pointer delivery', () => { }) }) + it('points a second report after its filtered waiter expires during first-notice acceptance', async () => { + const waiters = new RuntimeMessageWaiters() + const { delivery, send, receive, markAsDelivered } = harness({ + mailbox: 'run:run_1', + waiters + }) + const waiting = waiters.wait('run:run_1', { typeFilter: ['worker_done'], timeoutMs: 40 }) + const accepted = Promise.withResolvers() + send.mockImplementationOnce(async () => { + await accepted.promise + return { kind: 'sent', state: 'accepted' } + }) + delivery.deliverForHandle('run:run_1', new Set(['worker_done'])) + await flush() + expect(send).toHaveBeenCalledTimes(1) + expect(await waiting).toBe('timed_out') + receive('m2', 4, 'worker_done') + delivery.deliverForHandle('run:run_1') + await flush() + expect(send).toHaveBeenCalledTimes(1) + accepted.resolve() + await vi.waitFor(() => expect(send).toHaveBeenCalledTimes(2)) + expect(send.mock.calls[1]![0].body.from).toMatchObject({ + orchestration: { messages: [{ messageId: 'm2' }] } + }) + expect(markAsDelivered.mock.calls).toEqual([[['m1']], [['m2']]]) + delivery.onJournalActivity('session-1') + delivery.deliverForHandle('run:run_1') + await flush() + expect(send).toHaveBeenCalledTimes(2) + }) + it('nudges the worker while its coordinator holds an unacked Run delivery', async () => { // The exact window in which a coordinator replies to its workers: it checked, is acting on the // batch, and has not acked yet. The gate is keyed on the handle being nudged, so the @@ -442,6 +504,7 @@ describe('forgetting one settled worker', () => { function twoWorkerHarness() { let resolves = true let attached = false + const pointed = new Set() const sessionByMailbox: Record = { 'dispatch:d1': 'session-1', 'dispatch:d2': 'session-2' @@ -453,10 +516,11 @@ describe('forgetting one settled worker', () => { const db = { getDispatchContextById: () => ({ run_id: 'run_1' }), hasOutstandingMailboxDelivery: () => false, - getUndeliveredUnreadMessages: () => [ - { id: 'm1', type: 'status', sequence: 3, from_handle: 'term_coord', run_id: 'run_1' } - ], - markAsDelivered: vi.fn(), + getUndeliveredUnreadMessages: () => + pointed.has('m1') + ? [] + : [{ id: 'm1', type: 'status', sequence: 3, from_handle: 'term_coord', run_id: 'run_1' }], + markAsDelivered: vi.fn((ids: string[]) => ids.forEach((id) => pointed.add(id))), getStructuredPointerOperation: () => undefined, putStructuredPointerOperation: () => {}, deleteStructuredPointerOperation: () => {} @@ -474,6 +538,7 @@ describe('forgetting one settled worker', () => { senderName: () => null, host: { readSessionFacts: async () => (attached ? { submissions: [] } : null), + hasActionablePromptNow: () => false, currentFence: () => 4, currentContextClearOperationId: () => undefined, send @@ -575,6 +640,7 @@ describe('a mailbox a /clear moves while its nudge is in flight', () => { senderName: () => null, host: { readSessionFacts: async () => ({ submissions: [] }), + hasActionablePromptNow: () => false, currentFence: () => 4, currentContextClearOperationId: () => contextClearOperationId, send diff --git a/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.ts b/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.ts index 7376e9d63366..8d206dca015f 100644 --- a/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.ts +++ b/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.ts @@ -51,6 +51,7 @@ type ParkedPointerDelivery = { export type StructuredPointerSendOutcome = | { kind: 'sent'; state: StructuredDispatchState } + | { kind: 'held' } /** The chat's queue took it, as it takes a person's message. */ | { kind: 'queued' } | { kind: 'unattached' } @@ -63,6 +64,7 @@ export type StructuredPointerSessionFacts = { export type StructuredMailboxPointerHost = { /** `null` when the session cannot be read. */ readSessionFacts: (sessionId: string) => Promise + hasActionablePromptNow: (sessionId: string) => boolean | null send: (input: { sessionId: string dispatchId: string | null @@ -168,56 +170,92 @@ export class OrchestrationStructuredMailboxPointerDelivery< reservedTypes?: ReadonlySet, attemptedContexts = new Set() ): Promise { - const db = this.deps.getDb() - if (!db || this.inFlight.has(mailboxHandle)) { - return - } - // 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 - }) - if (unread.length === 0) { + if (this.inFlight.has(mailboxHandle)) { return } this.inFlight.add(mailboxHandle) + let current = target + let currentReservedTypes = reservedTypes let contextKey: string | undefined + let progressed = false try { - const session = await this.deps.host.readSessionFacts(target.sessionId) - const contextClearOperationId = this.deps.host.currentContextClearOperationId( - target.sessionId - ) - contextKey = JSON.stringify([target.sessionId, contextClearOperationId]) - if (attemptedContexts.has(contextKey)) { - return + for (;;) { + const db = this.deps.getDb() + // The consumer's own unacknowledged batch suppresses nudges only for this exact mailbox. + if (!db || db.hasOutstandingMailboxDelivery?.(mailboxHandle)) { + return + } + const unread = selectOrchestrationPointerBatch({ + db, + mailboxHandle, + waiters: this.deps.getMessageWaiters(mailboxHandle), + reservedTypes: currentReservedTypes + }) + if (unread.length === 0) { + return + } + this.parkedUntilJournalEdge.set(mailboxHandle, { + sessionId: current.sessionId, + reservedTypes: currentReservedTypes + }) + const session = await this.deps.host.readSessionFacts(current.sessionId) + const contextClearOperationId = this.deps.host.currentContextClearOperationId( + current.sessionId + ) + contextKey = JSON.stringify([current.sessionId, contextClearOperationId]) + if (!progressed && attemptedContexts.has(contextKey)) { + return + } + attemptedContexts.add(contextKey) + let attempt = await this.attempt( + db, + mailboxHandle, + current, + unread, + currentReservedTypes, + session, + contextClearOperationId + ) + while ( + attempt === 'held' && + this.deps.host.hasActionablePromptNow(current.sessionId) === false + ) { + attempt = await this.attempt( + db, + mailboxHandle, + current, + unread, + currentReservedTypes, + session, + contextClearOperationId + ) + } + if (attempt === 'held') { + this.retain(mailboxHandle, current.sessionId, 'turn-unsettled', currentReservedTypes) + return + } + progressed = attempt + if (!progressed) { + return + } + // The arrival snapshot protects its first handoff; successors derive live waiter ownership. + currentReservedTypes = undefined + // Acceptance releases the snapshotted batch; rederive any mail that arrived meanwhile. + this.parkedUntilJournalEdge.delete(mailboxHandle) + const next = this.deps.resolveStructuredTarget(mailboxHandle) + if (!next) { + return + } + current = next } - attemptedContexts.add(contextKey) - await this.attempt( - db, - mailboxHandle, - target, - unread, - reservedTypes, - session, - contextClearOperationId - ) } finally { this.inFlight.delete(mailboxHandle) // A thrown attempt follows too; its own failure is what still propagates. if (contextKey !== undefined) { await this.followChangedContext( mailboxHandle, - target, - reservedTypes, + current, + currentReservedTypes, attemptedContexts ).catch(() => undefined) } @@ -258,16 +296,12 @@ export class OrchestrationStructuredMailboxPointerDelivery< reservedTypes: ReadonlySet | undefined, session: StructuredPointerSessionFacts | null, contextClearOperationId: string | undefined - ): Promise { + ): Promise { const sessionId = target.sessionId - if (!session) { - this.retain(mailboxHandle, sessionId, 'session-not-attached', reservedTypes) - return - } const fence = this.deps.host.currentFence(sessionId) - if (fence === null) { + if (!session || fence === null) { this.retain(mailboxHandle, sessionId, 'session-not-attached', reservedTypes) - return + return false } const body: AgentJournalMessageItem = { kind: 'message', @@ -301,11 +335,11 @@ export class OrchestrationStructuredMailboxPointerDelivery< db.markAsDelivered(staged) db.deleteStructuredPointerOperation(mailboxHandle) this.sentOperationIds.delete(mailboxHandle) - return + return true } if (operation.kind === 'park') { this.retain(mailboxHandle, sessionId, 'turn-unsettled', reservedTypes) - return + return false } this.sentOperationIds.set(mailboxHandle, operation.operationId) const outcome = await this.deps.host.send({ @@ -315,21 +349,23 @@ export class OrchestrationStructuredMailboxPointerDelivery< expectedRuntimeFence: fence, body }) + if (outcome.kind === 'held') { + return 'held' + } if (outcome.kind === 'unattached') { this.retain(mailboxHandle, sessionId, 'session-not-attached', reservedTypes) - return + return false } // A queued pointer is the chat's queue's to send, as a person's queued message is. if (outcome.kind === 'sent' && !structuredDispatchDelivered(outcome.state)) { // The row stays: resending under its id replays this verdict and starts nothing. this.retain(mailboxHandle, sessionId, retainReasonForDispatch(outcome.state), reservedTypes) - return + return false } db.markAsDelivered(staged) - // The nudge landed as its own turn, so the next settle edge is the natural retry point for - // anything that arrives while it runs. db.deleteStructuredPointerOperation(mailboxHandle) this.sentOperationIds.delete(mailboxHandle) + return true } /** diff --git a/src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts b/src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts index 05683fcc505c..d7b33145c36b 100644 --- a/src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts +++ b/src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts @@ -87,7 +87,8 @@ describe('structured mailbox pointer host', () => { const submissions = [{ clientMessageId: 'op1', dispatchState: 'unknown' }] hostRef.current = { deps: { store: { getRecord: () => null } }, - journalSnapshot: () => ({ items: [], submissions }) + journalSnapshot: () => ({ items: [], submissions }), + currentWork: () => null } expect(await createStructuredMailboxPointerHost().readSessionFacts('s1')).toEqual({ submissions @@ -115,10 +116,16 @@ describe('structured mailbox pointer host', () => { getRecord: () => ({ providerContextBoundary: { operationId: 'clear-1', afterFence: 2 } }) } }, - journalSnapshot: () => ({ items, submissions: [pointerSubmission('earlier', 2, 6), current] }) + journalSnapshot: () => ({ + items, + submissions: [pointerSubmission('earlier', 2, 6), current] + }), + currentWork: () => null } const host = createStructuredMailboxPointerHost() - expect(await host.readSessionFacts('s1')).toEqual({ submissions: [current] }) + expect(await host.readSessionFacts('s1')).toEqual({ + submissions: [current] + }) expect(host.currentContextClearOperationId('s1')).toBe('clear-1') }) @@ -133,10 +140,13 @@ describe('structured mailbox pointer host', () => { journalSnapshot: () => ({ items: [], submissions: [pointerSubmission('earlier', 2, 2), current] - }) + }), + currentWork: () => null } const host = createStructuredMailboxPointerHost() - expect(await host.readSessionFacts('s1')).toEqual({ submissions: [current] }) + expect(await host.readSessionFacts('s1')).toEqual({ + submissions: [current] + }) expect(host.currentContextClearOperationId('s1')).toBe('clear-1') }) @@ -192,14 +202,11 @@ describe('structured mailbox pointer host', () => { expect(send.mock.calls[0]![1]!.retryUnknown).toBeUndefined() }) - it('asks a busy chat to queue the pointer as a card, with who it is from', async () => { + it('sends the pointer as a plain send for every agent, with who it is from', async () => { const send = vi.fn( async (_caller: unknown, _payload: { delivery?: string; body?: unknown }) => ({ ok: true, - value: { - clientMessageId: 'op1', - queued: { messageId: 'op1', position: 0, state: 'waiting' } - } + value: { clientMessageId: 'op1', submission: { dispatchState: 'accepted' } } }) ) hostRef.current = { send } @@ -211,11 +218,9 @@ describe('structured mailbox pointer host', () => { expectedRuntimeFence: 1, body: { kind: 'message', role: 'user', blocks: [], from: NOTICE_SOURCE } }) - ).resolves.toEqual({ kind: 'queued' }) - expect(send.mock.calls[0]![1]).toMatchObject({ - delivery: 'queue-if-active', - body: { from: NOTICE_SOURCE } - }) + ).resolves.toEqual({ kind: 'sent', state: 'accepted' }) + expect(send.mock.calls[0]![1]).toMatchObject({ body: { from: NOTICE_SOURCE } }) + expect(send.mock.calls[0]![1]!.delivery).toBeUndefined() }) it('consumes mail once an accepted nudge is delivered while the worker starts (W10)', async () => { diff --git a/src/main/runtime/orchestration/structured-mailbox-pointer-host.ts b/src/main/runtime/orchestration/structured-mailbox-pointer-host.ts index 92b1e0594823..78ecf621c22e 100644 --- a/src/main/runtime/orchestration/structured-mailbox-pointer-host.ts +++ b/src/main/runtime/orchestration/structured-mailbox-pointer-host.ts @@ -17,7 +17,7 @@ import { structuredSessionGateFacts, type StructuredSessionGateFacts } from './structured-session-pointer-delivery' -import { sendAgentTurn } from './send-agent-turn' +import { sendAgentTurn, type StructuredSessionTurnOutcome } from './send-agent-turn' /** Names the dispatch a nudge belongs to on the operation row it writes. */ export function structuredPointerCallerKey(dispatchId: string): string { @@ -63,8 +63,8 @@ async function readPointerSessionFacts( if (!snapshot) { return null } - const boundary = - getStructuredAgentSessionHost()?.deps.store.getRecord(sessionId)?.providerContextBoundary + const host = getStructuredAgentSessionHost() + const boundary = host?.deps.store.getRecord(sessionId)?.providerContextBoundary return { submissions: boundary ? snapshot.submissions.filter((submission) => submission.fence > boundary.afterFence) @@ -95,6 +95,10 @@ export function createStructuredMailboxPointerHost(): StructuredMailboxPointerHo return readPointerSessionFacts(sessionId) }, + hasActionablePromptNow(sessionId) { + return getStructuredAgentSessionHost()?.currentWork(sessionId)?.hasActionablePrompt() ?? null + }, + currentFence(sessionId) { return ( getStructuredAgentSessionHost()?.deps.store.getRecord(sessionId)?.lease.runtimeFence ?? null @@ -111,21 +115,37 @@ export function createStructuredMailboxPointerHost(): StructuredMailboxPointerHo if (!host) { return { kind: 'unattached' } } - const outcome = await sendAgentTurn({ - kind: 'structured-session', - host, - sessionId: input.sessionId, - callerKey: input.dispatchId - ? structuredPointerCallerKey(input.dispatchId) - : structuredSessionPointerCallerKey(input.sessionId), - turn: { - body: input.body, - // As a person's message is: a busy chat queues it as a card, sent when the turn ends. - delivery: 'queue', - operationId: input.operationId, - expectedRuntimeFence: input.expectedRuntimeFence + let outcome: StructuredSessionTurnOutcome + try { + outcome = await sendAgentTurn({ + kind: 'structured-session', + host, + sessionId: input.sessionId, + callerKey: input.dispatchId + ? structuredPointerCallerKey(input.dispatchId) + : structuredSessionPointerCallerKey(input.sessionId), + turn: { + body: input.body, + // A person's plain send: every agent takes it the way it takes their own mid-turn message. + delivery: 'now', + operationId: input.operationId, + expectedRuntimeFence: input.expectedRuntimeFence, + beforeRun: () => { + if ( + input.body.from?.orchestration?.message === 'mail-notice' && + host.currentWork(input.sessionId)?.hasActionablePrompt() + ) { + throw new StructuredMailHeldError() + } + } + } + }) + } catch (error) { + if (error instanceof StructuredMailHeldError) { + return { kind: 'held' } } - }) + throw error + } switch (outcome.kind) { case 'refused': return outcome.refusal.code === AGENT_SESSION_NOT_ATTACHED.code @@ -146,3 +166,5 @@ export function createStructuredMailboxPointerHost(): StructuredMailboxPointerHo } } } + +class StructuredMailHeldError extends Error {} diff --git a/src/main/runtime/orchestration/structured-session-pointer-delivery.ts b/src/main/runtime/orchestration/structured-session-pointer-delivery.ts index 8a99deb39950..289f4a9cb2e3 100644 --- a/src/main/runtime/orchestration/structured-session-pointer-delivery.ts +++ b/src/main/runtime/orchestration/structured-session-pointer-delivery.ts @@ -2,7 +2,8 @@ * What orchestration mail delivery reads of a host-owned structured ("native") agent session. * * A structured session has no PTY the pointer can be typed into, so the nudge travels as a session - * turn instead of as bytes, and a busy session's own queue holds it until the turn ends. + * turn instead of as bytes: steered into a busy session's running turn, or held by its queue until + * the turn ends when its agent cannot take a message mid-turn. * Everything here is pure. Orchestration's database stays the source of truth: nothing here * consumes mail. */ diff --git a/src/main/runtime/orchestration/terminal-pane-mail-routing.ts b/src/main/runtime/orchestration/terminal-pane-mail-routing.ts index 1d8c7a0598e7..c8ee0e84edc9 100644 --- a/src/main/runtime/orchestration/terminal-pane-mail-routing.ts +++ b/src/main/runtime/orchestration/terminal-pane-mail-routing.ts @@ -195,20 +195,24 @@ export abstract class OrcaRuntimeWithGetPtyRecordForPaneKey extends OrcaRuntimeW } /** - * Every structured session's status change reaches here. At its idle edge, retry what is parked - * on it and re-derive the mailboxes it owns, so mail it could not take earlier (mid-turn, closed) - * is pointed again. Workers and chats alike: this is not per-dispatch. + * Every structured session's status change reaches here. At its working and idle edges, retry + * what is parked on it; at idle also re-derive the mailboxes it owns, so mail it could not take + * earlier (closed) is pointed again. Workers and chats alike: this is not per-dispatch. */ onStructuredSessionStatusForMail(summary: { sessionId: string status: 'working' | 'attention' | 'idle' | null }): void { - if (summary.status === 'working' || summary.status === 'attention') { + if (summary.status === 'attention') { return } // Logged, never thrown: the same status callback goes on to the first-turn workspace rename. try { + // Working also follows an answered prompt, so mail held behind the prompt is retried too. this.notifyStructuredSessionJournalActivity(summary.sessionId) + if (summary.status === 'working') { + return + } const openDb = () => this.getExistingOrchestrationDb() const deliver = (mailbox: string) => this.deliverPendingMessagesForHandle(mailbox) structuredSessionIdleEdgeMailboxes(summary.sessionId, openDb).forEach(deliver) diff --git a/src/main/runtime/structured-chat-coordinator-fake-codex-fixture.ts b/src/main/runtime/structured-chat-coordinator-fake-codex-fixture.ts index 05f5c064ab50..3987165c00f0 100644 --- a/src/main/runtime/structured-chat-coordinator-fake-codex-fixture.ts +++ b/src/main/runtime/structured-chat-coordinator-fake-codex-fixture.ts @@ -6,7 +6,6 @@ import type { CodexAppServerConnectionHandlers, openCodexAppServerConnection } from '../codex/codex-app-server-connection' -import { refuseUnroutedSteer } from '../codex/codex-structured-dispatch-test-support' import { computeAgentSessionPayloadFingerprint } from '../../shared/agent-session-mutation-envelope' import { attachFingerprintFields } from '../native-chat/agent-session-wire/structured-agent-session-attach' import { isRecord } from './rpc/orchestration-session-caller-test-fixture' @@ -20,6 +19,8 @@ export type FakeConnection = Omit & { /** Every JSON-RPC method the adapter called, in order. */ methods: string[] turns: { clientUserMessageId: string; text: string }[] + /** Messages steered into a running turn, as Codex takes them. */ + steers: { expectedTurnId: string; clientUserMessageId: string; text: string }[] } /** How the fake provider misbehaves; reset before each test. */ @@ -61,6 +62,7 @@ export function fakeCodex() { threadId: null, methods: [], turns: [], + steers: [], pid: 4321, closed: false, request: async (method, params) => { @@ -114,7 +116,12 @@ export function fakeCodex() { } } if (method === 'turn/steer') { - return refuseUnroutedSteer(undefined) + connection.steers.push({ + expectedTurnId: String(input.expectedTurnId), + clientUserMessageId: String(input.clientUserMessageId), + text: JSON.stringify(input.input) + }) + return { turnId: input.expectedTurnId } } return {} }, diff --git a/src/main/runtime/structured-chat-coordinator-mail-queue.test.ts b/src/main/runtime/structured-chat-coordinator-mail-queue.test.ts deleted file mode 100644 index 789fe3a5c908..000000000000 --- a/src/main/runtime/structured-chat-coordinator-mail-queue.test.ts +++ /dev/null @@ -1,193 +0,0 @@ -import './rpc/unused-default-rpc-methods.test-fixture' -// A busy structured chat holds the orchestration pointer as a card in its own queue, sent when the -// turn ends, as it holds a message the person sends then; the queue does nothing else with it. End -// to end on the coordinator-mail rig. - -import { describe, expect, it, vi } from 'vitest' -import { structuredAgentSessionMessageSendMutation } from '../../shared/structured-agent-session-send-mutation' -import { operationId, type FakeConnection } from './structured-chat-coordinator-fake-codex-fixture' -import { idOf } from './rpc/orchestration-session-caller-test-fixture' -import { - COORDINATOR, - WORKER_2_PANE, - WAIT, - call, - coordinatorRunAndTask, - db, - finishWorker, - host, - observationClock, - openChat, - ptyPointer, - queuedCardTexts, - runtime, - sendUserMessage, - settleTurn, - turnText -} from './structured-chat-coordinator-mail-rig.test-fixture' - -/** The person's turn, started and still running; resolves to its end. */ -async function runningUserTurn(chat: FakeConnection): Promise<() => Promise> { - expect(await sendUserMessage(COORDINATOR, 'go')).toMatchObject({ ok: true }) - await vi.waitFor(() => expect(chat.turns).toHaveLength(1), WAIT) - const notify = (method: string, params: unknown) => chat.handlers.onNotification?.(method, params) - notify('turn/started', { turn: { id: 'turn-1' } }) - notify('item/completed', { - item: { - type: 'userMessage', - id: 'echo-go', - clientId: chat.turns[0]!.clientUserMessageId, - content: [{ type: 'text', text: 'go' }] - } - }) - await host.flushStreamedEvents(COORDINATOR) - return async () => { - notify('turn/completed', { turn: { id: 'turn-1' } }) - await host.flushStreamedEvents(COORDINATOR) - } -} - -/** Idle edges with nothing owed: whatever they would send gets the time to show. */ -async function idleEdgesSettled(): Promise { - for (let edge = 0; edge < 3; edge += 1) { - runtime.onStructuredSessionStatusForMail({ sessionId: COORDINATOR, status: 'idle' }) - await observationClock.observe(100) - } -} - -/** The chat's queue as its journal stores it. */ -function queuedRows() { - return host.collaboratorsForTests().sessions.get(COORDINATOR)?.journal.queuedMessages.list() ?? [] -} - -/** A second task, for a second worker result. */ -async function secondTask(): Promise { - return idOf( - (await call('orchestration.taskCreate', { spec: 'more' }, { sessionId: COORDINATOR })).task - ) -} - -describe("a busy chat's orchestration pointer waits in its queue", () => { - it('queues the pointer as a card, with who it is from, and sends it once when the turn ends', async () => { - observationClock.start() - const chat = await openChat(COORDINATOR) - const { runId, taskId } = await coordinatorRunAndTask() - const endTurn = await runningUserTurn(chat) - const dispatchId = await finishWorker(taskId) - // The report was accepted, so its dispatch settled before its mail was named. - expect(db.getDispatchContextById(dispatchId)?.status).toBe('completed') - await vi.waitFor( - async () => expect(await queuedCardTexts()).toEqual([ptyPointer(`run:${runId}`)]), - WAIT - ) - expect(chat.turns).toHaveLength(1) - const [card] = queuedRows() - const [mail] = db.getAllMessages(`run:${runId}`) - const from = { - kind: 'agent', - senders: [ - { - party: { address: 'term_worker', terminalHandle: 'term_worker', orcaSessionId: null }, - // Named by the task of the dispatch it just finished, through the real runtime's naming. - name: 'build it' - } - ], - orchestration: { - message: 'mail-notice', - mailbox: `run:${runId}`, - dispatchId: null, - messages: [{ messageId: mail!.id, runId, from: 'term_worker' }] - } - } - // On the card's body: the turn the queue sends carries it, and the provider never sees it. - expect(card?.body.from).toEqual(from) - - await endTurn() - await vi.waitFor(() => expect(chat.turns).toHaveLength(2), WAIT) - expect(turnText(chat.turns[1]!)).toBe(ptyPointer(`run:${runId}`)) - expect(JSON.stringify(chat.turns[1])).not.toContain('term_worker') - const sent = (await host.journalSnapshot(COORDINATOR)).items.filter( - (item) => item.body.kind === 'message' && item.body.from - ) - expect(sent.map((item) => item.body.kind === 'message' && item.body.from)).toEqual([from]) - expect(await queuedCardTexts()).toEqual([]) - await settleTurn(COORDINATOR, 1) - await idleEdgesSettled() - expect(chat.turns).toHaveLength(2) - }) - - it('queues a second card for mail that arrives while the first waits, each counting its own mail', async () => { - observationClock.start() - const chat = await openChat(COORDINATOR) - const { runId, taskId } = await coordinatorRunAndTask() - const second = await secondTask() - const endTurn = await runningUserTurn(chat) - await finishWorker(taskId) - await vi.waitFor(async () => expect(await queuedCardTexts()).toHaveLength(1), WAIT) - await finishWorker(second, { handle: 'term_worker_2', paneKey: WORKER_2_PANE }) - const pointer = ptyPointer(`run:${runId}`) - await vi.waitFor(async () => expect(await queuedCardTexts()).toEqual([pointer, pointer]), WAIT) - - await endTurn() - await vi.waitFor(() => expect(chat.turns).toHaveLength(2), WAIT) - await settleTurn(COORDINATOR, 1) - await vi.waitFor(() => expect(chat.turns).toHaveLength(3), WAIT) - expect(turnText(chat.turns[2]!)).toBe(pointer) - await settleTurn(COORDINATOR, 2) - await idleEdgesSettled() - expect(chat.turns).toHaveLength(3) - expect(await queuedCardTexts()).toEqual([]) - }) - - it("leaves the chat's own `check` as it is: the mail stays readable, and the card stays", async () => { - const chat = await openChat(COORDINATOR) - const { runId, taskId } = await coordinatorRunAndTask() - const endTurn = await runningUserTurn(chat) - await finishWorker(taskId) - await vi.waitFor(async () => expect(await queuedCardTexts()).toHaveLength(1), WAIT) - const [mail] = db.getAllMessages(`run:${runId}`) - expect(await call('orchestration.check', {}, { sessionId: COORDINATOR })).toMatchObject({ - count: 1, - messages: [{ id: mail!.id }] - }) - expect(await queuedCardTexts()).toEqual([ptyPointer(`run:${runId}`)]) - await endTurn() - }) - - // The queue once refused every send past twenty cards, so a busy coordinator's mail locked the - // person out of their own chat. - it("still queues the person's message behind more than twenty mail cards", async () => { - const chat = await openChat(COORDINATOR) - const { runId, taskId } = await coordinatorRunAndTask() - await runningUserTurn(chat) - await finishWorker(taskId) - await vi.waitFor(async () => expect(await queuedCardTexts()).toHaveLength(1), WAIT) - for (let index = 1; index <= 24; index += 1) { - await call('orchestration.send', { - from: 'term_worker', - to: `run:${runId}`, - subject: `mail ${index}` - }) - await vi.waitFor(async () => expect(await queuedCardTexts()).toHaveLength(index + 1), WAIT) - } - const sent = await host.send( - { callerKey: 'test-surface' }, - { - ...structuredAgentSessionMessageSendMutation({ - sessionId: COORDINATOR, - clientOperationId: operationId(), - expectedRuntimeFence: host.deps.store.getRecord(COORDINATOR)!.lease.runtimeFence, - body: { - kind: 'message', - role: 'user', - blocks: [{ type: 'text', text: 'human message' }] - }, - delivery: 'queue-if-active' - }), - userSend: true - } - ) - expect(sent).toMatchObject({ ok: true, value: { queued: { position: 26, state: 'waiting' } } }) - expect((await queuedCardTexts()).at(-1)).toBe('human message') - }) -}) diff --git a/src/main/runtime/structured-chat-coordinator-mail.test.ts b/src/main/runtime/structured-chat-coordinator-mail.test.ts index 411229a05707..f67e1c53a3b4 100644 --- a/src/main/runtime/structured-chat-coordinator-mail.test.ts +++ b/src/main/runtime/structured-chat-coordinator-mail.test.ts @@ -410,10 +410,10 @@ describe('a worker result reaches the structured chat that coordinates it', () = } }) - it("holds mail a refused turn left in doubt, then queues the next pointer behind it as the person's message would wait", async () => { + it('points mail a refused turn left in doubt again with the next result, counting both', async () => { // A failed turn/start cannot prove the turn never started, so the host records it `unknown` - // and a resend under its id replays that. A live doubt counts as work still owed, so the next - // result's pointer waits in the chat's queue, as a message the person sent then would. + // and a resend under its id replays that. The next result's pointer does not wait behind + // that doubt: it names both messages, and the chat's own `check` reads them. const chat = await openChat(COORDINATOR) const { runId, taskId } = await coordinatorRunAndTask() const second = await call( @@ -429,14 +429,11 @@ describe('a worker result reaches the structured chat that coordinates it', () = expect(chat.turns).toHaveLength(0) await finishWorker(idOf(second.task), { handle: 'term_worker_2', paneKey: WORKER_2_PANE }) - await vi.waitFor( - async () => - expect(await queuedCardTexts()).toEqual([ - formatMessagePointer(2, `run:${runId}`, localOrchestrationCliCommand()).trim() - ]), - WAIT + await vi.waitFor(() => expect(chat.turns).toHaveLength(1), WAIT) + expect(turnText(chat.turns[0]!)).toBe( + formatMessagePointer(2, `run:${runId}`, localOrchestrationCliCommand()).trim() ) - expect(chat.turns).toHaveLength(0) + expect(await queuedCardTexts()).toEqual([]) expect(codex.connections.length).toBe(before) }) @@ -459,53 +456,45 @@ describe('a worker result reaches the structured chat that coordinates it', () = expect(await userTexts(COORDINATOR)).toEqual([expect.stringMatching(POINTER)]) }) - it('points mail at the idle edge when it arrived mid-turn', async () => { + it('steers mail that arrives mid-turn into the running turn, with who it is from', async () => { const chat = await openChat(COORDINATOR) const { runId, taskId } = await coordinatorRunAndTask() - const first = await host.send( - { callerKey: 'test-surface' }, - { - envelope: { - sessionId: COORDINATOR, - clientOperationId: operationId(), - expectedRuntimeFence: host.deps.store.getRecord(COORDINATOR)!.lease.runtimeFence, - payloadFingerprint: computeAgentSessionPayloadFingerprint({ - method: 'agentSession.send', - sessionId: COORDINATOR, - fields: { - body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'go' }] } - } - }) - }, - body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'go' }] } - } - ) - expect(first).toMatchObject({ ok: true }) - // Accepted, then delivered: the provider sees the turn once the host hands it over. + expect(await sendUserMessage(COORDINATOR, 'go')).toMatchObject({ ok: true }) await vi.waitFor(() => expect(chat.turns).toHaveLength(1), WAIT) const notify = (method: string, params: unknown) => chat.handlers.onNotification?.(method, params) + const echo = (id: string, clientId: string, text: string) => + notify('item/completed', { + item: { type: 'userMessage', id, clientId, content: [{ type: 'text', text }] } + }) notify('turn/started', { turn: { id: 'turn-1' } }) - notify('item/completed', { - item: { - type: 'userMessage', - id: 'echo-go', - clientId: chat.turns[0]!.clientUserMessageId, - content: [{ type: 'text', text: 'go' }] - } - }) + echo('echo-go', chat.turns[0]!.clientUserMessageId, 'go') await host.flushStreamedEvents(COORDINATOR) await finishWorker(taskId) - await new Promise((resolve) => setTimeout(resolve, 20)) - // The coordinator is mid-turn, so nothing is folded into that turn. - expect(chat.turns).toHaveLength(1) + // Into the turn that is running, as a terminal agent takes its typed nudge: no card, no + // second turn. + await vi.waitFor(() => expect(chat.steers).toHaveLength(1), WAIT) + const [steer] = chat.steers + expect(steer!.expectedTurnId).toBe('turn-1') + expect(steer!.text).toContain(ptyPointer(`run:${runId}`)) + expect(await queuedCardTexts()).toEqual([]) + const sent = (await host.journalSnapshot(COORDINATOR)).items.find( + (item) => item.body.kind === 'message' && item.body.from + ) + expect(sent?.body).toMatchObject({ + from: { kind: 'agent', orchestration: { message: 'mail-notice', mailbox: `run:${runId}` } } + }) + + echo('echo-pointer', steer!.clientUserMessageId, ptyPointer(`run:${runId}`)) notify('turn/completed', { turn: { id: 'turn-1' } }) await host.flushStreamedEvents(COORDINATOR) - await vi.waitFor(() => expect(chat.turns).toHaveLength(2), WAIT) - expect(chat.turns[1]!.text).toMatch(POINTER) - expect(chat.turns[1]!.text).toContain(runId) + // The idle edge finds the mail delivered: nothing is resent. + runtime.onStructuredSessionStatusForMail({ sessionId: COORDINATOR, status: 'idle' }) + await new Promise((resolve) => setTimeout(resolve, 200)) + expect(chat.turns).toHaveLength(1) + expect(chat.steers).toHaveLength(1) }) }) diff --git a/src/renderer/src/components/native-chat/structured-agent-session-send-runtime.ts b/src/renderer/src/components/native-chat/structured-agent-session-send-runtime.ts index e15628df6c0b..889a4c848d67 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-send-runtime.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-send-runtime.ts @@ -104,7 +104,9 @@ export function settleRecorded( finish( entry, 'recorded', - keep ? { phase: 'recorded', issued: true, ...(cursor ? { recordedCursor: cursor } : {}) } : undefined + keep + ? { phase: 'recorded', issued: true, ...(cursor ? { recordedCursor: cursor } : {}) } + : undefined ) } diff --git a/src/renderer/src/components/right-sidebar/AiVaultScanProgressStrip.test.tsx b/src/renderer/src/components/right-sidebar/AiVaultScanProgressStrip.test.tsx index 3f3976031a45..2cb42ca09076 100644 --- a/src/renderer/src/components/right-sidebar/AiVaultScanProgressStrip.test.tsx +++ b/src/renderer/src/components/right-sidebar/AiVaultScanProgressStrip.test.tsx @@ -35,9 +35,7 @@ const FINISHED: AiVaultScanProgress = { it('says the finished total in the strip and opens to one row per agent', async () => { render() - expect(screen.getByText('102,452 files').parentElement?.textContent).toBe( - 'Scanned 102,452 files' - ) + expect(screen.getByText('102,452 files').parentElement?.textContent).toBe('Scanned 102,452 files') expect(screen.queryByText('Codex')).toBeNull() await userEvent.setup().click(screen.getByRole('button')) @@ -120,7 +118,12 @@ it('shows a running scan at once but a finished one only with the list it produc act(() => push({ scanKey: 'scope-a', progress: FINISHED })) expect(aView.getByText('Scanned', { exact: false })).toBeTruthy() - act(() => push({ scanKey: 'scope-b', progress: { ...FINISHED, scanId: 'scan-b', stage: 'finding', settledAt: null } })) + act(() => + push({ + scanKey: 'scope-b', + progress: { ...FINISHED, scanId: 'scan-b', stage: 'finding', settledAt: null } + }) + ) expect(bView.getByText('Scanning', { exact: false })).toBeTruthy() expect(aView.getByText('Scanned', { exact: false })).toBeTruthy() diff --git a/src/shared/structured-agent-session-send-settlement.ts b/src/shared/structured-agent-session-send-settlement.ts index 41a9877d51c1..7cde699e1ff7 100644 --- a/src/shared/structured-agent-session-send-settlement.ts +++ b/src/shared/structured-agent-session-send-settlement.ts @@ -8,9 +8,9 @@ export function structuredAgentSessionReadCoversSend( ): boolean { return Boolean( cursor && - acknowledged && - cursor.epoch === acknowledged.epoch && - cursor.sequence >= acknowledged.sequence + acknowledged && + cursor.epoch === acknowledged.epoch && + cursor.sequence >= acknowledged.sequence ) }