diff --git a/src/main/runtime/rpc/methods/orchestration-send.test.ts b/src/main/runtime/rpc/methods/orchestration-send.test.ts index d8a4212dbef..7b3ed47b38e 100644 --- a/src/main/runtime/rpc/methods/orchestration-send.test.ts +++ b/src/main/runtime/rpc/methods/orchestration-send.test.ts @@ -8,7 +8,9 @@ import type { OrcaRuntimeService } from '../../orca-runtime' import type { RuntimeTerminalSummary } from '../../../../shared/runtime-types' import { ORCHESTRATION_CONTRACT_VERSION } from '../../../../shared/protocol-version' -function lifecycleGroupRecipientError(type: 'worker_done' | 'heartbeat'): string { +function lifecycleGroupRecipientError( + type: 'worker_done' | 'heartbeat' | 'escalation' | 'decision_gate' +): string { return `${type} messages belong to one exact Dispatch and cannot target a group address.` } @@ -634,6 +636,27 @@ describe('orchestration RPC methods', () => { expect(db.getInbox(100)).toHaveLength(0) }) + it.each(['escalation', 'decision_gate'] as const)( + 'rejects %s group sends before inserting rows', + async (type) => { + setup() + const listTerminals = vi.spyOn(runtime, 'listTerminals') + + await expect( + call('orchestration.send', { + from: 'term_worker', + to: '@all', + subject: `${type} broadcast`, + type, + payload: JSON.stringify({ taskId: 'task_1' }) + }) + ).rejects.toThrow(lifecycleGroupRecipientError(type)) + + expect(listTerminals).not.toHaveBeenCalled() + expect(db.getInbox(100)).toHaveLength(0) + } + ) + it('continues to send worker_done to a concrete terminal handle', async () => { setup() const task = db.createTask({ spec: 'work' }) diff --git a/src/main/runtime/rpc/methods/orchestration.ts b/src/main/runtime/rpc/methods/orchestration.ts index c0849c620d5..6280921bc18 100644 --- a/src/main/runtime/rpc/methods/orchestration.ts +++ b/src/main/runtime/rpc/methods/orchestration.ts @@ -64,7 +64,20 @@ async function routeAllMailboxPages( } } -function getLifecycleGroupRecipientError(type: 'worker_done' | 'heartbeat'): string { +type DispatchMutationMessageType = 'worker_done' | 'heartbeat' | 'escalation' | 'decision_gate' + +function isDispatchMutationMessageType( + type: string | undefined +): type is DispatchMutationMessageType { + return ( + type === 'worker_done' || + type === 'heartbeat' || + type === 'escalation' || + type === 'decision_gate' + ) +} + +function getLifecycleGroupRecipientError(type: DispatchMutationMessageType): string { return `${type} messages belong to one exact Dispatch and cannot target a group address.` } @@ -115,11 +128,7 @@ const SendParams = z devMode: OptionalBoolean }) .superRefine((params, ctx) => { - if ( - (params.type !== 'worker_done' && params.type !== 'heartbeat') || - !params.to || - !isGroupAddress(params.to) - ) { + if (!isDispatchMutationMessageType(params.type) || !params.to || !isGroupAddress(params.to)) { return } // Why: dispatch lifecycle messages are authority/liveness signals for one coordinator; fanout would create lifecycle mail in unrelated terminals. @@ -639,11 +648,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ const dispatch = routing.dispatchId ? db.getDispatchContextById(routing.dispatchId) : undefined - const dispatchMutationMessage = - msg.type === 'worker_done' || - msg.type === 'heartbeat' || - msg.type === 'escalation' || - msg.type === 'decision_gate' + const dispatchMutationMessage = isDispatchMutationMessageType(msg.type) if ( dispatchMutationMessage && dispatch &&