mirror of
https://github.com/stablyai/orca.git
synced 2026-10-22 16:03:47 +00:00
fix(orchestration): reject control-message fanout
This commit is contained in:
@@ -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' })
|
||||
|
||||
@@ -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 &&
|
||||
|
||||
Reference in New Issue
Block a user