Files
orca/src/main/runtime/orchestration-cli-subprocess.test.ts
T
JinjingandOrca 26934b11bf fix(orchestration): complete tasks on worker_done + coordinator UX fixes (#8030)
* fix(orchestration): complete worker tasks and improve coordinator UX

* Fix orchestration lifecycle sender resolution and peek/check compat hand

- Lifecycle sends (worker_done/heartbeat) now use ORCA_TERMINAL_HANDLE
  verbatim, skipping the liveness probe and pane remint that could
  block delivery during restarts or mismatch stale-runtime assignee
  handles.
- --peek now round-trips as {peek:true, unread:false} so older runtimes
  that strip unknown params degrade to non-destructive "all" instead of
  mark-read, with client-side filtering to restore peek semantics and a
  clear error when --peek --wait can't be honored.
- Reject combined read-mode flags (--unread/--peek/--all) before calling
  the runtime.
- Distinguish suppressed (already-consumed) lifecycle messages from
  ignored ones so send doesn't wake --wait waiters for stale heartbeats.
- Fix task summary truncation to avoid splitting UTF-16 surrogate pairs
  and to not misreport whitespace normalization as truncation.

* Add shared helper to abbreviate orchestration task specs for brief listi

- Normalizes whitespace and caps spec length at 160 chars, flagging
  truncation separately from whitespace-only changes
- Truncates on UTF-16 code point boundaries to avoid splitting
  surrogate pairs and emitting malformed strings

* Add pane-key identity to worker_done/heartbeat reconciliation and server

- Records the sender's pane key on messages and dispatch contexts so
  worker_done/heartbeat ownership can be verified by the remint-stable
  pane leaf instead of the terminal handle, which is reissued across
  restarts.
- Rejects lifecycle messages from a genuinely foreign pane while still
  tolerating handle remints, tab break-outs, and older CLIs that lack
  pane identity.
- Moves task-spec abbreviation server-side (orchestration.taskList
  --brief) so full specs no longer cross SSH/relay transports, with a
  client-side fallback for older runtimes; consolidates the shared
  abbreviation helper under src/shared.
- Adds a stderr warning when a pre-peek runtime's --peek response hits
  the 100-row cap, since older unread messages may be missing.

* Isolate ORCA_PANE_KEY in CLI test beforeEach to fix leaked senderPaneKey

Co-authored-by: Orca <help@stably.ai>

* Fix pane-key remint bypassing dispatch mutual-exclusion lock

- Dispatch locking only matched on assignee_handle, so a reminted
  terminal handle (tab break-out) could open a second concurrent
  dispatch on the same pane.
- Add leaf-UUID-based pane key comparison (parsePaneKey) as a
  secondary lock, falling back to exact handle match for legacy
  rows without pane keys.

* Update orchestration skill docs for lifecycle authority and CLI flag add

- Clarify that dispatch lifecycle is tied to taskId+dispatchId verified against
  the dispatched pane, not the terminal handle, since handles can be reminted
  after restart
- Document new `check --peek`/`--all` and `task-list --brief` flags, with
  fallback guidance for older CLIs that reject them
- Note that a valid worker_done auto-completes the task/dispatch, so workers
  shouldn't also call task-update manually

---------

Co-authored-by: Orca <help@stably.ai>
2026-07-12 02:15:11 -07:00

279 lines
11 KiB
TypeScript

// 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 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 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 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
// missing) so `pnpm test` works on a fresh checkout without requiring a prior
// `pnpm run build:cli`. The verification gate explicitly builds the CLI
// before running this file.
import { spawn } from 'node:child_process'
import { existsSync, mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { describe, expect, it } from 'vitest'
import { OrcaRuntimeService } from './orca-runtime'
import { OrchestrationDb } from './orchestration/db'
import { OrcaRuntimeRpcServer } from './runtime-rpc'
// Why: Vitest runs tests with `process.cwd()` pinned to the repo root, so
// join against it to locate the compiled CLI regardless of where this test
// file itself lives.
const CLI_PATH = join(process.cwd(), 'out', 'cli', 'index.js')
const describeIfBuilt = existsSync(CLI_PATH) ? describe : describe.skip
async function runBuiltCli(
userDataPath: string,
args: string[],
extraEnv: Record<string, string> = {}
): Promise<{ exitCode: number; stdout: string; stderr: string }> {
const child = spawn(process.execPath, [CLI_PATH, ...args], {
env: {
...process.env,
ORCA_USER_DATA_PATH: userDataPath,
ORCA_TERMINAL_HANDLE: 'term_cli',
...extraEnv
},
stdio: ['ignore', 'pipe', 'pipe']
})
const stdoutChunks: string[] = []
const stderrChunks: string[] = []
child.stdout.setEncoding('utf8')
child.stderr.setEncoding('utf8')
child.stdout.on('data', (d) => stdoutChunks.push(d))
child.stderr.on('data', (d) => stderrChunks.push(d))
const exitCode = await new Promise<number>((resolveExit, rejectExit) => {
child.once('exit', (code) => resolveExit(code ?? 1))
child.once('error', rejectExit)
})
return {
exitCode,
stdout: stdoutChunks.join(''),
stderr: stderrChunks.join('')
}
}
describeIfBuilt('orca orchestration check --wait subprocess (§3.4)', () => {
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:')
runtime.setOrchestrationDb(db)
const server = new OrcaRuntimeRpcServer({ runtime, userDataPath })
await server.start()
try {
// 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 keepaliveMs = 200
const waitTimeoutMs = 1200
const child = spawn(
process.execPath,
[
CLI_PATH,
'orchestration',
'check',
'--wait',
'--timeout-ms',
String(waitTimeoutMs),
'--json'
],
{
env: {
...process.env,
ORCA_USER_DATA_PATH: userDataPath,
ORCA_TERMINAL_HANDLE: 'term_nobody',
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).
stdio: ['ignore', 'pipe', 'pipe']
}
)
const stderrChunks: { at: number; data: string }[] = []
const stdoutChunks: { at: number; data: string }[] = []
const startedAt = Date.now()
child.stderr.setEncoding('utf8')
child.stdout.setEncoding('utf8')
child.stderr.on('data', (d) => stderrChunks.push({ at: Date.now() - startedAt, data: d }))
child.stdout.on('data', (d) => stdoutChunks.push({ at: Date.now() - startedAt, data: d }))
const exitCode = await new Promise<number>((resolveExit, rejectExit) => {
child.once('exit', (code) => resolveExit(code ?? 1))
child.once('error', rejectExit)
})
expect(exitCode).toBe(0)
const stderr = stderrChunks.map((c) => c.data).join('')
const stdout = stdoutChunks.map((c) => c.data).join('')
const keepaliveLines = stderr
.split('\n')
.filter((line) => line.trim().length > 0)
.map((line) => {
try {
return JSON.parse(line) as Record<string, unknown>
} catch {
return null
}
})
.filter((p): p is Record<string, unknown> => p !== null && p._keepalive === true)
// ≥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 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* keepalive chunk must arrive
// strictly before the exit chunk; i.e. we got at least two separate
// 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 keepalives
// leak, and the content parses as valid JSON.
const stdoutTrimmed = stdout.trim()
const stdoutPayload = JSON.parse(stdoutTrimmed) as Record<string, unknown>
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 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
// keepalive line on stderr is a standalone JSON object (so jq can
// match it) and doesn't span multiple lines — assert that each
// 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('_keepalive')) {
expect(() => JSON.parse(line)).not.toThrow()
expect(line).not.toContain('\n')
}
}
} finally {
db.close()
await server.stop()
}
}, 30_000)
})
describeIfBuilt('orca orchestration reset subprocess', () => {
it('validates reset scopes against an isolated runtime through the built CLI', async () => {
const userDataPath = mkdtempSync(join(tmpdir(), 'orca-cli-reset-'))
const runtime = new OrcaRuntimeService()
const db = new OrchestrationDb(':memory:')
runtime.setOrchestrationDb(db)
const server = new OrcaRuntimeRpcServer({ runtime, userDataPath })
await server.start()
try {
const send = await runBuiltCli(userDataPath, [
'orchestration',
'send',
'--to',
'term_target',
'--subject',
'hello',
'--json'
])
expect(send.exitCode, send.stderr).toBe(0)
const create = await runBuiltCli(userDataPath, [
'orchestration',
'task-create',
'--spec',
'throwaway task',
'--json'
])
expect(create.exitCode, create.stderr).toBe(0)
expect(db.getInbox()).toHaveLength(1)
expect(db.listTasks()).toHaveLength(1)
const invalid = await runBuiltCli(userDataPath, [
'orchestration',
'reset',
'--tasks',
'--messages',
'--json'
])
expect(invalid.exitCode).toBe(1)
const invalidPayload = JSON.parse(invalid.stdout) as {
ok: boolean
error: { code: string; message: string }
}
expect(invalidPayload.ok).toBe(false)
expect(invalidPayload.error.code).toBe('invalid_argument')
expect(invalidPayload.error.message).toContain('Choose exactly one reset scope')
expect(db.getInbox()).toHaveLength(1)
expect(db.listTasks()).toHaveLength(1)
const resetTasks = await runBuiltCli(userDataPath, [
'orchestration',
'reset',
'--tasks',
'--json'
])
expect(resetTasks.exitCode, resetTasks.stderr).toBe(0)
expect(JSON.parse(resetTasks.stdout)).toMatchObject({ ok: true, result: { reset: 'tasks' } })
expect(db.getInbox()).toHaveLength(1)
expect(db.listTasks()).toHaveLength(0)
const recreate = await runBuiltCli(userDataPath, [
'orchestration',
'task-create',
'--spec',
'throwaway task after partial reset',
'--json'
])
expect(recreate.exitCode, recreate.stderr).toBe(0)
expect(db.getInbox()).toHaveLength(1)
expect(db.listTasks()).toHaveLength(1)
const resetAll = await runBuiltCli(userDataPath, ['orchestration', 'reset', '--json'])
expect(resetAll.exitCode, resetAll.stderr).toBe(0)
expect(JSON.parse(resetAll.stdout)).toMatchObject({ ok: true, result: { reset: 'all' } })
expect(db.getInbox()).toHaveLength(0)
expect(db.listTasks()).toHaveLength(0)
} finally {
db.close()
await server.stop()
rmSync(userDataPath, { recursive: true, force: true })
}
}, 30_000)
})