From 2acf8549f30900cb77e59e9c3d5bec082f152ed3 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Sat, 13 Jun 2026 21:28:12 -0700 Subject: [PATCH] Prevent worker completion reports from reaching unrelated agents (#5333) --- skills/orchestration/SKILL.md | 3 + src/cli/handlers/orchestration.test.ts | 70 ++++++++++ src/cli/handlers/orchestration.ts | 17 ++- src/cli/specs/orchestration.ts | 1 + .../runtime/rpc/methods/orchestration.test.ts | 124 +++++++++++++++++- src/main/runtime/rpc/methods/orchestration.ts | 64 +++++---- 6 files changed, 254 insertions(+), 25 deletions(-) diff --git a/skills/orchestration/SKILL.md b/skills/orchestration/SKILL.md index fd5b22e454a..089cce6e987 100644 --- a/skills/orchestration/SKILL.md +++ b/skills/orchestration/SKILL.md @@ -69,6 +69,9 @@ Rules: - `check --wait` returns one message at a time. If N workers may finish together, loop N times and dispatch newly ready tasks after each completion. - Group addresses include `@all`, `@idle`, `@claude`, `@codex`, `@opencode`, `@gemini`, `@droid`, and `@worktree:`. - Message types include `status`, `dispatch`, `worker_done`, `merge_ready`, `escalation`, `handoff`, `decision_gate`, and `heartbeat`. +- Use group addresses only for messages that are genuinely useful to many terminals, such as `status` broadcasts or intentional fan-out questions. Do not send dispatch lifecycle messages to groups. +- `worker_done` must target the concrete coordinator handle from the live preamble. It is completion authority for one dispatch; group fanout would create false lifecycle mail in unrelated terminals. +- `heartbeat` is also dispatch-scoped. Send it only to the concrete coordinator handle with both `taskId` and `dispatchId`; use `status` for broad progress updates. ## Tasks And Dispatch diff --git a/src/cli/handlers/orchestration.test.ts b/src/cli/handlers/orchestration.test.ts index 8f1cc501db2..15a4791c2df 100644 --- a/src/cli/handlers/orchestration.test.ts +++ b/src/cli/handlers/orchestration.test.ts @@ -1,14 +1,20 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' const callMock = vi.fn() +const getTerminalHandleMock = vi.hoisted(() => vi.fn()) const originalTerminalHandle = process.env.ORCA_TERMINAL_HANDLE +function lifecycleGroupRecipientError(type: 'worker_done' | 'heartbeat'): string { + return `${type} messages must be sent to a concrete coordinator terminal handle, not a group address.` +} // Why: isolate the handler's flag-to-param mapping; printResult only writes output. vi.mock('../format', () => ({ printResult: vi.fn() })) +vi.mock('../selectors', () => ({ getTerminalHandle: getTerminalHandleMock })) import { ORCHESTRATION_HANDLERS } from './orchestration' afterEach(() => { + getTerminalHandleMock.mockReset() if (originalTerminalHandle === undefined) { delete process.env.ORCA_TERMINAL_HANDLE } else { @@ -59,6 +65,7 @@ describe('orchestration reset CLI handler', () => { describe('orchestration send structured payload flags', () => { beforeEach(() => { callMock.mockReset().mockResolvedValue({ result: { message: { id: 'msg_1' } } }) + getTerminalHandleMock.mockReset() delete process.env.ORCA_TERMINAL_HANDLE }) @@ -116,6 +123,69 @@ describe('orchestration send structured payload flags', () => { ).rejects.toThrow(/structured payload/) expect(callMock).not.toHaveBeenCalled() }) + + it('rejects worker_done group sends before resolving a sender handle', async () => { + getTerminalHandleMock.mockRejectedValue(new Error('sender resolution should not run')) + + await expect( + invokeSend( + new Map([ + ['to', '@all'], + ['subject', 'done'], + ['type', 'worker_done'] + ]) + ) + ).rejects.toMatchObject({ + code: 'invalid_argument', + message: lifecycleGroupRecipientError('worker_done') + }) + + expect(getTerminalHandleMock).not.toHaveBeenCalled() + expect(callMock).not.toHaveBeenCalled() + }) + + it('rejects heartbeat group sends before resolving a sender handle', async () => { + getTerminalHandleMock.mockRejectedValue(new Error('sender resolution should not run')) + + await expect( + invokeSend( + new Map([ + ['to', '@idle'], + ['subject', 'alive'], + ['type', 'heartbeat'] + ]) + ) + ).rejects.toMatchObject({ + code: 'invalid_argument', + message: lifecycleGroupRecipientError('heartbeat') + }) + + expect(getTerminalHandleMock).not.toHaveBeenCalled() + expect(callMock).not.toHaveBeenCalled() + }) + + it('continues to allow worker_done to a concrete terminal handle', async () => { + await invokeSend( + new Map([ + ['from', 'term_worker'], + ['to', 'term_coord'], + ['subject', 'done'], + ['type', 'worker_done'] + ]) + ) + + expect(callMock).toHaveBeenCalledWith('orchestration.send', { + from: 'term_worker', + to: 'term_coord', + subject: 'done', + body: undefined, + type: 'worker_done', + priority: undefined, + threadId: undefined, + payload: undefined, + devMode: false + }) + }) }) describe('orchestration timeout flag validation', () => { diff --git a/src/cli/handlers/orchestration.ts b/src/cli/handlers/orchestration.ts index a697842aa1e..2064eca9c5a 100644 --- a/src/cli/handlers/orchestration.ts +++ b/src/cli/handlers/orchestration.ts @@ -14,6 +14,9 @@ import { getTerminalHandle } from '../selectors' // parent process the subprocess is alive without flooding logs. See design // doc ยง3.4. const DEFAULT_HEARTBEAT_INTERVAL_MS = 15_000 +function getLifecycleGroupRecipientError(type: 'worker_done' | 'heartbeat'): string { + return `${type} messages must be sent to a concrete coordinator terminal handle, not a group address.` +} // Why: test-only escape hatch so subprocess tests can verify the feature in // under 10 s rather than needing a full 15 s silence window. Production users @@ -162,17 +165,27 @@ function getOptionalPositiveIntegerValueFlag( return value } +function rejectLifecycleGroupRecipient(type: string | undefined, to: string): void { + if ((type === 'worker_done' || type === 'heartbeat') && to.startsWith('@')) { + throw new RuntimeClientError('invalid_argument', getLifecycleGroupRecipientError(type)) + } +} + export const ORCHESTRATION_HANDLERS: Record = { 'orchestration send': async ({ flags, client, cwd, json }) => { + const to = getRequiredStringFlag(flags, 'to') + const type = getOptionalStringFlag(flags, 'type') + rejectLifecycleGroupRecipient(type, to) + const from = await resolveOrchestrationTerminalHandle(flags, cwd, client, 'from') const result = await client.call< { message: { id: string } } | { messages: { id: string }[]; recipients: number } >('orchestration.send', { from, - to: getRequiredStringFlag(flags, 'to'), + to, subject: getRequiredStringFlag(flags, 'subject'), body: getOptionalStringFlag(flags, 'body'), - type: getOptionalStringFlag(flags, 'type'), + type, priority: getOptionalStringFlag(flags, 'priority'), threadId: getOptionalStringFlag(flags, 'thread-id'), payload: getOptionalStructuredMessagePayload(flags), diff --git a/src/cli/specs/orchestration.ts b/src/cli/specs/orchestration.ts index b05bb60a319..da78131d785 100644 --- a/src/cli/specs/orchestration.ts +++ b/src/cli/specs/orchestration.ts @@ -25,6 +25,7 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ ], notes: [ 'On Windows PowerShell, quote group addresses such as --to "@all" or --to "@worktree:".', + 'worker_done and heartbeat must target a concrete coordinator terminal handle; use status for broadcast updates.', 'Prefer --task-id/--dispatch-id/etc. over raw --payload JSON in worker commands; PowerShell strips JSON quotes easily.' ] }, diff --git a/src/main/runtime/rpc/methods/orchestration.test.ts b/src/main/runtime/rpc/methods/orchestration.test.ts index 57687ede314..7c858ad3286 100644 --- a/src/main/runtime/rpc/methods/orchestration.test.ts +++ b/src/main/runtime/rpc/methods/orchestration.test.ts @@ -1,11 +1,16 @@ /* eslint-disable max-lines -- Why: orchestration tests share a mock runtime factory; splitting by method would duplicate 40 lines of setup per file without improving clarity. */ import { afterEach, describe, expect, it, vi } from 'vitest' import { ORCHESTRATION_METHODS } from './orchestration' -import { buildRegistry, type RpcContext } from '../core' +import { RpcDispatcher } from '../dispatcher' +import { buildRegistry, type RpcContext, type RpcRequest } from '../core' import { OrchestrationDb } from '../../orchestration/db' import { OrcaRuntimeService } from '../../orca-runtime' import type { RuntimeTerminalSummary } from '../../../../shared/runtime-types' +function lifecycleGroupRecipientError(type: 'worker_done' | 'heartbeat'): string { + return `${type} messages must be sent to a concrete coordinator terminal handle, not a group address.` +} + describe('orchestration RPC methods', () => { let db: OrchestrationDb let dbOpen = false @@ -45,6 +50,10 @@ describe('orchestration RPC methods', () => { return method.handler(parsed, ctx) } + function makeRequest(method: string, params: Record): RpcRequest { + return { id: 'req_1', authToken: 'token', method, params } + } + it('registers all expected methods', () => { const registry = buildRegistry(ORCHESTRATION_METHODS) expect(registry.size).toBe(16) @@ -97,6 +106,70 @@ describe('orchestration RPC methods', () => { expect(() => method.params!.parse({ to: 'b', subject: 'hi', priority: 'medium' })).toThrow() }) + it.each(['@all', '@idle', '@worktree:wt_1', '@codex', '@nobody'])( + 'rejects worker_done to group recipient %s without inserting rows', + async (to) => { + setup() + const listTerminals = vi.spyOn(runtime, 'listTerminals') + + await expect( + call('orchestration.send', { + from: 'term_worker', + to, + subject: 'done', + type: 'worker_done' + }) + ).rejects.toThrow(lifecycleGroupRecipientError('worker_done')) + + expect(db.getInbox(100)).toHaveLength(0) + expect(listTerminals).not.toHaveBeenCalled() + } + ) + + it('rejects worker_done groups before terminal listing failures can win', async () => { + setup() + const listTerminals = vi + .spyOn(runtime, 'listTerminals') + .mockRejectedValue(new Error('terminal listing failed')) + + await expect( + call('orchestration.send', { + from: 'term_worker', + to: '@all', + subject: 'done', + type: 'worker_done' + }) + ).rejects.toThrow(lifecycleGroupRecipientError('worker_done')) + + expect(listTerminals).not.toHaveBeenCalled() + expect(db.getInbox(100)).toHaveLength(0) + }) + + it('returns invalid_argument for worker_done group sends through the dispatcher', async () => { + setup() + const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }) + const listTerminals = vi.spyOn(runtime, 'listTerminals') + + const response = await dispatcher.dispatch( + makeRequest('orchestration.send', { + from: 'term_worker', + to: '@all', + subject: 'done', + type: 'worker_done' + }) + ) + + expect(response).toMatchObject({ + ok: false, + error: { + code: 'invalid_argument', + message: lifecycleGroupRecipientError('worker_done') + } + }) + expect(listTerminals).not.toHaveBeenCalled() + expect(db.getInbox(100)).toHaveLength(0) + }) + function makeSummary( handle: string, opts: Partial = {} @@ -146,6 +219,55 @@ describe('orchestration RPC methods', () => { expect(recipients).toEqual(['term_b', 'term_c']) }) + it('continues to fan out status messages to groups', async () => { + setupWithTerminals([makeSummary('term_a'), makeSummary('term_b'), makeSummary('term_c')]) + + const result = (await call('orchestration.send', { + from: 'term_a', + to: '@all', + subject: 'status broadcast', + type: 'status' + })) as { messages: { to_handle: string; type: string }[]; recipients: number } + + expect(result.recipients).toBe(2) + expect(result.messages.map((m) => m.to_handle).sort()).toEqual(['term_b', 'term_c']) + expect(result.messages.every((m) => m.type === 'status')).toBe(true) + }) + + it('rejects heartbeat group sends before inserting rows', async () => { + setup() + const listTerminals = vi.spyOn(runtime, 'listTerminals') + + await expect( + call('orchestration.send', { + from: 'term_worker', + to: '@all', + subject: 'alive', + type: 'heartbeat', + payload: JSON.stringify({ taskId: 'task_1', dispatchId: 'ctx_1' }) + }) + ).rejects.toThrow(lifecycleGroupRecipientError('heartbeat')) + + expect(listTerminals).not.toHaveBeenCalled() + expect(db.getInbox(100)).toHaveLength(0) + }) + + it('continues to send worker_done to a concrete terminal handle', async () => { + setup() + + const result = (await call('orchestration.send', { + from: 'term_worker', + to: 'term_coord', + subject: 'done', + type: 'worker_done', + payload: JSON.stringify({ taskId: 'task_1', dispatchId: 'ctx_1' }) + })) as { message: { to_handle: string; type: string; payload: string | null } } + + expect(result.message.to_handle).toBe('term_coord') + expect(result.message.type).toBe('worker_done') + expect(result.message.payload).toBe(JSON.stringify({ taskId: 'task_1', dispatchId: 'ctx_1' })) + }) + it('fans out @idle to only idle agents', async () => { setupWithTerminals([makeSummary('term_a'), makeSummary('term_b'), makeSummary('term_c')], { term_b: 'idle', diff --git a/src/main/runtime/rpc/methods/orchestration.ts b/src/main/runtime/rpc/methods/orchestration.ts index a72120fc879..1e5b3669ffb 100644 --- a/src/main/runtime/rpc/methods/orchestration.ts +++ b/src/main/runtime/rpc/methods/orchestration.ts @@ -28,28 +28,48 @@ const TASK_STATUSES: TaskStatus[] = [ 'blocked' ] -const SendParams = z.object({ - to: requiredString('Missing --to'), - subject: requiredString('Missing --subject'), - from: OptionalString, - body: OptionalString, - type: z - .enum([ - 'status', - 'dispatch', - 'worker_done', - 'merge_ready', - 'escalation', - 'handoff', - 'decision_gate', - 'heartbeat' - ]) - .optional(), - priority: z.enum(['normal', 'high', 'urgent']).optional(), - threadId: OptionalString, - payload: OptionalString, - devMode: OptionalBoolean -}) +function getLifecycleGroupRecipientError(type: 'worker_done' | 'heartbeat'): string { + return `${type} messages must be sent to a concrete coordinator terminal handle, not a group address.` +} + +const SendParams = z + .object({ + to: requiredString('Missing --to'), + subject: requiredString('Missing --subject'), + from: OptionalString, + body: OptionalString, + type: z + .enum([ + 'status', + 'dispatch', + 'worker_done', + 'merge_ready', + 'escalation', + 'handoff', + 'decision_gate', + 'heartbeat' + ]) + .optional(), + priority: z.enum(['normal', 'high', 'urgent']).optional(), + threadId: OptionalString, + payload: OptionalString, + devMode: OptionalBoolean + }) + .superRefine((params, ctx) => { + if ( + (params.type !== 'worker_done' && params.type !== 'heartbeat') || + !isGroupAddress(params.to) + ) { + return + } + // Why: dispatch lifecycle messages are authority/liveness signals for one + // coordinator. Fanout creates lifecycle mail in unrelated terminals. + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: getLifecycleGroupRecipientError(params.type), + path: ['to'] + }) + }) const CheckParams = z.object({ terminal: OptionalString,