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.
This commit is contained in:
Jinjing
2026-07-11 19:27:35 -07:00
parent 929d9971e8
commit dddc6cc452
10 changed files with 246 additions and 75 deletions
+92 -47
View File
@@ -214,7 +214,7 @@ describe('orchestration send structured payload flags', () => {
})
})
it('uses a live ORCA_TERMINAL_HANDLE as worker lifecycle sender metadata', async () => {
it('sends lifecycle messages from ORCA_TERMINAL_HANDLE without a liveness probe', async () => {
process.env.ORCA_TERMINAL_HANDLE = 'term_worker_env'
await invokeSend(
@@ -225,11 +225,8 @@ describe('orchestration send structured payload flags', () => {
])
)
expect(callMock).toHaveBeenCalledTimes(2)
expect(callMock).toHaveBeenNthCalledWith(1, 'terminal.show', {
terminal: 'term_worker_env'
})
expect(callMock).toHaveBeenNthCalledWith(2, 'orchestration.send', {
expect(callMock).toHaveBeenCalledTimes(1)
expect(callMock).toHaveBeenCalledWith('orchestration.send', {
from: 'term_worker_env',
to: 'term_coord',
subject: 'done',
@@ -242,48 +239,31 @@ 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' } } })
it.each(['worker_done', 'heartbeat'] as const)(
'never probes or remints a %s sender even when a pane key is set',
async (type) => {
process.env.ORCA_TERMINAL_HANDLE = 'term_worker_env'
process.env.ORCA_PANE_KEY = 'tab_worker:leaf_worker'
await invokeSend(
new Map<string, string | boolean>([
['to', 'term_coord'],
['subject', 'done'],
['type', 'worker_done']
])
)
await invokeSend(
new Map<string, string | boolean>([
['to', 'term_coord'],
['subject', 'update'],
['type', type]
])
)
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<string, string | boolean>([
['to', 'term_coord'],
['subject', 'done'],
['type', 'worker_done']
])
)
expect(callMock).toHaveBeenCalledTimes(2)
expect(callMock).toHaveBeenNthCalledWith(
2,
'orchestration.send',
expect.objectContaining({ from: 'term_worker_stale' })
)
})
// Why: pre-payload-authority runtimes only complete a worker_done whose
// sender equals the recorded (equally stale) assignee handle, and
// coordinator replies route to the sender row the worker's env-handle
// `check` actually reads — so lifecycle sends must stay env-verbatim.
expect(callMock).toHaveBeenCalledTimes(1)
expect(callMock).toHaveBeenCalledWith(
'orchestration.send',
expect.objectContaining({ from: 'term_worker_env' })
)
}
)
it('reports sender resolution failure instead of raw no_active_terminal', async () => {
getTerminalHandleMock.mockRejectedValue(
@@ -690,9 +670,11 @@ describe('orchestration timeout flag validation', () => {
])
)
// Why: --peek rides with unread:false so pre-peek runtimes fall back to
// the non-consuming all mode instead of the destructive mark-read default.
expect(callMock).toHaveBeenCalledWith('orchestration.check', {
terminal: 'term_worker',
unread: undefined,
unread: false,
peek: true,
all: undefined,
types: undefined,
@@ -702,6 +684,69 @@ describe('orchestration timeout flag validation', () => {
})
})
it('filters already-read rows from a peek response for pre-peek runtimes', async () => {
process.env.ORCA_TERMINAL_HANDLE = 'term_worker'
callMock.mockResolvedValue({
result: {
messages: [
{ id: 'msg_old', from_handle: 'a', subject: 'seen', read: 1 },
{ id: 'msg_new', from_handle: 'a', subject: 'fresh', read: 0 }
],
count: 2,
formatted: 'banners built from all rows'
}
})
vi.mocked(printResult).mockClear()
await invokeCheck(new Map<string, string | boolean>([['peek', true]]))
const response = vi.mocked(printResult).mock.calls[0]?.[0] as {
result: { messages: { id: string }[]; count: number; formatted?: string }
}
expect(response.result.messages.map((m) => m.id)).toEqual(['msg_new'])
expect(response.result.count).toBe(1)
// Why: the pre-peek runtime built `formatted` from all rows, including
// the read one the filter just removed.
expect(response.result.formatted).toBeUndefined()
})
it('rejects combined read modes before calling the runtime', async () => {
process.env.ORCA_TERMINAL_HANDLE = 'term_worker'
callMock.mockClear()
await expect(
invokeCheck(
new Map<string, string | boolean>([
['unread', true],
['peek', true]
])
)
).rejects.toMatchObject({
code: 'invalid_argument',
message: expect.stringContaining('read mode')
})
expect(callMock).not.toHaveBeenCalled()
})
it('fails --peek --wait against a runtime that returned only read rows', async () => {
process.env.ORCA_TERMINAL_HANDLE = 'term_worker'
callMock.mockResolvedValue({
result: {
messages: [{ id: 'msg_old', from_handle: 'a', subject: 'seen', read: 1 }],
count: 1
}
})
await expect(
invokeCheck(
new Map<string, string | boolean>([
['peek', true],
['wait', true]
])
)
).rejects.toMatchObject({ code: 'peek_wait_unsupported' })
})
it.each(invalidTimeoutValues)('rejects invalid ask --timeout-ms: %s', async (_label, value) => {
const flags = new Map<string, string | boolean>([
['to', 'term_coord'],
+51 -15
View File
@@ -77,6 +77,7 @@ type MessageSummary = {
type?: string
body?: string
payload?: string | null
read?: number
}
function getOptionalStructuredMessagePayload(
@@ -132,7 +133,7 @@ async function resolveOrchestrationTerminalHandle(
cwd: string,
client: Parameters<CommandHandler>[0]['client'],
flagName: 'from' | 'terminal',
options: { validateEnvHandle?: boolean; allowStaleEnvHandle?: boolean } = {}
options: { validateEnvHandle?: boolean } = {}
): Promise<string> {
const explicit = getOptionalStringFlag(flags, flagName)
if (explicit) {
@@ -146,15 +147,10 @@ async function resolveOrchestrationTerminalHandle(
// coordinator preambles.
const live = await isLiveTerminalHandle(envHandle, client)
if (!live) {
const reminted = await resolveOrchestrationPaneTerminalHandle(client, {
optional: options.allowStaleEnvHandle
})
const reminted = await resolveOrchestrationPaneTerminalHandle(client)
if (reminted) {
return reminted
}
if (options.allowStaleEnvHandle) {
return envHandle
}
throwNoActiveSenderTerminal()
}
}
@@ -351,12 +347,13 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
const type = getOptionalStringFlag(flags, 'type')
rejectLifecycleGroupRecipient(type, to)
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'
})
// Why: lifecycle senders keep ORCA_TERMINAL_HANDLE verbatim — no liveness
// probe (terminal.show throws runtime_unavailable in the exact mid-restart
// window worker_done must survive) and no pane remint (pre-payload-
// authority runtimes require from === the equally stale assignee_handle,
// and coordinator replies route to the sender row while the worker's own
// `check` reads its env-handle inbox).
const from = await resolveOrchestrationTerminalHandle(flags, cwd, client, 'from')
const result = await client.call<
{ message: { id: string } } | { messages: { id: string }[]; recipients: number }
>('orchestration.send', {
@@ -380,6 +377,16 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
'orchestration check': async ({ flags, client, cwd, json }) => {
const wait = flags.has('wait')
const peek = flags.has('peek')
// Why: enforce mode exclusivity client-side too — an older runtime strips
// the unknown `peek` param and would otherwise execute --unread --peek as
// a destructive mark-read.
if ([flags.has('unread'), peek, flags.has('all')].filter(Boolean).length > 1) {
throw new RuntimeClientError(
'invalid_argument',
'Choose at most one message read mode: --unread, --peek, or --all.'
)
}
const timeoutMs = getOptionalPositiveIntegerValueFlag(flags, 'timeout-ms')
const terminal = await resolveOrchestrationTerminalHandle(flags, cwd, client, 'terminal')
@@ -400,8 +407,12 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
try {
result = await client.call<CheckResult>('orchestration.check', {
terminal,
unread: flags.has('unread') ? true : undefined,
peek: flags.has('peek') ? true : undefined,
// Why: --peek also sends unread:false so runtimes that predate the
// peek param (which their non-strict schema strips) degrade to the
// non-consuming all-messages mode instead of the destructive
// mark-read default; the read filter below restores peek semantics.
unread: flags.has('unread') ? true : peek ? false : undefined,
peek: peek ? true : undefined,
all: flags.has('all') ? true : undefined,
types: getOptionalStringFlag(flags, 'types'),
inject: flags.has('inject') ? true : undefined,
@@ -411,6 +422,31 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
} finally {
stopKeepalive?.()
}
if (peek) {
const unreadOnly = result.result.messages.filter((m) => m.read !== 1)
const removedReadRows = unreadOnly.length !== result.result.messages.length
// Why: read rows in a peek response are the pre-peek-runtime signature
// (its schema stripped `peek` and it ran the all mode). Such a runtime
// returned instead of blocking, so honoring --wait is impossible —
// failing beats silently returning empty before the deadline.
if (wait && removedReadRows && unreadOnly.length === 0) {
throw new RuntimeClientError(
'peek_wait_unsupported',
'The connected runtime does not support --peek with --wait; upgrade the runtime or use --wait without --peek.'
)
}
result = {
...result,
result: {
...result.result,
// Why: a pre-peek runtime builds `formatted` from all rows; drop it
// when the read filter removed any so output matches the peek set.
...(removedReadRows ? { formatted: undefined } : {}),
messages: unreadOnly,
count: unreadOnly.length
}
}
}
printResult(result, json, (r) => {
if (r.formatted) {
return r.formatted
@@ -19,4 +19,20 @@ describe('abbreviateOrchestrationTasks', () => {
expect(task).toEqual({ spec: 'Short task', spec_truncated: false })
})
it('does not report whitespace normalization as truncation', () => {
const [task] = abbreviateOrchestrationTasks([{ spec: 'Short\n\n task' }])
expect(task).toEqual({ spec: 'Short task', spec_truncated: false })
})
it('does not split a surrogate pair at the truncation boundary', () => {
// 158 chars + an astral emoji spanning UTF-16 units 158-159: a naive
// slice(0, 159) would cut the pair and leave a lone high surrogate.
const [task] = abbreviateOrchestrationTasks([{ spec: `${'a'.repeat(158)}😀${'b'.repeat(40)}` }])
expect(task.spec_truncated).toBe(true)
expect(task.spec.isWellFormed()).toBe(true)
expect(task.spec.endsWith('…')).toBe(true)
})
})
+13 -6
View File
@@ -5,14 +5,21 @@ export function abbreviateOrchestrationTasks<T extends { spec: string }>(
): (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
const truncated = spec.length > TASK_SPEC_BRIEF_LENGTH
return {
...task,
spec: abbreviated,
spec_truncated: abbreviated !== task.spec
spec: truncated ? `${truncateAtCodePoint(spec).trimEnd()}…` : spec,
// Why: whitespace normalization alone is not truncation; flagging it
// would make agents re-fetch full specs that --brief already shows.
spec_truncated: truncated
}
})
}
function truncateAtCodePoint(spec: string): string {
const sliced = spec.slice(0, TASK_SPEC_BRIEF_LENGTH - 1)
// Why: a cut through a surrogate pair leaves a lone high surrogate that
// strict JSON consumers reject; drop it rather than emit malformed UTF-16.
const lastUnit = sliced.charCodeAt(sliced.length - 1)
return lastUnit >= 0xd800 && lastUnit <= 0xdbff ? sliced.slice(0, -1) : sliced
}
+1 -1
View File
@@ -116,7 +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).',
'With --agent --json, read the new agent handle from result.agentTerminalHandle; older runtimes return only result.startupTerminal.handle, and may return neither for folder-based repos.',
'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.'
+1 -1
View File
@@ -26,7 +26,7 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [
notes: [
'On Windows PowerShell, quote group addresses such as --to "@all" or --to "@worktree:<id>".',
'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.',
'A worker_done with the active task/dispatch IDs completes that task automatically; on older runtimes the sender must still match the dispatch assignee handle, so avoid overriding --from.',
'Prefer --task-id/--dispatch-id/etc. over raw --payload JSON in worker commands; PowerShell strips JSON quotes easily.'
]
},
@@ -68,7 +68,7 @@ describe('lifecycle reconciliation', () => {
expect(archived?.delivered_at).not.toBeNull()
const lateHeartbeat = insertHeartbeat(dispatch.id, 'term_worker')
expect(reconcileLifecycleMessage(db, lateHeartbeat)).toEqual({ action: 'ignored' })
expect(reconcileLifecycleMessage(db, lateHeartbeat)).toEqual({ action: 'suppressed' })
expect(db.getMessageById(lateHeartbeat.id)).toMatchObject({ read: 1 })
expect(db.getMessageById(lateHeartbeat.id)?.delivered_at).not.toBeNull()
})
@@ -3,6 +3,10 @@ import type { MessageRow } from './types'
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 }
@@ -69,7 +73,7 @@ function reconcileHeartbeatMessage(
// audit history without surfacing obsolete liveness to the coordinator.
db.markAsReadAndDelivered([msg.id])
onLog(`Heartbeat for inactive dispatch ${dispatchId} suppressed`)
return { action: 'ignored' }
return { action: 'suppressed' }
}
// Why: dispatchId-specific writes let the DB ignore late heartbeats for
@@ -90,6 +90,44 @@ describe('orchestration RPC methods', () => {
expect(runtime.deliverPendingMessagesForHandle).toHaveBeenCalledWith('term_b')
})
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()
@@ -704,6 +742,23 @@ describe('orchestration RPC methods', () => {
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' })
+11 -3
View File
@@ -87,10 +87,13 @@ const CheckParams = z
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.all === true || (params.unread === false && params.peek !== true)
].filter(Boolean)
if (modes.length > 1) {
ctx.addIssue({
@@ -217,7 +220,12 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
// 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)
@@ -279,7 +287,7 @@ 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 showAll = params.all === true || (params.unread === false && params.peek !== true)
const consumeUnread = !showAll && params.peek !== true
const readAndReturn = () => {