diff --git a/src/cli/handlers/orchestration.test.ts b/src/cli/handlers/orchestration.test.ts index d19d486e209..9d8e40378b3 100644 --- a/src/cli/handlers/orchestration.test.ts +++ b/src/cli/handlers/orchestration.test.ts @@ -14,6 +14,7 @@ vi.mock('../selectors', () => ({ getTerminalHandle: getTerminalHandleMock })) import { ORCHESTRATION_HANDLERS } from './orchestration' import { RuntimeClientError } from '../runtime-client' +import { printResult } from '../format' function staleHandleError(): RuntimeClientError { return new RuntimeClientError('terminal_handle_stale', 'terminal_handle_stale') @@ -213,7 +214,7 @@ describe('orchestration send structured payload flags', () => { }) }) - it('continues to use ORCA_TERMINAL_HANDLE as worker lifecycle sender authority', async () => { + it('uses a live ORCA_TERMINAL_HANDLE as worker lifecycle sender metadata', async () => { process.env.ORCA_TERMINAL_HANDLE = 'term_worker_env' await invokeSend( @@ -224,8 +225,11 @@ describe('orchestration send structured payload flags', () => { ]) ) - expect(callMock).toHaveBeenCalledTimes(1) - expect(callMock).toHaveBeenCalledWith('orchestration.send', { + expect(callMock).toHaveBeenCalledTimes(2) + expect(callMock).toHaveBeenNthCalledWith(1, 'terminal.show', { + terminal: 'term_worker_env' + }) + expect(callMock).toHaveBeenNthCalledWith(2, 'orchestration.send', { from: 'term_worker_env', to: 'term_coord', subject: 'done', @@ -238,6 +242,49 @@ describe('orchestration send structured payload flags', () => { }) }) + it('remints a stale lifecycle sender from its pane key', async () => { + process.env.ORCA_TERMINAL_HANDLE = 'term_worker_stale' + process.env.ORCA_PANE_KEY = 'tab_worker:leaf_worker' + stubStaleHandleRemint('term_worker_live', { result: { message: { id: 'msg_1' } } }) + + await invokeSend( + new Map([ + ['to', 'term_coord'], + ['subject', 'done'], + ['type', 'worker_done'] + ]) + ) + + expect(callMock).toHaveBeenNthCalledWith(2, 'terminal.resolvePane', { + paneKey: 'tab_worker:leaf_worker' + }) + expect(callMock).toHaveBeenNthCalledWith( + 3, + 'orchestration.send', + expect.objectContaining({ from: 'term_worker_live' }) + ) + }) + + it('keeps a stale lifecycle sender when no pane key is available', async () => { + process.env.ORCA_TERMINAL_HANDLE = 'term_worker_stale' + callMock.mockRejectedValueOnce(staleHandleError()) + + await invokeSend( + new Map([ + ['to', 'term_coord'], + ['subject', 'done'], + ['type', 'worker_done'] + ]) + ) + + expect(callMock).toHaveBeenCalledTimes(2) + expect(callMock).toHaveBeenNthCalledWith( + 2, + 'orchestration.send', + expect.objectContaining({ from: 'term_worker_stale' }) + ) + }) + it('reports sender resolution failure instead of raw no_active_terminal', async () => { getTerminalHandleMock.mockRejectedValue( new RuntimeClientError('no_active_terminal', 'no_active_terminal') @@ -631,13 +678,14 @@ describe('orchestration timeout flag validation', () => { expect(callMock).not.toHaveBeenCalled() }) - it('passes a parsed check timeout into the RPC payload', async () => { + it('passes a parsed check timeout and peek mode into the RPC payload', async () => { process.env.ORCA_TERMINAL_HANDLE = 'term_worker' callMock.mockResolvedValue({ result: { messages: [], count: 0 } }) await invokeCheck( new Map([ ['wait', true], + ['peek', true], ['timeout-ms', '250'] ]) ) @@ -645,6 +693,7 @@ describe('orchestration timeout flag validation', () => { expect(callMock).toHaveBeenCalledWith('orchestration.check', { terminal: 'term_worker', unread: undefined, + peek: true, all: undefined, types: undefined, inject: undefined, @@ -697,3 +746,27 @@ describe('orchestration timeout flag validation', () => { ) }) }) + +describe('orchestration task-list brief output', () => { + it('abbreviates task specs in the JSON response', async () => { + callMock.mockReset().mockResolvedValue({ + result: { + tasks: [{ id: 'task_1', spec: `First line\n${'detail '.repeat(40)}`, status: 'ready' }], + count: 1 + } + }) + vi.mocked(printResult).mockClear() + + await ORCHESTRATION_HANDLERS['orchestration task-list']({ + flags: new Map([['brief', true]]), + client: { call: callMock }, + json: true + } as never) + + const response = vi.mocked(printResult).mock.calls[0]?.[0] as { + result: { tasks: { spec: string; spec_truncated: boolean }[] } + } + expect(response.result.tasks[0].spec).toHaveLength(160) + expect(response.result.tasks[0].spec_truncated).toBe(true) + }) +}) diff --git a/src/cli/handlers/orchestration.ts b/src/cli/handlers/orchestration.ts index 4006c809cd1..166a1c3ece1 100644 --- a/src/cli/handlers/orchestration.ts +++ b/src/cli/handlers/orchestration.ts @@ -8,12 +8,13 @@ import { } from '../flags' import { RuntimeClientError } from '../runtime-client' import { getTerminalHandle } from '../selectors' +import { abbreviateOrchestrationTasks } from '../orchestration-task-summary' // Why: 15 s is well under Claude Code's empirical ~2 min Bash-tool silence // budget and generates only ~40 lines per 10 min wait — enough to assure the // parent process the subprocess is alive without flooding logs. See design // doc §3.4. -const DEFAULT_HEARTBEAT_INTERVAL_MS = 15_000 +const DEFAULT_KEEPALIVE_INTERVAL_MS = 15_000 function getLifecycleGroupRecipientError(type: 'worker_done' | 'heartbeat'): string { return `${type} messages must be sent to a concrete coordinator terminal handle, not a group address.` } @@ -21,33 +22,36 @@ function getLifecycleGroupRecipientError(type: 'worker_done' | 'heartbeat'): str // Why: test-only escape hatch so subprocess tests can verify the feature in // under 10 s rather than needing a full 15 s silence window. Production users // should never set this — there is no surface documentation. A bogus value -// falls back to the default rather than disabling the heartbeat. -function resolveHeartbeatIntervalMs(): number { - const raw = process.env.ORCA_HEARTBEAT_INTERVAL_MS +// falls back to the default rather than disabling the keepalive. +function resolveKeepaliveIntervalMs(): number { + const raw = process.env.ORCA_KEEPALIVE_INTERVAL_MS ?? process.env.ORCA_HEARTBEAT_INTERVAL_MS if (!raw) { - return DEFAULT_HEARTBEAT_INTERVAL_MS + return DEFAULT_KEEPALIVE_INTERVAL_MS } const parsed = Number(raw) if (!Number.isFinite(parsed) || parsed <= 0) { - return DEFAULT_HEARTBEAT_INTERVAL_MS + return DEFAULT_KEEPALIVE_INTERVAL_MS } return parsed } -function startCheckHeartbeat(deadlineMs: number | undefined): () => void { +function startCheckKeepalive(deadlineMs: number | undefined): () => void { const startedAt = Date.now() const interval = setInterval(() => { const payload = { + _keepalive: true, + // Why: retain the old marker for scripts filtering merged stderr while + // callers migrate to the unambiguous _keepalive field. _heartbeat: true, elapsedMs: Date.now() - startedAt, deadlineMs: deadlineMs ?? null } // Why: `process.stderr.write` is line-flushed per-call in Node, whereas a - // fully-buffered writer would hold all heartbeat lines until exit and + // fully-buffered writer would hold all keepalive lines until exit and // silently defeat the whole point of the ping. Subprocess test asserts // this by reading stderr incrementally. See §3.4. process.stderr.write(`${JSON.stringify(payload)}\n`) - }, resolveHeartbeatIntervalMs()) + }, resolveKeepaliveIntervalMs()) if (typeof interval.unref === 'function') { interval.unref() } @@ -128,7 +132,7 @@ async function resolveOrchestrationTerminalHandle( cwd: string, client: Parameters[0]['client'], flagName: 'from' | 'terminal', - options: { validateEnvHandle?: boolean } = {} + options: { validateEnvHandle?: boolean; allowStaleEnvHandle?: boolean } = {} ): Promise { const explicit = getOptionalStringFlag(flags, flagName) if (explicit) { @@ -142,7 +146,16 @@ async function resolveOrchestrationTerminalHandle( // coordinator preambles. const live = await isLiveTerminalHandle(envHandle, client) if (!live) { - return await resolveStaleOrchestrationSender(client) + const reminted = await resolveOrchestrationPaneTerminalHandle(client, { + optional: options.allowStaleEnvHandle + }) + if (reminted) { + return reminted + } + if (options.allowStaleEnvHandle) { + return envHandle + } + throwNoActiveSenderTerminal() } } return envHandle @@ -268,16 +281,6 @@ function getClientErrorMessage(err: unknown): string | undefined { return typeof message === 'string' ? message : undefined } -async function resolveStaleOrchestrationSender( - client: Parameters[0]['client'] -): Promise { - const paneHandle = await resolveOrchestrationPaneTerminalHandle(client) - if (paneHandle) { - return paneHandle - } - throwNoActiveSenderTerminal() -} - async function resolveCoordinatorTerminalHandle( flags: Map, cwd: string, @@ -348,7 +351,12 @@ export const ORCHESTRATION_HANDLERS: Record = { const type = getOptionalStringFlag(flags, 'type') rejectLifecycleGroupRecipient(type, to) - const from = await resolveOrchestrationTerminalHandle(flags, cwd, client, 'from') + const from = await resolveOrchestrationTerminalHandle(flags, cwd, client, 'from', { + validateEnvHandle: type === 'worker_done' || type === 'heartbeat', + // Why: payload IDs remain authoritative when an older/SSH session lacks + // a pane key, so handle repair must not prevent lifecycle delivery. + allowStaleEnvHandle: type === 'worker_done' || type === 'heartbeat' + }) const result = await client.call< { message: { id: string } } | { messages: { id: string }[]; recipients: number } >('orchestration.send', { @@ -377,12 +385,12 @@ export const ORCHESTRATION_HANDLERS: Record = { // Why: Claude Code's Bash tool auto-backgrounds subprocesses that produce // no output for ~2 min (shorter on the non-interactive path). Emit a - // heartbeat line to stderr every HEARTBEAT_INTERVAL_MS while the wait is + // keepalive line to stderr every KEEPALIVE_INTERVAL_MS while the wait is // active so the parent process can see the subprocess is still alive. // Stderr rather than stdout so stdout stays a single final JSON payload, // and JSON-shaped rather than `# …` so `2>&1 | jq` pipelines still work // (jq refuses `#`-prefixed lines). See design doc §3.4. - const stopHeartbeat = wait ? startCheckHeartbeat(timeoutMs) : null + const stopKeepalive = wait ? startCheckKeepalive(timeoutMs) : null type CheckResult = { messages: MessageSummary[] count: number @@ -393,6 +401,7 @@ export const ORCHESTRATION_HANDLERS: Record = { result = await client.call('orchestration.check', { terminal, unread: flags.has('unread') ? true : undefined, + peek: flags.has('peek') ? true : undefined, all: flags.has('all') ? true : undefined, types: getOptionalStringFlag(flags, 'types'), inject: flags.has('inject') ? true : undefined, @@ -400,7 +409,7 @@ export const ORCHESTRATION_HANDLERS: Record = { timeoutMs }) } finally { - stopHeartbeat?.() + stopKeepalive?.() } printResult(result, json, (r) => { if (r.formatted) { @@ -491,7 +500,13 @@ export const ORCHESTRATION_HANDLERS: Record = { status: getOptionalStringFlag(flags, 'status'), ready: flags.has('ready') ? true : undefined }) - printResult(result, json, (r) => { + const output = flags.has('brief') + ? { + ...result, + result: { ...result.result, tasks: abbreviateOrchestrationTasks(result.result.tasks) } + } + : result + printResult(output, json, (r) => { if (r.count === 0) { return 'No tasks.' } diff --git a/src/cli/orchestration-task-summary.test.ts b/src/cli/orchestration-task-summary.test.ts new file mode 100644 index 00000000000..69c7a19131f --- /dev/null +++ b/src/cli/orchestration-task-summary.test.ts @@ -0,0 +1,22 @@ +import { describe, expect, it } from 'vitest' +import { abbreviateOrchestrationTasks } from './orchestration-task-summary' + +describe('abbreviateOrchestrationTasks', () => { + it('collapses whitespace and caps long task specs', () => { + const [task] = abbreviateOrchestrationTasks([ + { id: 'task_1', spec: `First line\n\n${'detail '.repeat(40)}` } + ]) + + expect(task.id).toBe('task_1') + expect(task.spec).not.toContain('\n') + expect(task.spec).toHaveLength(160) + expect(task.spec.endsWith('…')).toBe(true) + expect(task.spec_truncated).toBe(true) + }) + + it('preserves a short one-line spec', () => { + const [task] = abbreviateOrchestrationTasks([{ spec: 'Short task' }]) + + expect(task).toEqual({ spec: 'Short task', spec_truncated: false }) + }) +}) diff --git a/src/cli/orchestration-task-summary.ts b/src/cli/orchestration-task-summary.ts new file mode 100644 index 00000000000..930041893b7 --- /dev/null +++ b/src/cli/orchestration-task-summary.ts @@ -0,0 +1,18 @@ +const TASK_SPEC_BRIEF_LENGTH = 160 + +export function abbreviateOrchestrationTasks( + tasks: readonly T[] +): (T & { spec_truncated: boolean })[] { + return tasks.map((task) => { + const spec = task.spec.replace(/\s+/g, ' ').trim() + const abbreviated = + spec.length > TASK_SPEC_BRIEF_LENGTH + ? `${spec.slice(0, TASK_SPEC_BRIEF_LENGTH - 1).trimEnd()}…` + : spec + return { + ...task, + spec: abbreviated, + spec_truncated: abbreviated !== task.spec + } + }) +} diff --git a/src/cli/specs/core.ts b/src/cli/specs/core.ts index e4f2ffe9f91..73c677a0f9b 100644 --- a/src/cli/specs/core.ts +++ b/src/cli/specs/core.ts @@ -116,6 +116,7 @@ export const CORE_COMMAND_SPECS: CommandSpec[] = [ '--no-parent only affects Orca lineage; omit --base-branch to use the repo default base, or pass the default base ref explicitly for independent top-level work.', 'By default this creates the worktree and its first terminal without switching the active Orca view.', 'Pass --agent to launch an agent in the first terminal; --prompt sends initial work to that agent.', + 'With --agent --json, read the new agent handle from result.agentTerminalHandle (or result.startupTerminal.handle on older runtimes).', 'Repo-defined setup hooks follow the repository setup policy; pass --setup run to force them.', 'Pass --activate when the CLI caller intentionally wants to reveal the new worktree in the app.', 'Passing --run-hooks is kept as a legacy alias for --setup run and reveals the worktree.' diff --git a/src/cli/specs/orchestration.ts b/src/cli/specs/orchestration.ts index d3708a3dc77..86971bfd8dc 100644 --- a/src/cli/specs/orchestration.ts +++ b/src/cli/specs/orchestration.ts @@ -26,6 +26,7 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ notes: [ 'On Windows PowerShell, quote group addresses such as --to "@all" or --to "@worktree:".', 'worker_done and heartbeat must target a concrete coordinator terminal handle; use status for broadcast updates.', + 'A worker_done with the active task/dispatch IDs completes that task automatically; handles are routing metadata.', 'Prefer --task-id/--dispatch-id/etc. over raw --payload JSON in worker commands; PowerShell strips JSON quotes easily.' ] }, @@ -33,17 +34,20 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ path: ['orchestration', 'check'], summary: 'Check messages for a terminal', usage: - 'orca orchestration check [--terminal ] [--unread | --all] [--types ] [--inject] [--wait] [--timeout-ms ] [--json]\n' + + 'orca orchestration check [--terminal ] [--unread | --peek | --all] [--types ] [--inject] [--wait] [--timeout-ms ] [--json]\n' + ' --unread (default): return only unread messages and mark them read.\n' + + ' --peek: return only unread messages without marking them read.\n' + ' --all: return every message for the handle; does not mark read.\n' + ' --wait: block until a matching message arrives or --timeout-ms expires.\n' + - ' Emits JSON heartbeat lines to stderr every 15s so the caller can\n' + - ' tell the process is alive. Filter with `grep -v _heartbeat` or\n' + - ' `jq "select(._heartbeat|not)"` when merging streams with 2>&1.', + ' Emits JSON keepalive lines to stderr every 15s so the caller can\n' + + ' tell the process is alive. `_keepalive` is unrelated to heartbeat\n' + + ' messages; `_heartbeat` remains as a deprecated compatibility alias.\n' + + ' Filter with `jq "select(._keepalive|not)"` when merging streams.', allowedFlags: [ ...GLOBAL_FLAGS, 'terminal', 'unread', + 'peek', 'all', 'types', 'inject', @@ -76,8 +80,9 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ { path: ['orchestration', 'task-list'], summary: 'List orchestration tasks', - usage: 'orca orchestration task-list [--status ] [--ready] [--json]', - allowedFlags: [...GLOBAL_FLAGS, 'status', 'ready'] + usage: 'orca orchestration task-list [--status ] [--ready] [--brief] [--json]', + allowedFlags: [...GLOBAL_FLAGS, 'status', 'ready', 'brief'], + notes: ['--brief collapses whitespace and caps each spec at 160 characters.'] }, { path: ['orchestration', 'task-update'], diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 19d08a619ba..0f08bcd0960 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -14618,6 +14618,7 @@ export class OrcaRuntimeService { const shouldActivate = args.activate === true || args.runHooks === true let warning: string | undefined let didSpawnStartup = false + let startupTerminal: CreateWorktreeResult['startupTerminal'] if (effectiveStartup && this.ptyController?.spawn) { try { const startupTrustAgent = effectiveDraftPaste?.agent ?? effectiveCreatedWithAgent @@ -14641,6 +14642,14 @@ export class OrcaRuntimeService { this.sendStartupFollowupWhenReady(terminal.handle, effectiveStartupFollowup) } didSpawnStartup = true + startupTerminal = { + spawned: true, + handle: terminal.handle, + ...(terminal.tabId ? { tabId: terminal.tabId } : {}), + ...(terminal.paneKey ? { paneKey: terminal.paneKey } : {}), + ...(terminal.ptyId ? { ptyId: terminal.ptyId } : {}), + surface: 'background' + } } catch (err) { const message = err instanceof Error ? err.message : String(err) warning = `Failed to create the startup terminal for ${worktree.path}: ${message}` @@ -14678,6 +14687,7 @@ export class OrcaRuntimeService { isMainWorktree: worktree.isMainWorktree } }, + ...(startupTerminal ? { startupTerminal } : {}), ...(warning ? { warning } : {}) } } diff --git a/src/main/runtime/orchestration-cli-subprocess.test.ts b/src/main/runtime/orchestration-cli-subprocess.test.ts index 3b94acbb3df..6ec938a7e28 100644 --- a/src/main/runtime/orchestration-cli-subprocess.test.ts +++ b/src/main/runtime/orchestration-cli-subprocess.test.ts @@ -1,14 +1,14 @@ -// Why: subprocess-level test for the CLI heartbeat behavior described in +// Why: subprocess-level test for the CLI keepalive behavior described in // design doc §3.4. Spawns the real compiled CLI with no TTY, points it at a // real in-process runtime via ORCA_USER_DATA_PATH, and asserts: -// - the first heartbeat line appears on stderr well under Claude Code's +// - the first keepalive line appears on stderr well under Claude Code's // ~2 min Bash-tool silence budget (we verify with a shortened interval; // production uses 15 s via the same code path) -// - ≥3 heartbeats arrive during the wait window -// - stderr is line-flushed (we observe each heartbeat as a separate chunk +// - ≥3 keepalives arrive during the wait window +// - stderr is line-flushed (we observe each keepalive as a separate chunk // before the process exits — not in one burst at the end) -// - stdout stays a single clean JSON payload (no heartbeats leak to stdout) -// - a `jq "select(._heartbeat|not)"` filter on the merged stream would +// - stdout stays a single clean JSON payload (no keepalives leak to stdout) +// - a `jq "select(._keepalive|not)"` filter on the merged stream would // yield exactly the final result // // This test is skipped if the CLI hasn't been built yet (out/cli/index.js @@ -66,7 +66,7 @@ async function runBuiltCli( } describeIfBuilt('orca orchestration check --wait subprocess (§3.4)', () => { - it('emits newline-flushed JSON heartbeats to stderr while waiting', async () => { + it('emits newline-flushed JSON keepalives to stderr while waiting', async () => { const userDataPath = mkdtempSync(join(tmpdir(), 'orca-cli-sub-')) const runtime = new OrcaRuntimeService() const db = new OrchestrationDb(':memory:') @@ -75,12 +75,12 @@ describeIfBuilt('orca orchestration check --wait subprocess (§3.4)', () => { await server.start() try { - // Why: use the ORCA_HEARTBEAT_INTERVAL_MS escape hatch to shrink the + // Why: use the ORCA_KEEPALIVE_INTERVAL_MS escape hatch to shrink the // test to ~1 s wall time. Production callers never set this; the // production default (15 s) is exercised by §3.4's own unit tests // and by the fact that this same code path runs with the real // constant when the env var is absent. - const heartbeatMs = 200 + const keepaliveMs = 200 const waitTimeoutMs = 1200 const child = spawn( @@ -99,7 +99,7 @@ describeIfBuilt('orca orchestration check --wait subprocess (§3.4)', () => { ...process.env, ORCA_USER_DATA_PATH: userDataPath, ORCA_TERMINAL_HANDLE: 'term_nobody', - ORCA_HEARTBEAT_INTERVAL_MS: String(heartbeatMs) + ORCA_KEEPALIVE_INTERVAL_MS: String(keepaliveMs) }, // Why: explicit pipe for all three fds so we can watch stderr // in real time; no TTY attached (Bash-tool parity). @@ -125,7 +125,7 @@ describeIfBuilt('orca orchestration check --wait subprocess (§3.4)', () => { const stderr = stderrChunks.map((c) => c.data).join('') const stdout = stdoutChunks.map((c) => c.data).join('') - const heartbeatLines = stderr + const keepaliveLines = stderr .split('\n') .filter((line) => line.trim().length > 0) .map((line) => { @@ -135,51 +135,52 @@ describeIfBuilt('orca orchestration check --wait subprocess (§3.4)', () => { return null } }) - .filter((p): p is Record => p !== null && p._heartbeat === true) + .filter((p): p is Record => p !== null && p._keepalive === true) - // ≥3 heartbeats in a 1.2s window with a 200ms interval - expect(heartbeatLines.length).toBeGreaterThanOrEqual(3) - expect(heartbeatLines[0]).toHaveProperty('elapsedMs') - expect(heartbeatLines[0]).toHaveProperty('deadlineMs', waitTimeoutMs) + // ≥3 keepalives in a 1.2s window with a 200ms interval + expect(keepaliveLines.length).toBeGreaterThanOrEqual(3) + expect(keepaliveLines[0]).toMatchObject({ _keepalive: true, _heartbeat: true }) + expect(keepaliveLines[0]).toHaveProperty('elapsedMs') + expect(keepaliveLines[0]).toHaveProperty('deadlineMs', waitTimeoutMs) // Why: under full-suite load the child process startup may take longer - // than one heartbeat interval. The invariant that matters is that at - // least one heartbeat is observed before the terminal stdout payload. - const firstHeartbeatChunk = stderrChunks.find((c) => c.data.includes('_heartbeat')) - expect(firstHeartbeatChunk).toBeDefined() - expect(firstHeartbeatChunk!.at).toBeLessThan(stdoutChunks[0]?.at ?? Number.POSITIVE_INFINITY) + // than one keepalive interval. The invariant that matters is that at + // least one keepalive is observed before the terminal stdout payload. + const firstKeepaliveChunk = stderrChunks.find((c) => c.data.includes('_keepalive')) + expect(firstKeepaliveChunk).toBeDefined() + expect(firstKeepaliveChunk!.at).toBeLessThan(stdoutChunks[0]?.at ?? Number.POSITIVE_INFINITY) - // Why: line-flushing proof — the *first* heartbeat chunk must arrive + // Why: line-flushing proof — the *first* keepalive chunk must arrive // strictly before the exit chunk; i.e. we got at least two separate - // stderr data events (heartbeat + final). A single-chunk delivery + // stderr data events (keepalive + final). A single-chunk delivery // would indicate stderr was buffered until exit. const lastStderrAt = stderrChunks.at(-1)?.at ?? 0 const firstStderrAt = stderrChunks.at(0)?.at ?? 0 expect(lastStderrAt).toBeGreaterThan(firstStderrAt) - // Stdout: exactly one JSON payload, the terminal result. No heartbeats + // Stdout: exactly one JSON payload, the terminal result. No keepalives // leak, and the content parses as valid JSON. const stdoutTrimmed = stdout.trim() const stdoutPayload = JSON.parse(stdoutTrimmed) as Record - expect(stdoutPayload).not.toHaveProperty('_heartbeat') - expect(stdoutTrimmed).not.toContain('_heartbeat') + expect(stdoutPayload).not.toHaveProperty('_keepalive') + expect(stdoutTrimmed).not.toContain('_keepalive') // Why: result should be an RPC success envelope with the expected // shape. `count: 0` and `messages: []` because the wait timed out // with no message for term_nobody. expect(stdoutPayload).toMatchObject({ ok: true }) - // Why: the heartbeats-on-stderr design is meant to pair with shell - // filters like `2>&1 | jq "select(._heartbeat|not)"`. jq is + // Why: the keepalives-on-stderr design is meant to pair with shell + // filters like `2>&1 | jq "select(._keepalive|not)"`. jq is // line-oriented by default, but also accepts pretty-printed JSON // across multiple lines. What matters here is that every - // heartbeat line on stderr is a standalone JSON object (so jq can + // keepalive line on stderr is a standalone JSON object (so jq can // match it) and doesn't span multiple lines — assert that each - // heartbeat is a single-line JSON with no embedded newlines. + // keepalive is a single-line JSON with no embedded newlines. for (const line of stderr.split('\n')) { if (line.trim().length === 0) { continue } - if (line.includes('_heartbeat')) { + if (line.includes('_keepalive')) { expect(() => JSON.parse(line)).not.toThrow() expect(line).not.toContain('\n') } diff --git a/src/main/runtime/orchestration/coordinator.test.ts b/src/main/runtime/orchestration/coordinator.test.ts index 8593fd81622..34f50b78d18 100644 --- a/src/main/runtime/orchestration/coordinator.test.ts +++ b/src/main/runtime/orchestration/coordinator.test.ts @@ -570,7 +570,7 @@ describe('Coordinator', () => { expect(db.getDispatchContextById(activeCtx.id)?.status).toBe('completed') }) - it('ignores worker_done sent by a terminal that does not own the dispatch', async () => { + it('accepts worker_done payload provenance after an assignee handle changes', async () => { db = new OrchestrationDb(':memory:') const runtime = createMockRuntime() const logs: string[] = [] @@ -579,9 +579,9 @@ describe('Coordinator', () => { const ctx = db.createDispatchContext(task.id, 'term_owner') db.insertMessage({ - from: 'term_intruder', + from: 'term_reminted', to: 'coord', - subject: 'Spoofed done', + subject: 'Done after restart', type: 'worker_done', payload: JSON.stringify({ taskId: task.id, dispatchId: ctx.id }) }) @@ -592,16 +592,12 @@ describe('Coordinator', () => { pollIntervalMs: 20, onLog: (m) => logs.push(m) }) - const runPromise = coordinator.run() - await new Promise((r) => { - setTimeout(r, 80) - }) - coordinator.stop() - await runPromise + const result = await coordinator.run() - expect(db.getTask(task.id)?.status).toBe('dispatched') - expect(db.getDispatchContextById(ctx.id)?.status).toBe('dispatched') - expect(logs.some((m) => m.includes('expected term_owner'))).toBe(true) + expect(result.status).toBe('completed') + expect(db.getTask(task.id)?.status).toBe('completed') + expect(db.getDispatchContextById(ctx.id)?.status).toBe('completed') + expect(logs.some((m) => m.includes('accepting payload provenance'))).toBe(true) }) it('can be stopped', async () => { diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index 0c137fe78a5..a0f0c8a13d7 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -382,6 +382,20 @@ export class OrchestrationDb { .run(...ids) } + markAsReadAndDelivered(ids: string[]): void { + if (ids.length === 0) { + return + } + const placeholders = ids.map(() => '?').join(',') + // Why: superseded lifecycle messages stay queryable through history but + // must not be consumed or injected after their dispatch has finished. + this.db + .prepare( + `UPDATE messages SET read = 1, delivered_at = COALESCE(delivered_at, datetime('now')) WHERE id IN (${placeholders})` + ) + .run(...ids) + } + getInbox(limit = 20): MessageRow[] { return this.db .prepare('SELECT * FROM messages ORDER BY sequence DESC LIMIT ?') diff --git a/src/main/runtime/orchestration/lifecycle-reconciliation.test.ts b/src/main/runtime/orchestration/lifecycle-reconciliation.test.ts new file mode 100644 index 00000000000..c7e47beab0c --- /dev/null +++ b/src/main/runtime/orchestration/lifecycle-reconciliation.test.ts @@ -0,0 +1,75 @@ +import { afterEach, describe, expect, it } from 'vitest' +import { OrchestrationDb } from './db' +import { reconcileLifecycleMessage } from './lifecycle-reconciliation' + +describe('lifecycle reconciliation', () => { + let db: OrchestrationDb + + afterEach(() => db?.close()) + + it('completes an active dispatch from payload IDs after its terminal handle is reminted', () => { + db = new OrchestrationDb(':memory:') + const task = db.createTask({ spec: 'work' }) + const dispatch = db.createDispatchContext(task.id, 'term_before_restart') + const logs: string[] = [] + const message = db.insertMessage({ + from: 'term_after_restart', + to: 'term_coordinator', + subject: 'Done', + type: 'worker_done', + payload: JSON.stringify({ taskId: task.id, dispatchId: dispatch.id }) + }) + + expect(reconcileLifecycleMessage(db, message, (line) => logs.push(line))).toEqual({ + action: 'completed', + taskId: task.id, + dispatchId: dispatch.id + }) + expect(db.getTask(task.id)?.status).toBe('completed') + expect(db.getDispatchContextById(dispatch.id)?.status).toBe('completed') + expect(logs.some((line) => line.includes('accepting payload provenance'))).toBe(true) + }) + + it('suppresses same-dispatch heartbeats once worker_done is reconciled', () => { + db = new OrchestrationDb(':memory:') + const task = db.createTask({ spec: 'work' }) + const dispatch = db.createDispatchContext(task.id, 'term_worker') + const otherTask = db.createTask({ spec: 'other work' }) + const otherDispatch = db.createDispatchContext(otherTask.id, 'term_other') + const insertHeartbeat = (dispatchId: string, from: string) => + db.insertMessage({ + from, + to: 'term_coordinator', + subject: 'alive', + type: 'heartbeat', + payload: JSON.stringify({ dispatchId }) + }) + const staleHeartbeat = insertHeartbeat(dispatch.id, 'term_worker') + const otherHeartbeat = insertHeartbeat(otherDispatch.id, 'term_other') + reconcileLifecycleMessage(db, staleHeartbeat) + reconcileLifecycleMessage(db, otherHeartbeat) + const done = db.insertMessage({ + from: 'term_worker', + to: 'term_coordinator', + subject: 'Done', + type: 'worker_done', + payload: JSON.stringify({ taskId: task.id, dispatchId: dispatch.id }) + }) + + reconcileLifecycleMessage(db, done) + + expect(db.getUnreadMessages('term_coordinator', ['heartbeat']).map((row) => row.id)).toEqual([ + otherHeartbeat.id + ]) + const archived = db + .getAllMessagesForHandle('term_coordinator') + .find((row) => row.id === staleHeartbeat.id) + expect(archived).toMatchObject({ read: 1 }) + expect(archived?.delivered_at).not.toBeNull() + + const lateHeartbeat = insertHeartbeat(dispatch.id, 'term_worker') + expect(reconcileLifecycleMessage(db, lateHeartbeat)).toEqual({ action: 'ignored' }) + expect(db.getMessageById(lateHeartbeat.id)).toMatchObject({ read: 1 }) + expect(db.getMessageById(lateHeartbeat.id)?.delivered_at).not.toBeNull() + }) +}) diff --git a/src/main/runtime/orchestration/lifecycle-reconciliation.ts b/src/main/runtime/orchestration/lifecycle-reconciliation.ts index a413924c59a..12f828b5d5c 100644 --- a/src/main/runtime/orchestration/lifecycle-reconciliation.ts +++ b/src/main/runtime/orchestration/lifecycle-reconciliation.ts @@ -63,6 +63,15 @@ function reconcileHeartbeatMessage( return { action: 'ignored' } } + const dispatch = db.getDispatchContextById(dispatchId) + if (!dispatch || dispatch.status !== 'dispatched') { + // Why: an in-flight heartbeat can arrive after completion; retain it for + // audit history without surfacing obsolete liveness to the coordinator. + db.markAsReadAndDelivered([msg.id]) + onLog(`Heartbeat for inactive dispatch ${dispatchId} suppressed`) + return { action: 'ignored' } + } + // Why: dispatchId-specific writes let the DB ignore late heartbeats for // completed/failed retries without masking a newer hung dispatch. db.recordHeartbeat(dispatchId, msg.created_at) @@ -113,9 +122,8 @@ function reconcileWorkerDoneMessage( } if (dispatch.assignee_handle !== msg.from_handle) { onLog( - `Warning: worker_done for dispatch ${dispatchId} came from ${msg.from_handle}, expected ${dispatch.assignee_handle ?? ''}` + `Warning: worker_done for dispatch ${dispatchId} came from ${msg.from_handle}, expected ${dispatch.assignee_handle ?? ''}; accepting payload provenance` ) - return { action: 'ignored' } } // Why: `orchestration.send` can release the DB lock before waking the // coordinator; the later coordinator read still needs to observe completion. @@ -143,7 +151,26 @@ function reconcileWorkerDoneMessage( completedAt: new Date().toISOString() }) db.updateTaskStatus(taskId, 'completed', result) + suppressEarlierHeartbeats(db, msg, dispatchId) onLog(`Task ${taskId} completed`) return { action: 'completed', taskId, dispatchId } } + +function suppressEarlierHeartbeats( + db: OrchestrationDb, + workerDone: MessageRow, + dispatchId: string +): void { + const heartbeatIds = db + .getUnreadMessages(workerDone.to_handle, ['heartbeat']) + .filter((message) => { + if (message.sequence >= workerDone.sequence) { + return false + } + const payload = parseObjectPayload(message, () => undefined) + return payload.dispatchId === dispatchId + }) + .map((message) => message.id) + db.markAsReadAndDelivered(heartbeatIds) +} diff --git a/src/main/runtime/rpc/methods/orchestration.test.ts b/src/main/runtime/rpc/methods/orchestration.test.ts index e5255595a6a..0f0c237b00d 100644 --- a/src/main/runtime/rpc/methods/orchestration.test.ts +++ b/src/main/runtime/rpc/methods/orchestration.test.ts @@ -598,11 +598,11 @@ describe('orchestration RPC methods', () => { expect(db.getDispatchContextById(dispatch.id)?.status).toBe('dispatched') }) - it('does not complete worker_done from a terminal that does not own the dispatch', async () => { + it('completes worker_done by payload IDs when the sender handle changed', async () => { setup() const { task, dispatch } = createDispatchedTask('term_owner') insertWorkerDone({ - from: 'term_intruder', + from: 'term_reminted', taskId: task.id, dispatchId: dispatch.id }) @@ -613,8 +613,8 @@ describe('orchestration RPC methods', () => { })) as { count: number } expect(result.count).toBe(1) - expect(db.getTask(task.id)?.status).toBe('dispatched') - expect(db.getDispatchContextById(dispatch.id)?.status).toBe('dispatched') + expect(db.getTask(task.id)?.status).toBe('completed') + expect(db.getDispatchContextById(dispatch.id)?.status).toBe('completed') }) it('does not complete worker_done for a stale inactive dispatch', async () => { @@ -670,6 +670,11 @@ describe('orchestration RPC methods', () => { ).rejects.toThrow('Invalid --types') }) + it('rejects conflicting message read modes', () => { + const method = findMethod('orchestration.check') + expect(() => method.params!.parse({ unread: true, peek: true })).toThrow(/read mode/) + }) + it('default (unread only) marks returned rows as read', async () => { setup() db.insertMessage({ from: 'a', to: 'b', subject: 'one' }) @@ -686,6 +691,19 @@ describe('orchestration RPC methods', () => { expect(second.count).toBe(0) }) + it('--peek returns unread messages without marking them read', async () => { + setup() + db.insertMessage({ from: 'a', to: 'b', subject: 'one' }) + + const result = (await call('orchestration.check', { + terminal: 'b', + peek: true + })) as { count: number } + + expect(result.count).toBe(1) + expect(db.getUnreadMessages('b')).toHaveLength(1) + }) + it('--all returns every message for the handle without marking read', async () => { setup() db.insertMessage({ from: 'a', to: 'b', subject: 'one' }) diff --git a/src/main/runtime/rpc/methods/orchestration.ts b/src/main/runtime/rpc/methods/orchestration.ts index f22ad62d8f4..d33f003e24f 100644 --- a/src/main/runtime/rpc/methods/orchestration.ts +++ b/src/main/runtime/rpc/methods/orchestration.ts @@ -72,18 +72,33 @@ const SendParams = z }) }) -const CheckParams = z.object({ - terminal: OptionalString, - unread: OptionalBoolean, - // Why: `all` surfaces every message for the handle and skips mark-read. - // Previously the only way to ask for "all" was the hidden RPC trick - // `{unread: false}`. See design doc §3.2 / §3.3. - all: OptionalBoolean, - types: OptionalString, - inject: OptionalBoolean, - wait: OptionalBoolean, - timeoutMs: OptionalFiniteNumber -}) +const CheckParams = z + .object({ + terminal: OptionalString, + unread: OptionalBoolean, + peek: OptionalBoolean, + // Why: `all` surfaces every message for the handle and skips mark-read. + // Previously the only way to ask for "all" was the hidden RPC trick + // `{unread: false}`. See design doc §3.2 / §3.3. + all: OptionalBoolean, + types: OptionalString, + inject: OptionalBoolean, + wait: OptionalBoolean, + timeoutMs: OptionalFiniteNumber + }) + .superRefine((params, ctx) => { + const modes = [ + params.unread === true, + params.peek === true, + params.all === true || params.unread === false + ].filter(Boolean) + if (modes.length > 1) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: 'Choose at most one message read mode: --unread, --peek, or --all.' + }) + } + }) const ReplyParams = z.object({ id: requiredString('Missing --id'), @@ -265,14 +280,14 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ // shim so in-flight callers don't break (see design doc §5). Otherwise // today's behavior is preserved: default is unread-only + mark-read. const showAll = params.all === true || params.unread === false - const showUnread = !showAll + const consumeUnread = !showAll && params.peek !== true const readAndReturn = () => { - const messages = showUnread - ? db.getUnreadMessages(handle, typeFilter) - : db.getAllMessagesForHandle(handle, undefined, typeFilter) + const messages = showAll + ? db.getAllMessagesForHandle(handle, undefined, typeFilter) + : db.getUnreadMessages(handle, typeFilter) - if (showUnread && messages.length > 0) { + if (consumeUnread && messages.length > 0) { // Why: manual coordinators can consume lifecycle messages before // the coordinator loop sees them, but unread `check` is still an // authoritative read path for worker_done/heartbeat. diff --git a/src/main/runtime/rpc/methods/worktree.test.ts b/src/main/runtime/rpc/methods/worktree.test.ts index ef7ef3910f0..8bc68319d2d 100644 --- a/src/main/runtime/rpc/methods/worktree.test.ts +++ b/src/main/runtime/rpc/methods/worktree.test.ts @@ -472,14 +472,18 @@ describe('worktree RPC methods', () => { const runtime = { getRuntimeId: () => 'test-runtime', showRepo: vi.fn().mockResolvedValue(repo), - createManagedWorktree: vi.fn().mockResolvedValue({ worktree: { id: 'wt-1' } }) + createManagedWorktree: vi.fn().mockResolvedValue({ + worktree: { id: 'wt-1' }, + startupTerminal: { spawned: true, handle: 'term_agent' } + }) } as unknown as OrcaRuntimeService const dispatcher = new RpcDispatcher({ runtime, methods: WORKTREE_METHODS }) - await dispatcher.dispatch( + const response = await dispatcher.dispatch( makeRequest('worktree.create', { repo: 'repo-1', name: 'agent-startup', + startupAgent: 'codex', startupCommand: "codex 'summarize repo'", startupCommandDelivery: 'shell-ready', startupEnv: { ORCA_AGENT_MODE: 'direct' }, @@ -492,11 +496,17 @@ describe('worktree RPC methods', () => { }) ) + expect(response).toMatchObject({ + ok: true, + result: { agentTerminalHandle: 'term_agent' } + }) + expect(runtime.createManagedWorktree).toHaveBeenCalledWith( expect.objectContaining({ repoSelector: 'repo-1', name: 'agent-startup', activate: true, + startupAgent: 'codex', startup: { command: "codex 'summarize repo'", startupCommandDelivery: 'shell-ready', diff --git a/src/main/runtime/rpc/methods/worktree.ts b/src/main/runtime/rpc/methods/worktree.ts index f84a26d1d87..8132a96e3f8 100644 --- a/src/main/runtime/rpc/methods/worktree.ts +++ b/src/main/runtime/rpc/methods/worktree.ts @@ -133,7 +133,11 @@ export const WORKTREE_METHODS: RpcMethod[] = [ } }) finishAutomationWorkspaceProvenanceRequest(params.automationProvenanceRequest) - return result + // Why: agent callers need a stable dispatch target without traversing + // terminal-list layout duplicates after creating the worktree. + return params.startupAgent && result.startupTerminal?.handle + ? { ...result, agentTerminalHandle: result.startupTerminal.handle } + : result } catch (error) { releaseAutomationWorkspaceProvenanceRequest(params.automationProvenanceRequest) throw error diff --git a/src/shared/runtime-types.ts b/src/shared/runtime-types.ts index f3328ab66dc..9f91e7dcc8e 100644 --- a/src/shared/runtime-types.ts +++ b/src/shared/runtime-types.ts @@ -10,6 +10,7 @@ import type { BrowserCookieImportResult, BrowserSessionProfile, BrowserSessionProfileSource, + CreateWorktreeResult, GitWorktreeInfo, RemoveWorktreeResult, Repo, @@ -673,6 +674,8 @@ export type RuntimeWorktreeCreateResult = { workspaceLineage?: WorkspaceLineage | null warnings: WorktreeLineageWarning[] warning?: string + startupTerminal?: CreateWorktreeResult['startupTerminal'] + agentTerminalHandle?: string } export type RuntimeWorktreeRemoveResult = RemoveWorktreeResult & {