mirror of
https://github.com/stablyai/orca.git
synced 2026-10-20 16:03:20 +00:00
fix(orchestration): preserve active workers across updates (#11271)
* fix(orchestration): preserve active workers across updates * test(ssh): model absent legacy adoption * test(orchestration): align compatibility contracts * fix(windows): escape updater PowerShell booleans * fix(windows): restore stock uninstall process check * fix(orchestration): keep recovery off renderer startup barrier * fix(orchestration): harden legacy recovery migration * fix(orchestration): close recovery review gaps * fix(orchestration): complete legacy worker cutover recovery * fix(orchestration): preserve legacy workers across updates --------- Co-authored-by: OrcaWin <293788423+OrcaWin@users.noreply.github.com>
This commit is contained in:
committed by
Neil
co-authored by
OrcaWin
parent
3240c26bce
commit
959e8d63b0
File diff suppressed because one or more lines are too long
@@ -71,4 +71,24 @@ describe('orchestration check identity', () => {
|
||||
})
|
||||
)
|
||||
})
|
||||
|
||||
it('preserves the pinned legacy --inject check signature', async () => {
|
||||
process.env.ORCA_TERMINAL_HANDLE = 'term_legacy_worker'
|
||||
|
||||
await invokeCheck(
|
||||
new Map<string, string | boolean>([
|
||||
['unread', true],
|
||||
['inject', true]
|
||||
])
|
||||
)
|
||||
|
||||
expect(callMock).toHaveBeenCalledWith(
|
||||
'orchestration.check',
|
||||
expect.objectContaining({
|
||||
terminal: 'term_legacy_worker',
|
||||
unread: true,
|
||||
inject: true
|
||||
})
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -179,6 +179,7 @@ describe('legacy orchestration CLI inspection', () => {
|
||||
id: 'msg_current',
|
||||
run_id: 'run_current',
|
||||
from_handle: 'term_current',
|
||||
to_handle: 'run:run_current',
|
||||
subject: 'current question',
|
||||
type: 'question',
|
||||
body: 'May I continue?',
|
||||
@@ -217,7 +218,8 @@ describe('legacy orchestration CLI inspection', () => {
|
||||
expect(response.result.formatted).not.toContain('RUNTIME_SENTINEL')
|
||||
expect(
|
||||
response.result.formatted.split('\n').filter((line) => line.startsWith('[Reply:'))
|
||||
).toEqual(['[Reply: orca orchestration reply --id msg_current --from term_coord --body "..."]'])
|
||||
).toEqual(['[Reply: orca orchestration reply --id msg_current --body "..."]'])
|
||||
expect(response.result.formatted).not.toContain('--from run:run_current')
|
||||
})
|
||||
|
||||
it('preserves runtime formatting when every message belongs to a current Run', async () => {
|
||||
|
||||
@@ -1,28 +1,44 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { ORCHESTRATION_HANDLERS } from './orchestration'
|
||||
|
||||
describe('orchestration CLI migration recovery', () => {
|
||||
it('redirects worker_done without an outcome before resolving or calling the runtime', async () => {
|
||||
const call = vi.fn()
|
||||
const originalPaneKey = process.env.ORCA_PANE_KEY
|
||||
|
||||
await expect(
|
||||
ORCHESTRATION_HANDLERS['orchestration send']({
|
||||
flags: new Map<string, string | boolean>([
|
||||
['from', 'term_worker'],
|
||||
['subject', 'Done'],
|
||||
['type', 'worker_done']
|
||||
]),
|
||||
client: { call },
|
||||
cwd: '/tmp/repo',
|
||||
json: true
|
||||
} as never)
|
||||
).rejects.toMatchObject({
|
||||
code: 'invalid_argument',
|
||||
data: {
|
||||
effectsApplied: false,
|
||||
nextCommandArgs: ['skills', 'get', 'orchestration', '--full']
|
||||
}
|
||||
afterEach(() => {
|
||||
if (originalPaneKey === undefined) {
|
||||
delete process.env.ORCA_PANE_KEY
|
||||
} else {
|
||||
process.env.ORCA_PANE_KEY = originalPaneKey
|
||||
}
|
||||
})
|
||||
|
||||
describe('orchestration CLI migration recovery', () => {
|
||||
it('forwards legacy worker_done without inventing a completion outcome', async () => {
|
||||
process.env.ORCA_PANE_KEY = 'tab-worker:leaf-worker'
|
||||
const call = vi.fn().mockResolvedValue({ result: { message: { id: 'msg_done' } } })
|
||||
|
||||
await ORCHESTRATION_HANDLERS['orchestration send']({
|
||||
flags: new Map<string, string | boolean>([
|
||||
['from', 'term_worker'],
|
||||
['subject', 'Done'],
|
||||
['type', 'worker_done']
|
||||
]),
|
||||
client: { call },
|
||||
cwd: '/tmp/repo',
|
||||
json: true
|
||||
} as never)
|
||||
|
||||
expect(call).toHaveBeenCalledWith('orchestration.send', {
|
||||
from: 'term_worker',
|
||||
to: undefined,
|
||||
run: undefined,
|
||||
subject: 'Done',
|
||||
body: undefined,
|
||||
type: 'worker_done',
|
||||
priority: undefined,
|
||||
threadId: undefined,
|
||||
payload: undefined,
|
||||
senderPaneKey: 'tab-worker:leaf-worker',
|
||||
devMode: false
|
||||
})
|
||||
expect(call).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -62,6 +62,28 @@ describe('lightweight Run CLI handlers', () => {
|
||||
from: 'term_coord'
|
||||
})
|
||||
})
|
||||
|
||||
it('passes explicit legacy takeover only when requested', async () => {
|
||||
callMock.mockResolvedValue({
|
||||
result: { run: { id: 'run_adopted', objective: 'Recovered work' } }
|
||||
})
|
||||
await ORCHESTRATION_HANDLERS['orchestration run-use']({
|
||||
flags: new Map<string, string | boolean>([
|
||||
['id', 'run_adopted'],
|
||||
['from', 'term_current'],
|
||||
['takeover-legacy', true]
|
||||
]),
|
||||
client: { call: callMock },
|
||||
cwd: '/tmp/repo',
|
||||
json: true
|
||||
} as never)
|
||||
|
||||
expect(callMock).toHaveBeenCalledWith('orchestration.runUse', {
|
||||
id: 'run_adopted',
|
||||
from: 'term_current',
|
||||
takeoverLegacy: true
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
describe('orchestration reset CLI handler', () => {
|
||||
|
||||
@@ -64,11 +64,13 @@ describe('orchestration timeout flag validation', () => {
|
||||
// Why: unread:false makes pre-peek runtimes fall back to non-consuming all mode.
|
||||
expect(callMock).toHaveBeenCalledWith('orchestration.check', {
|
||||
terminal: 'term_worker',
|
||||
terminalPaneKey: undefined,
|
||||
unread: false,
|
||||
peek: true,
|
||||
all: undefined,
|
||||
types: undefined,
|
||||
format: undefined,
|
||||
compatibilityCliCommand: expect.stringMatching(/^orca(?:-ide)?$/),
|
||||
run: undefined,
|
||||
ack: undefined,
|
||||
wait: true,
|
||||
@@ -182,7 +184,9 @@ describe('orchestration timeout flag validation', () => {
|
||||
resume: undefined,
|
||||
options: undefined,
|
||||
timeoutMs: 123,
|
||||
from: 'term_worker'
|
||||
from: 'term_worker',
|
||||
compatibilityCliCommand: expect.stringMatching(/^orca(?:-ide)?$/),
|
||||
compatibilityWindowsCommand: undefined
|
||||
},
|
||||
{ timeoutMs: 5_123, orchestrationCapability: undefined }
|
||||
)
|
||||
@@ -209,7 +213,9 @@ describe('orchestration timeout flag validation', () => {
|
||||
resume: 'msg_question',
|
||||
options: undefined,
|
||||
timeoutMs: undefined,
|
||||
from: 'term_worker'
|
||||
from: 'term_worker',
|
||||
compatibilityCliCommand: expect.stringMatching(/^orca(?:-ide)?$/),
|
||||
compatibilityWindowsCommand: undefined
|
||||
},
|
||||
{ timeoutMs: 605_000, orchestrationCapability: undefined }
|
||||
)
|
||||
|
||||
@@ -86,11 +86,15 @@ describe('orchestration timeout flag validation', () => {
|
||||
|
||||
expect(callMock).toHaveBeenCalledWith('orchestration.check', {
|
||||
terminal: 'term_worker',
|
||||
terminalPaneKey: undefined,
|
||||
unread: false,
|
||||
peek: true,
|
||||
all: undefined,
|
||||
types: undefined,
|
||||
inject: undefined,
|
||||
format: undefined,
|
||||
compatibilityCliCommand: expect.stringMatching(/^orca(?:-ide)?$/),
|
||||
run: undefined,
|
||||
ack: undefined,
|
||||
wait: true,
|
||||
timeoutMs: 250
|
||||
})
|
||||
@@ -273,12 +277,16 @@ describe('orchestration timeout flag validation', () => {
|
||||
'orchestration.ask',
|
||||
{
|
||||
to: 'term_coord',
|
||||
run: undefined,
|
||||
question: 'Proceed?',
|
||||
resume: undefined,
|
||||
options: undefined,
|
||||
timeoutMs: 123,
|
||||
from: 'term_worker'
|
||||
from: 'term_worker',
|
||||
compatibilityCliCommand: expect.stringMatching(/^orca(?:-ide)?$/),
|
||||
compatibilityWindowsCommand: undefined
|
||||
},
|
||||
{ timeoutMs: 5_123 }
|
||||
{ timeoutMs: 5_123, orchestrationCapability: undefined }
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -0,0 +1,110 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
const callMock = vi.fn()
|
||||
const originalCliCommand = process.env.ORCA_CLI_COMMAND
|
||||
const originalPackagedLauncher = process.env.ORCA_WINDOWS_PACKAGED_CLI_LAUNCHER
|
||||
const originalTerminalHandle = process.env.ORCA_TERMINAL_HANDLE
|
||||
const originalExitCode = process.exitCode
|
||||
|
||||
vi.mock('../format', () => ({ printResult: vi.fn() }))
|
||||
vi.mock('../selectors', () => ({ getTerminalHandle: vi.fn() }))
|
||||
|
||||
import { ORCHESTRATION_HANDLERS } from './orchestration'
|
||||
|
||||
describe('packaged Windows legacy ask protocol', () => {
|
||||
beforeEach(() => {
|
||||
callMock.mockReset()
|
||||
process.env.ORCA_WINDOWS_PACKAGED_CLI_LAUNCHER = '1'
|
||||
process.env.ORCA_TERMINAL_HANDLE = 'term_worker'
|
||||
process.exitCode = undefined
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks()
|
||||
process.exitCode = originalExitCode
|
||||
restoreEnv('ORCA_CLI_COMMAND', originalCliCommand)
|
||||
restoreEnv('ORCA_WINDOWS_PACKAGED_CLI_LAUNCHER', originalPackagedLauncher)
|
||||
restoreEnv('ORCA_TERMINAL_HANDLE', originalTerminalHandle)
|
||||
})
|
||||
|
||||
it.each(['orca', 'orca-ide'] as const)(
|
||||
'commits with the %s launcher and exits 75 before resume',
|
||||
async (command) => {
|
||||
process.env.ORCA_CLI_COMMAND = command
|
||||
callMock.mockResolvedValue({
|
||||
result: {
|
||||
answer: null,
|
||||
messageId: 'msg_question',
|
||||
threadId: 'msg_question',
|
||||
timedOut: false,
|
||||
legacyCompatibility: {
|
||||
resumeRequired: true,
|
||||
resumeCommand: `${command} orchestration ask --resume msg_question`
|
||||
}
|
||||
}
|
||||
})
|
||||
const log = vi.spyOn(console, 'log').mockImplementation(() => {})
|
||||
|
||||
await invokeAsk(new Map([['question', 'Proceed?']]))
|
||||
|
||||
expect(callMock).toHaveBeenCalledWith(
|
||||
'orchestration.ask',
|
||||
expect.objectContaining({
|
||||
question: 'Proceed?',
|
||||
resume: undefined,
|
||||
compatibilityWindowsCommand: command
|
||||
}),
|
||||
expect.any(Object)
|
||||
)
|
||||
expect(log.mock.calls.map(([line]) => line)).toEqual([
|
||||
'Question msg_question committed.',
|
||||
`Resume with: ${command} orchestration ask --resume msg_question`
|
||||
])
|
||||
expect(process.exitCode).toBe(75)
|
||||
}
|
||||
)
|
||||
|
||||
it('resumes the committed question without another exit-75 handoff', async () => {
|
||||
process.env.ORCA_CLI_COMMAND = 'orca'
|
||||
callMock.mockResolvedValue({
|
||||
result: {
|
||||
answer: 'yes',
|
||||
messageId: 'msg_question',
|
||||
threadId: 'msg_question',
|
||||
timedOut: false
|
||||
}
|
||||
})
|
||||
const log = vi.spyOn(console, 'log').mockImplementation(() => {})
|
||||
|
||||
await invokeAsk(new Map([['resume', 'msg_question']]))
|
||||
|
||||
expect(callMock).toHaveBeenCalledWith(
|
||||
'orchestration.ask',
|
||||
expect.objectContaining({
|
||||
question: undefined,
|
||||
resume: 'msg_question',
|
||||
compatibilityWindowsCommand: 'orca'
|
||||
}),
|
||||
expect.any(Object)
|
||||
)
|
||||
expect(log).toHaveBeenCalledWith('yes')
|
||||
expect(process.exitCode).toBeUndefined()
|
||||
})
|
||||
})
|
||||
|
||||
function invokeAsk(flags: Map<string, string | boolean>): Promise<void> {
|
||||
return ORCHESTRATION_HANDLERS['orchestration ask']({
|
||||
flags,
|
||||
client: { call: callMock },
|
||||
cwd: '/tmp/repo',
|
||||
json: false
|
||||
} as never)
|
||||
}
|
||||
|
||||
function restoreEnv(name: string, value: string | undefined): void {
|
||||
if (value === undefined) {
|
||||
delete process.env[name]
|
||||
} else {
|
||||
process.env[name] = value
|
||||
}
|
||||
}
|
||||
@@ -692,6 +692,7 @@ describe('orchestration timeout flag validation', () => {
|
||||
all: undefined,
|
||||
types: undefined,
|
||||
format: undefined,
|
||||
compatibilityCliCommand: expect.stringMatching(/^orca(?:-ide)?$/),
|
||||
run: undefined,
|
||||
ack: undefined,
|
||||
wait: true,
|
||||
@@ -819,7 +820,9 @@ describe('orchestration timeout flag validation', () => {
|
||||
resume: undefined,
|
||||
options: undefined,
|
||||
timeoutMs: 123,
|
||||
from: 'term_worker'
|
||||
from: 'term_worker',
|
||||
compatibilityCliCommand: expect.stringMatching(/^orca(?:-ide)?$/),
|
||||
compatibilityWindowsCommand: undefined
|
||||
},
|
||||
{ timeoutMs: 5_123, orchestrationCapability: undefined }
|
||||
)
|
||||
@@ -848,7 +851,9 @@ describe('orchestration timeout flag validation', () => {
|
||||
resume: 'msg_question',
|
||||
options: undefined,
|
||||
timeoutMs: undefined,
|
||||
from: 'term_worker'
|
||||
from: 'term_worker',
|
||||
compatibilityCliCommand: expect.stringMatching(/^orca(?:-ide)?$/),
|
||||
compatibilityWindowsCommand: undefined
|
||||
},
|
||||
{ timeoutMs: 605_000, orchestrationCapability: undefined }
|
||||
)
|
||||
|
||||
@@ -21,11 +21,14 @@ import type {
|
||||
} from '../../shared/orchestration-worker-output'
|
||||
import type { NativeChatMessage } from '../../shared/native-chat-types'
|
||||
import type { RuntimeTerminalRead } from '../../shared/runtime-types'
|
||||
import { orchestrationMigrationData } from '../../shared/orchestration-rpc-contract'
|
||||
import {
|
||||
ORCHESTRATION_LEGACY_RUN_ID,
|
||||
orchestrationMigrationData,
|
||||
orchestrationSkillRecoveryData
|
||||
} from '../../shared/orchestration-rpc-contract'
|
||||
formatMessageReadOnlyTag,
|
||||
formatOrchestrationCheckText,
|
||||
prepareOrchestrationCheckOutput,
|
||||
type LegacyCompatibilityResult,
|
||||
type OrchestrationMessageSummary as MessageSummary
|
||||
} from '../../shared/orchestration-check-output'
|
||||
|
||||
// Why: 15 s is well under Claude Code's ~2 min Bash-tool silence budget while keeping log volume low. See design doc §3.4.
|
||||
const DEFAULT_KEEPALIVE_INTERVAL_MS = 15_000
|
||||
@@ -75,81 +78,6 @@ const TASK_STATUS_VALUES = [
|
||||
'blocked'
|
||||
] as const
|
||||
|
||||
type MessageSummary = {
|
||||
id: string
|
||||
run_id?: string
|
||||
from_handle: string
|
||||
to_handle?: string
|
||||
subject?: string
|
||||
type?: string
|
||||
body?: string
|
||||
payload?: string | null
|
||||
priority?: string
|
||||
read?: number
|
||||
}
|
||||
|
||||
function formatMessageReadOnlyTag(message: MessageSummary): string {
|
||||
return message.run_id === ORCHESTRATION_LEGACY_RUN_ID ? ' [legacy, read-only]' : ''
|
||||
}
|
||||
|
||||
function isLegacyReadOnlyMessage(message: MessageSummary): boolean {
|
||||
return message.run_id === ORCHESTRATION_LEGACY_RUN_ID
|
||||
}
|
||||
|
||||
function formatMessagePriorityTag(message: MessageSummary): string {
|
||||
return message.priority === 'urgent' ? ' [URGENT]' : message.priority === 'high' ? ' [HIGH]' : ''
|
||||
}
|
||||
|
||||
function escapeTerminalControlCharacters(value: string): string {
|
||||
return [...value]
|
||||
.map((character) => {
|
||||
const code = character.charCodeAt(0)
|
||||
if (character === '\n' || (code >= 0x20 && code < 0x7f) || code > 0x9f) {
|
||||
return character
|
||||
}
|
||||
return `\\x${code.toString(16).padStart(2, '0')}`
|
||||
})
|
||||
.join('')
|
||||
}
|
||||
|
||||
function formatQuotedMessageField(label: string, value?: string): string {
|
||||
return `[${label}]\n${escapeTerminalControlCharacters(value ?? '')
|
||||
.split('\n')
|
||||
.map((line) => ` ${line}`)
|
||||
.join('\n')}`
|
||||
}
|
||||
|
||||
function formatLegacyAwareCheckMessages(
|
||||
messages: MessageSummary[],
|
||||
checkedTerminal: string
|
||||
): string {
|
||||
return messages
|
||||
.map((message) => {
|
||||
const legacyReadOnly = isLegacyReadOnlyMessage(message)
|
||||
const lines = [
|
||||
`${message.id}${formatMessageReadOnlyTag(message)}${formatMessagePriorityTag(message)} [${message.type ?? 'status'}] from=${message.from_handle}`,
|
||||
formatQuotedMessageField('subject', message.subject)
|
||||
]
|
||||
if (legacyReadOnly) {
|
||||
lines.push('[Inspection only: reply and acknowledgment are unavailable.]')
|
||||
}
|
||||
if (message.body) {
|
||||
lines.push(formatQuotedMessageField('body', message.body))
|
||||
}
|
||||
if (message.payload) {
|
||||
lines.push(formatQuotedMessageField('payload', message.payload))
|
||||
}
|
||||
if (!legacyReadOnly) {
|
||||
const replyFrom = message.to_handle ?? checkedTerminal
|
||||
lines.push(
|
||||
`[Reply: orca orchestration reply --id ${message.id} --from ${replyFrom} --body "..."]`
|
||||
)
|
||||
}
|
||||
return lines.join('\n')
|
||||
})
|
||||
.join('\n\n')
|
||||
}
|
||||
|
||||
type LifecycleSendRejection = {
|
||||
action: 'rejected'
|
||||
code: string
|
||||
@@ -170,6 +98,40 @@ type OrchestrationSendResult =
|
||||
lifecycle?: { action: 'completed' | 'failed' }
|
||||
}
|
||||
|
||||
function resolveCompatibilityCliCommand(): 'orca' | 'orca-ide' | 'orca-dev' {
|
||||
const configured = process.env.ORCA_CLI_COMMAND
|
||||
if (configured === 'orca' || configured === 'orca-ide' || configured === 'orca-dev') {
|
||||
return configured
|
||||
}
|
||||
return process.platform === 'linux' ? 'orca-ide' : 'orca'
|
||||
}
|
||||
|
||||
function resolvePackagedWindowsCompatibilityCommand(): 'orca' | 'orca-ide' | undefined {
|
||||
if (process.env.ORCA_WINDOWS_PACKAGED_CLI_LAUNCHER !== '1') {
|
||||
return undefined
|
||||
}
|
||||
const command = process.env.ORCA_CLI_COMMAND
|
||||
if (command === 'orca' || command === 'orca-ide') {
|
||||
return command
|
||||
}
|
||||
throw new RuntimeClientError(
|
||||
'invalid_argument',
|
||||
'The packaged Orca launcher did not provide a valid resume command. No question was created.'
|
||||
)
|
||||
}
|
||||
|
||||
async function flushStdout(): Promise<void> {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
process.stdout.write('', (error) => {
|
||||
if (error) {
|
||||
reject(error)
|
||||
} else {
|
||||
resolve()
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
function getOptionalStructuredMessagePayload(
|
||||
flags: Map<string, string | boolean>
|
||||
): string | undefined {
|
||||
@@ -485,7 +447,8 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
|
||||
run: { id: string; objective: string; consumer_generation: number }
|
||||
}>(client, flags, 'orchestration.runUse', {
|
||||
id: getRequiredStringFlag(flags, 'id'),
|
||||
from
|
||||
from,
|
||||
...(flags.has('takeover-legacy') ? { takeoverLegacy: true } : {})
|
||||
})
|
||||
printResult(result, json, (r) => `Using Run ${r.run.id}: ${r.run.objective}`)
|
||||
},
|
||||
@@ -541,13 +504,6 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
|
||||
rejectLifecycleGroupRecipient(type, to)
|
||||
}
|
||||
const outcome = getOptionalStringFlag(flags, 'outcome')
|
||||
if (type === 'worker_done' && outcome === undefined && !flags.has('payload')) {
|
||||
throw new RuntimeClientError(
|
||||
'invalid_argument',
|
||||
'worker_done requires --outcome succeeded or --outcome failed. No effects were applied.',
|
||||
orchestrationSkillRecoveryData()
|
||||
)
|
||||
}
|
||||
if (type !== 'worker_done' && outcome !== undefined) {
|
||||
throw new RuntimeClientError(
|
||||
'invalid_argument',
|
||||
@@ -634,6 +590,7 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
|
||||
timedOut?: boolean
|
||||
cancelled?: boolean
|
||||
connectionLost?: boolean
|
||||
legacyCompatibility?: LegacyCompatibilityResult
|
||||
}
|
||||
let result: Awaited<ReturnType<typeof client.call<CheckResult>>>
|
||||
try {
|
||||
@@ -646,6 +603,8 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
|
||||
all: flags.has('all') ? true : undefined,
|
||||
types: getOptionalStringFlag(flags, 'types'),
|
||||
format: flags.has('format') ? true : undefined,
|
||||
inject: flags.has('inject') ? true : undefined,
|
||||
compatibilityCliCommand: resolveCompatibilityCliCommand(),
|
||||
run: getOptionalStringFlag(flags, 'run'),
|
||||
ack: getOptionalStringFlag(flags, 'ack'),
|
||||
wait: wait ? true : undefined,
|
||||
@@ -682,39 +641,25 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
|
||||
}
|
||||
}
|
||||
}
|
||||
if (flags.has('format') && result.result.messages.some(isLegacyReadOnlyMessage)) {
|
||||
// Why: formatted is one opaque batch with untrusted bodies, so selective banner parsing cannot safely remove legacy actions.
|
||||
result = {
|
||||
...result,
|
||||
result: {
|
||||
...result.result,
|
||||
formatted: formatLegacyAwareCheckMessages(result.result.messages, terminal)
|
||||
}
|
||||
}
|
||||
result = {
|
||||
...result,
|
||||
result: prepareOrchestrationCheckOutput(result.result, terminal, flags.has('format'))
|
||||
}
|
||||
printResult(result, json, (r) => formatOrchestrationCheckText(r, terminal))
|
||||
const compatibilityAck = result.result.legacyCompatibility?.ackMessageIds
|
||||
if (compatibilityAck && compatibilityAck.length > 0) {
|
||||
await flushStdout()
|
||||
await client.call('orchestration.check', {
|
||||
terminal,
|
||||
compatibilityAck: JSON.stringify({
|
||||
messageIds: compatibilityAck,
|
||||
types: getOptionalStringFlag(flags, 'types')
|
||||
?.split(',')
|
||||
.map((type) => type.trim())
|
||||
.filter(Boolean)
|
||||
})
|
||||
})
|
||||
}
|
||||
printResult(result, json, (r) => {
|
||||
if (r.formatted) {
|
||||
return r.formatted
|
||||
}
|
||||
if (r.count === 0) {
|
||||
if (r.timedOut) {
|
||||
return 'Wait timed out; no messages were consumed.'
|
||||
}
|
||||
if (r.cancelled) {
|
||||
return r.connectionLost
|
||||
? 'Wait cancelled because the connection closed; no messages were consumed.'
|
||||
: 'Wait cancelled; no messages were consumed.'
|
||||
}
|
||||
return 'No messages.'
|
||||
}
|
||||
const rendered = r.messages
|
||||
.map(
|
||||
(m) =>
|
||||
`${m.id}${formatMessageReadOnlyTag(m)} [${m.type ?? 'status'}] from=${m.from_handle} "${m.subject}"`
|
||||
)
|
||||
.join('\n')
|
||||
return r.deliveryId ? `Delivery ${r.deliveryId}\n${rendered}` : rendered
|
||||
})
|
||||
},
|
||||
|
||||
'orchestration reply': async ({ flags, client, cwd, json }) => {
|
||||
@@ -1034,6 +979,8 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
|
||||
timeoutMs?: number
|
||||
cancelled?: boolean
|
||||
connectionLost?: boolean
|
||||
answerMessageId?: string | null
|
||||
legacyCompatibility?: LegacyCompatibilityResult
|
||||
}>(
|
||||
client,
|
||||
flags,
|
||||
@@ -1045,7 +992,9 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
|
||||
resume,
|
||||
options: getOptionalStringFlag(flags, 'options'),
|
||||
timeoutMs: parsedTimeoutMs === undefined ? undefined : timeoutMs,
|
||||
from
|
||||
from,
|
||||
compatibilityCliCommand: resolveCompatibilityCliCommand(),
|
||||
compatibilityWindowsCommand: resolvePackagedWindowsCompatibilityCommand()
|
||||
},
|
||||
// Why: extend past timeoutMs so the RPC transport's 60s default doesn't abort before the runtime's own timeout resolves.
|
||||
{
|
||||
@@ -1056,9 +1005,25 @@ export const ORCHESTRATION_HANDLERS: Record<string, CommandHandler> = {
|
||||
// Why: bypass printResult so --json emits a bare JSON object (no envelope) pipeable via `jq -r .answer`, unlike other verbs.
|
||||
if (json) {
|
||||
console.log(JSON.stringify(result.result))
|
||||
} else if (result.result.legacyCompatibility?.resumeRequired) {
|
||||
console.log(`Question ${result.result.messageId} committed.`)
|
||||
console.log(`Resume with: ${result.result.legacyCompatibility.resumeCommand}`)
|
||||
} else if (result.result.answer !== null) {
|
||||
console.log(result.result.answer)
|
||||
}
|
||||
if (result.result.legacyCompatibility?.resumeRequired) {
|
||||
await flushStdout()
|
||||
process.exitCode = 75
|
||||
return
|
||||
}
|
||||
const answerAck = result.result.legacyCompatibility?.answerAcknowledgement
|
||||
if (answerAck && result.result.answer !== null) {
|
||||
await flushStdout()
|
||||
await client.call('orchestration.check', {
|
||||
terminal: from,
|
||||
compatibilityQuestionAck: JSON.stringify(answerAck)
|
||||
})
|
||||
}
|
||||
if (result.result.timedOut) {
|
||||
if (!json) {
|
||||
// Why: report the server's effective budget — it clamps large values, so the requested one would overstate the wait.
|
||||
|
||||
@@ -103,20 +103,40 @@ describe.skipIf(process.platform === 'win32')('RuntimeClient', () => {
|
||||
await new Promise<void>((resolve) => server.listen(endpoint, resolve))
|
||||
writeMetadata(userDataPath, endpoint)
|
||||
|
||||
const priorLaunchToken = process.env.ORCA_AGENT_LAUNCH_TOKEN
|
||||
process.env.ORCA_AGENT_LAUNCH_TOKEN = 'launch-secret'
|
||||
const client = new RuntimeClient(userDataPath, 500)
|
||||
await client.call(
|
||||
'orchestration.send',
|
||||
{ subject: 'hello' },
|
||||
{
|
||||
orchestrationRequestId: 'mutation_explicit'
|
||||
try {
|
||||
await client.call(
|
||||
'orchestration.send',
|
||||
{ subject: 'hello' },
|
||||
{
|
||||
orchestrationRequestId: 'mutation_explicit'
|
||||
}
|
||||
)
|
||||
await client.call('orchestration.taskList', {})
|
||||
const secondClient = new RuntimeClient(userDataPath, 500)
|
||||
await secondClient.call('orchestration.taskList', {})
|
||||
} finally {
|
||||
if (priorLaunchToken === undefined) {
|
||||
delete process.env.ORCA_AGENT_LAUNCH_TOKEN
|
||||
} else {
|
||||
process.env.ORCA_AGENT_LAUNCH_TOKEN = priorLaunchToken
|
||||
}
|
||||
)
|
||||
await client.call('orchestration.taskList', {})
|
||||
}
|
||||
|
||||
expect(requests[0]?.method).toBe('status.get')
|
||||
expect(requests[0]?.compatibilityInvocationId).toBeUndefined()
|
||||
expect(requests[1]?.orchestrationRequestId).toBe('mutation_explicit')
|
||||
expect(requests[1]?.orchestrationContractVersion).toBe(1)
|
||||
expect(requests[1]?.compatibilityInvocationId).toBe('mutation_explicit')
|
||||
expect(requests[1]?.orchestrationCompatibilityEvidence).toMatchObject({
|
||||
launchToken: 'launch-secret'
|
||||
})
|
||||
expect(requests[2]?.orchestrationRequestId).toBeUndefined()
|
||||
expect(requests[2]?.compatibilityInvocationId).not.toBe(requests[1]?.compatibilityInvocationId)
|
||||
expect(requests[3]?.method).toBe('orchestration.taskList')
|
||||
expect(requests[3]?.compatibilityInvocationId).not.toBe(requests[1]?.compatibilityInvocationId)
|
||||
})
|
||||
|
||||
it('rejects an old local runtime before sending an orchestration mutation', async () => {
|
||||
|
||||
+14
-17
@@ -20,6 +20,8 @@ import {
|
||||
ORCHESTRATION_CONTRACT_VERSION,
|
||||
RUNTIME_PROTOCOL_VERSION
|
||||
} from '../../shared/protocol-version'
|
||||
import { createOrchestrationCompatibilityEnvelope } from './orchestration-compatibility-envelope'
|
||||
import { getTimeoutMsParam, isWaitingCheck } from './runtime-request-timeout'
|
||||
|
||||
// Why: for long-poll methods the caller's method-level
|
||||
// `params.timeoutMs` is the inner waiter budget; we extend the client-side
|
||||
@@ -43,6 +45,9 @@ export class RuntimeClient {
|
||||
private readonly environmentSelector: string | null
|
||||
private remoteCompatChecked = false
|
||||
private orchestrationContractCheck: Promise<void> | null = null
|
||||
private readonly orchestrationCompatibility = createOrchestrationCompatibilityEnvelope(
|
||||
process.env
|
||||
)
|
||||
|
||||
// Why: browser commands trigger first-time session init (agent-browser connect +
|
||||
// CDP proxy setup) which can take 15-30s. 60s accommodates cold start without
|
||||
@@ -76,12 +81,20 @@ export class RuntimeClient {
|
||||
const orchestrationRequestId = orchestrationMutation
|
||||
? (options?.orchestrationRequestId ?? randomUUID())
|
||||
: undefined
|
||||
const compatibilityEnvelope = method.startsWith('orchestration.')
|
||||
? {
|
||||
...this.orchestrationCompatibility,
|
||||
compatibilityInvocationId:
|
||||
orchestrationRequestId ?? this.orchestrationCompatibility.compatibilityInvocationId
|
||||
}
|
||||
: {}
|
||||
const envelope = {
|
||||
orchestrationCapability: options?.orchestrationCapability,
|
||||
orchestrationContractVersion: method.startsWith('orchestration.')
|
||||
? ORCHESTRATION_CONTRACT_VERSION
|
||||
: undefined,
|
||||
orchestrationRequestId
|
||||
orchestrationRequestId,
|
||||
...compatibilityEnvelope
|
||||
}
|
||||
if (this.remotePairing) {
|
||||
if (method !== 'status.get') {
|
||||
@@ -327,19 +340,3 @@ function resolveRemotePairing(
|
||||
function delay(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms))
|
||||
}
|
||||
|
||||
function isWaitingCheck(params: unknown): boolean {
|
||||
return (
|
||||
typeof params === 'object' &&
|
||||
params !== null &&
|
||||
'wait' in params &&
|
||||
(params as { wait: unknown }).wait === true
|
||||
)
|
||||
}
|
||||
|
||||
function getTimeoutMsParam(params: unknown): unknown {
|
||||
if (typeof params !== 'object' || params === null || !('timeoutMs' in params)) {
|
||||
return undefined
|
||||
}
|
||||
return (params as { timeoutMs?: unknown }).timeoutMs
|
||||
}
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import { readOrchestrationCompatibilityEvidence } from '../../shared/orchestration-compatibility-evidence'
|
||||
import type { RuntimeOrchestrationEnvelope } from '../../shared/runtime-rpc-envelope'
|
||||
|
||||
export function createOrchestrationCompatibilityEnvelope(
|
||||
env: NodeJS.ProcessEnv
|
||||
): RuntimeOrchestrationEnvelope {
|
||||
return {
|
||||
compatibilityInvocationId: randomUUID(),
|
||||
orchestrationCompatibilityEvidence: readOrchestrationCompatibilityEvidence(env)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
export function isWaitingCheck(params: unknown): boolean {
|
||||
return (
|
||||
typeof params === 'object' &&
|
||||
params !== null &&
|
||||
'wait' in params &&
|
||||
(params as { wait: unknown }).wait === true
|
||||
)
|
||||
}
|
||||
|
||||
export function getTimeoutMsParam(params: unknown): unknown {
|
||||
if (typeof params !== 'object' || params === null || !('timeoutMs' in params)) {
|
||||
return undefined
|
||||
}
|
||||
return (params as { timeoutMs?: unknown }).timeoutMs
|
||||
}
|
||||
@@ -188,7 +188,9 @@ export async function sendRequest<TResult>(
|
||||
params,
|
||||
orchestrationCapability: envelope?.orchestrationCapability,
|
||||
orchestrationContractVersion: envelope?.orchestrationContractVersion,
|
||||
orchestrationRequestId: envelope?.orchestrationRequestId
|
||||
orchestrationRequestId: envelope?.orchestrationRequestId,
|
||||
compatibilityInvocationId: envelope?.compatibilityInvocationId,
|
||||
orchestrationCompatibilityEvidence: envelope?.orchestrationCompatibilityEvidence
|
||||
})}\n`
|
||||
)
|
||||
})
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { RuntimeRpcFailureError } from './types'
|
||||
|
||||
describe('RuntimeRpcFailureError compatibility redaction', () => {
|
||||
it('removes compatibility evidence from CLI JSON errors', () => {
|
||||
const error = new RuntimeRpcFailureError({
|
||||
id: 'request-1',
|
||||
ok: false,
|
||||
error: {
|
||||
code: 'compatibility_rejected',
|
||||
message: 'Rejected',
|
||||
data: {
|
||||
orchestrationCompatibilityEvidence: {
|
||||
launchToken: 'launch-secret'
|
||||
}
|
||||
}
|
||||
},
|
||||
_meta: { runtimeId: 'runtime-1' }
|
||||
})
|
||||
|
||||
expect(error.data).toEqual({
|
||||
orchestrationCompatibilityEvidence: '[redacted]'
|
||||
})
|
||||
expect(JSON.stringify(error.response)).not.toContain('launch-secret')
|
||||
})
|
||||
})
|
||||
@@ -1,4 +1,5 @@
|
||||
import type { RuntimeRpcFailure } from '../../shared/runtime-rpc-envelope'
|
||||
import { redactOrchestrationCompatibilitySecrets } from '../../shared/orchestration-compatibility-evidence'
|
||||
|
||||
export type {
|
||||
RuntimeRpcFailure,
|
||||
@@ -15,7 +16,7 @@ export class RuntimeClientError extends Error {
|
||||
constructor(code: string, message: string, data?: unknown) {
|
||||
super(message)
|
||||
this.code = code
|
||||
this.data = data
|
||||
this.data = redactOrchestrationCompatibilitySecrets(data)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -25,6 +26,12 @@ export class RuntimeRpcFailureError extends RuntimeClientError {
|
||||
constructor(response: RuntimeRpcFailure) {
|
||||
// Why: all client errors expose recovery through the same inherited channel.
|
||||
super(response.error.code, response.error.message, response.error.data)
|
||||
this.response = response
|
||||
this.response = {
|
||||
...response,
|
||||
error: {
|
||||
...response.error,
|
||||
...(response.error.data === undefined ? {} : { data: this.data })
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -16,8 +16,12 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [
|
||||
{
|
||||
path: ['orchestration', 'run-use'],
|
||||
summary: 'Bind this coordinator terminal to an existing Run',
|
||||
usage: 'orca orchestration run-use --id <run_id> [--from <handle>] [--json]',
|
||||
allowedFlags: [...GLOBAL_FLAGS, 'id', 'from', 'retry-request']
|
||||
usage:
|
||||
'orca orchestration run-use --id <run_id> [--from <handle>] [--takeover-legacy] [--json]',
|
||||
allowedFlags: [...GLOBAL_FLAGS, 'id', 'from', 'takeover-legacy', 'retry-request'],
|
||||
notes: [
|
||||
'--takeover-legacy must run in the live coordinator agent terminal it binds; it preserves existing worker assignments.'
|
||||
]
|
||||
},
|
||||
{
|
||||
path: ['orchestration', 'run-current'],
|
||||
|
||||
Reference in New Issue
Block a user