From 8535efd658b5b65320526c2ebaeba46551dd9203 Mon Sep 17 00:00:00 2001 From: Jinjing <6427696+AmethystLiang@users.noreply.github.com> Date: Fri, 10 Jul 2026 22:36:11 -0700 Subject: [PATCH] Add event-driven lifecycle delivery for coordinator wake messages Coordinators previously had to poll `check --wait` for worker_done, escalation, and decision_gate messages. Push these directly into a working agent's PTY via a bracketed-paste wake submit, with waiter deduplication, in-flight serialization, and safety checks (foreground process, permission-state) to avoid double delivery or unsafe submits. Also persist the dispatching coordinator on dispatch contexts so a worker crash escalates even without an active automated coordinator run, and update the orchestration skill/reliability-gate docs to match. --- config/reliability-gates.jsonc | 36 +- .../orchestration-skill-guidance.test.mjs | 16 +- skills/orchestration/SKILL.md | 20 +- src/main/runtime/orca-runtime.test.ts | 482 ++++++++++++++++++ src/main/runtime/orca-runtime.ts | 263 ++++++++-- src/main/runtime/orchestration/coordinator.ts | 6 +- src/main/runtime/orchestration/db.test.ts | 10 + src/main/runtime/orchestration/db.ts | 23 +- .../orchestration/lifecycle-delivery.ts | 41 ++ src/main/runtime/orchestration/types.ts | 1 + .../runtime/rpc/methods/orchestration.test.ts | 74 ++- src/main/runtime/rpc/methods/orchestration.ts | 21 +- 12 files changed, 923 insertions(+), 70 deletions(-) create mode 100644 src/main/runtime/orchestration/lifecycle-delivery.ts diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index 30940476103..9eed51e3240 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -1793,7 +1793,7 @@ }, { "id": "terminal-input.agent-prompt-injection", - "title": "Orchestration agent prompts arrive as bracketed paste before submit", + "title": "Orchestration agent prompts and lifecycle wakes arrive as bracketed paste before submit", "maturity": "experimental", "protection": "partial", "owner": "terminal-input", @@ -1802,6 +1802,7 @@ "terminal input", "agent prompt injection", "orchestration dispatch", + "orchestration lifecycle delivery", "PTY writes", "bracketed paste" ], @@ -1822,12 +1823,12 @@ "coveredProviders": [ "local" ], - "coverageNotes": "Local macOS evidence covers the runtime PTY write contract and a live dev-runtime CLI repro. SSH, daemon, remote-runtime, Linux, and Windows remain provider/platform gaps; the product path stays provider-owned and does not add local filesystem or process assumptions.", + "coverageNotes": "Local macOS evidence covers the runtime PTY write contract, event-driven lifecycle wake delivery, and a live dev-runtime CLI repro. SSH, daemon, remote-runtime, Linux, and Windows remain provider/platform gaps; both product paths stay provider-owned and add no local filesystem or process assumptions.", "motivatingLinks": [ "https://github.com/stablyai/orca/issues/7226" ], - "invariant": "Injected orchestration task prompts for recognized agent CLIs must send the prompt body inside one bracketed-paste frame, sanitize embedded ESC bytes, preserve chunk boundaries without losing the frame, and send Enter only after the paste frame completes.", - "oracle": "Runtime tests assert the exact PTY write sequence and failure cleanup; orchestration tests assert dispatch/coordinator use the agent prompt path; the live CLI harness dispatches a 32KB task to a fake Codex-like TUI and requires marker present, bracketed paste present, zero unframed line breaks, and submit observed.", + "invariant": "Injected orchestration task prompts and working-coordinator lifecycle wakes for recognized agent CLIs must send the prompt body inside one bracketed-paste frame, sanitize embedded ESC bytes, preserve chunk boundaries without losing the frame, and send Enter only after the paste frame completes. A lifecycle message must choose exactly one delivery path: a matching compatibility waiter, a submitted agent prompt, or a concurrent check read — never more than one, including reads issued during the in-flight claim window. Wake submits verify a live recognized agent (a shell foreground process is definitive negative evidence) and withhold the delayed Enter when a permission state appears mid-submit.", + "oracle": "Runtime tests assert the exact PTY write sequence and failure cleanup; lifecycle tests assert one pushed prompt for a working coordinator, waiter deduplication, serialized concurrent completions, and persisted fallback for unsafe prompt owners; orchestration tests assert dispatch/coordinator use the agent prompt path and classify only worker_done, escalation, and decision_gate as wake messages. The live CLI harness dispatches a 32KB task to a fake Codex-like TUI and requires marker present, bracketed paste present, zero unframed line breaks, and submit observed.", "commands": [ "pnpm exec vitest run --config config/vitest.config.ts src/shared/agent-prompt-injection.test.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/rpc/methods/orchestration.test.ts src/main/runtime/orchestration/coordinator.test.ts", "node tools/repro-orchestration-long-prompt.mjs --cli out/bin/orca-dev --mode codex-like --size-kb 32 --timeout-ms 20000" @@ -1854,7 +1855,18 @@ "assertions": [ "runtime writes bracketed paste before a delayed submit", "large prompt frames are chunked and reconstructed before submit", - "partial prompt write failure closes the paste frame and does not submit" + "partial prompt write failure closes the paste frame and does not submit", + "working coordinators receive lifecycle messages as submitted agent prompts", + "matching compatibility waiters prevent duplicate PTY injection", + "concurrent lifecycle completions serialize across the delayed submit window", + "permission, shell, Cursor, and automated-coordinator prompts retain persisted fallback", + "push-on-idle defers while a wake submit is in flight instead of double-delivering", + "a stale working title with a shell foreground process keeps messages queued", + "a permission prompt appearing in the paste-settle window withholds the Enter", + "pty-backed handles without renderer leaves receive wake delivery", + "wake rows are claimed during submit and released after delivery", + "worker exits escalate to the persisted dispatching coordinator with a wake", + "deliverOrchestrationMessage routes an insert to exactly one delivery path" ] }, { @@ -1862,7 +1874,10 @@ "assertions": [ "orchestration.dispatch uses the agent prompt path for injected preambles", "raw terminal.send is not called for injected task prompts", - "failed prompt injection rolls back the active dispatch" + "failed prompt injection rolls back the active dispatch", + "worker_done, escalation, and decision_gate request coordinator wakes while status and heartbeat do not", + "wake-claimed rows are hidden from concurrent unread checks but not from the audit view", + "dispatch persists the coordinating terminal for worker-exit escalation" ] }, { @@ -1907,15 +1922,15 @@ }, "flakeHistory": { "status": "unknown", - "evidence": "First local macOS evidence only; needs repeated dev-runtime harness runs and provider matrix evidence before promotion." + "evidence": "The existing CLI harness and one manual real-agent lifecycle push passed locally; repeated runs and provider matrix evidence are still required before promotion." }, "redGreenEvidence": { "status": "partial", - "evidence": "The live harness reproduced the unsafe raw multiline contract before the fix and passes after the fix; intentional-break evidence is local only and not yet in CI." + "evidence": "The live harness reproduced the unsafe raw multiline contract before the fix and passes after the fix. The lifecycle regression test reproduced zero PTY writes for a working coordinator before the fix and passes with one framed prompt plus submit; waiter and concurrent-completion branches have local deterministic coverage. A manual isolated dev instance also showed a real working Codex turn receive the worker_done banner without check --wait, with delivered_at stamped in the same second. Intentional-break evidence is local only and not yet in CI." }, "performanceBudget": { "required": true, - "evidence": "Agent prompt dispatch remains O(prompt bytes), uses the existing 16KB terminal input chunking and one existing 500ms submit delay, and adds no polling, provider listing, subprocess churn, hidden-pane wakeups, or renderer work." + "evidence": "Agent prompt dispatch remains O(prompt bytes), uses the existing 16KB terminal input chunking and one existing 500ms submit delay. Lifecycle delivery adds one handle/waiter lookup and one unread-message query on the message-arrival event, serializes overlap with bounded handle sets, and adds no polling, retry loop, provider listing, subprocess churn, hidden-pane wakeups, or renderer work." }, "promotionCriteria": [ "Run the live harness in soak with a self-starting dev runtime or provider-contract fixture.", @@ -1925,7 +1940,8 @@ "knownGaps": [ "Live harness command currently expects an already-running dev runtime and generated out/bin/orca-dev wrapper.", "No Windows ConPTY, Linux PTY, SSH, daemon, or remote-runtime live evidence yet.", - "Push-on-idle orchestration message banners remain outside this dispatch-prompt gate." + "Non-submitting push-on-idle banners remain outside this gate; lifecycle wake coverage applies to working coordinators that use the submitted agent-prompt path.", + "Foreground-process verification before wake submits is best-effort: providers without process inspection fall back to title evidence, so a stale working title there is not blocked." ], "demotionRule": "Demote or quarantine if the live harness flakes without a product bug or harness bug filed to the terminal-input owner." }, diff --git a/config/scripts/orchestration-skill-guidance.test.mjs b/config/scripts/orchestration-skill-guidance.test.mjs index 78a51ad10a7..0c6e42a3667 100644 --- a/config/scripts/orchestration-skill-guidance.test.mjs +++ b/config/scripts/orchestration-skill-guidance.test.mjs @@ -41,15 +41,19 @@ describe('orchestration skill guidance', () => { ) }) - it('treats long-running worker waits as liveness checkpoints, not failures', () => { + it('uses pushed lifecycle delivery instead of coordinator polling', () => { const skill = readSkill() - expect(skill).toContain('Treat a `check --wait` timeout or `{count:0}` as a checkpoint') - expect(skill).toContain('Do not stop, close, kill, or restart a worker') - expect(skill).toContain('keep waiting instead of retrying the task') - expect(skill).not.toContain( - 'If `check --wait` times out with no `worker_done` or `escalation`, fall back to `terminal wait --for tui-idle`, then `terminal read`.' + expect(skill).toContain( + 'Do not call `orca orchestration check --wait` after dispatching supervised work.' ) + expect(skill).toContain( + 'Orca pushes `worker_done`, `escalation`, and `decision_gate` messages into the coordinator agent input' + ) + expect(skill).toContain('end the turn and let Orca re-engage the coordinator') + expect(skill).not.toContain('keep using rolling waits') + expect(skill).not.toContain('During supervision, use rolling `check --wait` windows.') + expect(skill).not.toContain('orca orchestration check --wait --types') }) it('keeps full handoffs out of dispatch lifecycle and off the active branch base', () => { diff --git a/skills/orchestration/SKILL.md b/skills/orchestration/SKILL.md index 90aeb093c09..daad12975e8 100644 --- a/skills/orchestration/SKILL.md +++ b/skills/orchestration/SKILL.md @@ -3,7 +3,7 @@ name: orchestration description: >- Use Orca orchestration for structured multi-agent coordination: threaded messages, blocking ask/reply flows, task dispatch, worker_done/escalation - waits, task DAGs, decision gates, coordinator loops, or decomposing work + delivery, task DAGs, decision gates, coordinator loops, or decomposing work across agents. Use `orca-cli` instead for full ownership handoffs, including requests phrased as "hand off", "handoff", "handover", "give this to another agent", or "another worktree" when the user did not explicitly ask to @@ -38,7 +38,7 @@ If the work was accidentally run outside Orca orchestration, say so plainly. To ## When To Use - Send/reply/ask between agent terminals with persistent messages. -- Dispatch structured tasks to workers and wait for `worker_done` or `escalation`. +- Dispatch structured tasks to workers and react to pushed `worker_done` or `escalation` messages. - Track task DAGs with dependencies. - Run coordinator loops or decision gates. @@ -57,7 +57,7 @@ Orchestration messages and tasks are runtime-global. Completion authority comes Classify inherited context before sending lifecycle messages: -- Coordinated subtask: a live coordinator owns the DAG and waits on this dispatch. Follow the preamble exactly, including `worker_done`, heartbeat/status, `ask`, and `escalation`. +- Coordinated subtask: a live coordinator owns the DAG and expects pushed results from this dispatch. Follow the preamble exactly, including `worker_done`, heartbeat/status, `ask`, and `escalation`. - Full handoff means ownership transfer, not supervised dispatch. The original actor is not monitoring a DAG, so do not create lifecycle obligations unless the user explicitly asks you to supervise. - Classify requests containing "hand off", "handoff", "handover", "give this to another agent", "give this to another worktree", "another agent", or "another worktree" as full handoffs by default, even when the user names a custom model or reasoning effort. - Use supervised orchestration only when the user explicitly asks you to "supervise", "monitor", "wait", "track completion", "wait for worker_done", return results, coordinate a DAG, use a decision gate, or manage ask/reply flow. @@ -88,11 +88,13 @@ orca orchestration inbox [--limit ] [--json] Rules: - Omit `--from` unless impersonating another terminal; Orca auto-resolves it from the current terminal. -- While supervising workers manually, use `check --wait --types worker_done,escalation,decision_gate --timeout-ms ` instead of sleep/poll loops. Reply to `decision_gate` messages with `orca orchestration reply --id --body --json`, then keep waiting. -- Treat a `check --wait` timeout or `{count:0}` as a checkpoint, not a worker failure. Long coding tasks routinely run 15-60 minutes; keep using rolling waits unless you receive `worker_done`/`escalation`, the terminal exits or disappears, or the user explicitly asks you to stop. +- Do not call `orca orchestration check --wait` after dispatching supervised work. +- Orca pushes `worker_done`, `escalation`, and `decision_gate` messages into the coordinator agent input as soon as they arrive. Finish any independent coordinator work already in progress, then end the turn and let Orca re-engage the coordinator. +- Reply to a pushed `decision_gate` with `orca orchestration reply --id --body --json`. Use a non-waiting `check` only for explicit inbox recovery or auditing, not as a supervision loop. - Heartbeats and visible terminal activity mean the worker is alive, not done. Do not stop, close, kill, or restart a worker just because it has not produced a completion message yet. +- If a worker terminal exits while its dispatch is active, Orca fails the dispatch and pushes an `escalation` to the dispatching coordinator, so crashes also arrive without polling. - Use `ask` when a worker needs a blocking answer from the coordinator; it waits for the reply and returns the answer directly. -- `check --wait` returns one message at a time. If N workers may finish together, loop N times and dispatch newly ready tasks after each completion. +- If several workers finish together, Orca queues their pushed messages and submits them to the coordinator without requiring one wait call per worker. - Group addresses include `@all`, `@idle`, `@claude`, `@codex`, `@opencode`, `@gemini`, `@droid`, and `@worktree:`. - Message types include `status`, `dispatch`, `worker_done`, `merge_ready`, `escalation`, `handoff`, `decision_gate`, and `heartbeat`. - Use group addresses only for messages that are genuinely useful to many terminals, such as `status` broadcasts or intentional fan-out questions. Do not send dispatch lifecycle messages to groups. @@ -210,7 +212,7 @@ orca terminal send --terminal --text --enter --json If an older CLI rejects `worktree create --agent`, create the worktree normally, then run `orca terminal create --worktree --command "codex" --json` or `--command "claude"`. -Wait for `tui-idle` before dispatching. Always pass `--timeout-ms`; real coding tasks can take 15-60 minutes. During supervision, use rolling `check --wait` windows. If a window returns no matching message, inspect `task-list`, `terminal read`, or `terminal wait --for tui-idle` as a liveness checkpoint; if the terminal is still working or producing activity, keep waiting instead of retrying the task. +Wait for `tui-idle` before dispatching so the initial prompt is not lost. After dispatch, continue useful coordinator work or end the turn; lifecycle messages wake the coordinator automatically. Use `task-list`, `terminal read`, or a non-waiting `check` only for explicit diagnosis or recovery, not periodic liveness checks. ## Agent Guidance @@ -231,11 +233,11 @@ orca terminal create --worktree active --title login-css-worker --command "claud orca terminal wait --terminal --for tui-idle --timeout-ms 60000 --json orca orchestration task-create --spec "Fix the login button CSS" --json orca orchestration dispatch --task --to --inject --json -orca orchestration check --wait --types worker_done,escalation,decision_gate --timeout-ms 900000 --json +# Continue independent work or end the turn. Orca pushes lifecycle messages here. ``` ## Next Action -Coordinator: confirm `orca status --json`, inspect `task-list`/`dispatch-show` if inheriting state, then choose either a manual loop (`task-create` -> worker -> `dispatch --inject` -> `check --wait`) or `orchestration run`. +Coordinator: confirm `orca status --json`, inspect `task-list`/`dispatch-show` if inheriting state, then choose manual dispatch (`task-create` -> worker -> `dispatch --inject`) or `orchestration run`. Do not add a wait loop after dispatch; react when Orca pushes lifecycle input. Worker: if the current prompt contains a live dispatch preamble, do the task, use `ask` for blocking questions, and send `worker_done` once with the required payload. If the preamble is stale or absent, do not send lifecycle messages; inspect state or treat the prompt as an ordinary handoff. diff --git a/src/main/runtime/orca-runtime.test.ts b/src/main/runtime/orca-runtime.test.ts index 157442e548b..b7a4b29f707 100644 --- a/src/main/runtime/orca-runtime.test.ts +++ b/src/main/runtime/orca-runtime.test.ts @@ -43,6 +43,7 @@ import { } from '../hooks' import { getBranchConflictKind, getDefaultBaseRef } from '../git/repo' import type { OrchestrationDb } from './orchestration/db' +import { deliverOrchestrationMessage } from './orchestration/lifecycle-delivery' import type { MessagePriority, MessageRow, MessageType } from './orchestration/types' import { appendNormalizedToTailBuffer, @@ -867,6 +868,15 @@ function antigravityPromptBeforeModelReadyScreen(model = 'Gemini 3.5 Flash (High // Why: these runtime feature tests only need message-queue semantics; using // SQLite here makes them fail on unrelated native runtime ABI drift. +type InMemoryDispatchContext = { + id: string + task_id: string + assignee_handle: string | null + coordinator_handle: string | null + status: string + last_failure: string | null +} + class InMemoryOrchestrationMessages { private sequence = 0 @@ -927,6 +937,42 @@ class InMemoryOrchestrationMessages { return this.activeCoordinatorRun } + markAsRead(ids: string[]): void { + const readIds = new Set(ids) + for (const message of this.messages) { + if (readIds.has(message.id)) { + message.read = 1 + } + } + } + + private dispatches: InMemoryDispatchContext[] = [] + + setDispatch(dispatch: InMemoryDispatchContext): void { + this.dispatches.push(dispatch) + } + + getDispatch(id: string): InMemoryDispatchContext | undefined { + return this.dispatches.find((dispatch) => dispatch.id === id) + } + + getActiveDispatchForTerminal(handle: string): InMemoryDispatchContext | undefined { + return this.dispatches.find( + (dispatch) => + dispatch.assignee_handle === handle && + (dispatch.status === 'pending' || dispatch.status === 'dispatched') + ) + } + + failDispatch(id: string, error: string): InMemoryDispatchContext | undefined { + const dispatch = this.getDispatch(id) + if (dispatch) { + dispatch.status = 'failed' + dispatch.last_failure = error + } + return dispatch + } + markAsDelivered(ids: string[]): void { const deliveredIds = new Set(ids) for (const message of this.messages) { @@ -11661,6 +11707,20 @@ describe('OrcaRuntimeService', () => { expect(preview.nextCursor).toBe('2000') }) + function setupOrchestrationDeliveryRuntime() { + const runtime = new OrcaRuntimeService(store) + const db = new InMemoryOrchestrationMessages() + const write = vi.fn().mockReturnValue(true) + setInMemoryOrchestrationMessages(runtime, db) + runtime.setPtyController({ + write, + kill: vi.fn(), + getForegroundProcess: async () => null + }) + syncSinglePty(runtime) + return { runtime, db, write } + } + it('delivers pending orchestration messages to an already-idle agent', async () => { vi.useFakeTimers() try { @@ -11704,6 +11764,428 @@ describe('OrcaRuntimeService', () => { } }) + it('pushes a lifecycle message directly to a working coordinator agent', async () => { + vi.useFakeTimers() + try { + const { runtime, db, write } = setupOrchestrationDeliveryRuntime() + + const [terminal] = (await runtime.listTerminals()).terminals + runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100) + db.insertMessage({ + from: 'term_worker', + to: terminal.handle, + subject: 'implementation complete', + type: 'worker_done' + }) + + runtime.deliverPendingMessagesForHandle(terminal.handle, { + wakeAgent: true, + messageType: 'worker_done' + }) + + await vi.advanceTimersByTimeAsync(0) + expect(write).toHaveBeenCalledTimes(1) + expect(write).toHaveBeenCalledWith( + 'pty-1', + expect.stringContaining('Subject: implementation complete') + ) + await vi.advanceTimersByTimeAsync(500) + expect(write).toHaveBeenCalledTimes(2) + expect(write).toHaveBeenLastCalledWith('pty-1', '\r') + expect(db.getUnreadMessages(terminal.handle)[0].delivered_at).not.toBeNull() + db.close() + } finally { + vi.useRealTimers() + } + }) + + it('lets a matching legacy waiter consume a lifecycle message without PTY injection', async () => { + vi.useFakeTimers() + try { + const { runtime, db, write } = setupOrchestrationDeliveryRuntime() + + const [terminal] = (await runtime.listTerminals()).terminals + runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100) + const waiter = runtime.waitForMessage(terminal.handle, { + typeFilter: ['worker_done'], + timeoutMs: 60_000 + }) + db.insertMessage({ + from: 'term_worker', + to: terminal.handle, + subject: 'legacy completion', + type: 'worker_done' + }) + + runtime.deliverPendingMessagesForHandle(terminal.handle, { + wakeAgent: true, + messageType: 'worker_done' + }) + runtime.notifyMessageArrived(terminal.handle, 'worker_done') + await waiter + + expect(write).not.toHaveBeenCalled() + expect(db.getUnreadMessages(terminal.handle)[0].delivered_at).toBeNull() + db.close() + } finally { + vi.useRealTimers() + } + }) + + it('serializes lifecycle messages that arrive during agent prompt submission', async () => { + vi.useFakeTimers() + try { + const { runtime, db, write } = setupOrchestrationDeliveryRuntime() + + const [terminal] = (await runtime.listTerminals()).terminals + runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100) + db.insertMessage({ + from: 'term_worker_1', + to: terminal.handle, + subject: 'first completion', + type: 'worker_done' + }) + runtime.deliverPendingMessagesForHandle(terminal.handle, { + wakeAgent: true, + messageType: 'worker_done' + }) + await vi.advanceTimersByTimeAsync(0) + + db.insertMessage({ + from: 'term_worker_2', + to: terminal.handle, + subject: 'second completion', + type: 'worker_done' + }) + runtime.deliverPendingMessagesForHandle(terminal.handle, { + wakeAgent: true, + messageType: 'worker_done' + }) + + expect(write.mock.calls.filter(([, text]) => text.includes('first completion'))).toHaveLength( + 1 + ) + expect( + write.mock.calls.filter(([, text]) => text.includes('second completion')) + ).toHaveLength(0) + + await vi.advanceTimersByTimeAsync(500) + expect( + write.mock.calls.filter(([, text]) => text.includes('second completion')) + ).toHaveLength(1) + await vi.advanceTimersByTimeAsync(500) + expect(write.mock.calls.filter(([, text]) => text === '\r')).toHaveLength(2) + expect(db.getUnreadMessages(terminal.handle).every((message) => message.delivered_at)).toBe( + true + ) + db.close() + } finally { + vi.useRealTimers() + } + }) + + it.each([ + ['permission prompt', 'Codex waiting for permission', false], + ['bare shell', 'zsh', false], + ['Cursor Agent', '\u280b Cursor Agent', false], + ['automated coordinator', 'Codex working', true] + ])( + 'keeps lifecycle messages queued when the recipient is a %s', + async (_name, title, activeRun) => { + vi.useFakeTimers() + try { + const { runtime, db, write } = setupOrchestrationDeliveryRuntime() + + const [terminal] = (await runtime.listTerminals()).terminals + runtime.onPtyData('pty-1', `\x1b]0;${title}\x07`, 100) + if (activeRun) { + db.setActiveCoordinatorRun({ coordinator_handle: terminal.handle }) + } + db.insertMessage({ + from: 'term_worker', + to: terminal.handle, + subject: 'queued completion', + type: 'worker_done' + }) + + runtime.deliverPendingMessagesForHandle(terminal.handle, { + wakeAgent: true, + messageType: 'worker_done' + }) + await vi.advanceTimersByTimeAsync(500) + + expect(write).not.toHaveBeenCalled() + expect(db.getUnreadMessages(terminal.handle)[0].delivered_at).toBeNull() + db.close() + } finally { + vi.useRealTimers() + } + } + ) + + it('defers push-on-idle while a wake submit is in flight instead of double-delivering', async () => { + vi.useFakeTimers() + try { + const { runtime, db, write } = setupOrchestrationDeliveryRuntime() + + const [terminal] = (await runtime.listTerminals()).terminals + runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100) + db.insertMessage({ + from: 'term_worker', + to: terminal.handle, + subject: 'finished mid-turn', + type: 'worker_done' + }) + runtime.deliverPendingMessagesForHandle(terminal.handle, { + wakeAgent: true, + messageType: 'worker_done' + }) + await vi.advanceTimersByTimeAsync(0) + expect( + write.mock.calls.filter(([, text]) => text.includes('finished mid-turn')) + ).toHaveLength(1) + + // The coordinator's turn ends inside the paste-settle window. The idle + // transition must defer to the in-flight submit, not re-read the + // still-unstamped row and paste it a second time. + runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 200) + await vi.advanceTimersByTimeAsync(500) + + expect( + write.mock.calls.filter(([, text]) => text.includes('finished mid-turn')) + ).toHaveLength(1) + expect(write.mock.calls.filter(([, text]) => text === '\r')).toHaveLength(1) + expect(db.getUnreadMessages(terminal.handle)[0].delivered_at).not.toBeNull() + db.close() + } finally { + vi.useRealTimers() + } + }) + + it('keeps lifecycle messages queued when a stale working title hides a shell foreground', async () => { + vi.useFakeTimers() + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + try { + const runtime = new OrcaRuntimeService(store) + const db = new InMemoryOrchestrationMessages() + const write = vi.fn().mockReturnValue(true) + setInMemoryOrchestrationMessages(runtime, db) + runtime.setPtyController({ + write, + kill: vi.fn(), + // The agent crashed and the shell took over without repainting the + // OSC title, so lastAgentStatus is a stale 'working'. + getForegroundProcess: async () => '/bin/zsh' + }) + syncSinglePty(runtime) + + const [terminal] = (await runtime.listTerminals()).terminals + runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100) + db.insertMessage({ + from: 'term_worker', + to: terminal.handle, + subject: 'shell must not run this', + type: 'worker_done' + }) + + runtime.deliverPendingMessagesForHandle(terminal.handle, { + wakeAgent: true, + messageType: 'worker_done' + }) + await vi.advanceTimersByTimeAsync(500) + + expect(write).not.toHaveBeenCalled() + expect(db.getUnreadMessages(terminal.handle)[0].delivered_at).toBeNull() + db.close() + } finally { + warn.mockRestore() + vi.useRealTimers() + } + }) + + it('withholds the wake Enter when a permission prompt appears in the paste-settle window', async () => { + vi.useFakeTimers() + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + try { + const { runtime, db, write } = setupOrchestrationDeliveryRuntime() + + const [terminal] = (await runtime.listTerminals()).terminals + runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100) + db.insertMessage({ + from: 'term_worker', + to: terminal.handle, + subject: 'racing a permission prompt', + type: 'worker_done' + }) + runtime.deliverPendingMessagesForHandle(terminal.handle, { + wakeAgent: true, + messageType: 'worker_done' + }) + await vi.advanceTimersByTimeAsync(0) + expect( + write.mock.calls.filter(([, text]) => text.includes('racing a permission prompt')) + ).toHaveLength(1) + + // The agent raises a permission dialog before the delayed Enter fires; + // sending \r now could approve the pending request. + runtime.onPtyData('pty-1', '\x1b]0;Codex waiting for permission\x07', 200) + await vi.advanceTimersByTimeAsync(500) + + expect(write.mock.calls.filter(([, text]) => text === '\r')).toHaveLength(0) + expect(db.getUnreadMessages(terminal.handle)[0].delivered_at).toBeNull() + db.close() + } finally { + warn.mockRestore() + vi.useRealTimers() + } + }) + + it('claims wake rows during submit and releases them after delivery', async () => { + vi.useFakeTimers() + try { + const { runtime, db } = setupOrchestrationDeliveryRuntime() + + const [terminal] = (await runtime.listTerminals()).terminals + runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100) + const message = db.insertMessage({ + from: 'term_worker', + to: terminal.handle, + subject: 'claimed while submitting', + type: 'worker_done' + }) + runtime.deliverPendingMessagesForHandle(terminal.handle, { + wakeAgent: true, + messageType: 'worker_done' + }) + await vi.advanceTimersByTimeAsync(0) + + expect(runtime.isMessageClaimedForWakeDelivery(message.id)).toBe(true) + await vi.advanceTimersByTimeAsync(500) + expect(runtime.isMessageClaimedForWakeDelivery(message.id)).toBe(false) + expect(db.getUnreadMessages(terminal.handle)[0].delivered_at).not.toBeNull() + db.close() + } finally { + vi.useRealTimers() + } + }) + + it('pushes lifecycle messages to a pty-backed handle with no renderer leaf', async () => { + vi.useFakeTimers() + try { + const { runtime, db, write } = setupOrchestrationDeliveryRuntime() + runtime.registerPty('pty-9', TEST_WORKTREE_ID) + runtime.onPtyData('pty-9', '\x1b]0;Codex working\x07', 100) + + const { terminals } = await runtime.listTerminals() + const ptyTerminal = terminals.find((terminal) => terminal.tabId === 'pty:pty-9') + expect(ptyTerminal).toBeDefined() + + db.insertMessage({ + from: 'term_worker', + to: ptyTerminal!.handle, + subject: 'headless coordinator wake', + type: 'worker_done' + }) + runtime.deliverPendingMessagesForHandle(ptyTerminal!.handle, { + wakeAgent: true, + messageType: 'worker_done' + }) + await vi.advanceTimersByTimeAsync(0) + expect( + write.mock.calls.filter( + ([ptyId, text]) => ptyId === 'pty-9' && text.includes('headless coordinator wake') + ) + ).toHaveLength(1) + await vi.advanceTimersByTimeAsync(500) + expect( + write.mock.calls.filter(([ptyId, text]) => ptyId === 'pty-9' && text === '\r') + ).toHaveLength(1) + expect(db.getUnreadMessages(ptyTerminal!.handle)[0].delivered_at).not.toBeNull() + db.close() + } finally { + vi.useRealTimers() + } + }) + + it('escalates a worker exit to the persisted dispatching coordinator', async () => { + const { runtime, db } = setupOrchestrationDeliveryRuntime() + + const [terminal] = (await runtime.listTerminals()).terminals + db.setDispatch({ + id: 'ctx_test', + task_id: 'task_test', + assignee_handle: terminal.handle, + coordinator_handle: 'term_manual_coordinator', + status: 'dispatched', + last_failure: null + }) + const deliver = vi + .spyOn(runtime, 'deliverPendingMessagesForHandle') + .mockImplementation(() => {}) + const notify = vi.spyOn(runtime, 'notifyMessageArrived').mockImplementation(() => {}) + + runtime.onPtyExit('pty-1', 137) + + expect(db.getDispatch('ctx_test')?.status).toBe('failed') + const [escalation] = db.getUnreadMessages('term_manual_coordinator') + expect(escalation).toBeDefined() + expect(escalation.type).toBe('escalation') + expect(JSON.parse(escalation.payload ?? '{}')).toMatchObject({ + taskId: 'task_test', + dispatchId: 'ctx_test', + exitCode: 137 + }) + expect(deliver).toHaveBeenCalledWith('term_manual_coordinator', { + wakeAgent: true, + messageType: 'escalation' + }) + expect(notify).toHaveBeenCalledWith('term_manual_coordinator', 'escalation') + db.close() + }) + + it('routes an insert through deliverOrchestrationMessage to exactly one delivery path', async () => { + vi.useFakeTimers() + try { + const { runtime, db, write } = setupOrchestrationDeliveryRuntime() + + const [terminal] = (await runtime.listTerminals()).terminals + runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100) + + // A blocked matching waiter owns delivery: no PTY injection. + const waiter = runtime.waitForMessage(terminal.handle, { + typeFilter: ['worker_done'], + timeoutMs: 60_000 + }) + const first = db.insertMessage({ + from: 'term_worker_1', + to: terminal.handle, + subject: 'first done', + type: 'worker_done' + }) + deliverOrchestrationMessage(runtime, first) + await waiter + await vi.advanceTimersByTimeAsync(500) + expect(write).not.toHaveBeenCalled() + db.markAsRead([first.id]) + + // No waiter: exactly one pushed prompt plus one Enter. + const second = db.insertMessage({ + from: 'term_worker_2', + to: terminal.handle, + subject: 'second done', + type: 'worker_done' + }) + deliverOrchestrationMessage(runtime, second) + await vi.advanceTimersByTimeAsync(0) + expect(write.mock.calls.filter(([, text]) => text.includes('second done'))).toHaveLength(1) + await vi.advanceTimersByTimeAsync(500) + expect(write.mock.calls.filter(([, text]) => text === '\r')).toHaveLength(1) + db.close() + } finally { + vi.useRealTimers() + } + }) + it('injects pending orchestration messages into the active coordinator without auto-submitting', async () => { vi.useFakeTimers() try { diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 325c72c362e..7f8c3528adb 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -56,8 +56,12 @@ import { isAbsolute, join, resolve } from 'node:path' import { mkdir, readFile, readdir, rm, stat } from 'node:fs/promises' import { resolveWorktreeCreateBase } from '../worktree-create-base' import { resolveWorktreeAddBaseRef } from '../../shared/worktree-base-ref' -import { OrchestrationDb } from './orchestration/db' +import { OrchestrationDb, type MessageRow, type MessageType } from './orchestration/db' import { formatMessagesForInjection } from './orchestration/formatter' +import { + COORDINATOR_WAKE_MESSAGE_TYPES, + deliverOrchestrationMessage +} from './orchestration/lifecycle-delivery' import type { Automation, AutomationCreateInput, @@ -1362,6 +1366,15 @@ type MessageWaiter = { abortCleanup: (() => void) | null } +function messageWaiterMatchesType(waiter: MessageWaiter, messageType?: string): boolean { + return !messageType || !waiter.typeFilter || waiter.typeFilter.includes(messageType) +} + +type PendingMessageDeliveryOptions = { + wakeAgent?: boolean + messageType?: MessageType +} + function omitUndefinedProperties>(value: T): Partial { return Object.fromEntries( Object.entries(value).filter(([, entry]) => entry !== undefined) @@ -2106,6 +2119,11 @@ export class OrcaRuntimeService { private ptyDelayedForegroundSnapshotTitleObservations = new Map() private _orchestrationDb: OrchestrationDb | null = null private messageWaitersByHandle = new Map>() + // Why: agent prompt submission has a short paste-settle delay. Serialize + // concurrent completions so one message cannot be injected twice in that gap. + private orchestrationWakeDeliveryInFlightHandles = new Set() + private orchestrationWakeDeliveryPendingHandles = new Set() + private orchestrationWakeDeliveryClaimedMessageIds = new Set() // Why: mobile clients subscribe to terminal output via terminal.subscribe. // These listeners fire on every onPtyData call, enabling real-time streaming // without polling. Keyed by ptyId for O(1) lookup per data event. @@ -8488,24 +8506,31 @@ export class OrcaRuntimeService { const errorContext = `Agent exited with code ${exitCode}` this._orchestrationDb.failDispatch(dispatch.id, errorContext) - // Why: create an escalation message so the coordinator is notified about - // the unexpected exit on its next check cycle, even if the circuit breaker - // hasn't tripped yet. - const run = this._orchestrationDb.getActiveCoordinatorRun() - if (run) { - this._orchestrationDb.insertMessage({ - from: handle, - to: run.coordinator_handle, - subject: `Agent exited unexpectedly (code ${exitCode})`, - type: 'escalation', - priority: 'high', - payload: JSON.stringify({ - taskId: dispatch.task_id, - exitCode, - handle - }) - }) + // Why: a crashed worker never sends worker_done, and event-driven + // coordinators no longer poll — the exit escalation is the only signal + // they get. Route it to whoever dispatched this context (the persisted + // handle covers manual coordinators; the active run covers automated + // ones and pre-migration rows) and push it like any lifecycle message. + const coordinatorHandle = + dispatch.coordinator_handle ?? + this._orchestrationDb.getActiveCoordinatorRun()?.coordinator_handle + if (!coordinatorHandle) { + return } + const escalation = this._orchestrationDb.insertMessage({ + from: handle, + to: coordinatorHandle, + subject: `Agent exited unexpectedly (code ${exitCode})`, + type: 'escalation', + priority: 'high', + payload: JSON.stringify({ + taskId: dispatch.task_id, + dispatchId: dispatch.id, + exitCode, + handle + }) + }) + deliverOrchestrationMessage(this, escalation) } async listTerminals( @@ -20016,18 +20041,79 @@ export class OrcaRuntimeService { return this.getLeavesForPty(ptyId)[0] ?? null } - deliverPendingMessagesForHandle(handle: string): void { + deliverPendingMessagesForHandle( + handle: string, + options: PendingMessageDeliveryOptions = {} + ): void { + if (options.wakeAgent && this.hasMatchingMessageWaiter(handle, options.messageType)) { + return + } try { const { leaf } = this.getLiveLeafForHandle(handle) if (leaf.lastAgentStatus === 'idle') { this.deliverPendingMessages(leaf) + } else if (options.wakeAgent && leaf.lastAgentStatus === 'working') { + this.deliverPendingMessages(leaf, { wakeAgent: true }) } + return + } catch { + // Renderer-graph leaf resolution fails for daemon/headless terminals; + // fall through to the runtime-owned PTY handle route below. + } + if (!options.wakeAgent) { + // Non-actionable messages never steer a terminal; the persisted row + // remains available via explicit check or future idle delivery. + return + } + try { + this.deliverPendingWakeMessagesForPtyHandle(handle) } catch { // Unknown or stale handles cannot be pushed immediately; the persisted // message remains available via explicit check or future idle delivery. } } + // Why: daemon/headless coordinators have no renderer-graph leaf, but their + // PTY handles are fully writable through sendTerminalAgentPrompt. Without + // this route those coordinators would follow the no-polling guidance, end + // their turn, and never be re-engaged. + private deliverPendingWakeMessagesForPtyHandle(handle: string): void { + const db = this._orchestrationDb + if (!db) { + return + } + const pty = this.getLivePtyForHandle(handle)?.pty + if (!pty || !pty.connected) { + return + } + if (pty.lastAgentStatus !== 'working' && pty.lastAgentStatus !== 'idle') { + return + } + if (this.orchestrationWakeDeliveryInFlightHandles.has(handle)) { + this.orchestrationWakeDeliveryPendingHandles.add(handle) + return + } + if (db.getActiveCoordinatorRun()?.coordinator_handle === handle) { + return + } + if ([pty.lastOscTitle, pty.title].some(isCursorAgentTitle)) { + return + } + const unread = db.getUndeliveredUnreadMessages(handle, COORDINATOR_WAKE_MESSAGE_TYPES) + if (unread.length === 0) { + return + } + this.submitWakeMessagesToAgent(handle, unread) + } + + private hasMatchingMessageWaiter(handle: string, messageType?: MessageType): boolean { + const waiters = this.messageWaitersByHandle.get(handle) + if (!waiters || waiters.size === 0) { + return false + } + return [...waiters].some((waiter) => messageWaiterMatchesType(waiter, messageType)) + } + // Why: after a message is inserted for a recipient, any blocking // orchestration.check --wait calls watching that handle must be woken // so they can return the new message immediately instead of polling. @@ -20039,7 +20125,7 @@ export class OrcaRuntimeService { for (const waiter of [...waiters]) { // Why: a coordinator waiting for worker_done/escalation should not be // woken by worker heartbeat noise and mistake that empty read for idleness. - if (messageType && waiter.typeFilter && !waiter.typeFilter.includes(messageType)) { + if (!messageWaiterMatchesType(waiter, messageType)) { continue } this.resolveMessageWaiter(waiter) @@ -20585,11 +20671,12 @@ export class OrcaRuntimeService { return null } - // Why: push-on-idle delivery — when an agent transitions working→idle, check - // for unread orchestration messages addressed to that terminal and inject them - // into the PTY. This is event-driven (no polling) because the runtime owns - // both the message store and terminal status detection. - private deliverPendingMessages(leaf: RuntimeLeafRecord): void { + // Why: the runtime owns message insertion and terminal status events, so both + // idle delivery and actionable working-agent wakes can stay event-driven. + private deliverPendingMessages( + leaf: RuntimeLeafRecord, + options: { wakeAgent?: boolean } = {} + ): void { if (!this._orchestrationDb) { return } @@ -20599,7 +20686,18 @@ export class OrcaRuntimeService { return } - const unread = this._orchestrationDb.getUndeliveredUnreadMessages(handle) + // Why: a wake submit may be mid paste-settle for this handle. Any delivery + // that reads the store now (including push-on-idle after a working→idle + // transition) would replay the not-yet-stamped rows; defer to the drain. + if (this.orchestrationWakeDeliveryInFlightHandles.has(handle)) { + this.orchestrationWakeDeliveryPendingHandles.add(handle) + return + } + + const unread = this._orchestrationDb.getUndeliveredUnreadMessages( + handle, + options.wakeAgent ? COORDINATOR_WAKE_MESSAGE_TYPES : undefined + ) if (unread.length === 0) { return } @@ -20608,6 +20706,21 @@ export class OrcaRuntimeService { return } + const tabTitle = this.tabs.get(leaf.tabId)?.title + const activeCoordinatorOwnsPrompt = + this._orchestrationDb.getActiveCoordinatorRun()?.coordinator_handle === handle + const cursorOwnsPrompt = isCursorAgentOrchestrationTarget(leaf, tabTitle) + if (options.wakeAgent && (activeCoordinatorOwnsPrompt || cursorOwnsPrompt)) { + // Working prompts owned by an automated coordinator or Cursor stay + // untouched; their persisted inbox is the safe delivery fallback. + return + } + + if (options.wakeAgent) { + this.submitWakeMessagesToAgent(handle, unread) + return + } + const payload = formatMessagesForInjection(unread) const wrote = this.ptyController?.write(leaf.ptyId, payload) ?? false if (!wrote) { @@ -20615,13 +20728,12 @@ export class OrcaRuntimeService { } // The active coordinator prompt is user-owned input, so push-on-idle must not synthesize Enter. - if (this._orchestrationDb.getActiveCoordinatorRun()?.coordinator_handle === handle) { + if (activeCoordinatorOwnsPrompt) { this._orchestrationDb.markAsDelivered(unread.map((m) => m.id)) return } - const tabTitle = this.tabs.get(leaf.tabId)?.title - if (isCursorAgentOrchestrationTarget(leaf, tabTitle)) { + if (cursorOwnsPrompt) { // Why: Cursor Agent treats injected PTY text as editable prompt input. // Push-on-idle may surface the message, but submitting it must stay // under user control. @@ -20658,6 +20770,99 @@ export class OrcaRuntimeService { }, 500) } + // Why: `lastAgentStatus` mirrors the newest observed title, which survives an + // agent crash in shells that never repaint the OSC title (#1437). A live + // shell foreground process is definitive negative evidence, and a permission + // state means the delayed Enter could approve a pending request — both must + // abort the submit. Also used as the beforeWrite re-check so the Enter is + // withheld when the state changes inside the paste-settle window. + private async assertWakeDeliveryTargetSafe(handle: string): Promise { + const status = await this.getTerminalAgentStatus(handle) + if (!status.isRunningAgent || status.status === 'permission') { + throw new Error('wake_target_not_safe_agent') + } + if (await this.terminalHasShellForegroundProcess(handle)) { + throw new Error('wake_target_foreground_shell') + } + } + + private submitWakeMessagesToAgent(handle: string, unread: MessageRow[]): void { + const payload = formatMessagesForInjection(unread) + const messageIds = unread.map((message) => message.id) + this.orchestrationWakeDeliveryInFlightHandles.add(handle) + for (const id of messageIds) { + this.orchestrationWakeDeliveryClaimedMessageIds.add(id) + } + void this.assertWakeDeliveryTargetSafe(handle) + .then(() => + this.sendTerminalAgentPrompt(handle, payload, { + beforeWrite: () => this.assertWakeDeliveryTargetSafe(handle) + }) + ) + .then(() => { + try { + this._orchestrationDb?.markAsDelivered(messageIds) + } catch (err) { + // The prompt was submitted but the stamp failed (DB swapped or + // closed mid-flight); the rows will replay on the next delivery + // event, so leave a diagnosable trace for that duplicate. + console.warn('[orchestration] failed to stamp delivered_at after wake submit', { + handle, + messageIds, + err + }) + } + }) + .catch((err) => { + // Failed submits leave the rows queued for a future delivery event. + console.warn('[orchestration] working-agent wake delivery failed; messages stay queued', { + handle, + err + }) + }) + .finally(() => { + for (const id of messageIds) { + this.orchestrationWakeDeliveryClaimedMessageIds.delete(id) + } + this.orchestrationWakeDeliveryInFlightHandles.delete(handle) + if (!this.orchestrationWakeDeliveryPendingHandles.delete(handle)) { + return + } + // Drain even when this submit failed: the deferred arrival was a real + // delivery event and gets exactly one attempt, not a retry loop. + this.drainPendingWakeDelivery(handle) + }) + } + + // Why: the drain re-derives the delivery mode from the CURRENT terminal + // state — a completion may have flipped working→idle while the prior submit + // was settling. + private drainPendingWakeDelivery(handle: string): void { + try { + const { leaf } = this.getLiveLeafForHandle(handle) + if (leaf.lastAgentStatus === 'idle') { + this.deliverPendingMessages(leaf) + } else if (leaf.lastAgentStatus === 'working') { + this.deliverPendingMessages(leaf, { wakeAgent: true }) + } + return + } catch { + // Leaf gone or renderer graph unavailable; try the PTY-handle route. + } + try { + this.deliverPendingWakeMessagesForPtyHandle(handle) + } catch { + // The terminal closed after submit; pending rows remain persisted. + } + } + + // Why: orchestration.check must not hand a caller a row that an in-flight + // working-agent wake is about to submit — the sub-second claim window is + // the only gap where both delivery paths could observe the same message. + isMessageClaimedForWakeDelivery(messageId: string): boolean { + return this.orchestrationWakeDeliveryClaimedMessageIds.has(messageId) + } + private resolveWaiter(waiter: TerminalWaiter, result: RuntimeTerminalWait): void { this.removeWaiter(waiter) waiter.resolve(result) diff --git a/src/main/runtime/orchestration/coordinator.ts b/src/main/runtime/orchestration/coordinator.ts index 2313dfdf320..ebbe7e4fccc 100644 --- a/src/main/runtime/orchestration/coordinator.ts +++ b/src/main/runtime/orchestration/coordinator.ts @@ -461,7 +461,11 @@ export class Coordinator { } } - const dispatch = this.db.createDispatchContext(task.id, targetHandle) + const dispatch = this.db.createDispatchContext( + task.id, + targetHandle, + this.opts.coordinatorHandle + ) // Why: agents dispatched by the coordinator must use orca-dev in dev mode // so they talk to the dev runtime's socket, not production (Section 6.4). diff --git a/src/main/runtime/orchestration/db.test.ts b/src/main/runtime/orchestration/db.test.ts index 8f43298a92c..625739c84e5 100644 --- a/src/main/runtime/orchestration/db.test.ts +++ b/src/main/runtime/orchestration/db.test.ts @@ -314,10 +314,20 @@ describe('OrchestrationDb', () => { expect(ctx.id).toMatch(/^ctx_/) expect(ctx.task_id).toBe(task.id) expect(ctx.assignee_handle).toBe('term_worker') + expect(ctx.coordinator_handle).toBeNull() expect(ctx.status).toBe('dispatched') expect(d.getTask(task.id)?.status).toBe('dispatched') }) + it('persists the dispatching coordinator so worker exits can be escalated', () => { + const d = createDb() + const task = d.createTask({ spec: 'work' }) + const ctx = d.createDispatchContext(task.id, 'term_worker', 'term_coord') + + expect(ctx.coordinator_handle).toBe('term_coord') + expect(d.getDispatchContextById(ctx.id)?.coordinator_handle).toBe('term_coord') + }) + it('rejects dispatch for non-ready tasks', () => { const d = createDb() const parent = d.createTask({ spec: 'parent' }) diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index 0c137fe78a5..ddd24a37c05 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -41,7 +41,10 @@ function generateId(prefix: string): string { // the terminal that created a task so task-record worktree creation can infer // the parent workspace even when no dispatch context exists. v4 → v5 adds // explicit task_title/display_name fields for orchestration worker UI labels. -const SCHEMA_VERSION = 5 +// v5 → v6 records the dispatching coordinator on dispatch contexts so a +// worker exit can be escalated to a manual coordinator, not only an active +// automated run. +const SCHEMA_VERSION = 6 export class OrchestrationDb { private db: Database.Database @@ -107,6 +110,7 @@ export class OrchestrationDb { id TEXT PRIMARY KEY, task_id TEXT NOT NULL, assignee_handle TEXT, + coordinator_handle TEXT, status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending', 'dispatched', 'completed', 'failed', 'circuit_broken')), failure_count INTEGER NOT NULL DEFAULT 0, @@ -246,6 +250,11 @@ export class OrchestrationDb { this.db.exec(`ALTER TABLE tasks ADD COLUMN display_name TEXT`) } } + if (current < 6) { + if (!this.hasColumn('dispatch_contexts', 'coordinator_handle')) { + this.db.exec(`ALTER TABLE dispatch_contexts ADD COLUMN coordinator_handle TEXT`) + } + } this.createUndeliveredInboxIndexIfPossible() this.db.pragma(`user_version = ${SCHEMA_VERSION}`) @@ -568,7 +577,11 @@ export class OrchestrationDb { // ── Dispatch Contexts ── - createDispatchContext(taskId: string, assigneeHandle: string): DispatchContextRow { + createDispatchContext( + taskId: string, + assigneeHandle: string, + coordinatorHandle: string | null = null + ): DispatchContextRow { const task = this.getTask(taskId) if (!task) { throw new Error(`Task not found: ${taskId}`) @@ -599,10 +612,10 @@ export class OrchestrationDb { const id = generateId('ctx') this.db .prepare( - `INSERT INTO dispatch_contexts (id, task_id, assignee_handle, status, failure_count, dispatched_at) - VALUES (?, ?, ?, 'dispatched', ?, datetime('now'))` + `INSERT INTO dispatch_contexts (id, task_id, assignee_handle, coordinator_handle, status, failure_count, dispatched_at) + VALUES (?, ?, ?, ?, 'dispatched', ?, datetime('now'))` ) - .run(id, taskId, assigneeHandle, priorFailures) + .run(id, taskId, assigneeHandle, coordinatorHandle, priorFailures) this.db.prepare("UPDATE tasks SET status = 'dispatched' WHERE id = ?").run(taskId) diff --git a/src/main/runtime/orchestration/lifecycle-delivery.ts b/src/main/runtime/orchestration/lifecycle-delivery.ts new file mode 100644 index 00000000000..788e42df1a8 --- /dev/null +++ b/src/main/runtime/orchestration/lifecycle-delivery.ts @@ -0,0 +1,41 @@ +import type { MessageRow, MessageType } from './types' + +// Why: only actionable lifecycle messages should steer an active coordinator; +// progress traffic remains in the persisted inbox until the next idle turn. +export const COORDINATOR_WAKE_MESSAGE_TYPES: MessageType[] = [ + 'worker_done', + 'escalation', + 'decision_gate' +] + +export function shouldWakeCoordinatorForMessage(type: MessageType): boolean { + return COORDINATOR_WAKE_MESSAGE_TYPES.includes(type) +} + +// Structural view of the runtime so every insert-time caller (RPC methods and +// the runtime's own exit escalation) shares one delivery entry point. +export type OrchestrationMessageDeliverer = { + deliverPendingMessagesForHandle( + handle: string, + options?: { wakeAgent?: boolean; messageType?: MessageType } + ): void + notifyMessageArrived(handle: string, messageType?: string): void +} + +// Why: delivery MUST run before notify. The push path treats a still-blocked +// matching waiter as the delivery owner; notify resolves (and removes) that +// waiter, so the reversed order would make both paths observe the message. +export function deliverOrchestrationMessage( + runtime: OrchestrationMessageDeliverer, + message: Pick +): void { + if (shouldWakeCoordinatorForMessage(message.type)) { + runtime.deliverPendingMessagesForHandle(message.to_handle, { + wakeAgent: true, + messageType: message.type + }) + } else { + runtime.deliverPendingMessagesForHandle(message.to_handle) + } + runtime.notifyMessageArrived(message.to_handle, message.type) +} diff --git a/src/main/runtime/orchestration/types.ts b/src/main/runtime/orchestration/types.ts index cf3c780287d..d4cf2fe4aa1 100644 --- a/src/main/runtime/orchestration/types.ts +++ b/src/main/runtime/orchestration/types.ts @@ -52,6 +52,7 @@ export type DispatchContextRow = { id: string task_id: string assignee_handle: string | null + coordinator_handle: string | null status: DispatchStatus failure_count: number last_failure: string | null diff --git a/src/main/runtime/rpc/methods/orchestration.test.ts b/src/main/runtime/rpc/methods/orchestration.test.ts index e5255595a6a..39e44ddf110 100644 --- a/src/main/runtime/rpc/methods/orchestration.test.ts +++ b/src/main/runtime/rpc/methods/orchestration.test.ts @@ -90,6 +90,28 @@ describe('orchestration RPC methods', () => { expect(runtime.deliverPendingMessagesForHandle).toHaveBeenCalledWith('term_b') }) + it.each(['worker_done', 'escalation', 'decision_gate'] as const)( + 'requests a coordinator wake for %s messages', + async (type) => { + setup() + const deliver = vi + .spyOn(runtime, 'deliverPendingMessagesForHandle') + .mockImplementation(() => {}) + + await call('orchestration.send', { + from: 'term_worker', + to: 'term_coord', + subject: 'attention needed', + type + }) + + expect(deliver).toHaveBeenCalledWith('term_coord', { + wakeAgent: true, + messageType: type + }) + } + ) + it('rejects missing --to', () => { const method = findMethod('orchestration.send') expect(() => method.params!.parse({ subject: 'hi' })).toThrow() @@ -406,7 +428,9 @@ describe('orchestration RPC methods', () => { setup() const task = db.createTask({ spec: 'heartbeat work' }) const dispatch = db.createDispatchContext(task.id, 'term_worker') - vi.spyOn(runtime, 'deliverPendingMessagesForHandle').mockImplementation(() => {}) + const deliver = vi + .spyOn(runtime, 'deliverPendingMessagesForHandle') + .mockImplementation(() => {}) await call('orchestration.send', { from: 'term_worker', @@ -420,6 +444,7 @@ describe('orchestration RPC methods', () => { expect(db.getDispatchContextById(dispatch.id)?.status).toBe('dispatched') expect(db.getDispatchContextById(dispatch.id)?.last_heartbeat_at).toBeTruthy() expect(db.getActiveDispatchForTerminal('term_worker')).toBeDefined() + expect(deliver).toHaveBeenCalledWith('term_coord') }) it('does not release dispatch lock for non-lifecycle sends', async () => { @@ -514,6 +539,32 @@ describe('orchestration RPC methods', () => { expect(result.count).toBe(1) }) + it('does not hand a wake-claimed message to a concurrent unread check', async () => { + setup() + const claimed = db.insertMessage({ + from: 'term_worker', + to: 'term_coord', + subject: 'mid-submit', + type: 'worker_done' + }) + vi.spyOn(runtime, 'isMessageClaimedForWakeDelivery').mockImplementation( + (id) => id === claimed.id + ) + + const unread = (await call('orchestration.check', { terminal: 'term_coord' })) as { + count: number + } + expect(unread.count).toBe(0) + // The claimed row stays unread so the wake path keeps sole ownership. + expect(db.getUnreadMessages('term_coord')).toHaveLength(1) + + // The audit view is unaffected by claims. + const all = (await call('orchestration.check', { terminal: 'term_coord', all: true })) as { + count: number + } + expect(all.count).toBe(1) + }) + it('reconciles worker_done returned by a waiting manual check', async () => { setup() const { task, dispatch } = createDispatchedTask() @@ -1050,6 +1101,19 @@ describe('orchestration RPC methods', () => { expect(result.dispatch.status).toBe('dispatched') }) + it('persists the dispatching coordinator on the dispatch context', async () => { + setup() + const task = db.createTask({ spec: 'work' }) + + const result = (await call('orchestration.dispatch', { + task: task.id, + to: 'term_worker', + from: 'term_coord' + })) as { dispatch: { coordinator_handle: string | null } } + + expect(result.dispatch.coordinator_handle).toBe('term_coord') + }) + it('rejects dispatch for a pending task', async () => { setup() const parent = db.createTask({ spec: 'parent' }) @@ -1365,7 +1429,9 @@ describe('orchestration RPC methods', () => { describe('orchestration.ask', () => { it('sends a decision_gate and returns the first thread reply', async () => { setup() - vi.spyOn(runtime, 'deliverPendingMessagesForHandle').mockImplementation(() => {}) + const deliver = vi + .spyOn(runtime, 'deliverPendingMessagesForHandle') + .mockImplementation(() => {}) vi.spyOn(runtime, 'notifyMessageArrived').mockImplementation(() => {}) vi.spyOn(runtime, 'waitForMessage').mockImplementation(async () => { // Simulate coordinator replying in the thread during the wait @@ -1406,6 +1472,10 @@ describe('orchestration RPC methods', () => { const payload = JSON.parse(outbound!.payload ?? '{}') expect(payload.question).toBe('proceed?') expect(payload.options).toEqual(['yes', 'no']) + expect(deliver).toHaveBeenCalledWith('term_coord', { + wakeAgent: true, + messageType: 'decision_gate' + }) }) it('returns timedOut when no reply arrives in the window', async () => { diff --git a/src/main/runtime/rpc/methods/orchestration.ts b/src/main/runtime/rpc/methods/orchestration.ts index f22ad62d8f4..ec974ee5402 100644 --- a/src/main/runtime/rpc/methods/orchestration.ts +++ b/src/main/runtime/rpc/methods/orchestration.ts @@ -7,6 +7,7 @@ import { buildDispatchPreamble } from '../../orchestration/preamble' import { formatMessageBanner } from '../../orchestration/formatter' import { isGroupAddress, resolveGroupAddress } from '../../orchestration/groups' import { reconcileLifecycleMessage } from '../../orchestration/lifecycle-reconciliation' +import { deliverOrchestrationMessage } from '../../orchestration/lifecycle-delivery' import { ORCHESTRATION_GATE_METHODS } from './orchestration-gates' const MESSAGE_TYPES: MessageType[] = [ @@ -204,8 +205,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ if (msg.type === 'worker_done' || msg.type === 'heartbeat') { reconcileLifecycleMessage(db, msg) } - runtime.deliverPendingMessagesForHandle(params.to) - runtime.notifyMessageArrived(params.to, msg.type) + deliverOrchestrationMessage(runtime, msg) return { message: msg } } @@ -235,8 +235,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ }) ) for (const message of messages) { - runtime.deliverPendingMessagesForHandle(message.to_handle) - runtime.notifyMessageArrived(message.to_handle, message.type) + deliverOrchestrationMessage(runtime, message) } return { messages, recipients: handles.length } @@ -268,8 +267,13 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ const showUnread = !showAll const readAndReturn = () => { + // Why: rows a working-agent wake is mid-submit for must not also be + // handed to a concurrent check — exactly one immediate delivery path. + // The audit view (--all) still shows everything. const messages = showUnread - ? db.getUnreadMessages(handle, typeFilter) + ? db + .getUnreadMessages(handle, typeFilter) + .filter((m) => !runtime.isMessageClaimedForWakeDelivery(m.id)) : db.getAllMessagesForHandle(handle, undefined, typeFilter) if (showUnread && messages.length > 0) { @@ -475,7 +479,9 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ } } - const ctx = db.createDispatchContext(params.task, to) + // Why: the dispatching coordinator is persisted so a worker exit can be + // escalated to it even when no automated coordinator run is active. + const ctx = db.createDispatchContext(params.task, to, params.from ?? null) // Why: preamble is built here (not before ctx) so `dispatchId` can be // the real ctx.id — the preamble-hardening PR made dispatchId required @@ -580,8 +586,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ type: 'decision_gate', payload }) - runtime.deliverPendingMessagesForHandle(params.to) - runtime.notifyMessageArrived(params.to, outbound.type) + deliverOrchestrationMessage(runtime, outbound) const threadId = outbound.id const deadline = Date.now() + timeoutMs