Prevent worker completion reports from reaching unrelated agents (#5333)

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:<id>`.
 - 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
 

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<string, string | boolean>([
+          ['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<string, string | boolean>([
+          ['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<string, string | boolean>([
+        ['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', () => {

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<string, CommandHandler> = {
   '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),

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:<id>".',
+      '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.'
     ]
   },

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<string, unknown>): 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<RuntimeTerminalSummary> = {}
@@ -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',

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,