mirror of
https://github.com/stablyai/orca.git
synced 2026-10-19 08:03:13 +00:00
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>
This commit is contained in:
@@ -943,7 +943,8 @@ class InMemoryOrchestrationMessages {
|
||||
read: 0,
|
||||
sequence: this.sequence,
|
||||
created_at: '1970-01-01 00:00:00',
|
||||
delivered_at: null
|
||||
delivered_at: null,
|
||||
sender_pane_key: null
|
||||
}
|
||||
this.messages.push(row)
|
||||
return row
|
||||
|
||||
@@ -10092,6 +10092,13 @@ export class OrcaRuntimeService {
|
||||
throw new Error('no_active_terminal')
|
||||
}
|
||||
|
||||
// Why: orchestration records the pane key as the remint-stable assignee
|
||||
// identity at dispatch time; null (best-effort) rather than throwing so
|
||||
// dispatch still works for handles without a resolvable pane.
|
||||
getTerminalPaneKey(handle: string): string | null {
|
||||
return this.getPaneKeyForTerminalHandle(handle)
|
||||
}
|
||||
|
||||
resolveTerminalPane(paneKey: string): RuntimeTerminalResolvePane {
|
||||
// Why: the renderer context menu only knows the stable pane key; main owns
|
||||
// the runtime terminal handle that agents and CLI commands can address.
|
||||
@@ -14629,6 +14636,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
|
||||
@@ -14652,6 +14660,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}`
|
||||
@@ -14689,6 +14705,7 @@ export class OrcaRuntimeService {
|
||||
isMainWorktree: worktree.isMainWorktree
|
||||
}
|
||||
},
|
||||
...(startupTerminal ? { startupTerminal } : {}),
|
||||
...(warning ? { warning } : {})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<string, unknown> => p !== null && p._heartbeat === true)
|
||||
.filter((p): p is Record<string, unknown> => 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<string, unknown>
|
||||
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')
|
||||
}
|
||||
|
||||
@@ -144,6 +144,31 @@ describe('Coordinator', () => {
|
||||
expect(runtime.sentMessages.length).toBeGreaterThan(0)
|
||||
})
|
||||
|
||||
it('records the assignee pane key when the runtime can resolve one', async () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const runtime = createMockRuntime()
|
||||
runtime.terminals = [{ handle: 'term_a', worktreeId: 'wt1', connected: true, writable: true }]
|
||||
const withPaneLookup = Object.assign(runtime, {
|
||||
getTerminalPaneKey: (handle: string) => (handle === 'term_a' ? 'tab_a:leaf_a' : null)
|
||||
})
|
||||
|
||||
const task = db.createTask({ spec: 'implement feature' })
|
||||
const coordinator = new Coordinator(db, withPaneLookup, {
|
||||
spec: 'build it',
|
||||
coordinatorHandle: 'coord',
|
||||
pollIntervalMs: 50
|
||||
})
|
||||
const runPromise = coordinator.run()
|
||||
await new Promise((r) => {
|
||||
setTimeout(r, 100)
|
||||
})
|
||||
|
||||
expect(db.getDispatchContext(task.id)?.assignee_pane_key).toBe('tab_a:leaf_a')
|
||||
|
||||
insertWorkerDone(db, { taskId: task.id })
|
||||
await runPromise
|
||||
})
|
||||
|
||||
it('records completedTasks when send reconciled worker_done before coordinator read', async () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const runtime = createMockRuntime()
|
||||
@@ -570,7 +595,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 +604,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 +617,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 () => {
|
||||
|
||||
@@ -29,6 +29,9 @@ export type CoordinatorRuntime = {
|
||||
behind: number
|
||||
recentSubjects: string[]
|
||||
} | null>
|
||||
// Why: optional so lightweight runtime fakes keep compiling; when present,
|
||||
// dispatch records the remint-stable pane identity of the assignee.
|
||||
getTerminalPaneKey?(handle: string): string | null
|
||||
}
|
||||
|
||||
// Why (§3.1): single threshold, no warn/refuse split. Coordinator picked 20
|
||||
@@ -461,7 +464,11 @@ export class Coordinator {
|
||||
}
|
||||
}
|
||||
|
||||
const dispatch = this.db.createDispatchContext(task.id, targetHandle)
|
||||
const dispatch = this.db.createDispatchContext(
|
||||
task.id,
|
||||
targetHandle,
|
||||
this.runtime.getTerminalPaneKey?.(targetHandle) ?? undefined
|
||||
)
|
||||
|
||||
// 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).
|
||||
|
||||
@@ -339,6 +339,53 @@ describe('OrchestrationDb', () => {
|
||||
)
|
||||
})
|
||||
|
||||
// Real leaf UUIDs: pane keys are `${tabId}:${leafUuid}`; only the leaf is
|
||||
// remint-stable identity (tab half can change on pane break-out).
|
||||
const LEAF_A = '11111111-1111-1111-8111-111111111111'
|
||||
const LEAF_B = '22222222-2222-4222-9222-222222222222'
|
||||
|
||||
it('rejects dispatch to a reminted handle on a pane with an active dispatch', () => {
|
||||
const d = createDb()
|
||||
const t1 = d.createTask({ spec: 'first' })
|
||||
const t2 = d.createTask({ spec: 'second' })
|
||||
d.createDispatchContext(t1.id, 'term_old', `tab_1:${LEAF_A}`)
|
||||
|
||||
expect(() => d.createDispatchContext(t2.id, 'term_new', `tab_1:${LEAF_A}`)).toThrow(
|
||||
/already has an active dispatch/
|
||||
)
|
||||
})
|
||||
|
||||
it('rejects dispatch when pane keys share a leaf after break-out', () => {
|
||||
const d = createDb()
|
||||
const t1 = d.createTask({ spec: 'first' })
|
||||
const t2 = d.createTask({ spec: 'second' })
|
||||
d.createDispatchContext(t1.id, 'term_old', `tab_1:${LEAF_A}`)
|
||||
|
||||
expect(() => d.createDispatchContext(t2.id, 'term_new', `tab_2:${LEAF_A}`)).toThrow(
|
||||
/already has an active dispatch/
|
||||
)
|
||||
})
|
||||
|
||||
it('allows concurrent dispatches to different panes', () => {
|
||||
const d = createDb()
|
||||
const t1 = d.createTask({ spec: 'first' })
|
||||
const t2 = d.createTask({ spec: 'second' })
|
||||
d.createDispatchContext(t1.id, 'term_a', `tab_1:${LEAF_A}`)
|
||||
|
||||
expect(() => d.createDispatchContext(t2.id, 'term_b', `tab_1:${LEAF_B}`)).not.toThrow()
|
||||
})
|
||||
|
||||
it('falls back to handle lock when pane keys are missing', () => {
|
||||
const d = createDb()
|
||||
const t1 = d.createTask({ spec: 'first' })
|
||||
const t2 = d.createTask({ spec: 'second' })
|
||||
d.createDispatchContext(t1.id, 'term_worker')
|
||||
|
||||
// New dispatch has a pane key but the active row is legacy (no pane key):
|
||||
// only handle identity can lock; a different handle is free.
|
||||
expect(() => d.createDispatchContext(t2.id, 'term_other', `tab_1:${LEAF_A}`)).not.toThrow()
|
||||
})
|
||||
|
||||
it('allows dispatch to a terminal after previous dispatch completes', () => {
|
||||
const d = createDb()
|
||||
const t1 = d.createTask({ spec: 'first' })
|
||||
@@ -822,6 +869,25 @@ describe('OrchestrationDb', () => {
|
||||
expect(d.getMessageById('msg_v1')?.subject).toBe('pre-migration')
|
||||
})
|
||||
|
||||
it('adds pane-identity columns (v6) and persists them', () => {
|
||||
const path = createV1Snapshot()
|
||||
const d = new OrchestrationDb(path)
|
||||
db = d
|
||||
|
||||
const task = d.createTask({ spec: 'work' })
|
||||
const ctx = d.createDispatchContext(task.id, 'term_a', 'tab_1:leaf_1')
|
||||
expect(d.getDispatchContextById(ctx.id)?.assignee_pane_key).toBe('tab_1:leaf_1')
|
||||
|
||||
const msg = d.insertMessage({
|
||||
from: 'w',
|
||||
to: 'c',
|
||||
subject: 'done',
|
||||
type: 'worker_done',
|
||||
senderPaneKey: 'tab_1:leaf_1'
|
||||
})
|
||||
expect(d.getMessageById(msg.id)?.sender_pane_key).toBe('tab_1:leaf_1')
|
||||
})
|
||||
|
||||
it('is idempotent: opening an already-migrated DB is a no-op', () => {
|
||||
const path = createV1Snapshot()
|
||||
const first = new OrchestrationDb(path)
|
||||
|
||||
@@ -15,6 +15,18 @@ import type {
|
||||
CoordinatorRun
|
||||
} from './types'
|
||||
import { buildOrchestrationTaskDisplayMetadata } from '../../../shared/orchestration-task-display'
|
||||
import { parsePaneKey } from '../../../shared/stable-pane-id'
|
||||
|
||||
// Why: leaf UUID is the remint-stable pane identity; the tab half changes on
|
||||
// break-out. Exact string match covers legacy/unparseable keys.
|
||||
function isEquivalentPaneKey(a: string, b: string): boolean {
|
||||
if (a === b) {
|
||||
return true
|
||||
}
|
||||
const aLeaf = parsePaneKey(a)?.leafId
|
||||
const bLeaf = parsePaneKey(b)?.leafId
|
||||
return Boolean(aLeaf && bLeaf && aLeaf === bLeaf)
|
||||
}
|
||||
|
||||
export type {
|
||||
MessageType,
|
||||
@@ -41,7 +53,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 adds pane-identity columns (dispatch_contexts.assignee_pane_key,
|
||||
// messages.sender_pane_key) so worker_done ownership survives terminal handle
|
||||
// remints without accepting completions from unrelated panes.
|
||||
const SCHEMA_VERSION = 6
|
||||
|
||||
export class OrchestrationDb {
|
||||
private db: Database.Database
|
||||
@@ -75,7 +90,8 @@ export class OrchestrationDb {
|
||||
read INTEGER NOT NULL DEFAULT 0,
|
||||
sequence INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||||
delivered_at TEXT
|
||||
delivered_at TEXT,
|
||||
sender_pane_key TEXT
|
||||
);
|
||||
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_id ON messages(id);
|
||||
@@ -107,6 +123,7 @@ export class OrchestrationDb {
|
||||
id TEXT PRIMARY KEY,
|
||||
task_id TEXT NOT NULL,
|
||||
assignee_handle TEXT,
|
||||
assignee_pane_key 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 +263,14 @@ export class OrchestrationDb {
|
||||
this.db.exec(`ALTER TABLE tasks ADD COLUMN display_name TEXT`)
|
||||
}
|
||||
}
|
||||
if (current < 6) {
|
||||
if (!this.hasColumn('dispatch_contexts', 'assignee_pane_key')) {
|
||||
this.db.exec(`ALTER TABLE dispatch_contexts ADD COLUMN assignee_pane_key TEXT`)
|
||||
}
|
||||
if (!this.hasColumn('messages', 'sender_pane_key')) {
|
||||
this.db.exec(`ALTER TABLE messages ADD COLUMN sender_pane_key TEXT`)
|
||||
}
|
||||
}
|
||||
this.createUndeliveredInboxIndexIfPossible()
|
||||
|
||||
this.db.pragma(`user_version = ${SCHEMA_VERSION}`)
|
||||
@@ -293,11 +318,12 @@ export class OrchestrationDb {
|
||||
priority?: MessagePriority
|
||||
threadId?: string
|
||||
payload?: string
|
||||
senderPaneKey?: string
|
||||
}): MessageRow {
|
||||
const id = generateId('msg')
|
||||
const stmt = this.db.prepare(`
|
||||
INSERT INTO messages (id, from_handle, to_handle, subject, body, type, priority, thread_id, payload)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
INSERT INTO messages (id, from_handle, to_handle, subject, body, type, priority, thread_id, payload, sender_pane_key)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
`)
|
||||
stmt.run(
|
||||
id,
|
||||
@@ -308,7 +334,8 @@ export class OrchestrationDb {
|
||||
msg.type ?? 'status',
|
||||
msg.priority ?? 'normal',
|
||||
msg.threadId ?? null,
|
||||
msg.payload ?? null
|
||||
msg.payload ?? null,
|
||||
msg.senderPaneKey ?? null
|
||||
)
|
||||
return this.db.prepare('SELECT * FROM messages WHERE id = ?').get(id) as MessageRow
|
||||
}
|
||||
@@ -382,6 +409,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 ?')
|
||||
@@ -568,7 +609,14 @@ export class OrchestrationDb {
|
||||
|
||||
// ── Dispatch Contexts ──
|
||||
|
||||
createDispatchContext(taskId: string, assigneeHandle: string): DispatchContextRow {
|
||||
createDispatchContext(
|
||||
taskId: string,
|
||||
assigneeHandle: string,
|
||||
// Why: the pane key is the remint-stable identity behind the handle;
|
||||
// recording it at dispatch time lets worker_done ownership survive
|
||||
// restarts that reissue the handle.
|
||||
assigneePaneKey?: string
|
||||
): DispatchContextRow {
|
||||
const task = this.getTask(taskId)
|
||||
if (!task) {
|
||||
throw new Error(`Task not found: ${taskId}`)
|
||||
@@ -577,11 +625,11 @@ export class OrchestrationDb {
|
||||
throw new Error(`Task ${taskId} is ${task.status}; only ready tasks can be dispatched`)
|
||||
}
|
||||
|
||||
const existing = this.db
|
||||
.prepare(
|
||||
"SELECT * FROM dispatch_contexts WHERE assignee_handle = ? AND status IN ('pending', 'dispatched')"
|
||||
)
|
||||
.get(assigneeHandle) as DispatchContextRow | undefined
|
||||
// Why: handle match covers legacy rows without pane keys; when both the
|
||||
// new assignee and an active row have usable pane keys, also lock on
|
||||
// equivalent pane identity so a reminted handle cannot open a second
|
||||
// concurrent dispatch on the same pane.
|
||||
const existing = this.findActiveDispatchForAssignee(assigneeHandle, assigneePaneKey)
|
||||
|
||||
if (existing) {
|
||||
throw new Error(
|
||||
@@ -599,10 +647,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, assignee_pane_key, status, failure_count, dispatched_at)
|
||||
VALUES (?, ?, ?, ?, 'dispatched', ?, datetime('now'))`
|
||||
)
|
||||
.run(id, taskId, assigneeHandle, priorFailures)
|
||||
.run(id, taskId, assigneeHandle, assigneePaneKey ?? null, priorFailures)
|
||||
|
||||
this.db.prepare("UPDATE tasks SET status = 'dispatched' WHERE id = ?").run(taskId)
|
||||
|
||||
@@ -624,11 +672,38 @@ export class OrchestrationDb {
|
||||
}
|
||||
|
||||
getActiveDispatchForTerminal(handle: string): DispatchContextRow | undefined {
|
||||
return this.db
|
||||
return this.findActiveDispatchForAssignee(handle)
|
||||
}
|
||||
|
||||
private findActiveDispatchForAssignee(
|
||||
assigneeHandle: string,
|
||||
assigneePaneKey?: string
|
||||
): DispatchContextRow | undefined {
|
||||
const byHandle = this.db
|
||||
.prepare(
|
||||
"SELECT * FROM dispatch_contexts WHERE assignee_handle = ? AND status IN ('pending', 'dispatched') LIMIT 1"
|
||||
)
|
||||
.get(handle) as DispatchContextRow | undefined
|
||||
.get(assigneeHandle) as DispatchContextRow | undefined
|
||||
if (byHandle) {
|
||||
return byHandle
|
||||
}
|
||||
|
||||
if (!assigneePaneKey) {
|
||||
return undefined
|
||||
}
|
||||
|
||||
const actives = this.db
|
||||
.prepare(
|
||||
"SELECT * FROM dispatch_contexts WHERE assignee_pane_key IS NOT NULL AND status IN ('pending', 'dispatched')"
|
||||
)
|
||||
.all() as DispatchContextRow[]
|
||||
|
||||
for (const row of actives) {
|
||||
if (row.assignee_pane_key && isEquivalentPaneKey(row.assignee_pane_key, assigneePaneKey)) {
|
||||
return row
|
||||
}
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
|
||||
getLatestDispatchForTerminal(handle: string): DispatchContextRow | undefined {
|
||||
|
||||
@@ -17,6 +17,7 @@ function makeMessage(overrides: Partial<MessageRow> = {}): MessageRow {
|
||||
sequence: 1,
|
||||
created_at: '2026-01-01T00:00:00Z',
|
||||
delivered_at: null,
|
||||
sender_pane_key: null,
|
||||
...overrides
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,191 @@
|
||||
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)
|
||||
})
|
||||
|
||||
// Real leaf UUIDs: pane keys are `${tabId}:${leafUuid}` and only the leaf
|
||||
// half is identity (the tab half changes on pane break-out).
|
||||
const LEAF_A = '11111111-1111-1111-8111-111111111111'
|
||||
const LEAF_B = '22222222-2222-4222-9222-222222222222'
|
||||
|
||||
it('completes worker_done from the dispatched pane after a handle remint', () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const task = db.createTask({ spec: 'work' })
|
||||
const dispatch = db.createDispatchContext(task.id, 'term_before_restart', `tab_w:${LEAF_A}`)
|
||||
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 }),
|
||||
senderPaneKey: `tab_w:${LEAF_A}`
|
||||
})
|
||||
|
||||
expect(reconcileLifecycleMessage(db, message).action).toBe('completed')
|
||||
expect(db.getTask(task.id)?.status).toBe('completed')
|
||||
})
|
||||
|
||||
it('completes worker_done from the same leaf after a pane break-out changed the tab half', () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const task = db.createTask({ spec: 'work' })
|
||||
// Dispatch recorded the post-break-out pane key; the worker shell still
|
||||
// holds the spawn-time key with the old tab id.
|
||||
const dispatch = db.createDispatchContext(task.id, 'term_before_restart', `tab_new:${LEAF_A}`)
|
||||
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 }),
|
||||
senderPaneKey: `tab_old:${LEAF_A}`
|
||||
})
|
||||
|
||||
expect(reconcileLifecycleMessage(db, message).action).toBe('completed')
|
||||
expect(db.getTask(task.id)?.status).toBe('completed')
|
||||
})
|
||||
|
||||
it('completes worker_done when a pane key is unparseable (legacy format)', () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const task = db.createTask({ spec: 'work' })
|
||||
const dispatch = db.createDispatchContext(task.id, 'term_owner', `tab_w:${LEAF_A}`)
|
||||
const message = db.insertMessage({
|
||||
from: 'term_reminted',
|
||||
to: 'term_coordinator',
|
||||
subject: 'Done',
|
||||
type: 'worker_done',
|
||||
payload: JSON.stringify({ taskId: task.id, dispatchId: dispatch.id }),
|
||||
senderPaneKey: 'tab_w:42'
|
||||
})
|
||||
|
||||
expect(reconcileLifecycleMessage(db, message).action).toBe('completed')
|
||||
})
|
||||
|
||||
it('ignores worker_done sent from a different pane', () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const task = db.createTask({ spec: 'work' })
|
||||
const dispatch = db.createDispatchContext(task.id, 'term_owner', `tab_w1:${LEAF_A}`)
|
||||
const logs: string[] = []
|
||||
const message = db.insertMessage({
|
||||
from: 'term_other_worker',
|
||||
to: 'term_coordinator',
|
||||
subject: 'Done',
|
||||
type: 'worker_done',
|
||||
payload: JSON.stringify({ taskId: task.id, dispatchId: dispatch.id }),
|
||||
senderPaneKey: `tab_w2:${LEAF_B}`
|
||||
})
|
||||
|
||||
expect(reconcileLifecycleMessage(db, message, (line) => logs.push(line))).toEqual({
|
||||
action: 'ignored'
|
||||
})
|
||||
expect(db.getTask(task.id)?.status).toBe('dispatched')
|
||||
expect(db.getDispatchContextById(dispatch.id)?.status).toBe('dispatched')
|
||||
expect(logs.some((line) => line.includes(`expected pane tab_w1:${LEAF_A}`))).toBe(true)
|
||||
})
|
||||
|
||||
it('ignores a heartbeat sent from a different pane without recording liveness', () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const task = db.createTask({ spec: 'work' })
|
||||
const dispatch = db.createDispatchContext(task.id, 'term_owner', `tab_w1:${LEAF_A}`)
|
||||
const heartbeat = db.insertMessage({
|
||||
from: 'term_other_worker',
|
||||
to: 'term_coordinator',
|
||||
subject: 'alive',
|
||||
type: 'heartbeat',
|
||||
payload: JSON.stringify({ dispatchId: dispatch.id }),
|
||||
senderPaneKey: `tab_w2:${LEAF_B}`
|
||||
})
|
||||
|
||||
expect(reconcileLifecycleMessage(db, heartbeat)).toEqual({ action: 'ignored' })
|
||||
expect(db.getDispatchContextById(dispatch.id)?.last_heartbeat_at).toBeNull()
|
||||
})
|
||||
|
||||
it('records a heartbeat whose pane key drifted only in the tab half', () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const task = db.createTask({ spec: 'work' })
|
||||
const dispatch = db.createDispatchContext(task.id, 'term_owner', `tab_new:${LEAF_A}`)
|
||||
const heartbeat = db.insertMessage({
|
||||
from: 'term_owner',
|
||||
to: 'term_coordinator',
|
||||
subject: 'alive',
|
||||
type: 'heartbeat',
|
||||
payload: JSON.stringify({ dispatchId: dispatch.id }),
|
||||
senderPaneKey: `tab_old:${LEAF_A}`
|
||||
})
|
||||
|
||||
expect(reconcileLifecycleMessage(db, heartbeat)).toEqual({
|
||||
action: 'heartbeat_recorded',
|
||||
dispatchId: dispatch.id
|
||||
})
|
||||
expect(db.getDispatchContextById(dispatch.id)?.last_heartbeat_at).not.toBeNull()
|
||||
})
|
||||
|
||||
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: 'suppressed' })
|
||||
expect(db.getMessageById(lateHeartbeat.id)).toMatchObject({ read: 1 })
|
||||
expect(db.getMessageById(lateHeartbeat.id)?.delivered_at).not.toBeNull()
|
||||
})
|
||||
})
|
||||
@@ -1,8 +1,26 @@
|
||||
import type { OrchestrationDb } from './db'
|
||||
import type { MessageRow } from './types'
|
||||
import { parsePaneKey } from '../../../shared/stable-pane-id'
|
||||
|
||||
// Why: the tab half of a pane key changes when a pane is broken out into its
|
||||
// own tab, so only the leaf UUID is identity. Reject only when both keys
|
||||
// parse and name different leaves; an unparseable (e.g. legacy numeric) key
|
||||
// degrades to payload authority rather than stranding a completion.
|
||||
function isForeignPane(assigneePaneKey: string, senderPaneKey: string): boolean {
|
||||
if (assigneePaneKey === senderPaneKey) {
|
||||
return false
|
||||
}
|
||||
const assigneeLeaf = parsePaneKey(assigneePaneKey)?.leafId
|
||||
const senderLeaf = parsePaneKey(senderPaneKey)?.leafId
|
||||
return Boolean(assigneeLeaf && senderLeaf && assigneeLeaf !== senderLeaf)
|
||||
}
|
||||
|
||||
export type LifecycleReconciliationResult =
|
||||
| { action: 'ignored' }
|
||||
// Why: `suppressed` means the message was consumed at reconcile time (marked
|
||||
// read); senders must not wake waiters for it, unlike `ignored` rows that
|
||||
// stay unread and still need delivery.
|
||||
| { action: 'suppressed' }
|
||||
| { action: 'completed'; taskId: string; dispatchId: string }
|
||||
| { action: 'heartbeat_recorded'; dispatchId: string }
|
||||
|
||||
@@ -63,6 +81,28 @@ 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: 'suppressed' }
|
||||
}
|
||||
|
||||
if (
|
||||
dispatch.assignee_pane_key &&
|
||||
msg.sender_pane_key &&
|
||||
isForeignPane(dispatch.assignee_pane_key, msg.sender_pane_key)
|
||||
) {
|
||||
// Why: a wrong-pane heartbeat must not refresh liveness — it would mask
|
||||
// a hung assignee behind another agent's timer.
|
||||
onLog(
|
||||
`Heartbeat for dispatch ${dispatchId} came from pane ${msg.sender_pane_key}, expected pane ${dispatch.assignee_pane_key}; ignored`
|
||||
)
|
||||
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)
|
||||
@@ -112,10 +152,25 @@ function reconcileWorkerDoneMessage(
|
||||
return { action: 'ignored' }
|
||||
}
|
||||
if (dispatch.assignee_handle !== msg.from_handle) {
|
||||
// Why: pane leaves are the remint-stable identity behind handles. When
|
||||
// both sides carry one, a foreign leaf is a different pane completing
|
||||
// someone else's task — reject it; the same leaf is the same pane after
|
||||
// a handle remint or tab break-out. Without pane data (older CLI,
|
||||
// sessions without ORCA_PANE_KEY) payload IDs stay the completion
|
||||
// authority.
|
||||
if (
|
||||
dispatch.assignee_pane_key &&
|
||||
msg.sender_pane_key &&
|
||||
isForeignPane(dispatch.assignee_pane_key, msg.sender_pane_key)
|
||||
) {
|
||||
onLog(
|
||||
`Warning: worker_done for dispatch ${dispatchId} came from pane ${msg.sender_pane_key}, expected pane ${dispatch.assignee_pane_key}; ignored`
|
||||
)
|
||||
return { action: 'ignored' }
|
||||
}
|
||||
onLog(
|
||||
`Warning: worker_done for dispatch ${dispatchId} came from ${msg.from_handle}, expected ${dispatch.assignee_handle ?? '<unknown>'}`
|
||||
`Warning: worker_done for dispatch ${dispatchId} came from ${msg.from_handle}, expected ${dispatch.assignee_handle ?? '<unknown>'}; 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 +198,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)
|
||||
}
|
||||
|
||||
@@ -32,6 +32,7 @@ export type MessageRow = {
|
||||
sequence: number
|
||||
created_at: string
|
||||
delivered_at: string | null
|
||||
sender_pane_key: string | null
|
||||
}
|
||||
|
||||
export type TaskRow = {
|
||||
@@ -52,6 +53,7 @@ export type DispatchContextRow = {
|
||||
id: string
|
||||
task_id: string
|
||||
assignee_handle: string | null
|
||||
assignee_pane_key: string | null
|
||||
status: DispatchStatus
|
||||
failure_count: number
|
||||
last_failure: string | null
|
||||
|
||||
@@ -90,6 +90,59 @@ describe('orchestration RPC methods', () => {
|
||||
expect(runtime.deliverPendingMessagesForHandle).toHaveBeenCalledWith('term_b')
|
||||
})
|
||||
|
||||
it('stores the sender pane key on the message row', async () => {
|
||||
setup()
|
||||
vi.spyOn(runtime, 'deliverPendingMessagesForHandle').mockImplementation(() => {})
|
||||
vi.spyOn(runtime, 'notifyMessageArrived').mockImplementation(() => {})
|
||||
|
||||
const result = (await call('orchestration.send', {
|
||||
from: 'term_a',
|
||||
to: 'term_b',
|
||||
subject: 'hello',
|
||||
senderPaneKey: 'tab_a:leaf_a'
|
||||
})) as { message: { id: string } }
|
||||
|
||||
expect(db.getMessageById(result.message.id)?.sender_pane_key).toBe('tab_a:leaf_a')
|
||||
})
|
||||
|
||||
it('does not wake waiters for a heartbeat suppressed at send time', async () => {
|
||||
setup()
|
||||
const task = db.createTask({ spec: 'work' })
|
||||
const dispatch = db.createDispatchContext(task.id, 'term_worker')
|
||||
db.updateTaskStatus(task.id, 'completed')
|
||||
vi.spyOn(runtime, 'deliverPendingMessagesForHandle').mockImplementation(() => {})
|
||||
const notify = vi.spyOn(runtime, 'notifyMessageArrived').mockImplementation(() => {})
|
||||
|
||||
const result = (await call('orchestration.send', {
|
||||
from: 'term_worker',
|
||||
to: 'term_coord',
|
||||
subject: 'alive',
|
||||
type: 'heartbeat',
|
||||
payload: JSON.stringify({ dispatchId: dispatch.id })
|
||||
})) as { message: { id: string } }
|
||||
|
||||
expect(notify).not.toHaveBeenCalled()
|
||||
expect(db.getMessageById(result.message.id)).toMatchObject({ read: 1 })
|
||||
})
|
||||
|
||||
it('still wakes waiters for a heartbeat on an active dispatch', async () => {
|
||||
setup()
|
||||
const task = db.createTask({ spec: 'work' })
|
||||
const dispatch = db.createDispatchContext(task.id, 'term_worker')
|
||||
vi.spyOn(runtime, 'deliverPendingMessagesForHandle').mockImplementation(() => {})
|
||||
const notify = vi.spyOn(runtime, 'notifyMessageArrived').mockImplementation(() => {})
|
||||
|
||||
await call('orchestration.send', {
|
||||
from: 'term_worker',
|
||||
to: 'term_coord',
|
||||
subject: 'alive',
|
||||
type: 'heartbeat',
|
||||
payload: JSON.stringify({ dispatchId: dispatch.id })
|
||||
})
|
||||
|
||||
expect(notify).toHaveBeenCalledWith('term_coord', 'heartbeat')
|
||||
})
|
||||
|
||||
it('rejects missing --to', () => {
|
||||
const method = findMethod('orchestration.send')
|
||||
expect(() => method.params!.parse({ subject: 'hi' })).toThrow()
|
||||
@@ -598,11 +651,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 +666,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 +723,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 +744,36 @@ 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("treats the CLI's {peek, unread:false} compat pair as peek, not all", async () => {
|
||||
setup()
|
||||
const seen = db.insertMessage({ from: 'a', to: 'b', subject: 'seen' })
|
||||
db.markAsRead([seen.id])
|
||||
db.insertMessage({ from: 'a', to: 'b', subject: 'fresh' })
|
||||
|
||||
const result = (await call('orchestration.check', {
|
||||
terminal: 'b',
|
||||
peek: true,
|
||||
unread: false
|
||||
})) as { messages: { subject: string }[]; count: number }
|
||||
|
||||
expect(result.count).toBe(1)
|
||||
expect(result.messages[0]?.subject).toBe('fresh')
|
||||
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' })
|
||||
@@ -1000,6 +1088,24 @@ describe('orchestration RPC methods', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('orchestration.taskList --brief', () => {
|
||||
it('abbreviates specs server-side so full text never crosses the wire', async () => {
|
||||
setup()
|
||||
db.createTask({ spec: `First line\n${'detail '.repeat(40)}` })
|
||||
db.createTask({ spec: 'Short task' })
|
||||
|
||||
const result = (await call('orchestration.taskList', { brief: true })) as {
|
||||
tasks: { spec: string; spec_truncated: boolean }[]
|
||||
}
|
||||
|
||||
const [long, short] = result.tasks
|
||||
expect(long.spec).toHaveLength(160)
|
||||
expect(long.spec_truncated).toBe(true)
|
||||
expect(short.spec).toBe('Short task')
|
||||
expect(short.spec_truncated).toBe(false)
|
||||
})
|
||||
})
|
||||
|
||||
describe('orchestration.taskUpdate', () => {
|
||||
it('updates task status', async () => {
|
||||
setup()
|
||||
@@ -1050,6 +1156,20 @@ describe('orchestration RPC methods', () => {
|
||||
expect(result.dispatch.status).toBe('dispatched')
|
||||
})
|
||||
|
||||
it('records the assignee pane key on the dispatch context', async () => {
|
||||
setup()
|
||||
vi.spyOn(runtime, 'getTerminalPaneKey').mockReturnValue('tab_w:leaf_w')
|
||||
const task = db.createTask({ spec: 'work' })
|
||||
|
||||
const result = (await call('orchestration.dispatch', {
|
||||
task: task.id,
|
||||
to: 'term_a'
|
||||
})) as { dispatch: { id: string } }
|
||||
|
||||
expect(runtime.getTerminalPaneKey).toHaveBeenCalledWith('term_a')
|
||||
expect(db.getDispatchContextById(result.dispatch.id)?.assignee_pane_key).toBe('tab_w:leaf_w')
|
||||
})
|
||||
|
||||
it('rejects dispatch for a pending task', async () => {
|
||||
setup()
|
||||
const parent = db.createTask({ spec: 'parent' })
|
||||
|
||||
@@ -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 { abbreviateOrchestrationTasks } from '../../../../shared/orchestration-task-summary'
|
||||
import { ORCHESTRATION_GATE_METHODS } from './orchestration-gates'
|
||||
|
||||
const MESSAGE_TYPES: MessageType[] = [
|
||||
@@ -54,6 +55,9 @@ const SendParams = z
|
||||
priority: z.enum(['normal', 'high', 'urgent']).optional(),
|
||||
threadId: OptionalString,
|
||||
payload: OptionalString,
|
||||
// Why: the sender's pane key is the remint-stable identity used to verify
|
||||
// worker_done/heartbeat ownership; the from handle stays routing metadata.
|
||||
senderPaneKey: OptionalString,
|
||||
devMode: OptionalBoolean
|
||||
})
|
||||
.superRefine((params, ctx) => {
|
||||
@@ -72,18 +76,36 @@ 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) => {
|
||||
// Why: the CLI encodes --peek as {peek:true, unread:false} so pre-peek
|
||||
// runtimes degrade to the non-consuming all mode; that pair is one mode,
|
||||
// not a conflict.
|
||||
const modes = [
|
||||
params.unread === true,
|
||||
params.peek === true,
|
||||
params.all === true || (params.unread === false && params.peek !== true)
|
||||
].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'),
|
||||
@@ -110,7 +132,10 @@ const TaskCreateParams = z.object({
|
||||
|
||||
const TaskListParams = z.object({
|
||||
status: z.enum(['pending', 'ready', 'dispatched', 'completed', 'failed', 'blocked']).optional(),
|
||||
ready: OptionalBoolean
|
||||
ready: OptionalBoolean,
|
||||
// Why: truncating specs server-side keeps `--brief` cheap over SSH/relay
|
||||
// transports instead of shipping full specs the CLI then throws away.
|
||||
brief: OptionalBoolean
|
||||
})
|
||||
|
||||
const TaskUpdateParams = z.object({
|
||||
@@ -195,14 +220,20 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
type: params.type as MessageType,
|
||||
priority: params.priority as MessagePriority,
|
||||
threadId: params.threadId,
|
||||
payload: params.payload
|
||||
payload: params.payload,
|
||||
senderPaneKey: params.senderPaneKey
|
||||
})
|
||||
// Why: worker_done/heartbeat sent via `send` must release the dispatch
|
||||
// lock before waking recipients — a coordinator woken by delivery may
|
||||
// immediately dispatch to the same terminal, which fails if the lock
|
||||
// is still held.
|
||||
if (msg.type === 'worker_done' || msg.type === 'heartbeat') {
|
||||
reconcileLifecycleMessage(db, msg)
|
||||
const reconciled = reconcileLifecycleMessage(db, msg)
|
||||
// Why: a suppressed message is already read; waking a `check --wait`
|
||||
// waiter for it would return an empty result before the deadline.
|
||||
if (reconciled.action === 'suppressed') {
|
||||
return { message: msg }
|
||||
}
|
||||
}
|
||||
runtime.deliverPendingMessagesForHandle(params.to)
|
||||
runtime.notifyMessageArrived(params.to, msg.type)
|
||||
@@ -231,7 +262,8 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
type: params.type as MessageType,
|
||||
priority: params.priority as MessagePriority,
|
||||
threadId,
|
||||
payload: params.payload
|
||||
payload: params.payload,
|
||||
senderPaneKey: params.senderPaneKey
|
||||
})
|
||||
)
|
||||
for (const message of messages) {
|
||||
@@ -264,15 +296,15 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
// Explicit `unread: false` is also honored for one release as a compat
|
||||
// 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 showAll = params.all === true || (params.unread === false && params.peek !== true)
|
||||
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.
|
||||
@@ -406,7 +438,10 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
}
|
||||
return base
|
||||
})
|
||||
return { tasks, count: tasks.length }
|
||||
return {
|
||||
tasks: params.brief ? abbreviateOrchestrationTasks(tasks) : tasks,
|
||||
count: tasks.length
|
||||
}
|
||||
}
|
||||
}),
|
||||
|
||||
@@ -475,7 +510,11 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
}
|
||||
}
|
||||
|
||||
const ctx = db.createDispatchContext(params.task, to)
|
||||
const ctx = db.createDispatchContext(
|
||||
params.task,
|
||||
to,
|
||||
runtime.getTerminalPaneKey(to) ?? undefined
|
||||
)
|
||||
|
||||
// Why: preamble is built here (not before ctx) so `dispatchId` can be
|
||||
// the real ctx.id — the preamble-hardening PR made dispatchId required
|
||||
|
||||
@@ -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',
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user