mirror of
https://github.com/stablyai/orca.git
synced 2026-09-25 08:02:31 +00:00
fix(orchestration): enforce honest recipient routing (#14964)
This commit is contained in:
Jinwoo Hong
@@ -0,0 +1,87 @@
|
||||
import { expect, it, vi } from 'vitest'
|
||||
|
||||
const callMock = vi.fn()
|
||||
vi.mock('../format', () => ({ printResult: vi.fn() }))
|
||||
vi.mock('../selectors', () => ({ getTerminalHandle: vi.fn() }))
|
||||
|
||||
import { printResult } from '../format'
|
||||
import { ORCHESTRATION_HANDLERS } from './orchestration'
|
||||
|
||||
async function formatSend(result: unknown): Promise<string> {
|
||||
callMock.mockReset().mockResolvedValueOnce({ result })
|
||||
await ORCHESTRATION_HANDLERS['orchestration send']({
|
||||
flags: new Map([
|
||||
['from', 'term_sender'],
|
||||
['to', 'term_recipient'],
|
||||
['subject', 'ping']
|
||||
]),
|
||||
client: { call: callMock },
|
||||
cwd: '/workspace',
|
||||
json: false
|
||||
} as never)
|
||||
const printCall = vi.mocked(printResult).mock.calls.at(-1)
|
||||
const formatter = printCall?.[2] as (value: unknown) => string
|
||||
return formatter(result)
|
||||
}
|
||||
|
||||
it('prints a live terminal-only delivery limitation', async () => {
|
||||
const line = await formatSend({
|
||||
message: { id: 'msg_1' },
|
||||
warnings: [
|
||||
{
|
||||
code: 'legacy_terminal_recipient',
|
||||
recipient: 'term_recipient',
|
||||
message: 'term_recipient is live now, but its mailbox is not restart-durable.'
|
||||
}
|
||||
]
|
||||
})
|
||||
|
||||
expect(line).toBe(
|
||||
'Sent msg_1\nWarning: term_recipient is live now, but its mailbox is not restart-durable.'
|
||||
)
|
||||
})
|
||||
|
||||
it('shows partial fan-out omissions without hiding delivered recipients', async () => {
|
||||
const line = await formatSend({
|
||||
messages: [{ id: 'msg_1' }],
|
||||
recipients: 1,
|
||||
warnings: [
|
||||
{
|
||||
code: 'recipient_unreachable',
|
||||
recipient: 'term_gone',
|
||||
message: 'Terminal term_gone has no live pane or durable mailbox.'
|
||||
}
|
||||
]
|
||||
})
|
||||
|
||||
expect(line).toBe(
|
||||
'Sent 1 messages to 1 recipients\nWarning: Terminal term_gone has no live pane or durable mailbox.'
|
||||
)
|
||||
})
|
||||
|
||||
it('prints delivery limitations on relayed receipts', async () => {
|
||||
const line = await formatSend({
|
||||
relay: {
|
||||
messageId: 'relay_1',
|
||||
sequence: 1,
|
||||
dispatchId: 'ctx_remote',
|
||||
destination: 'worker',
|
||||
accepted: true
|
||||
},
|
||||
warnings: [
|
||||
{
|
||||
code: 'legacy_terminal_recipient',
|
||||
recipient: 'term_remote',
|
||||
message: 'term_remote is reachable through a compatibility address.'
|
||||
}
|
||||
]
|
||||
})
|
||||
|
||||
expect(line).toBe(
|
||||
'Queued relay_1 for worker Dispatch ctx_remote\nWarning: term_remote is reachable through a compatibility address.'
|
||||
)
|
||||
})
|
||||
|
||||
it('leaves a canonical receipt unchanged', async () => {
|
||||
await expect(formatSend({ message: { id: 'msg_1' } })).resolves.toBe('Sent msg_1')
|
||||
})
|
||||
@@ -100,9 +100,23 @@ type LifecycleSendResult =
|
||||
| { action: 'settled'; outcome: 'succeeded' | 'failed'; duplicate?: boolean }
|
||||
| { action: 'rejected'; code: string; reason: string }
|
||||
|
||||
type SendRecipientWarning = {
|
||||
code: string
|
||||
recipient: string
|
||||
message: string
|
||||
}
|
||||
|
||||
type OrchestrationSendResult =
|
||||
| { message: { id: string; run_id?: string }; lifecycle?: LifecycleSendResult }
|
||||
| { messages: { id: string }[]; recipients: number }
|
||||
| {
|
||||
message: { id: string; run_id?: string }
|
||||
lifecycle?: LifecycleSendResult
|
||||
warnings?: SendRecipientWarning[]
|
||||
}
|
||||
| {
|
||||
messages: { id: string }[]
|
||||
recipients: number
|
||||
warnings?: SendRecipientWarning[]
|
||||
}
|
||||
| {
|
||||
relay: {
|
||||
messageId: string
|
||||
@@ -112,6 +126,7 @@ type OrchestrationSendResult =
|
||||
accepted: true
|
||||
}
|
||||
lifecycle?: LifecycleSendResult
|
||||
warnings?: SendRecipientWarning[]
|
||||
}
|
||||
|
||||
function resolveCompatibilityCliCommand(): 'orca' | 'orca-ide' | 'orca-dev' {
|
||||
@@ -598,16 +613,25 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
|
||||
throw new RuntimeClientError(result.result.lifecycle.code, result.result.lifecycle.reason)
|
||||
}
|
||||
printResult(result, json, (r) => {
|
||||
const warnings = 'warnings' in r ? (r.warnings ?? []) : []
|
||||
const withWarnings = (line: string): string =>
|
||||
warnings.length > 0
|
||||
? [line, ...warnings.map((warning) => `Warning: ${warning.message}`)].join('\n')
|
||||
: line
|
||||
if ('message' in r) {
|
||||
return `Sent ${r.message.id}`
|
||||
return withWarnings(`Sent ${r.message.id}`)
|
||||
}
|
||||
if ('relay' in r) {
|
||||
if (r.relay.destination === 'worker') {
|
||||
return `Queued ${r.relay.messageId} for worker Dispatch ${r.relay.dispatchId}`
|
||||
return withWarnings(
|
||||
`Queued ${r.relay.messageId} for worker Dispatch ${r.relay.dispatchId}`
|
||||
)
|
||||
}
|
||||
return `Queued ${r.relay.messageId} for Run home (Dispatch ${r.relay.dispatchId})`
|
||||
return withWarnings(
|
||||
`Queued ${r.relay.messageId} for Run home (Dispatch ${r.relay.dispatchId})`
|
||||
)
|
||||
}
|
||||
return `Sent ${r.messages.length} messages to ${r.recipients} recipients`
|
||||
return withWarnings(`Sent ${r.messages.length} messages to ${r.recipients} recipients`)
|
||||
})
|
||||
},
|
||||
|
||||
|
||||
@@ -49884,6 +49884,7 @@ describe('OrcaRuntimeService', () => {
|
||||
expect(runtime.resolveLeafForHandle(handle)).toEqual({ ptyId: 'pty-b' })
|
||||
// The guarded resolver surfaces the staleness so clients can re-derive.
|
||||
expect(() => runtime.resolveLiveLeafForHandle(handle)).toThrow('terminal_handle_stale')
|
||||
expect(runtime.getLiveTerminalPaneKey(handle)).toBeNull()
|
||||
})
|
||||
|
||||
it('lets a handle issued before its first PTY adopt that PTY without erroring', async () => {
|
||||
@@ -49904,6 +49905,23 @@ describe('OrcaRuntimeService', () => {
|
||||
|
||||
expect(runtime.resolveLiveLeafForHandle(handle)).toEqual({ ptyId: 'pty-a' })
|
||||
})
|
||||
|
||||
it('keeps terminal cwd resolution fail-soft when the provider is unavailable', async () => {
|
||||
const runtime = new OrcaRuntimeService(store)
|
||||
runtime.attachWindow(1)
|
||||
syncSingleTerminalGraph(runtime, 'pty-a')
|
||||
const handle = issueLeafHandle(runtime, 'pty-a')
|
||||
runtime.setPtyController({
|
||||
write: () => true,
|
||||
kill: () => true,
|
||||
getForegroundProcess: async () => null,
|
||||
getCwd: async () => {
|
||||
throw new Error('ssh disconnected')
|
||||
}
|
||||
})
|
||||
|
||||
await expect(runtime.resolveTerminalCwd(handle)).resolves.toBeNull()
|
||||
})
|
||||
})
|
||||
|
||||
describe('mobile terminal create resilience (#7718)', () => {
|
||||
|
||||
@@ -17765,6 +17765,23 @@ export class OrcaRuntimeService {
|
||||
return this.getPaneKeyForTerminalHandle(handle)
|
||||
}
|
||||
|
||||
getLiveTerminalPaneKey(handle: string): string | null {
|
||||
const runtimePty = this.getLivePtyForHandle(handle)
|
||||
if (runtimePty) {
|
||||
return runtimePty.pty.connected ? (runtimePty.pty.paneKey ?? null) : null
|
||||
}
|
||||
try {
|
||||
const leaf = this.resolveLiveLeafForHandle(handle)
|
||||
if (!leaf?.ptyId) {
|
||||
return null
|
||||
}
|
||||
const pty = this.ptysById.get(leaf.ptyId)
|
||||
return pty?.connected === false ? null : this.getPaneKeyForTerminalHandle(handle)
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
getTerminalWorktreeIdForPaneKey(paneKey: string): string | null {
|
||||
const parsed = parsePaneKey(paneKey)
|
||||
const leaf = parsed ? this.leaves.get(this.getLeafKey(parsed.tabId, parsed.leafId)) : null
|
||||
|
||||
@@ -200,7 +200,14 @@ describeIfBuilt('orca orchestration reset subprocess', () => {
|
||||
runtime.setOrchestrationDb(db)
|
||||
const coordinatorPaneKey = 'tab_cli:11111111-1111-4111-8111-111111111111'
|
||||
vi.spyOn(runtime, 'getTerminalPaneKey').mockImplementation((handle) =>
|
||||
handle === 'term_cli' ? coordinatorPaneKey : null
|
||||
handle === 'term_cli'
|
||||
? coordinatorPaneKey
|
||||
: handle === 'term_target'
|
||||
? 'tab_target:33333333-3333-4333-8333-333333333333'
|
||||
: null
|
||||
)
|
||||
vi.spyOn(runtime, 'getLiveTerminalPaneKey').mockImplementation((handle) =>
|
||||
runtime.getTerminalPaneKey(handle)
|
||||
)
|
||||
db.createRun({
|
||||
objective: 'CLI reset subprocess fixture',
|
||||
|
||||
@@ -11,6 +11,8 @@ export type {
|
||||
ForeignDirectMailboxRoutingPage,
|
||||
MailboxRoutingPage
|
||||
} from './db/messages/mailbox-routing-page'
|
||||
export type { MessageInsert } from './db/messages/message-insert'
|
||||
export type { LegacyAdoptedMailboxOwner } from './db/runs/run-lookup'
|
||||
|
||||
export type {
|
||||
MessageType,
|
||||
|
||||
@@ -37,6 +37,49 @@ export function getActiveDispatchForIdentity(
|
||||
return this.findActiveDispatchForAssignee(handle, paneKey)
|
||||
}
|
||||
|
||||
export function getActiveDispatchMailboxOwners(
|
||||
this: OrchestrationDb,
|
||||
handle: string,
|
||||
paneKey?: string
|
||||
): DispatchContextRow[] {
|
||||
const byHandle = this.db
|
||||
.prepare(
|
||||
`SELECT * FROM dispatch_contexts
|
||||
WHERE assignee_handle = ? AND status IN ('pending', 'dispatched')
|
||||
ORDER BY rowid DESC`
|
||||
)
|
||||
.all(handle) as DispatchContextRow[]
|
||||
if (byHandle.length > 0 || !paneKey) {
|
||||
return byHandle
|
||||
}
|
||||
|
||||
const byExactPane = this.db
|
||||
.prepare(
|
||||
`SELECT * FROM dispatch_contexts
|
||||
WHERE assignee_pane_key = ? AND status IN ('pending', 'dispatched')
|
||||
ORDER BY rowid DESC`
|
||||
)
|
||||
.all(paneKey) as DispatchContextRow[]
|
||||
if (byExactPane.length > 0 || !parsePaneKey(paneKey)) {
|
||||
return byExactPane
|
||||
}
|
||||
return (
|
||||
this.db
|
||||
.prepare(
|
||||
`SELECT * FROM dispatch_contexts
|
||||
WHERE assignee_pane_key IS NOT NULL
|
||||
AND status IN ('pending', 'dispatched') AND instr(assignee_pane_key, ':') > 1
|
||||
AND ${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL} = ?
|
||||
ORDER BY rowid DESC`
|
||||
)
|
||||
.all(paneKeyMatchSuffix(paneKey)) as DispatchContextRow[]
|
||||
).filter(
|
||||
(dispatch) =>
|
||||
dispatch.assignee_pane_key !== null &&
|
||||
isEquivalentPaneKey(dispatch.assignee_pane_key, paneKey)
|
||||
)
|
||||
}
|
||||
|
||||
export function isDispatchMessageSender(
|
||||
this: OrchestrationDb,
|
||||
params: {
|
||||
@@ -125,6 +168,7 @@ export type DispatchLookupMethods = {
|
||||
getActiveDispatchForTerminal: typeof getActiveDispatchForTerminal
|
||||
hasAnyDispatchContexts: typeof hasAnyDispatchContexts
|
||||
getActiveDispatchForIdentity: typeof getActiveDispatchForIdentity
|
||||
getActiveDispatchMailboxOwners: typeof getActiveDispatchMailboxOwners
|
||||
isDispatchMessageSender: typeof isDispatchMessageSender
|
||||
findActiveDispatchForAssignee: typeof findActiveDispatchForAssignee
|
||||
getLatestDispatchForTerminal: typeof getLatestDispatchForTerminal
|
||||
@@ -135,6 +179,7 @@ export function attachDispatchLookup(ctor: { prototype: object }): void {
|
||||
getActiveDispatchForTerminal,
|
||||
hasAnyDispatchContexts,
|
||||
getActiveDispatchForIdentity,
|
||||
getActiveDispatchMailboxOwners,
|
||||
isDispatchMessageSender,
|
||||
findActiveDispatchForAssignee,
|
||||
getLatestDispatchForTerminal
|
||||
|
||||
@@ -6,23 +6,24 @@ import type { OrchestrationDb } from '../orchestration-db'
|
||||
|
||||
// ── Messages ──
|
||||
|
||||
export function insertMessage(
|
||||
this: OrchestrationDb,
|
||||
msg: {
|
||||
id?: string
|
||||
from: string
|
||||
to: string
|
||||
subject: string
|
||||
body?: string
|
||||
type?: MessageType
|
||||
priority?: MessagePriority
|
||||
threadId?: string
|
||||
payload?: string
|
||||
senderPaneKey?: string
|
||||
runId?: string
|
||||
deliveryContract?: MessageDeliveryContract
|
||||
}
|
||||
): MessageRow {
|
||||
const MESSAGE_INSERT_SAVEPOINT = 'message_insert_batch'
|
||||
|
||||
export type MessageInsert = {
|
||||
id?: string
|
||||
from: string
|
||||
to: string
|
||||
subject: string
|
||||
body?: string
|
||||
type?: MessageType
|
||||
priority?: MessagePriority
|
||||
threadId?: string
|
||||
payload?: string
|
||||
senderPaneKey?: string
|
||||
runId?: string
|
||||
deliveryContract?: MessageDeliveryContract
|
||||
}
|
||||
|
||||
export function insertMessage(this: OrchestrationDb, msg: MessageInsert): MessageRow {
|
||||
const runId = msg.runId ?? LEGACY_RUN_ID
|
||||
const deliveryContract = msg.deliveryContract ?? 'current_delivery'
|
||||
this.requireRun(runId)
|
||||
@@ -53,12 +54,27 @@ export function insertMessage(
|
||||
)
|
||||
}
|
||||
|
||||
export function insertMessages(this: OrchestrationDb, messages: MessageInsert[]): MessageRow[] {
|
||||
this.db.exec(`SAVEPOINT ${MESSAGE_INSERT_SAVEPOINT}`)
|
||||
try {
|
||||
const inserted = messages.map((message) => this.insertMessage(message))
|
||||
this.db.exec(`RELEASE ${MESSAGE_INSERT_SAVEPOINT}`)
|
||||
return inserted
|
||||
} catch (error) {
|
||||
this.db.exec(`ROLLBACK TO ${MESSAGE_INSERT_SAVEPOINT}`)
|
||||
this.db.exec(`RELEASE ${MESSAGE_INSERT_SAVEPOINT}`)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
export type MessageInsertMethods = {
|
||||
insertMessage: typeof insertMessage
|
||||
insertMessages: typeof insertMessages
|
||||
}
|
||||
|
||||
export function attachMessageInsert(ctor: { prototype: object }): void {
|
||||
Object.assign(ctor.prototype, {
|
||||
insertMessage
|
||||
insertMessage,
|
||||
insertMessages
|
||||
})
|
||||
}
|
||||
|
||||
@@ -83,15 +83,14 @@ export function getUniqueLegacyCoordinatorHandle(
|
||||
.prepare(
|
||||
`SELECT DISTINCT assignee_handle AS handle
|
||||
FROM dispatch_contexts
|
||||
WHERE run_id = ? AND contract_version = ?
|
||||
AND assignee_handle IS NOT NULL
|
||||
WHERE run_id = ? AND assignee_handle IS NOT NULL
|
||||
UNION
|
||||
SELECT DISTINCT terminal_handle AS handle
|
||||
FROM legacy_compatibility_principals
|
||||
WHERE run_id = ? AND role = 'worker'
|
||||
AND status IN ('committed', 'settled')`
|
||||
)
|
||||
.all(runId, LEGACY_CONTRACT_VERSION, runId) as { handle: string }[]
|
||||
.all(runId, runId) as { handle: string }[]
|
||||
).map((row) => row.handle)
|
||||
)
|
||||
const durableRows = this.db
|
||||
@@ -107,10 +106,9 @@ export function getUniqueLegacyCoordinatorHandle(
|
||||
AND EXISTS(
|
||||
SELECT 1 FROM dispatch_contexts d
|
||||
WHERE d.task_id = t.id AND d.run_id = t.run_id
|
||||
AND d.contract_version = ?
|
||||
)`
|
||||
)
|
||||
.all(adoption.adopted_at, runId, adoption.adopted_at, LEGACY_CONTRACT_VERSION) as {
|
||||
.all(adoption.adopted_at, runId, adoption.adopted_at) as {
|
||||
handle: string
|
||||
}[]
|
||||
if (durableRows.some((row) => workerHandles.has(row.handle))) {
|
||||
|
||||
@@ -10,11 +10,53 @@ import { encodeRunListCursor, decodeRunListCursor } from '../run-list-cursor'
|
||||
import type { RunListPage } from '../run-list-page'
|
||||
import type { OrchestrationDb } from '../orchestration-db'
|
||||
|
||||
export type LegacyAdoptedMailboxOwner = {
|
||||
runId: string
|
||||
terminalHandle: string
|
||||
}
|
||||
|
||||
export function getRun(this: OrchestrationDb, id: string): RunRow | undefined {
|
||||
const run = this.getRunRaw(id)
|
||||
return run ? exposeRunTimestamps(run) : undefined
|
||||
}
|
||||
|
||||
export function getLegacyAdoptedRunMailboxOwner(
|
||||
this: OrchestrationDb
|
||||
): LegacyAdoptedMailboxOwner | null {
|
||||
const adoption = this.getLegacyAdoption()
|
||||
if (!adoption) {
|
||||
return null
|
||||
}
|
||||
const terminalHandle = this.getUniqueLegacyCoordinatorHandle(adoption.adopted_run_id)
|
||||
return terminalHandle ? { runId: adoption.adopted_run_id, terminalHandle } : null
|
||||
}
|
||||
|
||||
export function getRunMailboxOwnerIdsForHandle(
|
||||
this: OrchestrationDb,
|
||||
terminalHandle: string,
|
||||
legacyAdoptedMailboxOwner?: LegacyAdoptedMailboxOwner | null
|
||||
): string[] {
|
||||
const runIds = (
|
||||
this.db
|
||||
.prepare(
|
||||
`SELECT coordinator.run_id
|
||||
FROM run_coordinator_handles AS coordinator
|
||||
JOIN runs ON runs.id = coordinator.run_id AND runs.legacy = 0
|
||||
WHERE coordinator.terminal_handle = ?
|
||||
ORDER BY coordinator.run_id`
|
||||
)
|
||||
.all(terminalHandle) as { run_id: string }[]
|
||||
).map((row) => row.run_id)
|
||||
const adoptedOwner =
|
||||
legacyAdoptedMailboxOwner === undefined
|
||||
? this.getLegacyAdoptedRunMailboxOwner()
|
||||
: legacyAdoptedMailboxOwner
|
||||
if (adoptedOwner?.terminalHandle === terminalHandle) {
|
||||
runIds.push(adoptedOwner.runId)
|
||||
}
|
||||
return [...new Set(runIds)].sort()
|
||||
}
|
||||
|
||||
export function listRuns(
|
||||
this: OrchestrationDb,
|
||||
params: { limit?: number; cursor?: string } = {}
|
||||
@@ -117,6 +159,8 @@ export function fenceOutstandingDelivery(this: OrchestrationDb, runId: string):
|
||||
|
||||
export type RunLookupMethods = {
|
||||
getRun: typeof getRun
|
||||
getLegacyAdoptedRunMailboxOwner: typeof getLegacyAdoptedRunMailboxOwner
|
||||
getRunMailboxOwnerIdsForHandle: typeof getRunMailboxOwnerIdsForHandle
|
||||
listRuns: typeof listRuns
|
||||
getCurrentRunForPane: typeof getCurrentRunForPane
|
||||
runsBoundToPane: typeof runsBoundToPane
|
||||
@@ -129,6 +173,8 @@ export type RunLookupMethods = {
|
||||
export function attachRunLookup(ctor: { prototype: object }): void {
|
||||
Object.assign(ctor.prototype, {
|
||||
getRun,
|
||||
getLegacyAdoptedRunMailboxOwner,
|
||||
getRunMailboxOwnerIdsForHandle,
|
||||
listRuns,
|
||||
getCurrentRunForPane,
|
||||
runsBoundToPane,
|
||||
|
||||
@@ -91,4 +91,33 @@ describe('message batch atomicity', () => {
|
||||
}
|
||||
expect(first).toEqual({ subject: 'outer change', read: 0 })
|
||||
})
|
||||
|
||||
it('preserves an outer transaction when a message insert batch rolls back', () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const sqlite = (db as unknown as { db: Database.Database }).db
|
||||
sqlite.exec(`
|
||||
CREATE TRIGGER reject_second_message_insert
|
||||
BEFORE INSERT ON messages WHEN NEW.id = 'inner_second'
|
||||
BEGIN
|
||||
SELECT RAISE(ABORT, 'blocked');
|
||||
END;
|
||||
BEGIN IMMEDIATE;
|
||||
INSERT INTO messages (id, from_handle, to_handle, subject)
|
||||
VALUES ('outer', 'sender', 'recipient', 'outer change');
|
||||
`)
|
||||
|
||||
expect(() =>
|
||||
db?.insertMessages([
|
||||
{ id: 'inner_first', from: 'sender', to: 'recipient', subject: 'first' },
|
||||
{ id: 'inner_second', from: 'sender', to: 'recipient', subject: 'second' }
|
||||
])
|
||||
).toThrow('blocked')
|
||||
sqlite.exec('COMMIT')
|
||||
|
||||
expect(
|
||||
sqlite
|
||||
.prepare("SELECT id FROM messages WHERE id IN ('outer', 'inner_first') ORDER BY id")
|
||||
.all()
|
||||
).toEqual([{ id: 'outer' }])
|
||||
})
|
||||
})
|
||||
|
||||
@@ -79,6 +79,8 @@ describe('OrchestrationDb legacy contract storage', () => {
|
||||
scheduler_state_lost: 1
|
||||
})
|
||||
expect(db.getRun(adoptedRunId)).toMatchObject({ legacy: 0, consumer_generation: 0 })
|
||||
expect(db.getRunMailboxOwnerIdsForHandle('term_legacy_coord')).toEqual([adoptedRunId])
|
||||
expect(db.getRunMailboxOwnerIdsForHandle('term_invented')).toEqual([])
|
||||
expect(db.listTasks({ runId: LEGACY_RUN_ID })).toEqual([])
|
||||
expect(db.getDispatchContextById(fixture.legacyDispatchId)).toMatchObject({
|
||||
run_id: adoptedRunId,
|
||||
@@ -146,6 +148,21 @@ describe('OrchestrationDb legacy contract storage', () => {
|
||||
)
|
||||
})
|
||||
|
||||
it('fails closed when an adopted coordinator handle becomes a current-contract worker', () => {
|
||||
const state = openAdoptedFixture()
|
||||
expect(db!.getRunMailboxOwnerIdsForHandle('term_legacy_coord')).toEqual([state.adoptedRunId])
|
||||
const task = db!.createTask({
|
||||
runId: state.adoptedRunId,
|
||||
spec: 'mixed contract worker identity',
|
||||
createdByTerminalHandle: 'term_legacy_coord'
|
||||
})
|
||||
|
||||
const dispatch = db!.createDispatchContext(task.id, 'term_legacy_coord', 'tab_mixed:leaf_mixed')
|
||||
|
||||
expect(dispatch.contract_version).toBe(CURRENT_CONTRACT_VERSION)
|
||||
expect(db!.getRunMailboxOwnerIdsForHandle('term_legacy_coord')).toEqual([])
|
||||
})
|
||||
|
||||
it('does not synthesize an adopted Run or compatibility authority for a fresh database', () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
|
||||
|
||||
@@ -86,6 +86,9 @@ const STRUCTURED_RUNTIME_PASSTHROUGH_CODES: ReadonlySet<string> = new Set([
|
||||
'task_not_startable',
|
||||
'dispatch_not_found',
|
||||
'dispatch_run_mismatch',
|
||||
'terminal_not_found',
|
||||
'recipient_ambiguous',
|
||||
'recipient_run_mismatch',
|
||||
'dispatch_inactive',
|
||||
'worker_identity_changed',
|
||||
'cursor_invalid',
|
||||
|
||||
@@ -659,6 +659,9 @@ describe('orchestration RPC methods', () => {
|
||||
|
||||
it('keeps waiting for requested types when an unrelated status arrives', async () => {
|
||||
setup()
|
||||
vi.mocked(runtime.getTerminalPaneKey).mockImplementation((handle) =>
|
||||
handle === 'coord' ? 'tab_wait:leaf_wait' : null
|
||||
)
|
||||
|
||||
const waitPromise = call('orchestration.check', {
|
||||
terminal: 'coord',
|
||||
|
||||
@@ -0,0 +1,582 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { ORCHESTRATION_CONTRACT_VERSION } from '../../../../shared/protocol-version'
|
||||
import type { RuntimeTerminalSummary } from '../../../../shared/runtime-types'
|
||||
import type { OrchestrationDb } from '../../orchestration/db'
|
||||
import type { OrcaRuntimeService } from '../../orca-runtime'
|
||||
import type { RpcContext, RpcRequest } from '../core'
|
||||
import { RpcDispatcher } from '../dispatcher'
|
||||
import { ORCHESTRATION_METHODS } from './orchestration'
|
||||
import { createOrchestrationRpcHarness } from './orchestration-rpc-test-harness'
|
||||
|
||||
type SendWarning = { code: string; recipient: string; message: string }
|
||||
type SendResult = {
|
||||
message: { id: string; run_id: string; to_handle: string }
|
||||
warnings?: SendWarning[]
|
||||
}
|
||||
type GroupSendResult = {
|
||||
messages: { id: string; run_id: string; to_handle: string }[]
|
||||
recipients: number
|
||||
warnings?: SendWarning[]
|
||||
}
|
||||
|
||||
describe('orchestration recipient routing oracle', () => {
|
||||
const harness = createOrchestrationRpcHarness()
|
||||
let db: OrchestrationDb
|
||||
let runtime: OrcaRuntimeService
|
||||
let ctx: RpcContext
|
||||
let senderRunId: string
|
||||
|
||||
function setup(): void {
|
||||
const state = harness.setup()
|
||||
db = state.db
|
||||
runtime = state.runtime
|
||||
ctx = state.ctx
|
||||
senderRunId = state.activeRunId!
|
||||
}
|
||||
|
||||
async function call(params: Record<string, unknown>): Promise<unknown> {
|
||||
return harness.call('orchestration.send', params, ctx)
|
||||
}
|
||||
|
||||
function mockTerminalPaneKeys(resolve: (handle: string) => string | null): void {
|
||||
vi.mocked(runtime.getTerminalPaneKey).mockImplementation(resolve)
|
||||
vi.mocked(runtime.getLiveTerminalPaneKey).mockImplementation(resolve)
|
||||
}
|
||||
|
||||
afterEach(() => harness.cleanup())
|
||||
|
||||
it('rejects an unknown terminal without creating an unread row', async () => {
|
||||
setup()
|
||||
|
||||
await expect(
|
||||
call({ from: 'term_coord', to: 'term_invented', subject: 'unreachable' })
|
||||
).rejects.toMatchObject({ code: 'terminal_not_found' })
|
||||
expect(db.getInbox(100)).toEqual([])
|
||||
})
|
||||
|
||||
it('rejects a closed terminal that has no durable mailbox owner', async () => {
|
||||
setup()
|
||||
vi.mocked(runtime.getTerminalProcessIncarnation).mockReturnValue('retained-pty-record')
|
||||
|
||||
await expect(
|
||||
call({ from: 'term_coord', to: 'term_closed', subject: 'closed' })
|
||||
).rejects.toMatchObject({ code: 'terminal_not_found' })
|
||||
expect(db.getInbox(100)).toEqual([])
|
||||
})
|
||||
|
||||
it('keeps a live terminal-only recipient readable and reports its delivery limitation', async () => {
|
||||
setup()
|
||||
mockTerminalPaneKeys((handle) =>
|
||||
handle === 'term_coord'
|
||||
? harness.coordinatorPaneKey
|
||||
: handle === 'term_live'
|
||||
? 'tab_live:leaf_live'
|
||||
: null
|
||||
)
|
||||
|
||||
const result = (await call({
|
||||
from: 'term_coord',
|
||||
to: 'term_live',
|
||||
subject: 'compatibility'
|
||||
})) as SendResult
|
||||
|
||||
expect(result).toMatchObject({
|
||||
message: { run_id: senderRunId, to_handle: 'term_live' },
|
||||
warnings: [{ code: 'legacy_terminal_recipient', recipient: 'term_live' }]
|
||||
})
|
||||
const check = (await harness.call(
|
||||
'orchestration.check',
|
||||
{ terminal: 'term_live', peek: true },
|
||||
ctx
|
||||
)) as {
|
||||
messages: { id: string }[]
|
||||
}
|
||||
expect(check.messages.map((message) => message.id)).toEqual([result.message.id])
|
||||
})
|
||||
|
||||
it('normalizes a cross-Run coordinator handle to the recipient Run mailbox', async () => {
|
||||
setup()
|
||||
const recipientPane = 'tab_recipient:leaf_recipient'
|
||||
const recipientRun = db.createRun({
|
||||
objective: 'Recipient Run',
|
||||
coordinatorHandle: 'term_recipient',
|
||||
coordinatorPaneKey: recipientPane
|
||||
})
|
||||
mockTerminalPaneKeys((handle) =>
|
||||
handle === 'term_coord'
|
||||
? harness.coordinatorPaneKey
|
||||
: handle === 'term_recipient'
|
||||
? recipientPane
|
||||
: null
|
||||
)
|
||||
|
||||
const result = (await call({
|
||||
from: 'term_coord',
|
||||
to: 'term_recipient',
|
||||
subject: 'cross-run'
|
||||
})) as SendResult
|
||||
|
||||
expect(result.message).toMatchObject({
|
||||
run_id: recipientRun.id,
|
||||
to_handle: `run:${recipientRun.id}`
|
||||
})
|
||||
const check = (await harness.call(
|
||||
'orchestration.check',
|
||||
{ terminal: 'term_recipient', peek: true },
|
||||
ctx
|
||||
)) as { messages: { id: string }[] }
|
||||
expect(check.messages.map((message) => message.id)).toEqual([result.message.id])
|
||||
})
|
||||
|
||||
it('never lets a stale leaf handle adopt its replacement pane Run', async () => {
|
||||
setup()
|
||||
const staleOwner = db.createRun({
|
||||
objective: 'Original pane owner',
|
||||
coordinatorHandle: 'term_stale',
|
||||
coordinatorPaneKey: 'tab_original:leaf_shared'
|
||||
})
|
||||
const replacement = db.createRun({
|
||||
objective: 'Replacement pane owner',
|
||||
coordinatorHandle: 'term_replacement',
|
||||
coordinatorPaneKey: 'tab_replacement:leaf_shared'
|
||||
})
|
||||
vi.mocked(runtime.getTerminalPaneKey).mockImplementation((handle) =>
|
||||
handle === 'term_coord'
|
||||
? harness.coordinatorPaneKey
|
||||
: handle === 'term_stale' || handle === 'term_unowned_stale'
|
||||
? 'tab_replacement:leaf_shared'
|
||||
: null
|
||||
)
|
||||
vi.mocked(runtime.getLiveTerminalPaneKey).mockImplementation((handle) =>
|
||||
handle === 'term_coord' ? harness.coordinatorPaneKey : null
|
||||
)
|
||||
|
||||
const routed = (await call({
|
||||
from: 'term_coord',
|
||||
to: 'term_stale',
|
||||
subject: 'historical owner'
|
||||
})) as SendResult
|
||||
|
||||
expect(routed.message).toMatchObject({
|
||||
run_id: staleOwner.id,
|
||||
to_handle: `run:${staleOwner.id}`
|
||||
})
|
||||
expect(routed.message.run_id).not.toBe(replacement.id)
|
||||
await expect(
|
||||
call({ from: 'term_coord', to: 'term_unowned_stale', subject: 'no owner' })
|
||||
).rejects.toMatchObject({ code: 'terminal_not_found' })
|
||||
})
|
||||
|
||||
it('normalizes an active Dispatch owner even when no pane is live', async () => {
|
||||
setup()
|
||||
const task = db.createTask({ spec: 'detached worker' })
|
||||
const dispatch = db.createDispatchContext(task.id, 'term_detached', 'tab_gone:leaf_gone')
|
||||
|
||||
const result = (await call({
|
||||
from: 'term_coord',
|
||||
to: 'term_detached',
|
||||
subject: 'wait durably'
|
||||
})) as SendResult
|
||||
|
||||
expect(result.message).toMatchObject({
|
||||
run_id: senderRunId,
|
||||
to_handle: `dispatch:${dispatch.id}`
|
||||
})
|
||||
expect(result.warnings).toBeUndefined()
|
||||
expect(db.getUnreadMessages(`dispatch:${dispatch.id}`)).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('reports an explicit Run mismatch for one detached Dispatch owner', async () => {
|
||||
setup()
|
||||
const foreignRun = db.createRun({
|
||||
objective: 'Foreign worker Run',
|
||||
coordinatorHandle: 'term_foreign_coord',
|
||||
coordinatorPaneKey: 'tab_foreign:leaf_coord'
|
||||
})
|
||||
const task = db.createTask({ spec: 'detached foreign worker', runId: foreignRun.id })
|
||||
db.createDispatchContext(task.id, 'term_detached_foreign', 'tab_gone:leaf_gone')
|
||||
|
||||
await expect(
|
||||
call({
|
||||
from: 'term_coord',
|
||||
to: 'term_detached_foreign',
|
||||
run: senderRunId,
|
||||
subject: 'wrong Run'
|
||||
})
|
||||
).rejects.toMatchObject({ code: 'recipient_run_mismatch' })
|
||||
expect(db.getInbox(100)).toEqual([])
|
||||
})
|
||||
|
||||
it('matches check precedence when a live Run coordinator pane overlaps a Dispatch', async () => {
|
||||
setup()
|
||||
const overlapPane = 'tab_overlap:leaf_overlap'
|
||||
const task = db.createTask({ spec: 'overlapped worker' })
|
||||
db.createDispatchContext(task.id, 'term_overlap', overlapPane)
|
||||
const recipientRun = db.createRun({
|
||||
objective: 'Overlapping coordinator',
|
||||
coordinatorHandle: 'term_overlap',
|
||||
coordinatorPaneKey: overlapPane
|
||||
})
|
||||
mockTerminalPaneKeys((handle) =>
|
||||
handle === 'term_coord'
|
||||
? harness.coordinatorPaneKey
|
||||
: handle === 'term_overlap'
|
||||
? overlapPane
|
||||
: null
|
||||
)
|
||||
|
||||
const result = (await call({
|
||||
from: 'term_coord',
|
||||
to: 'term_overlap',
|
||||
subject: 'same read path'
|
||||
})) as SendResult
|
||||
|
||||
expect(result.message).toMatchObject({
|
||||
run_id: recipientRun.id,
|
||||
to_handle: `run:${recipientRun.id}`
|
||||
})
|
||||
const check = (await harness.call(
|
||||
'orchestration.check',
|
||||
{ terminal: 'term_overlap', peek: true },
|
||||
ctx
|
||||
)) as { messages: { id: string }[] }
|
||||
expect(check.messages.map((message) => message.id)).toEqual([result.message.id])
|
||||
})
|
||||
|
||||
it('keeps same-Run historical coordinator routing from the canonical mailbox change', async () => {
|
||||
setup()
|
||||
db.bindRun({
|
||||
runId: senderRunId,
|
||||
coordinatorHandle: 'term_current',
|
||||
coordinatorPaneKey: 'tab_current:leaf_current'
|
||||
})
|
||||
mockTerminalPaneKeys((handle) =>
|
||||
handle === 'term_current' ? 'tab_current:leaf_current' : null
|
||||
)
|
||||
|
||||
const result = (await call({
|
||||
from: 'term_current',
|
||||
to: 'term_coord',
|
||||
subject: 'historical'
|
||||
})) as SendResult
|
||||
|
||||
expect(result.message.to_handle).toBe(`run:${senderRunId}`)
|
||||
expect(result.warnings).toBeUndefined()
|
||||
})
|
||||
|
||||
it('rejects a historical handle that names more than one foreign Run', async () => {
|
||||
setup()
|
||||
db.createRun({
|
||||
objective: 'First owner',
|
||||
coordinatorHandle: 'term_ambiguous',
|
||||
coordinatorPaneKey: 'tab_first:leaf_first'
|
||||
})
|
||||
db.createRun({
|
||||
objective: 'Second owner',
|
||||
coordinatorHandle: 'term_ambiguous',
|
||||
coordinatorPaneKey: 'tab_second:leaf_second'
|
||||
})
|
||||
|
||||
await expect(
|
||||
call({ from: 'term_coord', to: 'term_ambiguous', subject: 'ambiguous' })
|
||||
).rejects.toMatchObject({ code: 'recipient_ambiguous' })
|
||||
expect(db.getInbox(100)).toEqual([])
|
||||
})
|
||||
|
||||
it.each(['@all', '@worktree:wt_target'])(
|
||||
'partially delivers %s when a listed recipient disappears before routing',
|
||||
async (address) => {
|
||||
setup()
|
||||
vi.spyOn(runtime, 'listTerminals').mockResolvedValue({
|
||||
terminals: [terminal('term_coord'), terminal('term_live'), terminal('term_disappeared')],
|
||||
totalCount: 3,
|
||||
truncated: false
|
||||
})
|
||||
mockTerminalPaneKeys((handle) =>
|
||||
handle === 'term_coord'
|
||||
? harness.coordinatorPaneKey
|
||||
: handle === 'term_live'
|
||||
? 'tab_live:leaf_live'
|
||||
: null
|
||||
)
|
||||
const adoptionLookup = vi.spyOn(db, 'getLegacyAdoptedRunMailboxOwner')
|
||||
|
||||
const result = (await call({
|
||||
from: 'term_coord',
|
||||
to: address,
|
||||
subject: 'fan-out'
|
||||
})) as GroupSendResult
|
||||
|
||||
expect(result.messages).toHaveLength(1)
|
||||
expect(result.messages[0]).toMatchObject({ to_handle: 'term_live' })
|
||||
expect(result.recipients).toBe(1)
|
||||
expect(result.warnings?.map((warning) => warning.code).sort()).toEqual([
|
||||
'legacy_terminal_recipient',
|
||||
'recipient_unreachable'
|
||||
])
|
||||
expect(db.getInbox(100)).toHaveLength(1)
|
||||
expect(adoptionLookup).toHaveBeenCalledTimes(1)
|
||||
}
|
||||
)
|
||||
|
||||
it('fans out once when historical handles resolve to the same Run mailbox', async () => {
|
||||
setup()
|
||||
const foreignRun = db.createRun({
|
||||
objective: 'Foreign Run',
|
||||
coordinatorHandle: 'term_foreign_first',
|
||||
coordinatorPaneKey: 'tab_foreign_first:leaf_foreign_first'
|
||||
})
|
||||
db.bindRun({
|
||||
runId: foreignRun.id,
|
||||
coordinatorHandle: 'term_foreign_second',
|
||||
coordinatorPaneKey: 'tab_foreign_second:leaf_foreign_second'
|
||||
})
|
||||
vi.spyOn(runtime, 'listTerminals').mockResolvedValue({
|
||||
terminals: [
|
||||
terminal('term_coord'),
|
||||
terminal('term_foreign_first'),
|
||||
terminal('term_foreign_second')
|
||||
],
|
||||
totalCount: 3,
|
||||
truncated: false
|
||||
})
|
||||
mockTerminalPaneKeys((handle) => (handle === 'term_coord' ? harness.coordinatorPaneKey : null))
|
||||
|
||||
const result = (await call({
|
||||
from: 'term_coord',
|
||||
to: '@all',
|
||||
subject: 'one mailbox'
|
||||
})) as GroupSendResult
|
||||
|
||||
expect(result.recipients).toBe(1)
|
||||
expect(result.messages).toHaveLength(1)
|
||||
expect(result.messages[0]).toMatchObject({
|
||||
run_id: foreignRun.id,
|
||||
to_handle: `run:${foreignRun.id}`
|
||||
})
|
||||
expect(db.getInbox(100)).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('excludes historical handles that resolve back to the sender mailbox', async () => {
|
||||
setup()
|
||||
db.bindRun({
|
||||
runId: senderRunId,
|
||||
coordinatorHandle: 'term_middle',
|
||||
coordinatorPaneKey: 'tab_middle:leaf_middle'
|
||||
})
|
||||
db.bindRun({
|
||||
runId: senderRunId,
|
||||
coordinatorHandle: 'term_sender',
|
||||
coordinatorPaneKey: 'tab_sender:leaf_sender'
|
||||
})
|
||||
vi.spyOn(runtime, 'listTerminals').mockResolvedValue({
|
||||
terminals: [
|
||||
terminal('term_sender'),
|
||||
terminal('term_coord'),
|
||||
terminal('term_middle'),
|
||||
terminal('term_live')
|
||||
],
|
||||
totalCount: 4,
|
||||
truncated: false
|
||||
})
|
||||
mockTerminalPaneKeys((handle) =>
|
||||
handle === 'term_sender'
|
||||
? 'tab_sender:leaf_sender'
|
||||
: handle === 'term_live'
|
||||
? 'tab_live:leaf_live'
|
||||
: null
|
||||
)
|
||||
|
||||
const result = (await call({
|
||||
from: 'term_sender',
|
||||
to: '@all',
|
||||
subject: 'exclude self aliases'
|
||||
})) as GroupSendResult
|
||||
|
||||
expect(result.messages).toHaveLength(1)
|
||||
expect(result.messages[0].to_handle).toBe('term_live')
|
||||
expect(result.warnings).toMatchObject([{ code: 'legacy_terminal_recipient' }])
|
||||
})
|
||||
|
||||
it('replays one honest receipt and discards retry receipts for rejected recipients', async () => {
|
||||
setup()
|
||||
mockTerminalPaneKeys((handle) =>
|
||||
handle === 'term_coord'
|
||||
? harness.coordinatorPaneKey
|
||||
: handle === 'term_live'
|
||||
? 'tab_live:leaf_live'
|
||||
: null
|
||||
)
|
||||
const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS })
|
||||
const rejected = request('rpc_reject_1', 'retry_reject', 'term_missing')
|
||||
|
||||
const firstRejected = await dispatcher.dispatch(rejected)
|
||||
const retriedRejected = await dispatcher.dispatch({ ...rejected, id: 'rpc_reject_2' })
|
||||
expect(firstRejected).toMatchObject({ ok: false, error: { code: 'terminal_not_found' } })
|
||||
expect(retriedRejected).toMatchObject({ ok: false, error: { code: 'terminal_not_found' } })
|
||||
expect(db.getInbox(100)).toEqual([])
|
||||
const callerFingerprint = db.getOrCreateLocalMutationCallerFingerprint()
|
||||
expect(db.getMutationReceipt(callerFingerprint, 'retry_reject')).toBeUndefined()
|
||||
|
||||
const accepted = request('rpc_accept_1', 'retry_accept', 'term_live')
|
||||
const firstAccepted = await dispatcher.dispatch(accepted)
|
||||
const replayed = await dispatcher.dispatch({ ...accepted, id: 'rpc_accept_2' })
|
||||
expect(firstAccepted).toMatchObject({
|
||||
ok: true,
|
||||
result: {
|
||||
message: { id: expect.stringMatching(/^msg_/) },
|
||||
warnings: [{ code: 'legacy_terminal_recipient' }],
|
||||
mutation: { requestId: 'retry_accept', replayed: false }
|
||||
}
|
||||
})
|
||||
expect(replayed).toMatchObject({
|
||||
ok: true,
|
||||
result: {
|
||||
message: { id: firstAccepted.ok ? (firstAccepted.result as SendResult).message.id : '' },
|
||||
warnings: [{ code: 'legacy_terminal_recipient' }],
|
||||
mutation: { requestId: 'retry_accept', replayed: true }
|
||||
}
|
||||
})
|
||||
expect(db.getInbox(100)).toHaveLength(1)
|
||||
expect(db.getMutationReceipt(callerFingerprint, 'retry_accept')).toMatchObject({
|
||||
state: 'completed',
|
||||
receipt: expect.stringContaining('legacy_terminal_recipient')
|
||||
})
|
||||
})
|
||||
|
||||
it('serializes ambiguity and explicit Run mismatch through the RPC boundary', async () => {
|
||||
setup()
|
||||
db.createRun({
|
||||
objective: 'First owner',
|
||||
coordinatorHandle: 'term_ambiguous',
|
||||
coordinatorPaneKey: 'tab_first:leaf_first'
|
||||
})
|
||||
db.createRun({
|
||||
objective: 'Second owner',
|
||||
coordinatorHandle: 'term_ambiguous',
|
||||
coordinatorPaneKey: 'tab_second:leaf_second'
|
||||
})
|
||||
const foreignRun = db.createRun({
|
||||
objective: 'Foreign owner',
|
||||
coordinatorHandle: 'term_foreign',
|
||||
coordinatorPaneKey: 'tab_foreign:leaf_foreign'
|
||||
})
|
||||
const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS })
|
||||
|
||||
const ambiguous = await dispatcher.dispatch(
|
||||
request('rpc_ambiguous', 'retry_ambiguous', 'term_ambiguous')
|
||||
)
|
||||
const mismatch = await dispatcher.dispatch(
|
||||
request('rpc_mismatch', 'retry_mismatch', 'term_foreign', { run: senderRunId })
|
||||
)
|
||||
|
||||
expect(ambiguous).toMatchObject({ ok: false, error: { code: 'recipient_ambiguous' } })
|
||||
expect(mismatch).toMatchObject({ ok: false, error: { code: 'recipient_run_mismatch' } })
|
||||
expect(foreignRun.id).not.toBe(senderRunId)
|
||||
expect(db.getInbox(100)).toEqual([])
|
||||
})
|
||||
|
||||
it('rolls back a partial group insert before an idempotent retry', async () => {
|
||||
setup()
|
||||
vi.spyOn(runtime, 'listTerminals').mockResolvedValue({
|
||||
terminals: [terminal('term_coord'), terminal('term_first'), terminal('term_second')],
|
||||
totalCount: 3,
|
||||
truncated: false
|
||||
})
|
||||
mockTerminalPaneKeys((handle) =>
|
||||
handle === 'term_coord'
|
||||
? harness.coordinatorPaneKey
|
||||
: handle === 'term_first'
|
||||
? 'tab_first:leaf_first'
|
||||
: handle === 'term_second'
|
||||
? 'tab_second:leaf_second'
|
||||
: null
|
||||
)
|
||||
const insertMessage = db.insertMessage.bind(db)
|
||||
vi.spyOn(db, 'insertMessage')
|
||||
.mockImplementationOnce(insertMessage)
|
||||
.mockImplementationOnce(() => {
|
||||
throw new Error('injected second insert failure')
|
||||
})
|
||||
const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS })
|
||||
const group = request('rpc_group_partial', 'retry_group_partial', '@all')
|
||||
|
||||
const failed = await dispatcher.dispatch(group)
|
||||
expect(failed.ok).toBe(false)
|
||||
expect(db.getInbox(100)).toEqual([])
|
||||
|
||||
const retried = await dispatcher.dispatch({ ...group, id: 'rpc_group_partial_retry' })
|
||||
expect(retried).toMatchObject({
|
||||
ok: true,
|
||||
result: { messages: [{ id: expect.any(String) }, { id: expect.any(String) }] }
|
||||
})
|
||||
expect(db.getInbox(100)).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('replays a completed group receipt when notification fails after durable insertion', async () => {
|
||||
setup()
|
||||
vi.spyOn(runtime, 'listTerminals').mockResolvedValue({
|
||||
terminals: [terminal('term_coord'), terminal('term_live')],
|
||||
totalCount: 2,
|
||||
truncated: false
|
||||
})
|
||||
mockTerminalPaneKeys((handle) =>
|
||||
handle === 'term_coord'
|
||||
? harness.coordinatorPaneKey
|
||||
: handle === 'term_live'
|
||||
? 'tab_live:leaf_live'
|
||||
: null
|
||||
)
|
||||
vi.spyOn(runtime, 'notifyMessageArrived').mockImplementationOnce(() => {
|
||||
throw new Error('injected notification failure')
|
||||
})
|
||||
const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS })
|
||||
const group = request('rpc_group_notify', 'retry_group_notify', '@all')
|
||||
|
||||
const failed = await dispatcher.dispatch(group)
|
||||
const retried = await dispatcher.dispatch({ ...group, id: 'rpc_group_notify_retry' })
|
||||
|
||||
expect(failed.ok).toBe(false)
|
||||
expect(retried).toMatchObject({
|
||||
ok: true,
|
||||
result: {
|
||||
messages: [{ id: expect.any(String) }],
|
||||
mutation: { requestId: 'retry_group_notify', replayed: true }
|
||||
}
|
||||
})
|
||||
expect(db.getInbox(100)).toHaveLength(1)
|
||||
})
|
||||
})
|
||||
|
||||
function terminal(handle: string): RuntimeTerminalSummary {
|
||||
return {
|
||||
handle,
|
||||
ptyId: handle,
|
||||
worktreeId: 'wt_target',
|
||||
worktreePath: '/workspace',
|
||||
branch: 'main',
|
||||
tabId: `tab_${handle}`,
|
||||
leafId: `leaf_${handle}`,
|
||||
title: null,
|
||||
connected: true,
|
||||
writable: true,
|
||||
lastOutputAt: null,
|
||||
preview: ''
|
||||
}
|
||||
}
|
||||
|
||||
function request(
|
||||
id: string,
|
||||
requestId: string,
|
||||
to: string,
|
||||
extraParams: Record<string, unknown> = {}
|
||||
): RpcRequest {
|
||||
return {
|
||||
id,
|
||||
authToken: 'test-token',
|
||||
method: 'orchestration.send',
|
||||
params: { from: 'term_coord', to, subject: 'retry', ...extraParams },
|
||||
orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION,
|
||||
orchestrationRequestId: requestId
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,145 @@
|
||||
import type { LegacyAdoptedMailboxOwner, OrchestrationDb } from '../../orchestration/db'
|
||||
import type { DispatchContextRow } from '../../orchestration/types'
|
||||
import type { OrcaRuntimeService } from '../../orca-runtime'
|
||||
|
||||
export type SendRecipientWarning = {
|
||||
code:
|
||||
| 'legacy_terminal_recipient'
|
||||
| 'recipient_unreachable'
|
||||
| 'recipient_ambiguous'
|
||||
| 'recipient_run_mismatch'
|
||||
recipient: string
|
||||
message: string
|
||||
}
|
||||
|
||||
export type BareRecipientResolution =
|
||||
| {
|
||||
ok: true
|
||||
to: string
|
||||
runId: string | undefined
|
||||
warning?: SendRecipientWarning
|
||||
}
|
||||
| {
|
||||
ok: false
|
||||
code: 'terminal_not_found' | 'recipient_ambiguous' | 'recipient_run_mismatch'
|
||||
message: string
|
||||
warning: SendRecipientWarning
|
||||
}
|
||||
|
||||
export function resolveBareOrchestrationRecipient(params: {
|
||||
runtime: OrcaRuntimeService
|
||||
db: OrchestrationDb
|
||||
handle: string
|
||||
senderRunId?: string
|
||||
explicitRunId?: string
|
||||
legacyAdoptedMailboxOwner?: LegacyAdoptedMailboxOwner | null
|
||||
}): BareRecipientResolution {
|
||||
const { runtime, db, handle } = params
|
||||
const paneKey = runtime.getLiveTerminalPaneKey(handle) ?? undefined
|
||||
const boundRun = paneKey ? db.getCurrentRunForPane(paneKey) : undefined
|
||||
if (boundRun) {
|
||||
const mismatch = runMismatch(handle, boundRun.id, params.explicitRunId)
|
||||
return mismatch ?? { ok: true, to: `run:${boundRun.id}`, runId: boundRun.id }
|
||||
}
|
||||
|
||||
const dispatches = db.getActiveDispatchMailboxOwners(handle, paneKey)
|
||||
const dispatch = selectDispatch(dispatches, params.explicitRunId)
|
||||
if (dispatches.length > 0 && !dispatch) {
|
||||
return ambiguous(
|
||||
handle,
|
||||
dispatches.map((candidate) => `dispatch:${candidate.id}`)
|
||||
)
|
||||
}
|
||||
if (dispatch) {
|
||||
const mismatch = runMismatch(handle, dispatch.run_id, params.explicitRunId)
|
||||
return mismatch ?? { ok: true, to: `dispatch:${dispatch.id}`, runId: dispatch.run_id }
|
||||
}
|
||||
|
||||
const ownerRunIds = db.getRunMailboxOwnerIdsForHandle(handle, params.legacyAdoptedMailboxOwner)
|
||||
const selectedRunId = selectHistoricalRun(ownerRunIds, params)
|
||||
if (ownerRunIds.length > 0 && !selectedRunId) {
|
||||
return ambiguous(
|
||||
handle,
|
||||
ownerRunIds.map((runId) => `run:${runId}`)
|
||||
)
|
||||
}
|
||||
if (selectedRunId) {
|
||||
const mismatch = runMismatch(handle, selectedRunId, params.explicitRunId)
|
||||
return mismatch ?? { ok: true, to: `run:${selectedRunId}`, runId: selectedRunId }
|
||||
}
|
||||
|
||||
if (paneKey) {
|
||||
return {
|
||||
ok: true,
|
||||
to: handle,
|
||||
runId: params.senderRunId,
|
||||
warning: {
|
||||
code: 'legacy_terminal_recipient',
|
||||
recipient: handle,
|
||||
message: `${handle} is a live terminal-only mailbox. Delivery is not durable after that terminal closes; prefer run:<id> or dispatch:<id>.`
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const message = `Terminal ${handle} has no live pane or durable Run/Dispatch mailbox.`
|
||||
return {
|
||||
ok: false,
|
||||
code: 'terminal_not_found',
|
||||
message,
|
||||
warning: { code: 'recipient_unreachable', recipient: handle, message }
|
||||
}
|
||||
}
|
||||
|
||||
function selectDispatch(
|
||||
dispatches: DispatchContextRow[],
|
||||
explicitRunId: string | undefined
|
||||
): DispatchContextRow | undefined {
|
||||
if (dispatches.length === 1) {
|
||||
return dispatches[0]
|
||||
}
|
||||
if (!explicitRunId) {
|
||||
return undefined
|
||||
}
|
||||
const matches = dispatches.filter((dispatch) => dispatch.run_id === explicitRunId)
|
||||
return matches.length === 1 ? matches[0] : undefined
|
||||
}
|
||||
|
||||
function selectHistoricalRun(
|
||||
ownerRunIds: string[],
|
||||
params: { senderRunId?: string; explicitRunId?: string }
|
||||
): string | undefined {
|
||||
if (params.explicitRunId && ownerRunIds.includes(params.explicitRunId)) {
|
||||
return params.explicitRunId
|
||||
}
|
||||
if (params.senderRunId && ownerRunIds.includes(params.senderRunId)) {
|
||||
return params.senderRunId
|
||||
}
|
||||
return ownerRunIds.length === 1 ? ownerRunIds[0] : undefined
|
||||
}
|
||||
|
||||
function ambiguous(handle: string, addresses: string[]): BareRecipientResolution {
|
||||
const message = `${handle} resolves to multiple durable mailboxes (${addresses.join(', ')}). Use an explicit canonical address.`
|
||||
return {
|
||||
ok: false,
|
||||
code: 'recipient_ambiguous',
|
||||
message,
|
||||
warning: { code: 'recipient_ambiguous', recipient: handle, message }
|
||||
}
|
||||
}
|
||||
|
||||
function runMismatch(
|
||||
handle: string,
|
||||
resolvedRunId: string,
|
||||
explicitRunId: string | undefined
|
||||
): BareRecipientResolution | undefined {
|
||||
if (!explicitRunId || explicitRunId === resolvedRunId) {
|
||||
return undefined
|
||||
}
|
||||
const message = `${handle} belongs to Run ${resolvedRunId}, not explicitly requested Run ${explicitRunId}.`
|
||||
return {
|
||||
ok: false,
|
||||
code: 'recipient_run_mismatch',
|
||||
message,
|
||||
warning: { code: 'recipient_run_mismatch', recipient: handle, message }
|
||||
}
|
||||
}
|
||||
@@ -30,6 +30,9 @@ export function createOrchestrationRpcHarness() {
|
||||
vi.spyOn(runtime, 'getTerminalPaneKey').mockImplementation((handle) =>
|
||||
handle === 'term_coord' ? coordinatorPaneKey : null
|
||||
)
|
||||
vi.spyOn(runtime, 'getLiveTerminalPaneKey').mockImplementation((handle) =>
|
||||
runtime.getTerminalPaneKey(handle)
|
||||
)
|
||||
vi.spyOn(runtime, 'getTerminalProcessIncarnation').mockImplementation((handle) =>
|
||||
handle.startsWith('term_') ? `runtime_test:${handle}:1` : null
|
||||
)
|
||||
|
||||
@@ -601,6 +601,13 @@ describe('orchestration RPC methods', () => {
|
||||
totalCount: terminals.length,
|
||||
truncated: false
|
||||
})
|
||||
vi.mocked(runtime.getTerminalPaneKey).mockImplementation((handle) => {
|
||||
if (handle === 'term_coord') {
|
||||
return coordinatorPaneKey
|
||||
}
|
||||
const terminal = terminals.find((candidate) => candidate.handle === handle)
|
||||
return terminal ? `${terminal.tabId}:${terminal.leafId}` : null
|
||||
})
|
||||
vi.spyOn(runtime, 'getAgentStatusForHandle').mockImplementation(
|
||||
(handle: string) => agentStatuses?.[handle] ?? null
|
||||
)
|
||||
|
||||
@@ -23,6 +23,10 @@ import {
|
||||
} from '../../../../shared/orchestration-rpc-contract'
|
||||
import { clampOrchestrationAskTimeoutMs } from '../../../../shared/orchestration-ask-timeout'
|
||||
import { ORCHESTRATION_GATE_METHODS } from './orchestration-gates'
|
||||
import {
|
||||
resolveBareOrchestrationRecipient,
|
||||
type SendRecipientWarning
|
||||
} from './orchestration-recipient-routing'
|
||||
import { resolveRunScope } from './orchestration-run-scope'
|
||||
import { ORCHESTRATION_RUN_METHODS } from './orchestration-runs'
|
||||
import { ORCHESTRATION_WORKER_METHODS } from './orchestration-worker-methods'
|
||||
@@ -445,6 +449,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
legacyCoordinatorRunId,
|
||||
revalidateLegacyCoordinator,
|
||||
orchestrationCompatibilityCallerAuthority,
|
||||
recordMutationReceipt,
|
||||
signal
|
||||
}
|
||||
) => {
|
||||
@@ -592,8 +597,35 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
)
|
||||
}
|
||||
|
||||
const sendWarnings: SendRecipientWarning[] = []
|
||||
let messageRunId = routing.run?.id
|
||||
if (!isGroupAddress(to) && !to.startsWith('run:') && !to.startsWith('dispatch:')) {
|
||||
const recipient = resolveBareOrchestrationRecipient({
|
||||
runtime,
|
||||
db,
|
||||
handle: to,
|
||||
senderRunId: routing.run?.id,
|
||||
explicitRunId: params.run
|
||||
})
|
||||
if (!recipient.ok) {
|
||||
throw new OrchestrationError(recipient.code, recipient.message)
|
||||
}
|
||||
to = recipient.to
|
||||
messageRunId = recipient.runId
|
||||
if (recipient.warning) {
|
||||
sendWarnings.push(recipient.warning)
|
||||
}
|
||||
}
|
||||
const withSendWarnings = <T extends object>(
|
||||
receipt: T
|
||||
): T & {
|
||||
warnings?: SendRecipientWarning[]
|
||||
} => (sendWarnings.length > 0 ? { ...receipt, warnings: sendWarnings } : receipt)
|
||||
|
||||
if (!isGroupAddress(to)) {
|
||||
const federatedDispatchId = routing.dispatchId
|
||||
const federatedDispatchId = to.startsWith('dispatch:')
|
||||
? to.slice('dispatch:'.length)
|
||||
: undefined
|
||||
const federatedTarget =
|
||||
federatedDispatchId && to === `dispatch:${federatedDispatchId}`
|
||||
? db.getFederatedDispatch(federatedDispatchId)
|
||||
@@ -636,8 +668,8 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
payload: params.payload ?? null
|
||||
})
|
||||
})
|
||||
runtime.ensureOrchestrationFederationRelay(routing.run?.id)
|
||||
return {
|
||||
runtime.ensureOrchestrationFederationRelay(messageRunId)
|
||||
return withSendWarnings({
|
||||
relay: {
|
||||
messageId: relay.message_id,
|
||||
sequence: relay.sequence,
|
||||
@@ -645,7 +677,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
destination: 'worker',
|
||||
accepted: true
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
// Point-to-point — existing single-recipient behavior
|
||||
revalidateLegacyCoordinator?.()
|
||||
@@ -665,10 +697,10 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
? bindCoordinatorMutationPayload(messageType, params.payload, dispatch.id)
|
||||
: params.payload,
|
||||
senderPaneKey,
|
||||
runId: routing.run?.id,
|
||||
runId: messageRunId,
|
||||
deliveryContract: legacyWorkerDeliveryContract(
|
||||
runtime,
|
||||
routing.run?.id ?? legacyCoordinatorRunId,
|
||||
messageRunId ?? legacyCoordinatorRunId,
|
||||
to
|
||||
)
|
||||
})
|
||||
@@ -738,14 +770,14 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
const rejection =
|
||||
db.convertLifecycleMessageToRejection(msg.id, code, authority.reason) ?? msg
|
||||
runtime.notifyMessageArrived(rejection.to_handle, rejection.type)
|
||||
return {
|
||||
return withSendWarnings({
|
||||
message: rejection,
|
||||
lifecycle: {
|
||||
action: 'rejected',
|
||||
code,
|
||||
reason: authority.reason
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
// Why: reconcile releases the dispatch lock before waking recipients, else a woken coordinator re-dispatches while the lock is still held.
|
||||
@@ -753,20 +785,20 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
const reconciled = reconcileLifecycleMessage(db, msg)
|
||||
// Why: a suppressed message is already read, so skip the notify that would wake a check --wait waiter to an empty result.
|
||||
if (reconciled.action === 'suppressed') {
|
||||
return { message: msg }
|
||||
return withSendWarnings({ message: msg })
|
||||
}
|
||||
if (reconciled.action === 'rejected') {
|
||||
const rejection = db.getMessageById(msg.id) ?? msg
|
||||
runtime.notifyMessageArrived(rejection.to_handle, rejection.type)
|
||||
return { message: rejection, lifecycle: reconciled }
|
||||
return withSendWarnings({ message: rejection, lifecycle: reconciled })
|
||||
}
|
||||
runtime.notifyMessageArrived(msg.to_handle, msg.type)
|
||||
return msg.type === 'worker_done'
|
||||
? { message: msg, lifecycle: reconciled }
|
||||
: { message: msg }
|
||||
return withSendWarnings(
|
||||
msg.type === 'worker_done' ? { message: msg, lifecycle: reconciled } : { message: msg }
|
||||
)
|
||||
}
|
||||
runtime.notifyMessageArrived(msg.to_handle, msg.type)
|
||||
return { message: msg }
|
||||
return withSendWarnings({ message: msg })
|
||||
}
|
||||
|
||||
// Why: fan out one message per recipient (independent read-tracking) but share a thread_id for correlation (Section 4.5).
|
||||
@@ -781,12 +813,57 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
throw new Error(`No recipients resolved for group address: ${to}`)
|
||||
}
|
||||
|
||||
const legacyAdoptedMailboxOwner = db.getLegacyAdoptedRunMailboxOwner()
|
||||
const resolvedRecipients = handles.map((handle) => ({
|
||||
handle,
|
||||
resolution: resolveBareOrchestrationRecipient({
|
||||
runtime,
|
||||
db,
|
||||
handle,
|
||||
senderRunId: routing.run?.id,
|
||||
explicitRunId: params.run,
|
||||
legacyAdoptedMailboxOwner
|
||||
})
|
||||
}))
|
||||
const deliverableRecipients = resolvedRecipients.filter(
|
||||
(
|
||||
recipient
|
||||
): recipient is typeof recipient & {
|
||||
resolution: { ok: true; to: string; runId?: string; warning?: SendRecipientWarning }
|
||||
} => recipient.resolution.ok
|
||||
)
|
||||
const senderRecipient = resolveBareOrchestrationRecipient({
|
||||
runtime,
|
||||
db,
|
||||
handle: from,
|
||||
senderRunId: routing.run?.id,
|
||||
legacyAdoptedMailboxOwner
|
||||
})
|
||||
const senderMailboxKey = senderRecipient.ok
|
||||
? `${senderRecipient.runId ?? ''}\u0000${senderRecipient.to}`
|
||||
: undefined
|
||||
const seenMailboxes = new Set<string>()
|
||||
const uniqueRecipients = deliverableRecipients.filter(({ resolution }) => {
|
||||
const mailboxKey = `${resolution.runId ?? ''}\u0000${resolution.to}`
|
||||
if (mailboxKey === senderMailboxKey || seenMailboxes.has(mailboxKey)) {
|
||||
return false
|
||||
}
|
||||
seenMailboxes.add(mailboxKey)
|
||||
return true
|
||||
})
|
||||
if (uniqueRecipients.length === 0) {
|
||||
throw new OrchestrationError(
|
||||
'terminal_not_found',
|
||||
`No recipient of ${to} resolved to a live terminal or durable Run/Dispatch mailbox.`
|
||||
)
|
||||
}
|
||||
|
||||
revalidateLegacyCoordinator?.()
|
||||
const threadId = params.threadId ?? `thread_${Date.now()}`
|
||||
const messages = handles.map((handle) =>
|
||||
db.insertMessage({
|
||||
const messages = db.insertMessages(
|
||||
uniqueRecipients.map(({ resolution }) => ({
|
||||
from,
|
||||
to: handle,
|
||||
to: resolution.to,
|
||||
subject: params.subject,
|
||||
body: params.body,
|
||||
type: params.type as MessageType,
|
||||
@@ -794,19 +871,27 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
threadId,
|
||||
payload: params.payload,
|
||||
senderPaneKey,
|
||||
runId: routing.run?.id,
|
||||
runId: resolution.runId,
|
||||
deliveryContract: legacyWorkerDeliveryContract(
|
||||
runtime,
|
||||
routing.run?.id ?? legacyCoordinatorRunId,
|
||||
handle
|
||||
resolution.runId ?? legacyCoordinatorRunId,
|
||||
resolution.to
|
||||
)
|
||||
})
|
||||
}))
|
||||
)
|
||||
const groupWarnings = resolvedRecipients.flatMap(({ resolution }) =>
|
||||
resolution.ok ? (resolution.warning ? [resolution.warning] : []) : [resolution.warning]
|
||||
)
|
||||
const receipt = {
|
||||
messages,
|
||||
recipients: messages.length,
|
||||
...(groupWarnings.length > 0 ? { warnings: groupWarnings } : {})
|
||||
}
|
||||
recordMutationReceipt?.(receipt)
|
||||
for (const message of messages) {
|
||||
runtime.notifyMessageArrived(message.to_handle, message.type)
|
||||
}
|
||||
|
||||
return { messages, recipients: handles.length }
|
||||
return receipt
|
||||
}
|
||||
}),
|
||||
|
||||
|
||||
@@ -68,6 +68,9 @@ function createHarness(): Harness {
|
||||
vi.spyOn(runtime, 'getTerminalPaneKey').mockImplementation((handle) =>
|
||||
handle === COORDINATOR_HANDLE ? COORDINATOR_PANE : handle === WORKER_HANDLE ? WORKER_PANE : null
|
||||
)
|
||||
vi.spyOn(runtime, 'getLiveTerminalPaneKey').mockImplementation((handle) =>
|
||||
runtime.getTerminalPaneKey(handle)
|
||||
)
|
||||
vi.spyOn(runtime, 'verifyOrchestrationCompatibilityCaller').mockImplementation(
|
||||
(compatibilityEvidence) => {
|
||||
const coordinator =
|
||||
@@ -199,7 +202,7 @@ describe('legacy coordinator takeover races', () => {
|
||||
result: {
|
||||
message: {
|
||||
run_id: harness.adoptedRunId,
|
||||
to_handle: WORKER_HANDLE,
|
||||
to_handle: `dispatch:${harness.dispatchId}`,
|
||||
delivery_contract: 'legacy_direct'
|
||||
}
|
||||
}
|
||||
@@ -310,6 +313,15 @@ describe('legacy coordinator takeover races', () => {
|
||||
|
||||
it('partitions a coordinator group send by legacy recipient contract', async () => {
|
||||
const harness = createHarness()
|
||||
vi.mocked(harness.runtime.getTerminalPaneKey).mockImplementation((handle) =>
|
||||
handle === COORDINATOR_HANDLE
|
||||
? COORDINATOR_PANE
|
||||
: handle === WORKER_HANDLE
|
||||
? WORKER_PANE
|
||||
: handle === 'term_current_worker'
|
||||
? 'tab_current_worker:leaf_current_worker'
|
||||
: null
|
||||
)
|
||||
vi.spyOn(harness.runtime, 'listTerminals').mockResolvedValue({
|
||||
terminals: [
|
||||
{ handle: COORDINATOR_HANDLE },
|
||||
@@ -336,7 +348,7 @@ describe('legacy coordinator takeover races', () => {
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
run_id: harness.adoptedRunId,
|
||||
to_handle: WORKER_HANDLE,
|
||||
to_handle: `dispatch:${harness.dispatchId}`,
|
||||
delivery_contract: 'legacy_direct'
|
||||
}),
|
||||
expect.objectContaining({
|
||||
|
||||
@@ -192,7 +192,7 @@ function entityCounts(db: OrchestrationDb): Record<string, number> {
|
||||
}
|
||||
|
||||
function resultOf(response: RpcResponse): Record<string, unknown> {
|
||||
expect(response.ok).toBe(true)
|
||||
expect(response.ok, JSON.stringify(response)).toBe(true)
|
||||
if (!response.ok) {
|
||||
throw new Error(response.error.message)
|
||||
}
|
||||
@@ -352,6 +352,9 @@ describe('orchestration runtime update settlement', () => {
|
||||
|
||||
it('routes ordinary mail with the same attested authority without settling work', async () => {
|
||||
const harness = createUpdateHarness()
|
||||
expect(harness.db.getRunMailboxOwnerIdsForHandle(COORDINATOR_HANDLE)).toEqual([
|
||||
harness.adoptedRunId
|
||||
])
|
||||
const response = await harness.createDispatcher().dispatch(
|
||||
request(
|
||||
'orchestration.send',
|
||||
|
||||
@@ -81,7 +81,9 @@ describe('runRemoteOrcaCli', () => {
|
||||
}),
|
||||
getLegacyAdoption: vi.fn(() => undefined),
|
||||
getActiveDispatchForIdentity: vi.fn(() => undefined),
|
||||
getActiveDispatchMailboxOwners: vi.fn(() => []),
|
||||
getCurrentRunForPane: vi.fn(() => undefined),
|
||||
getRunMailboxOwnerIdsForHandle: vi.fn(() => []),
|
||||
findActiveRemoteAttachmentForPane: vi.fn(() => undefined)
|
||||
}
|
||||
const runtime = {
|
||||
@@ -96,6 +98,8 @@ describe('runRemoteOrcaCli', () => {
|
||||
}),
|
||||
getOrchestrationDb: () => db,
|
||||
getTerminalPaneKey: () => null,
|
||||
getLiveTerminalPaneKey: (handle: string) =>
|
||||
handle === 'term_windows' ? 'tab_windows:leaf_windows' : null,
|
||||
deliverPendingMessagesForHandle: vi.fn(),
|
||||
notifyMessageArrived: vi.fn(),
|
||||
linearIssueContext: vi.fn(async (request: unknown) => ({
|
||||
@@ -188,9 +192,18 @@ describe('runRemoteOrcaCli', () => {
|
||||
LEGACY_FALLBACK_OPTIONS
|
||||
)
|
||||
|
||||
expect(result.exitCode).toBe(0)
|
||||
const payload = JSON.parse(result.stdout) as { ok: boolean }
|
||||
expect(payload.ok).toBe(true)
|
||||
expect(result.exitCode, result.stdout).toBe(0)
|
||||
expect(JSON.parse(result.stdout)).toMatchObject({
|
||||
ok: true,
|
||||
result: {
|
||||
warnings: [
|
||||
{
|
||||
code: 'legacy_terminal_recipient',
|
||||
recipient: 'term_windows'
|
||||
}
|
||||
]
|
||||
}
|
||||
})
|
||||
expect(db.getUnreadMessages('term_windows')[0]?.from_handle).toBe('term_ssh')
|
||||
})
|
||||
|
||||
@@ -210,7 +223,7 @@ describe('runRemoteOrcaCli', () => {
|
||||
LEGACY_FALLBACK_OPTIONS
|
||||
)
|
||||
|
||||
expect(result.exitCode).toBe(0)
|
||||
expect(result.exitCode, result.stdout).toBe(0)
|
||||
expect(db.insertMessage).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ senderPaneKey: undefined })
|
||||
)
|
||||
@@ -481,7 +494,7 @@ describe('runRemoteOrcaCli', () => {
|
||||
LEGACY_FALLBACK_OPTIONS
|
||||
)
|
||||
|
||||
expect(result.exitCode).toBe(0)
|
||||
expect(result.exitCode, result.stdout).toBe(0)
|
||||
const payload = JSON.parse(result.stdout) as { ok: boolean }
|
||||
expect(payload.ok).toBe(true)
|
||||
const message = db.getUnreadMessages('term_windows')[0]
|
||||
|
||||
Reference in New Issue
Block a user