From 45c1cb979ae6cf9d663aa925cedc8b0da0ad8f93 Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Sun, 9 Aug 2026 14:49:36 -0700 Subject: [PATCH] fix(orchestration): release context-only dispatches (#13376) * fix(orchestration): release context-only dispatches Refs #13005 * test(orchestration): align PTY readiness timeout --------- Co-authored-by: Jinwoo-H --- config/reliability-gates.jsonc | 121 ++++++++++++ .../handlers/orchestration-worker-cli.test.ts | 37 ++++ src/cli/handlers/orchestration.ts | 3 +- src/cli/help.ts | 2 +- src/cli/specs/orchestration-worker-specs.ts | 7 +- .../context-only-dispatch-release.ts | 44 +++++ src/main/runtime/orchestration/db.ts | 47 ++++- ...hestration-manual-dispatch-release.test.ts | 186 ++++++++++++++++++ .../methods/orchestration-worker-control.ts | 27 +++ .../rpc/methods/orchestration-worker-stop.ts | 25 +++ ...tration-low-level-dispatch-release.spec.ts | 143 ++++++++++++++ 11 files changed, 631 insertions(+), 11 deletions(-) create mode 100644 src/main/runtime/orchestration/context-only-dispatch-release.ts create mode 100644 src/main/runtime/rpc/methods/orchestration-manual-dispatch-release.test.ts create mode 100644 tests/e2e/orchestration-low-level-dispatch-release.spec.ts diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index e6ce1f9a557..028255fec95 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -7651,6 +7651,127 @@ ], "demotionRule": "Keep experimental or demote if either Electron journey flakes without a product or harness defect, if local worker-start can return before tab materialization without an explicit reveal warning, if focus moves to the worker, if restart or workspace re-entry spawns/resumes/duplicates/interferes with the worker, if retained output becomes unreadable, or if pane-stable delivery reads the wrong mailbox." }, + { + "id": "orchestration.context-only-dispatch-release", + "title": "Manual Dispatch release fences assignments without stopping unsupervised terminals", + "maturity": "experimental", + "protection": "partial", + "owner": "orchestration", + "layer": "cli-runtime-sqlite-terminal-lifecycle", + "surfaces": [ + "orchestration dispatch", + "dispatch-show", + "worker-abandon", + "worker-stop", + "manual terminal reuse" + ], + "platforms": ["macos", "linux", "windows"], + "providers": ["local", "daemon", "ssh", "wsl", "remote-runtime"], + "coveredPlatforms": ["macos"], + "coveredProviders": ["local"], + "coverageNotes": "Deterministic runtime tests cover both release verbs, durable outcomes, repeat and cross-verb idempotency, superseded Task ownership, unrelated and supervised Dispatch isolation, and unknown ids. An isolated Electron journey drives the real app runtime and proves both releases preserve the exact PTY incarnation and allow immediate same-pane redispatch.", + "motivatingLinks": [ + "https://github.com/stablyai/orca/issues/13005", + "https://github.com/stablyai/orca/pull/13250" + ], + "invariant": "A live Dispatch created by orchestration dispatch can be stopped or abandoned even though it has no supervised worker row. Release must durably record the requested outcome, revoke lifecycle authority, close questions, free the exact assignee identity, and block only the Task whose current Dispatch was released. It must never close the unsupervised terminal process, disturb unrelated or supervised workers, or let a repeat or opposite verb rewrite the persisted outcome.", + "oracle": "Create manual, unrelated, and supervised Dispatches through production runtime methods. Require dispatch-show to return the manual id while no worker row exists, then release it and require failed status with exact stopped or abandoned provenance, completion and revocation timestamps, one status notification, zero terminal closes, and immediate redispatch to the same terminal. Repeat through the opposite verb and require the first durable outcome. Create two active contexts for one Task through an explicit ready override, release the older context, and require only its identity to unlock while the newer context and Task remain dispatched. In an isolated Electron runtime, repeat both verbs against one real pane and require the same PTY/incarnation to survive before a third dispatch succeeds.", + "commands": [ + "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/methods/orchestration-manual-dispatch-release.test.ts src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts src/main/runtime/rpc/methods/orchestration-workers-recovery.test.ts src/main/runtime/rpc/methods/orchestration-worker-release.test.ts src/cli/handlers/orchestration-worker-cli.test.ts --reporter=dot", + "pnpm run ensure:electron-runtime && pnpm exec playwright test tests/e2e/orchestration-low-level-dispatch-release.spec.ts --config tests/playwright.config.ts --project electron-headless --workers=1", + "SKIP_BUILD=1 pnpm exec playwright test tests/e2e/orchestration-low-level-dispatch-release.spec.ts --config tests/playwright.config.ts --project electron-headless --workers=1" + ], + "testFiles": [ + "src/main/runtime/rpc/methods/orchestration-manual-dispatch-release.test.ts", + "src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts", + "src/main/runtime/rpc/methods/orchestration-workers-recovery.test.ts", + "src/main/runtime/rpc/methods/orchestration-worker-release.test.ts", + "src/cli/handlers/orchestration-worker-cli.test.ts", + "tests/e2e/orchestration-low-level-dispatch-release.spec.ts" + ], + "assertionRefs": [ + { + "file": "src/main/runtime/rpc/methods/orchestration-manual-dispatch-release.test.ts", + "assertions": [ + "worker-abandon and worker-stop durably release context-only Dispatches without closing terminals", + "repeat and cross-verb calls preserve the first stored outcome", + "superseded release frees only the old identity and leaves the current Task Dispatch active", + "unrelated and supervised workers remain unchanged" + ] + }, + { + "file": "src/cli/handlers/orchestration-worker-cli.test.ts", + "assertions": [ + "text-mode worker-stop reports that a manual Dispatch retained its unsupervised terminal process" + ] + }, + { + "file": "tests/e2e/orchestration-low-level-dispatch-release.spec.ts", + "assertions": [ + "dispatch-show returns the live manual Dispatch before release", + "abandon and stop record distinct durable outcomes", + "the exact PTY and incarnation survive worker-stop", + "the same pane accepts a third Dispatch after both releases" + ] + } + ], + "evidenceRuns": [ + { + "date": "2026-08-09", + "runner": "local", + "platform": "macos", + "command": "SKIP_BUILD=1 pnpm exec playwright test tests/e2e/orchestration-low-level-dispatch-release.spec.ts --config tests/playwright.config.ts --project electron-headless --workers=1", + "result": "failed", + "durationSeconds": 4.3, + "summary": "Current main 2dc172f666 reproduced issue #13005: dispatch-show returned the live id, then worker-abandon failed with dispatch_not_found before any release or pane reuse could occur." + }, + { + "date": "2026-08-09", + "runner": "local", + "platform": "macos", + "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/methods/orchestration-manual-dispatch-release.test.ts src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts src/main/runtime/rpc/methods/orchestration-workers-recovery.test.ts src/main/runtime/rpc/methods/orchestration-worker-release.test.ts src/cli/handlers/orchestration-worker-cli.test.ts --reporter=dot", + "result": "passed", + "durationSeconds": 3.38, + "summary": "Five focused files passed 60 tests, including both context-only release verbs, stale/current ownership, question closure, repeat and cross-verb idempotency, supervised controls, terminal-close negative assertions, and text-mode retained-process guidance." + }, + { + "date": "2026-08-09", + "runner": "local", + "platform": "macos", + "command": "SKIP_BUILD=1 pnpm exec playwright test tests/e2e/orchestration-low-level-dispatch-release.spec.ts --config tests/playwright.config.ts --project electron-headless --workers=1", + "result": "passed", + "durationSeconds": 4.1, + "summary": "The isolated Electron journey passed both release verbs, preserved the exact pane PTY/incarnation, and dispatched a third Task to the same terminal." + } + ], + "runtimeBudget": { + "p95Seconds": 30, + "scope": "focused runtime contracts plus one isolated Electron journey" + }, + "flakeHistory": { + "status": "unknown", + "evidence": "The deterministic unit and Electron journeys pass locally; CI and soak history are not yet available." + }, + "redGreenEvidence": { + "status": "complete", + "evidence": "The byte-identical Electron oracle failed on current main 2dc172f666 with dispatch_not_found and passed after the context-only release implementation. The focused runtime oracle additionally proves durable outcomes and stale Task ownership." + }, + "performanceBudget": { + "required": true, + "evidence": "Release is an explicit control-path transaction with indexed Dispatch and worker lookups, one latest-context lookup for the affected Task, bounded updates, and no polling, process inspection, provider listing, subprocess, renderer work, or terminal I/O." + }, + "promotionCriteria": [ + "Collect 100 consecutive focused CI passes or 14 days of soak history on macOS, Linux, and Windows.", + "Add isolated headed and headless paired-runtime journeys against mixed client and host versions.", + "Add Docker SSH and physical Windows WSL same-pane reuse evidence." + ], + "knownGaps": [ + "Live paired-runtime, headless serve, SSH, WSL, Linux, and Windows evidence is not yet attached.", + "The Electron journey calls production runtime RPC methods directly instead of spawning a separate CLI process.", + "The local journey covers a folder-backed seeded repository through the app fixture but not a git-independent folder workspace." + ], + "demotionRule": "Keep experimental or demote if release closes an unsupervised terminal, rewrites a durable outcome, blocks a newer Task Dispatch, leaves an assignee identity occupied, or the focused oracle flakes without a product or harness defect." + }, { "id": "orchestration.settled-worker-terminal-release", "title": "Settled worker cleanup preserves one exact terminal lease and immutable output", diff --git a/src/cli/handlers/orchestration-worker-cli.test.ts b/src/cli/handlers/orchestration-worker-cli.test.ts index a2777ec3fdb..d19666556c6 100644 --- a/src/cli/handlers/orchestration-worker-cli.test.ts +++ b/src/cli/handlers/orchestration-worker-cli.test.ts @@ -205,6 +205,43 @@ describe('orchestration worker-start CLI contract', () => { ).toContain('Warning: Terminal term_worker is running but could not be revealed.') }) + it('prints the retained-process warning for a manual worker-stop', async () => { + callMock.mockResolvedValue({ + result: { + dispatchId: 'ctx_manual', + state: 'stopped', + processAction: 'none', + warning: 'The assignment was stopped without closing its unsupervised terminal process.' + } + }) + + await ORCHESTRATION_HANDLERS['orchestration worker-stop']({ + flags: new Map([['dispatch', 'ctx_manual']]), + client: { call: callMock }, + cwd: '/tmp/repo', + json: false + } as never) + + const formatter = vi.mocked(printResult).mock.calls[0]?.[2] as + | ((result: { + dispatchId: string + state: string + processAction: string + warning?: string + }) => string) + | undefined + expect( + formatter?.({ + dispatchId: 'ctx_manual', + state: 'stopped', + processAction: 'none', + warning: 'The assignment was stopped without closing its unsupervised terminal process.' + }) + ).toContain( + 'Warning: The assignment was stopped without closing its unsupervised terminal process.' + ) + }) + it('allows the initial zero cursor when paging worker output', async () => { callMock.mockResolvedValue({ result: { diff --git a/src/cli/handlers/orchestration.ts b/src/cli/handlers/orchestration.ts index 6668b4e1906..5be6e62394d 100644 --- a/src/cli/handlers/orchestration.ts +++ b/src/cli/handlers/orchestration.ts @@ -956,6 +956,7 @@ export const ORCHESTRATION_HANDLERS: Record = { state: string processAction: string lastError?: string + warning?: string }>(client, flags, 'orchestration.workerStop', { dispatch: getRequiredStringFlag(flags, 'dispatch') }) @@ -966,7 +967,7 @@ export const ORCHESTRATION_HANDLERS: Record = { result, json, (value) => - `Worker ${value.dispatchId} [${value.state}] process=${value.processAction}${value.lastError ? `\n${value.lastError}` : ''}` + `Worker ${value.dispatchId} [${value.state}] process=${value.processAction}${value.lastError ? `\n${value.lastError}` : ''}${value.warning ? `\nWarning: ${value.warning}` : ''}` ) }, diff --git a/src/cli/help.ts b/src/cli/help.ts index 48df4b893e0..35cc2b5d796 100644 --- a/src/cli/help.ts +++ b/src/cli/help.ts @@ -109,7 +109,7 @@ Orchestration: orchestration worker-start Start a supervised worker locally or on a connected Orca server orchestration worker-show Inspect one supervised worker orchestration worker-read Read bounded output from one supervised worker - orchestration worker-stop Stop one supervised worker + orchestration worker-stop Fence one Dispatch; stop only its supervised worker orchestration worker-abandon Fence an uncertain worker without claiming it stopped orchestration worker-release Release a settled worker's terminal after archiving its output orchestration worker-retain Keep a worker terminal live for debugging diff --git a/src/cli/specs/orchestration-worker-specs.ts b/src/cli/specs/orchestration-worker-specs.ts index b7f6b583a73..d452350f401 100644 --- a/src/cli/specs/orchestration-worker-specs.ts +++ b/src/cli/specs/orchestration-worker-specs.ts @@ -57,11 +57,14 @@ export const ORCHESTRATION_WORKER_COMMAND_SPECS: CommandSpec[] = [ }, { path: ['orchestration', 'worker-stop'], - summary: 'Fence and stop only one supervised agent terminal', + summary: 'Fence one Dispatch and stop its supervised agent terminal', usage: 'orca orchestration worker-stop --dispatch [--retry-request ] [--json]', allowedFlags: [...GLOBAL_FLAGS, 'dispatch', 'retry-request'], - notes: ['Never deletes the worktree, setup terminal, configured tabs, or unrelated processes.'] + notes: [ + 'A Dispatch created by orchestration dispatch is fenced without closing its unsupervised terminal process.', + 'Never deletes the worktree, setup terminal, configured tabs, or unrelated processes.' + ] }, { path: ['orchestration', 'worker-abandon'], diff --git a/src/main/runtime/orchestration/context-only-dispatch-release.ts b/src/main/runtime/orchestration/context-only-dispatch-release.ts new file mode 100644 index 00000000000..ce4d7a7e6fe --- /dev/null +++ b/src/main/runtime/orchestration/context-only-dispatch-release.ts @@ -0,0 +1,44 @@ +import type Database from '../../sqlite/sync-database' +import type { DispatchContextRow, DispatchStatus } from './types' + +export type ContextOnlyDispatchReleaseState = 'abandoned' | 'stopped' | DispatchStatus + +export type ContextOnlyDispatchReleaseResult = { + state: ContextOnlyDispatchReleaseState + alreadySettled: boolean + releasedCurrentTask: boolean +} + +export function releaseContextOnlyDispatch( + db: Database.Database, + dispatch: DispatchContextRow, + latestDispatchId: string | undefined, + requestedState: 'abandoned' | 'stopped' +): ContextOnlyDispatchReleaseResult { + if (dispatch.status !== 'pending' && dispatch.status !== 'dispatched') { + return { + state: persistedReleaseState(dispatch), + alreadySettled: true, + releasedCurrentTask: latestDispatchId === dispatch.id + } + } + + const releasedCurrentTask = latestDispatchId === dispatch.id + db.prepare( + `UPDATE dispatch_contexts + SET status = 'failed', last_failure = ?, + capability_revoked_at = COALESCE(capability_revoked_at, datetime('now')), + completed_at = COALESCE(completed_at, datetime('now')) + WHERE id = ? AND status IN ('pending', 'dispatched')` + ).run(requestedState, dispatch.id) + if (releasedCurrentTask) { + db.prepare("UPDATE tasks SET status = 'blocked' WHERE id = ?").run(dispatch.task_id) + } + return { state: requestedState, alreadySettled: false, releasedCurrentTask } +} + +function persistedReleaseState(dispatch: DispatchContextRow): ContextOnlyDispatchReleaseState { + return dispatch.last_failure === 'abandoned' || dispatch.last_failure === 'stopped' + ? dispatch.last_failure + : dispatch.status +} diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index 3608ee7d754..7049d59469d 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -53,6 +53,10 @@ import { } from './worker-terminal-ownership' import { ORCHESTRATION_RUN_PAGE_LIMIT } from '../../../shared/orchestration-run-pagination' import { ORCHESTRATION_CONTRACT_VERSION } from '../../../shared/protocol-version' +import { + releaseContextOnlyDispatch, + type ContextOnlyDispatchReleaseResult +} from './context-only-dispatch-release' // Why: leaf UUID is the remint-stable pane identity (tab half changes on break-out); exact match covers legacy/unparseable keys. function isEquivalentPaneKey(a: string, b: string): boolean { @@ -5378,14 +5382,28 @@ export class OrchestrationDb { dispatchId: string ): | { disposition: 'stopping'; worker: WorkerDispatchRow; dispatch: DispatchContextRow } - | { disposition: 'already_settled'; worker: WorkerDispatchRow; dispatch: DispatchContextRow } { + | { disposition: 'already_settled'; worker: WorkerDispatchRow; dispatch: DispatchContextRow } + | ({ disposition: 'context_only' } & ContextOnlyDispatchReleaseResult) { this.db.exec('BEGIN IMMEDIATE') try { const dispatch = this.getDispatchContextById(dispatchId) const worker = this.getWorkerDispatch(dispatchId) - if (!dispatch || !worker) { + if (!dispatch) { throw new OrchestrationError('dispatch_not_found', `Dispatch ${dispatchId} was not found.`) } + if (!worker) { + const released = releaseContextOnlyDispatch( + this.db, + dispatch, + this.getDispatchContext(dispatch.task_id)?.id, + 'stopped' + ) + if (!released.alreadySettled) { + this.closeQuestionsForDispatch(dispatchId) + } + this.db.exec('COMMIT') + return { disposition: 'context_only', ...released } + } if (['succeeded', 'failed', 'stopped', 'abandoned'].includes(worker.state)) { this.db.exec('COMMIT') return { disposition: 'already_settled', worker, dispatch } @@ -5537,17 +5555,32 @@ export class OrchestrationDb { return this.getWorkerDispatch(dispatchId) as WorkerDispatchRow } - abandonWorkerDispatch(dispatchId: string): { - disposition: 'abandoned' | 'already_abandoned' | 'stale' - worker: WorkerDispatchRow - } { + abandonWorkerDispatch(dispatchId: string): + | { + disposition: 'abandoned' | 'already_abandoned' | 'stale' + worker: WorkerDispatchRow + } + | ({ disposition: 'context_only' } & ContextOnlyDispatchReleaseResult) { this.db.exec('BEGIN IMMEDIATE') try { const worker = this.getWorkerDispatch(dispatchId) const dispatch = this.getDispatchContextById(dispatchId) - if (!worker || !dispatch) { + if (!dispatch) { throw new OrchestrationError('dispatch_not_found', `Dispatch ${dispatchId} was not found.`) } + if (!worker) { + const released = releaseContextOnlyDispatch( + this.db, + dispatch, + this.getDispatchContext(dispatch.task_id)?.id, + 'abandoned' + ) + if (!released.alreadySettled) { + this.closeQuestionsForDispatch(dispatchId) + } + this.db.exec('COMMIT') + return { disposition: 'context_only', ...released } + } if (worker.state === 'abandoned') { this.db.exec('COMMIT') return { disposition: 'already_abandoned', worker } diff --git a/src/main/runtime/rpc/methods/orchestration-manual-dispatch-release.test.ts b/src/main/runtime/rpc/methods/orchestration-manual-dispatch-release.test.ts new file mode 100644 index 00000000000..d4a124cf895 --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration-manual-dispatch-release.test.ts @@ -0,0 +1,186 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { OrcaRuntimeService } from '../../orca-runtime' +import { OrchestrationDb } from '../../orchestration/db' +import { ORCHESTRATION_METHODS } from './orchestration' + +const COORDINATOR = 'term_coordinator' +const TARGET = 'term_target' +const OTHER = 'term_other' +const SUPERVISED = 'term_supervised' + +describe('manual Dispatch release', () => { + let db: OrchestrationDb + let runtime: OrcaRuntimeService + let runId: string + + beforeEach(() => { + db = new OrchestrationDb(':memory:') + runtime = new OrcaRuntimeService() + runtime.setOrchestrationDb(db) + vi.spyOn(runtime, 'getTerminalPaneKey').mockImplementation((handle) => paneKey(handle)) + vi.spyOn(runtime, 'getTerminalProcessIncarnation').mockImplementation( + (handle) => `${handle}:process` + ) + vi.spyOn(runtime, 'closeTerminal').mockResolvedValue({ closed: true } as never) + runId = db.createRun({ + objective: 'Release manual Dispatches', + coordinatorHandle: COORDINATOR, + coordinatorPaneKey: paneKey(COORDINATOR) + }).id + }) + + afterEach(() => db.close()) + + it.each([ + ['orchestration.workerAbandon', 'abandoned'], + ['orchestration.workerStop', 'stopped'] + ] as const)('releases a context-only Dispatch through %s', async (method, expectedState) => { + const unrelated = await dispatchNewTask(OTHER, 'unrelated') + const supervised = createSupervisedWorker() + const targetTask = createTask('target') + const targetDispatch = await dispatchTask(targetTask, TARGET) + const question = db.createQuestion({ + runId, + dispatchId: targetDispatch, + askerHandle: TARGET, + question: 'Can this assignment finish?' + }) + expect(db.getWorkerDispatch(targetDispatch)).toBeUndefined() + await expect(call('orchestration.dispatchShow', { task: targetTask })).resolves.toMatchObject({ + dispatch: { id: targetDispatch, status: 'dispatched' } + }) + + const notify = vi.spyOn(runtime, 'notifyMessageArrived') + notify.mockClear() + const released = (await call(method, { dispatch: targetDispatch })) as { + state: string + alreadySettled: boolean + processAction: string + residualResources?: unknown[] + } + expect(released).toMatchObject({ + state: expectedState, + alreadySettled: false, + processAction: 'none' + }) + if (method === 'orchestration.workerAbandon') { + expect(released.residualResources).toEqual([]) + } + + expect(db.getDispatchContextById(targetDispatch)).toMatchObject({ + status: 'failed', + last_failure: expectedState, + capability_revoked_at: expect.any(String), + completed_at: expect.any(String) + }) + expect(db.getTask(targetTask)?.status).toBe('blocked') + expect(db.getQuestion(question.message.id)?.status).toBe('closed') + expect(db.getActiveDispatchForIdentity(TARGET, paneKey(TARGET))).toBeUndefined() + expect(runtime.closeTerminal).not.toHaveBeenCalled() + expect(notify).toHaveBeenCalledTimes(1) + expect(notify).toHaveBeenCalledWith(`dispatch:${targetDispatch}`, 'status') + + expect(db.getDispatchContextById(unrelated)).toMatchObject({ status: 'dispatched' }) + expect(db.getWorkerDispatch(supervised)).toMatchObject({ state: 'ready' }) + expect(db.getDispatchContextById(supervised)).toMatchObject({ status: 'dispatched' }) + + const oppositeMethod = + method === 'orchestration.workerAbandon' + ? 'orchestration.workerStop' + : 'orchestration.workerAbandon' + await expect(call(oppositeMethod, { dispatch: targetDispatch })).resolves.toMatchObject({ + state: expectedState, + alreadySettled: true, + processAction: 'none' + }) + expect(notify).toHaveBeenCalledTimes(1) + + const replacement = await dispatchNewTask(TARGET, 'replacement') + expect(replacement).not.toBe(targetDispatch) + expect(db.getActiveDispatchForIdentity(TARGET, paneKey(TARGET))?.id).toBe(replacement) + }) + + it('fences a superseded context without blocking its current replacement', async () => { + const task = createTask('superseded') + const superseded = await dispatchTask(task, TARGET) + db.updateTaskStatus(task, 'ready') + const current = await dispatchTask(task, OTHER) + + await expect(call('orchestration.workerStop', { dispatch: superseded })).resolves.toMatchObject( + { + state: 'stopped', + alreadySettled: false, + processAction: 'none' + } + ) + + expect(db.getDispatchContextById(superseded)).toMatchObject({ + status: 'failed', + last_failure: 'stopped' + }) + expect(db.getDispatchContextById(current)).toMatchObject({ status: 'dispatched' }) + expect(db.getTask(task)?.status).toBe('dispatched') + expect(db.getActiveDispatchForIdentity(TARGET, paneKey(TARGET))).toBeUndefined() + expect(db.getActiveDispatchForIdentity(OTHER, paneKey(OTHER))?.id).toBe(current) + expect(runtime.closeTerminal).not.toHaveBeenCalled() + + await expect(dispatchNewTask(TARGET, 'reuses superseded terminal')).resolves.toMatch(/^ctx_/) + }) + + it('keeps unknown Dispatch errors honest', async () => { + await expect( + call('orchestration.workerAbandon', { dispatch: 'ctx_missing' }) + ).rejects.toMatchObject({ code: 'dispatch_not_found' }) + await expect( + call('orchestration.workerStop', { dispatch: 'ctx_missing' }) + ).rejects.toMatchObject({ code: 'dispatch_not_found' }) + }) + + function createTask(spec: string): string { + return db.createTask({ spec, runId }).id + } + + async function dispatchNewTask(handle: string, spec: string): Promise { + return dispatchTask(createTask(spec), handle) + } + + async function dispatchTask(taskId: string, handle: string): Promise { + const result = (await call('orchestration.dispatch', { + task: taskId, + run: runId, + from: COORDINATOR, + to: handle + })) as { dispatch: { id: string } } + return result.dispatch.id + } + + function createSupervisedWorker(): string { + const started = db.createStartingWorkerDispatch({ + taskId: createTask('supervised'), + startOptions: {} + }) + db.prepareStartingWorkerAuthority({ + dispatchId: started.dispatch.id, + handle: SUPERVISED, + paneKey: paneKey(SUPERVISED), + processIncarnation: `${SUPERVISED}:process`, + worktreeId: 'repo::worktree', + setupState: 'not_applicable', + effects: [] + }) + db.markWorkerDispatchReady(started.dispatch.id) + return started.dispatch.id + } + + async function call(name: string, params: Record): Promise { + const method = ORCHESTRATION_METHODS.find((candidate) => candidate.name === name) + if (!method) { + throw new Error(`Method not found: ${name}`) + } + return method.handler(method.params!.parse(params), { runtime }) + } +}) + +function paneKey(handle: string): string { + return `tab:${handle}` +} diff --git a/src/main/runtime/rpc/methods/orchestration-worker-control.ts b/src/main/runtime/rpc/methods/orchestration-worker-control.ts index 23233b63965..5798806d899 100644 --- a/src/main/runtime/rpc/methods/orchestration-worker-control.ts +++ b/src/main/runtime/rpc/methods/orchestration-worker-control.ts @@ -219,6 +219,20 @@ export const ORCHESTRATION_WORKER_CONTROL_METHODS: RpcMethod[] = [ params: WorkerDispatchParams, handler: (params, { runtime }) => { const abandoned = runtime.getOrchestrationDb().abandonWorkerDispatch(params.dispatch) + if (abandoned.disposition === 'context_only') { + if (!abandoned.alreadySettled) { + runtime.notifyMessageArrived(`dispatch:${params.dispatch}`, 'status') + } + return { + dispatchId: params.dispatch, + state: abandoned.state, + alreadySettled: abandoned.alreadySettled, + stale: !abandoned.releasedCurrentTask, + processAction: 'none', + warning: contextOnlyAbandonWarning(abandoned), + residualResources: [] + } + } const worker = abandoned.worker if (abandoned.disposition === 'abandoned') { runtime.notifyMessageArrived(`dispatch:${params.dispatch}`, 'status') @@ -238,3 +252,16 @@ export const ORCHESTRATION_WORKER_CONTROL_METHODS: RpcMethod[] = [ } }) ] + +function contextOnlyAbandonWarning(result: { + state: string + alreadySettled: boolean + releasedCurrentTask: boolean +}): string { + if (result.alreadySettled) { + return `Dispatch was already ${result.state}; no state or process changed.` + } + return result.releasedCurrentTask + ? 'The assignment was abandoned; its unsupervised terminal process was retained.' + : 'The superseded assignment was abandoned without changing the current Task or terminal process.' +} diff --git a/src/main/runtime/rpc/methods/orchestration-worker-stop.ts b/src/main/runtime/rpc/methods/orchestration-worker-stop.ts index 91984c03ee5..f393e65012a 100644 --- a/src/main/runtime/rpc/methods/orchestration-worker-stop.ts +++ b/src/main/runtime/rpc/methods/orchestration-worker-stop.ts @@ -79,6 +79,18 @@ export const ORCHESTRATION_WORKER_STOP_METHODS: RpcMethod[] = [ if (begun.disposition === 'already_settled') { return settledReceipt(params.dispatch, begun.worker.state) } + if (begun.disposition === 'context_only') { + if (!begun.alreadySettled) { + runtime.notifyMessageArrived(`dispatch:${params.dispatch}`, 'status') + } + return { + dispatchId: params.dispatch, + state: begun.state, + alreadySettled: begun.alreadySettled, + processAction: 'none' as const, + warning: contextOnlyStopWarning(begun) + } + } const handle = begun.worker.agent_terminal_handle if (!handle) { return unknownReceipt( @@ -133,6 +145,19 @@ function settledReceipt(dispatchId: string, state: string) { return { dispatchId, state, alreadySettled: true, processAction: 'none' } } +function contextOnlyStopWarning(result: { + state: string + alreadySettled: boolean + releasedCurrentTask: boolean +}): string { + if (result.alreadySettled) { + return `Dispatch was already ${result.state}; no terminal process changed.` + } + return result.releasedCurrentTask + ? 'The assignment was stopped without closing its unsupervised terminal process.' + : 'The superseded assignment was stopped without changing the current Task or terminal process.' +} + function unknownReceipt( dispatchId: string, worker: { state: string; last_error: string | null }, diff --git a/tests/e2e/orchestration-low-level-dispatch-release.spec.ts b/tests/e2e/orchestration-low-level-dispatch-release.spec.ts new file mode 100644 index 00000000000..6ffa4b57dc4 --- /dev/null +++ b/tests/e2e/orchestration-low-level-dispatch-release.spec.ts @@ -0,0 +1,143 @@ +import { test, expect } from './helpers/orca-app' +import { RuntimeClient } from '../../src/cli/runtime-client' +import type { RuntimeTerminalListResult } from '../../src/shared/runtime-types' +import { waitForSessionReady, ensureTerminalVisible } from './helpers/store' +import { + waitForActivePaneHookDescriptor, + waitForActivePanePtyId, + waitForActiveTerminalManager +} from './helpers/terminal' + +test('low-level Dispatches can be abandoned and stopped without closing their pane', async ({ + orcaPage, + electronApp +}) => { + await waitForSessionReady(orcaPage) + await ensureTerminalVisible(orcaPage) + await waitForActiveTerminalManager(orcaPage) + await waitForActivePanePtyId(orcaPage) + + const userDataDir = await electronApp.evaluate(({ app }) => app.getPath('userData')) + const client = new RuntimeClient(userDataDir, 30_000, null, null) + const pane = await waitForActivePaneHookDescriptor(orcaPage) + const resolved = await client.call<{ terminal: { handle: string } }>('terminal.resolvePane', { + paneKey: pane.paneKey + }) + const terminalHandle = resolved.result.terminal.handle + await expect + .poll(async () => (await findTerminal(client, terminalHandle)).incarnationId, { + timeout: 15_000 + }) + .toBeTruthy() + const before = await findTerminal(client, terminalHandle) + if (!before.incarnationId) { + throw new Error('The target terminal never published a process incarnation') + } + const run = await client.call<{ run: { id: string } }>('orchestration.runCreate', { + objective: 'Release low-level Dispatches', + from: terminalHandle + }) + + const abandonedTask = await createTask(client, run.result.run.id, terminalHandle, 'abandon') + const abandonedDispatch = await dispatchTask( + client, + run.result.run.id, + abandonedTask, + terminalHandle + ) + const shownBeforeAbandon = await showDispatch(client, abandonedTask) + expect(shownBeforeAbandon.id).toBe(abandonedDispatch) + expect(shownBeforeAbandon.status).toBe('dispatched') + + await expect( + client.call('orchestration.workerAbandon', { dispatch: abandonedDispatch }) + ).resolves.toMatchObject({ + result: { + dispatchId: abandonedDispatch, + state: 'abandoned', + alreadySettled: false, + processAction: 'none' + } + }) + expect(await showDispatch(client, abandonedTask)).toMatchObject({ + status: 'failed', + last_failure: 'abandoned' + }) + + const stoppedTask = await createTask(client, run.result.run.id, terminalHandle, 'stop') + const stoppedDispatch = await dispatchTask(client, run.result.run.id, stoppedTask, terminalHandle) + await expect( + client.call('orchestration.workerStop', { dispatch: stoppedDispatch }) + ).resolves.toMatchObject({ + result: { + dispatchId: stoppedDispatch, + state: 'stopped', + alreadySettled: false, + processAction: 'none', + warning: expect.stringContaining('without closing') + } + }) + expect(await showDispatch(client, stoppedTask)).toMatchObject({ + status: 'failed', + last_failure: 'stopped' + }) + + const after = await findTerminal(client, terminalHandle) + expect(after).toMatchObject({ + handle: before.handle, + ptyId: before.ptyId, + incarnationId: before.incarnationId, + connected: true + }) + + const reusableTask = await createTask(client, run.result.run.id, terminalHandle, 'reuse') + await expect( + dispatchTask(client, run.result.run.id, reusableTask, terminalHandle) + ).resolves.toMatch(/^ctx_/) +}) + +async function createTask( + client: RuntimeClient, + runId: string, + coordinatorHandle: string, + suffix: string +): Promise { + const created = await client.call<{ task: { id: string } }>('orchestration.taskCreate', { + spec: `low-level ${suffix}`, + run: runId, + callerTerminalHandle: coordinatorHandle + }) + return created.result.task.id +} + +async function dispatchTask( + client: RuntimeClient, + runId: string, + taskId: string, + terminalHandle: string +): Promise { + const dispatched = await client.call<{ dispatch: { id: string } }>('orchestration.dispatch', { + task: taskId, + run: runId, + from: terminalHandle, + to: terminalHandle + }) + return dispatched.result.dispatch.id +} + +async function showDispatch( + client: RuntimeClient, + taskId: string +): Promise<{ id: string; status: string; last_failure: string | null }> { + const shown = await client.call<{ + dispatch: { id: string; status: string; last_failure: string | null } + }>('orchestration.dispatchShow', { task: taskId }) + return shown.result.dispatch +} + +async function findTerminal(client: RuntimeClient, handle: string) { + const listed = await client.call('terminal.list') + const terminal = listed.result.terminals.find((candidate) => candidate.handle === handle) + expect(terminal).toBeDefined() + return terminal! +}