From 9cfa00d665f37fed876c3849c38973ebe63ed8dd Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Thu, 13 Aug 2026 02:31:56 -0700 Subject: [PATCH] Fix federation terminal settlement retries and legacy admission (#14105) --- config/reliability-gates.jsonc | 96 +++- ...ration-federated-legacy-settlement.test.ts | 75 +++ .../orchestration-worker-settlement.ts | 3 +- src/cli/handlers/orchestration.ts | 5 +- src/main/runtime/orca-runtime.ts | 64 ++- src/main/runtime/orchestration/db.ts | 161 +++++- ...ederation-acknowledgment-integrity.test.ts | 100 ++++ ...ederation-acknowledgment-migration.test.ts | 50 ++ .../federation-lifecycle-settlement.test.ts | 34 ++ .../federation-lifecycle-settlement.ts | 11 + .../orchestration/federation-sync.test.ts | 134 ++++- .../runtime/orchestration/federation-sync.ts | 81 ++- .../federation-terminal-recovery.test.ts | 102 ++++ .../federation-worker-report-payload.ts | 37 ++ .../mutation-receipt-capacity.test.ts | 2 +- ...hestration-db-retention-pagination.test.ts | 8 +- .../orchestration-schema-version-skew.ts | 12 +- src/main/runtime/orchestration/types.ts | 1 + ...hestration-federation-agent-launch.test.ts | 2 +- ...ration-federation-folder-placement.test.ts | 2 +- ...on-federation-lifecycle-settlement.test.ts | 514 +++++++++++++++--- .../methods/orchestration-federation-relay.ts | 39 +- .../methods/orchestration-federation.test.ts | 30 +- src/main/runtime/rpc/methods/orchestration.ts | 41 +- 24 files changed, 1418 insertions(+), 186 deletions(-) create mode 100644 src/cli/handlers/orchestration-federated-legacy-settlement.test.ts create mode 100644 src/main/runtime/orchestration/federation-acknowledgment-integrity.test.ts create mode 100644 src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts create mode 100644 src/main/runtime/orchestration/federation-lifecycle-settlement.test.ts create mode 100644 src/main/runtime/orchestration/federation-terminal-recovery.test.ts create mode 100644 src/main/runtime/orchestration/federation-worker-report-payload.ts diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index 81f587046f0..a72b093f4b6 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -8401,12 +8401,16 @@ "providers": ["local", "daemon", "ssh", "wsl", "remote-runtime"], "coveredPlatforms": ["macos"], "coveredProviders": ["local", "daemon", "ssh"], - "coverageNotes": "A deterministic service-state-machine oracle now models a current-contract worker and coordinator whose renderer graph identities disappear across an app/runtime update. It exercises the production verifier with restored PTY and hydrated hook commitments, proves authenticated completion replay across a fresh runtime, explicit takeover, ordinary mail routing, remote-attachment process fencing, retained Task/Dispatch/terminal identity, and unchanged fixture marker bytes, and rejects foreign pane evidence. Other deterministic units cover authority-aware legacy formatting, exact legacy worker identity planning, local worker presentation, retained-output reads after adoption, reveal-failure warnings, stable-pane Run/Dispatch routing, creator pane/process/Run-generation fencing, indexed retained-Run lookup scaling, the SSH in-process CLI fallback, exact legacy Task/Dispatch/report-receipt verification, JSON rejection envelopes, and capability-negotiated Run-home settlement with lost-ack replay. Three isolated macOS Electron journeys launch fake Codex workers through the real RPC path. The compiled-CLI journey spawns fresh out/cli processes for rejected and accepted worker_done plus worker-release, then independently verifies runtime and SQLite state. The presentation and restart journeys assert immediate inactive presentation, one live agent PID, stable PTY/incarnation/tab/leaf/worktree/Task/Dispatch identity, and no interruption after workspace re-entry; the restart journey additionally proves exact background adoption with readable ACK output and no resume replay. Distinct A/B artifacts plus live SSH, WSL, folder, remote-runtime, Linux, and Windows cutover journeys remain explicit gaps.", + "coverageNotes": "A deterministic service-state-machine oracle now models a current-contract worker and coordinator whose renderer graph identities disappear across an app/runtime update. It exercises the production verifier with restored PTY and hydrated hook commitments, proves authenticated completion replay across a fresh runtime, explicit takeover, ordinary mail routing, remote-attachment process fencing, retained Task/Dispatch/terminal identity, and unchanged fixture marker bytes, and rejects foreign pane evidence. Other deterministic units cover authority-aware legacy formatting, exact legacy worker identity planning, local worker presentation, retained-output reads after adoption, reveal-failure warnings, stable-pane Run/Dispatch routing, creator pane/process/Run-generation fencing, indexed retained-Run lookup scaling, the SSH in-process CLI fallback, exact legacy Task/Dispatch/report-receipt verification, JSON rejection envelopes, and capability-negotiated Run-home settlement. The federation oracle drops the first terminal ACK, restarts Run-home twice, verifies replay through the existing polling path and the persisted acknowledgment watermark, races identical reports, accepts preexisting same-outcome ACK batches, rejects conflicting batches, and covers protocol v1/v2 completion across independently updated peers. A migration unit upgrades a v26 database with a zeroed watermark without replaying older migrations. Three isolated macOS Electron journeys launch fake Codex workers through the real RPC path. Distinct installed A/B artifacts plus live SSH, WSL, folder, Linux, and Windows cutover journeys remain explicit gaps.", "motivatingLinks": [ "https://github.com/stablyai/orca/pull/11107#discussion_r3663321387", - "https://github.com/stablyai/orca/issues/13858" + "https://github.com/stablyai/orca/issues/13858", + "https://github.com/stablyai/orca/pull/14105", + "https://linear.app/stably/issue/STA-4020", + "https://linear.app/stably/issue/STA-4021", + "https://linear.app/stably/issue/STA-4022" ], - "invariant": "Starting a worker in the coordinator's current workspace must materialize one inactive terminal tab before worker-start returns, preserve coordinator focus, and remain exactly once after workspace re-entry. After an app update or restart, an exact live legacy worker must fence automatic provider resume, adopt its original PTY into its original background pane, retain readable output, and clear the resume record without spawning, writing, signalling, interrupting, replacing, or focusing the worker. A current-contract worker whose renderer graph identity is temporarily absent must retain its Dispatch capability and settle exactly once from exact hook-attested handle, pane, and process evidence; otherwise only an exact attested coordinator may take over. A worker_done caller may report success only after the owning runtime returns an explicit lifecycle verdict or authoritative reads prove that the exact Task, Dispatch, and worker report receipt settled the expected outcome. A federated worker may accept only the Run-home runtime's settlement verdict. An exact existing target workspace must receive a discoverable tab without stealing coordinator focus; if renderer reveal fails, worker-start must expose that the live worker remains background-only. Run and Dispatch checks must resolve through the caller's stable pane identity when a terminal handle is reminted, while a live handle outranks mismatched pane metadata. A nested worker's creator edge requires the current creator pane, process incarnation, and owning Run generation; reminting and rebinding that pane to another Run must remove the stale edge. Explicit legacy terminal inspection remains handle-scoped, and remote or headless worker presentation remains background-only.", + "invariant": "Starting a worker in the coordinator's current workspace must materialize one inactive terminal tab before worker-start returns, preserve coordinator focus, and remain exactly once after workspace re-entry. After an app update or restart, an exact live legacy worker must fence automatic provider resume, adopt its original PTY into its original background pane, retain readable output, and clear the resume record without spawning, writing, signalling, interrupting, replacing, or focusing the worker. A current-contract worker whose renderer graph identity is temporarily absent must retain its Dispatch capability and settle exactly once from exact hook-attested handle, pane, and process evidence; otherwise only an exact attested coordinator may take over. A worker_done caller may report success only after the owning runtime returns an explicit lifecycle verdict or authoritative reads prove that the exact Task, Dispatch, and worker report receipt settled the expected outcome. Federated terminal settlement must remain replay-eligible until the worker durably acknowledges it, and identical same-outcome retries must converge idempotently. Independently updated clients and worker servers must preserve the negotiated protocol: current peers use Run-home lifecycle settlement, while protocol v1/v2 peers retain their legacy completion path without receiving newer-only fields. A federated worker may accept only the authority defined by its negotiated protocol. An exact existing target workspace must receive a discoverable tab without stealing coordinator focus; if renderer reveal fails, worker-start must expose that the live worker remains background-only. Run and Dispatch checks must resolve through the caller's stable pane identity when a terminal handle is reminted, while a live handle outranks mismatched pane metadata. A nested worker's creator edge requires the current creator pane, process incarnation, and owning Run generation; reminting and rebinding that pane to another Run must remove the stale edge. Explicit legacy terminal inspection remains handle-scoped, and remote or headless worker presentation remains background-only.", "oracle": "Drive Run create, Task create, and worker-start through production Electron runtimes with a deterministic Codex fixture. Require append-only ledgers with one still-live PID and no interruption, a visible inactive worker tab while the coordinator stays active, Run delivery through stable pane identity, and stable PTY/incarnation, tab, leaf, worktree, Task, and Dispatch across workspace re-entry. In a restart journey, retain the original daemon PTY and PID, remove renderer ownership, retain sleeping-session evidence, mark the Dispatch legacy, relaunch, and require exact inactive tab adoption, readable ACK output, cleared resume state, one spawn, and no resume argv or Conversation interrupted text after another workspace round trip. The service oracle removes renderer lookup identity from current-contract callers while retaining real restored-PTY and hook commitments, replays authenticated completion and takeover across fresh runtimes, and requires one Task, Dispatch, terminal authority, message, mutation, ordinary-mail delivery, remote process fencing, and unchanged fixture marker bytes while foreign pane evidence remains rejected. Unit tests separately remint a creator pane and process from Run A into Run B, require the nested Run A worker to fall back to its current coordinator, require indexed query plans, and bound 300 Task reads with 50,000 retained Runs. They also assert authority-specific legacy affordances, exact identity and owner matching, retained-output fallback, pane-stable routing, federated non-activation, and SSH fallback parity.", "commands": [ "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/orchestration-runtime-update-settlement.test.ts --reporter=dot", @@ -8414,6 +8418,7 @@ "pnpm exec vitest run --config config/vitest.config.ts src/cli/handlers/orchestration-lifecycle-rejection.test.ts src/cli/handlers/orchestration-lifecycle-json-rejection.test.ts src/cli/handlers/orchestration-migration.test.ts", "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration/formatter.test.ts src/main/runtime/rpc/methods/orchestration-federation.test.ts", "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts --reporter=dot", + "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts --reporter=dot", "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration/orchestration-legacy-worker-terminal-recovery.test.ts", "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration/orchestration-creator-authority-performance.test.ts", "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orca-runtime.test.ts", @@ -8436,6 +8441,7 @@ "src/main/runtime/rpc/methods/orchestration.test.ts", "src/main/runtime/rpc/methods/orchestration-federation.test.ts", "src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts", + "src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts", "src/main/ssh/ssh-remote-orca-cli.test.ts", "tests/e2e/orchestration-worker-terminal-visibility.spec.ts", "tests/e2e/orchestration-legacy-worker-restart-recovery.spec.ts", @@ -8530,8 +8536,13 @@ "file": "src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts", "assertions": [ "a remote worker_done waits for Run-home settlement even when an older CLI omits the wait hint", - "older negotiated federation protocols reject before queueing", - "a committed settlement is replayed after an acknowledgment is lost", + "protocol v1/v2 clients can start fresh workers and complete success or failure on a current worker server", + "protocol v1/v2 completion ACK loss replays safely after a Run-home restart without newer-only wire fields", + "a current Run home still completes reports from legacy protocol v1/v2 worker servers", + "persisted protocol v1/v2 workers retain legacy-compatible completion after their worker server updates", + "the existing relay polling path retries a committed settlement across Run-home restarts until the worker acknowledgment watermark advances", + "identical pending reports coalesce to one relay and every caller receives the same Run-home verdict", + "preexisting same-outcome terminal ACK batches settle atomically while conflicting outcomes acknowledge nothing", "an aborted settlement wait returns operation_unknown instead of accepted success", "an already-aborted signal registers no settlement waiter", "a Run-home terminal verdict settles the remote attachment and revokes its capability", @@ -8540,6 +8551,14 @@ "a rejected report replays its exact verdict without mutating durable rejection mail twice" ] }, + { + "file": "src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts", + "assertions": [ + "a v26 database gains a zeroed durable home-to-worker acknowledgment watermark", + "a complete v26 database starts migration at v26 rather than replaying v7-v26", + "the migration advances schema version without losing the imported relay cursor" + ] + }, { "file": "src/main/runtime/rpc/methods/orchestration-federation.test.ts", "assertions": ["federated worker placement explicitly sets activate=false"] @@ -8582,6 +8601,69 @@ } ], "evidenceRuns": [ + { + "date": "2026-08-13", + "runner": "local", + "platform": "macos", + "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts --reporter=dot", + "result": "passed", + "durationSeconds": 6.02, + "summary": "The 70f1d52f mixed-version oracle passed all 21 cases. Protocol v1/v2 clients started fresh workers on a current server, completed success and failure with explicit legacy authority, and automatically retried a lost ACK after Run-home restart; current-protocol settlement and duplicate-report controls stayed green." + }, + { + "date": "2026-08-13", + "runner": "local", + "platform": "macos", + "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts --reporter=dot", + "result": "failed", + "durationSeconds": 4.21, + "summary": "The byte-identical 70f1d52f oracle failed 6 mixed-version cases while 15 controls passed when the fresh v1/v2 refusal was restored: success and failure through both negotiated versions plus both lost-ACK restart cases." + }, + { + "date": "2026-08-12", + "runner": "local", + "platform": "macos", + "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts --reporter=dot", + "result": "failed", + "durationSeconds": 5.05, + "summary": "The byte-identical ac7bdf4e federation oracle failed 7 of 17 tests on affected 09ec516ae5: fresh v1/v2 work started before completion rejection, persisted v1/v2 work could not finish after update, same-outcome ACKs rejected, duplicate reports remained pending, and a dropped ACK was not replayed." + }, + { + "date": "2026-08-12", + "runner": "local", + "platform": "macos", + "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts --reporter=dot", + "result": "failed", + "durationSeconds": 5.86, + "summary": "The same byte-identical oracle failed the same 7 of 17 tests on latest main 1136503c6a." + }, + { + "date": "2026-08-12", + "runner": "local", + "platform": "macos", + "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts --reporter=dot", + "result": "passed", + "durationSeconds": 4.28, + "summary": "The same byte-identical oracle passed all 17 tests on candidate 008f740161, including restart replay and both directions of v1/v2 update compatibility." + }, + { + "date": "2026-08-12", + "runner": "local", + "platform": "macos", + "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts --reporter=dot", + "result": "failed", + "durationSeconds": 19.84, + "summary": "With the claimed production files restored to latest main in 3a15d3ed5d, the same byte-identical oracle returned to the same 7 failures while 10 unaffected cases still passed." + }, + { + "date": "2026-08-12", + "runner": "local", + "platform": "macos", + "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts --reporter=dot", + "result": "passed", + "durationSeconds": 0.47, + "summary": "A complete v26 database resolved migration start at v26, upgraded to v27, retained its imported cursor, and initialized the durable acknowledgment watermark to zero." + }, { "date": "2026-08-12", "runner": "local", @@ -8692,11 +8774,11 @@ }, "redGreenEvidence": { "status": "complete", - "evidence": "The byte-identical compiled-CLI Electron oracle failed on latest main 5ea7df1a5b because a wrong worker identity exited zero with ok=true while Dispatch stayed dispatched, passed on combined candidate d697666ce8 with a nonzero ok=false rejection and explicit completed verdict for the exact worker, and reproduced the main failure when the claimed production files were disabled at merge-base 64aec94cb2. The earlier byte-identical service oracle was also red on main, green on candidate, red with authority propagation disabled, and green after restore. Distinct installed A/B and CI artifacts are still needed." + "evidence": "The byte-identical ac7bdf4e federation oracle failed 7 of 17 tests on affected 09ec516ae5 and latest main 1136503c6a, passed 17 of 17 on candidate 008f740161, and returned to the same 7 failures when the claimed production files were restored to latest main in 3a15d3ed5d. The follow-up byte-identical 70f1d52f oracle passes 21 of 21 with fresh v1/v2 compatibility and fails 6 mixed-version cases while 15 controls pass when only the refusal is restored. Together they prove dropped-ACK replay across Run-home restarts for current and legacy protocols, identical and preexisting same-outcome retry convergence, conflicting-outcome refusal, fresh and persisted old-client/current-server completion, current-client/old-server completion, and legacy wire isolation. The compiled-CLI Electron oracle separately failed on main 5ea7df1a5b, passed on candidate d697666ce8, and reproduced its original failure with the claimed production files disabled. Distinct installed A/B and CI artifacts are still needed." }, "performanceBudget": { "required": true, - "evidence": "Worker-start reuses the existing one-shot renderer reveal and adds no polling, provider listing, or output work. Startup recovery performs one bounded controller inventory per legacy candidate, exact owner/identity checks, one background reveal, and a provider/renderer snapshot only when an adopted worker's in-memory tail is empty and terminal.read is explicitly requested. Check adds one optional pane-key field and reuses the existing Run scan or bounded active-Dispatch lookup. Federated and explicitly background terminals are unchanged." + "evidence": "Worker-start reuses the existing one-shot renderer reveal and adds no polling, provider listing, or output work. Startup recovery performs one bounded controller inventory per legacy candidate, exact owner/identity checks, one background reveal, and a provider/renderer snapshot only when an adopted worker's in-memory tail is empty and terminal.read is explicitly requested. Check adds one optional pane-key field and reuses the existing Run scan or bounded active-Dispatch lookup. Federation replay keeps the existing per-Dispatch polling cadence and existing transport queue limit, retaining a terminal Dispatch only while its imported cursor exceeds one persisted acknowledgment watermark. Each successful ACK adds one primary-key SQLite update; exact eligibility is a primary-key lookup, and identical pending terminal lookup is bounded by the existing 256-item per-Dispatch relay quota and pending-relay index. A version-aware schema sentinel keeps valid v26 databases on the direct v26-to-v27 migration path." }, "promotionCriteria": [ "Collect 100 consecutive passes or 14 days of stable CI history on macOS, Linux, and Windows.", diff --git a/src/cli/handlers/orchestration-federated-legacy-settlement.test.ts b/src/cli/handlers/orchestration-federated-legacy-settlement.test.ts new file mode 100644 index 00000000000..0c12f30e8ea --- /dev/null +++ b/src/cli/handlers/orchestration-federated-legacy-settlement.test.ts @@ -0,0 +1,75 @@ +import { afterEach, expect, it, vi } from 'vitest' + +const callMock = vi.fn() +vi.mock('../format', () => ({ printResult: vi.fn() })) +vi.mock('../selectors', () => ({ getTerminalHandle: vi.fn() })) + +import { ORCHESTRATION_HANDLERS } from './orchestration' + +afterEach(() => callMock.mockReset()) + +it('accepts an explicitly attested legacy worker-server settlement', async () => { + callMock.mockResolvedValueOnce({ + result: { + relay: { + messageId: 'relay_legacy_done', + sequence: 1, + dispatchId: 'ctx_legacy', + destination: 'run_home', + accepted: true + }, + lifecycle: { + action: 'completed', + authority: 'worker_server_legacy' + } + } + }) + + await ORCHESTRATION_HANDLERS['orchestration send']({ + flags: new Map([ + ['from', 'term_worker'], + ['subject', 'done'], + ['type', 'worker_done'], + ['task-id', 'task_legacy'], + ['dispatch-id', 'ctx_legacy'], + ['outcome', 'succeeded'] + ]), + client: { call: callMock }, + cwd: '/tmp/repo', + json: true + } as never) + + expect(callMock).toHaveBeenCalledOnce() +}) + +it('rejects an unattested relay settlement', async () => { + callMock.mockResolvedValueOnce({ + result: { + relay: { + messageId: 'relay_unattested', + sequence: 1, + dispatchId: 'ctx_legacy', + destination: 'run_home', + accepted: true + }, + lifecycle: { action: 'completed' } + } + }) + + await expect( + ORCHESTRATION_HANDLERS['orchestration send']({ + flags: new Map([ + ['from', 'term_worker'], + ['subject', 'done'], + ['type', 'worker_done'], + ['task-id', 'task_legacy'], + ['dispatch-id', 'ctx_legacy'], + ['outcome', 'succeeded'] + ]), + client: { call: callMock }, + cwd: '/tmp/repo', + json: true + } as never) + ).rejects.toMatchObject({ code: 'operation_unknown' }) + expect(callMock).toHaveBeenCalledOnce() +}) diff --git a/src/cli/handlers/orchestration-worker-settlement.ts b/src/cli/handlers/orchestration-worker-settlement.ts index 8b9425391ee..9458e61a487 100644 --- a/src/cli/handlers/orchestration-worker-settlement.ts +++ b/src/cli/handlers/orchestration-worker-settlement.ts @@ -103,8 +103,9 @@ function hasLifecycleVerdict(result: unknown): boolean { } const action = (lifecycle as { action?: unknown }).action if ('relay' in result) { + const authority = (lifecycle as { authority?: unknown }).authority return ( - (lifecycle as { authority?: unknown }).authority === 'run_home' && + (authority === 'run_home' || authority === 'worker_server_legacy') && (action === 'completed' || action === 'failed' || action === 'rejected') ) } diff --git a/src/cli/handlers/orchestration.ts b/src/cli/handlers/orchestration.ts index 771b8367b88..337f026f9bf 100644 --- a/src/cli/handlers/orchestration.ts +++ b/src/cli/handlers/orchestration.ts @@ -92,7 +92,10 @@ const WORKER_TERMINAL_LIST_STATES = [ ] as const type LifecycleSendResult = - | { action: 'completed' | 'failed'; authority?: 'run_home' } + | { + action: 'completed' | 'failed' + authority?: 'run_home' | 'worker_server_legacy' + } | { action: 'settled'; outcome: 'succeeded' | 'failed'; duplicate?: boolean } | { action: 'rejected'; code: string; reason: string } diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 856b5960b32..b789248f44d 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -2721,6 +2721,10 @@ export class OrcaRuntimeService { private managedHookReconciliationTail: Promise = Promise.resolve() private readonly orchestrationEnvironmentTransport: OrchestrationEnvironmentTransport | null private readonly orchestrationFederationTimers = new Map>() + private orchestrationTerminalHistoryRecoveryTimer: ReturnType | null = null + private orchestrationTerminalHistoryRecoveryInFlight: Promise | null = null + private orchestrationTerminalRecoveryRowId = 0 + private orchestrationFederationRelayGeneration = 0 private readonly orchestrationFederationSyncs = new Map< string, { db: OrchestrationDb; promise: Promise } @@ -3851,15 +3855,15 @@ export class OrcaRuntimeService { const { app } = require('electron') const dbPath = join(app.getPath('userData'), 'orchestration.db') this._orchestrationDb = new OrchestrationDb(dbPath) + this.ensureOrchestrationFederationRelay() } return this._orchestrationDb } setOrchestrationDb(db: OrchestrationDb): void { - clearFederationAckCheckpoints(this) - this.orchestrationFederationSyncs.clear() - this.orchestrationFederationWarnings.clear() + this.stopOrchestrationFederationRelay() this._orchestrationDb = db + this.ensureOrchestrationFederationRelay() } private getLegacyWorkerTerminalRecoveryPlan(): LegacyWorkerTerminalRecoveryPlan { @@ -4738,8 +4742,7 @@ export class OrcaRuntimeService { continue } const tick = () => { - const worker = this.getOrchestrationDb().getWorkerDispatch(dispatch.dispatch_id) - if (!worker || !['starting', 'ready', 'stopping'].includes(worker.state)) { + if (!this.getOrchestrationDb().isFederatedDispatchRelayEligible(dispatch.dispatch_id)) { const activeTimer = this.orchestrationFederationTimers.get(dispatch.dispatch_id) if (activeTimer) { clearInterval(activeTimer) @@ -4755,13 +4758,64 @@ export class OrcaRuntimeService { this.orchestrationFederationTimers.set(dispatch.dispatch_id, timer) tick() } + this.ensureTerminalHistoryRecovery() + } + + private ensureTerminalHistoryRecovery(): void { + if ( + this.orchestrationTerminalHistoryRecoveryTimer || + this.orchestrationTerminalHistoryRecoveryInFlight + ) { + return + } + const generation = this.orchestrationFederationRelayGeneration + const recovery = this.recoverNextTerminalHistoryAcknowledgment(generation).catch((error) => { + console.warn('[orchestration] terminal federation acknowledgment recovery failed', error) + }) + this.orchestrationTerminalHistoryRecoveryInFlight = recovery + void recovery.finally(() => { + if (this.orchestrationTerminalHistoryRecoveryInFlight === recovery) { + this.orchestrationTerminalHistoryRecoveryInFlight = null + } + }) + } + + private async recoverNextTerminalHistoryAcknowledgment(generation: number): Promise { + const db = this.getOrchestrationDb() + let historical = db.findNextTerminalFederatedDispatchPendingAcknowledgment( + this.orchestrationTerminalRecoveryRowId + ) + if (!historical && this.orchestrationTerminalRecoveryRowId > 0) { + this.orchestrationTerminalRecoveryRowId = 0 + historical = db.findNextTerminalFederatedDispatchPendingAcknowledgment(0) + } + if (!historical) { + return + } + this.orchestrationTerminalRecoveryRowId = historical.rowId + await this.syncOrchestrationFederatedDispatch(historical.dispatchId).catch(() => undefined) + if (generation !== this.orchestrationFederationRelayGeneration) { + return + } + this.orchestrationTerminalHistoryRecoveryTimer = setTimeout(() => { + this.orchestrationTerminalHistoryRecoveryTimer = null + this.ensureTerminalHistoryRecovery() + }, 1_000) + this.orchestrationTerminalHistoryRecoveryTimer.unref?.() } stopOrchestrationFederationRelay(): void { + this.orchestrationFederationRelayGeneration += 1 for (const timer of this.orchestrationFederationTimers.values()) { clearInterval(timer) } this.orchestrationFederationTimers.clear() + if (this.orchestrationTerminalHistoryRecoveryTimer) { + clearTimeout(this.orchestrationTerminalHistoryRecoveryTimer) + this.orchestrationTerminalHistoryRecoveryTimer = null + } + this.orchestrationTerminalHistoryRecoveryInFlight = null + this.orchestrationTerminalRecoveryRowId = 0 this.orchestrationFederationWarnings.clear() this.orchestrationFederationSyncs.clear() clearFederationAckCheckpoints(this) diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index a1ab618847c..38b0b194c78 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -53,7 +53,10 @@ import { type WorkerTerminalRetainedReason } from './worker-terminal-ownership' import { ORCHESTRATION_RUN_PAGE_LIMIT } from '../../../shared/orchestration-run-pagination' -import { ORCHESTRATION_CONTRACT_VERSION } from '../../../shared/protocol-version' +import { + ORCHESTRATION_CONTRACT_VERSION, + ORCHESTRATION_FEDERATION_LIFECYCLE_SETTLEMENT_PROTOCOL_VERSION +} from '../../../shared/protocol-version' import { releaseContextOnlyDispatch, type ContextOnlyDispatchReleaseResult @@ -284,8 +287,8 @@ type RunListCursor = { id: string } -// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 legacy scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity. -const SCHEMA_VERSION = 26 +// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 legacy scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity, v27 durable federation acknowledgments. +const SCHEMA_VERSION = 27 function hardenOrchestrationDatabaseFiles(dbPath: (string & {}) | ':memory:'): void { if (dbPath === ':memory:' || process.platform === 'win32') { @@ -467,6 +470,7 @@ export class OrchestrationDb { remote_worktree_id TEXT, remote_terminal_handle TEXT, to_home_imported_sequence INTEGER NOT NULL DEFAULT 0, + to_home_acknowledged_sequence INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (datetime('now')) ); @@ -870,6 +874,7 @@ export class OrchestrationDb { remote_worktree_id TEXT, remote_terminal_handle TEXT, to_home_imported_sequence INTEGER NOT NULL DEFAULT 0, + to_home_acknowledged_sequence INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (datetime('now')) ); @@ -989,6 +994,14 @@ export class OrchestrationDb { if (current < 26) { migrateMutationReceiptCapacity(this.db) } + if ( + current < 27 && + !this.hasColumn('federated_dispatches', 'to_home_acknowledged_sequence') + ) { + this.db.exec( + 'ALTER TABLE federated_dispatches ADD COLUMN to_home_acknowledged_sequence INTEGER NOT NULL DEFAULT 0' + ) + } this.db.exec(` CREATE INDEX IF NOT EXISTS idx_dispatch_assignee_pane_leaf ON dispatch_contexts(${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL}) @@ -4550,6 +4563,42 @@ export class OrchestrationDb { .all(runId ?? null, runId ?? null) as FederatedDispatchRow[] } + findNextTerminalFederatedDispatchPendingAcknowledgment( + afterRowId: number + ): { dispatchId: string; rowId: number } | undefined { + return this.db + .prepare( + `SELECT fd.dispatch_id AS dispatchId, fd.rowid AS rowId + FROM federated_dispatches fd + INNER JOIN worker_dispatches wd ON wd.dispatch_id = fd.dispatch_id + WHERE wd.state NOT IN ('starting', 'ready', 'stopping', 'start_unknown', 'stop_unknown') + AND fd.to_home_acknowledged_sequence < fd.to_home_imported_sequence + AND fd.rowid > ? + ORDER BY fd.rowid + LIMIT 1` + ) + .get(afterRowId) as { dispatchId: string; rowId: number } | undefined + } + + isFederatedDispatchRelayEligible(dispatchId: string): boolean { + return Boolean( + this.db + .prepare( + `SELECT 1 + FROM federated_dispatches fd + INNER JOIN worker_dispatches wd ON wd.dispatch_id = fd.dispatch_id + WHERE fd.dispatch_id = ? + AND ( + wd.state IN ('starting', 'ready', 'stopping', 'start_unknown', 'stop_unknown') + OR ( + fd.to_home_acknowledged_sequence < fd.to_home_imported_sequence + ) + )` + ) + .get(dispatchId) + ) + } + updateFederatedDispatchResources(params: { dispatchId: string remoteRuntimeEpoch: string @@ -4964,6 +5013,26 @@ export class OrchestrationDb { ) as FederationRelayItemRow } } + if (params.kind === 'worker_done') { + const identicalReport = this.db + .prepare( + `SELECT * FROM federation_relay_items + WHERE dispatch_id = ? AND direction = ? AND kind = 'worker_done' + AND payload = ? AND acked_at IS NULL + ORDER BY sequence DESC LIMIT 1` + ) + .get(params.dispatchId, params.direction, params.payload) as + | FederationRelayItemRow + | undefined + if (identicalReport) { + this.settleRemoteAttachmentInRelayTransaction( + params.dispatchId, + params.settleRemoteOutcome + ) + this.db.exec('COMMIT') + return identicalReport + } + } const quota = this.db .prepare( `SELECT COUNT(*) AS count, COALESCE(SUM(byte_count), 0) AS bytes @@ -5096,30 +5165,66 @@ export class OrchestrationDb { dispatchId: string direction: FederationRelayDirection throughSequence: number - settleRemoteReport?: { sequence: number; outcome: WorkerReportOutcome } + settleRemoteReports?: { sequence: number; outcome?: WorkerReportOutcome }[] }): void { this.db.exec('BEGIN IMMEDIATE') try { - if (params.settleRemoteReport) { + const settledReports = params.settleRemoteReports ?? [] + for (const settledReport of settledReports) { const report = this.getFederationRelayItem( params.dispatchId, params.direction, - params.settleRemoteReport.sequence + settledReport.sequence ) if ( params.direction !== 'to_home' || - params.settleRemoteReport.sequence > params.throughSequence || + settledReport.sequence > params.throughSequence || report?.kind !== 'worker_done' || - parseFederatedWorkerReportOutcome(report.payload) !== params.settleRemoteReport.outcome + (settledReport.outcome !== undefined && + parseFederatedWorkerReportOutcome(report.payload) !== settledReport.outcome) ) { throw new OrchestrationError( 'request_mismatch', `Federation acknowledgment for ${params.dispatchId} does not match its queued worker_done.` ) } + } + const attachment = this.getRemoteDispatchAttachment(params.dispatchId) + if ( + params.direction === 'to_home' && + attachment !== undefined && + attachment.protocol_version >= + ORCHESTRATION_FEDERATION_LIFECYCLE_SETTLEMENT_PROTOCOL_VERSION + ) { + const acknowledgedReports = this.db + .prepare( + `SELECT sequence FROM federation_relay_items + WHERE dispatch_id = ? AND direction = 'to_home' AND kind = 'worker_done' + AND acked_at IS NULL AND sequence <= ?` + ) + .all(params.dispatchId, params.throughSequence) as { sequence: number }[] + const settledSequences = new Set(settledReports.map((report) => report.sequence)) + if (acknowledgedReports.some((report) => !settledSequences.has(report.sequence))) { + throw new OrchestrationError( + 'request_mismatch', + `Federation acknowledgment for ${params.dispatchId} omits a worker_done settlement.` + ) + } + } + const terminalOutcomes = new Set( + settledReports.flatMap((report) => (report.outcome ? [report.outcome] : [])) + ) + if (terminalOutcomes.size > 1) { + throw new OrchestrationError( + 'request_mismatch', + `Federation acknowledgment for ${params.dispatchId} contains conflicting settlements.` + ) + } + const terminalOutcome = settledReports.find((report) => report.outcome)?.outcome + if (terminalOutcome) { this.settleRemoteAttachmentInRelayTransaction( params.dispatchId, - params.settleRemoteReport.outcome, + terminalOutcome, 'worker_report_settled' ) } @@ -5146,6 +5251,44 @@ export class OrchestrationDb { .run(sequence, dispatchId, sequence) } + recordFederatedHomeAcknowledgment(params: { + dispatchId: string + remoteRuntimeEpoch: string + sequence: number + }): void { + const federated = this.getFederatedDispatch(params.dispatchId) + if ( + !federated || + !Number.isInteger(params.sequence) || + params.sequence < 0 || + params.sequence > federated.to_home_imported_sequence + ) { + throw new OrchestrationError( + 'request_mismatch', + `Federation acknowledgment for ${params.dispatchId} exceeds imported relay state.` + ) + } + this.db + .prepare( + `UPDATE federated_dispatches + SET remote_runtime_epoch = ?, + to_home_acknowledged_sequence = CASE + WHEN remote_runtime_epoch = ? + THEN MAX(to_home_acknowledged_sequence, ?) + ELSE ? + END, + updated_at = datetime('now') + WHERE dispatch_id = ?` + ) + .run( + params.remoteRuntimeEpoch, + params.remoteRuntimeEpoch, + params.sequence, + params.sequence, + params.dispatchId + ) + } + importFederatedRelayItem(params: { dispatchId: string sequence: number diff --git a/src/main/runtime/orchestration/federation-acknowledgment-integrity.test.ts b/src/main/runtime/orchestration/federation-acknowledgment-integrity.test.ts new file mode 100644 index 00000000000..3f8f4022393 --- /dev/null +++ b/src/main/runtime/orchestration/federation-acknowledgment-integrity.test.ts @@ -0,0 +1,100 @@ +import { afterEach, describe, expect, it } from 'vitest' +import { OrchestrationDb } from './db' + +describe('federation acknowledgment integrity', () => { + let db: OrchestrationDb | undefined + + afterEach(() => db?.close()) + + function createReadyAttachment(protocolVersion: number): { + db: OrchestrationDb + dispatchId: string + } { + db = new OrchestrationDb(':memory:') + const dispatchId = `ctx_protocol_${protocolVersion}` + db.createRemoteDispatchAttachment({ + dispatchId, + taskId: `task_protocol_${protocolVersion}`, + homePeerFingerprint: 'home_peer', + protocolVersion, + runtimeEpoch: 'worker_epoch', + mutationReceipt: { + callerFingerprint: 'home_peer', + requestId: `attach_protocol_${protocolVersion}`, + method: 'orchestration.federationAttachStart', + payloadHash: `attach_hash_${protocolVersion}` + } + }) + db.recordRemoteAttachmentStage({ + dispatchId, + stage: 'input_accepted', + state: 'ready' + }) + return { db, dispatchId } + } + + function enqueueCompletion( + target: OrchestrationDb, + dispatchId: string, + settleRemoteOutcome?: 'succeeded' + ) { + return target.enqueueFederationRelay({ + dispatchId, + direction: 'to_home', + kind: 'worker_done', + payload: JSON.stringify({ + payload: JSON.stringify({ + taskId: dispatchId.replace('ctx_', 'task_'), + dispatchId, + outcome: 'succeeded' + }) + }), + ...(settleRemoteOutcome ? { settleRemoteOutcome } : {}) + }) + } + + it('rejects a protocol-v3 acknowledgment without a lifecycle verdict atomically', () => { + const current = createReadyAttachment(3) + const report = enqueueCompletion(current.db, current.dispatchId) + + expect(() => + current.db.acknowledgeFederationRelay({ + dispatchId: current.dispatchId, + direction: 'to_home', + throughSequence: report.sequence + }) + ).toThrowError(expect.objectContaining({ code: 'request_mismatch' })) + expect(current.db.getRemoteDispatchAttachment(current.dispatchId)?.state).toBe('ready') + expect(current.db.listPendingFederationRelay(current.dispatchId, 'to_home')).toHaveLength(1) + }) + + it('keeps legacy acknowledgments optional after local completion', () => { + const legacy = createReadyAttachment(2) + const report = enqueueCompletion(legacy.db, legacy.dispatchId, 'succeeded') + + expect(() => + legacy.db.acknowledgeFederationRelay({ + dispatchId: legacy.dispatchId, + direction: 'to_home', + throughSequence: report.sequence + }) + ).not.toThrow() + expect(legacy.db.getRemoteDispatchAttachment(legacy.dispatchId)?.state).toBe('succeeded') + expect(legacy.db.listPendingFederationRelay(legacy.dispatchId, 'to_home')).toHaveLength(0) + }) + + it('acknowledges a protocol-v3 rejected report without settling the attachment', () => { + const current = createReadyAttachment(3) + const report = enqueueCompletion(current.db, current.dispatchId) + + current.db.acknowledgeFederationRelay({ + dispatchId: current.dispatchId, + direction: 'to_home', + throughSequence: report.sequence, + settleRemoteReports: [{ sequence: report.sequence }] + }) + + expect(current.db.getRemoteDispatchAttachment(current.dispatchId)?.state).toBe('ready') + expect(current.db.listPendingFederationRelay(current.dispatchId, 'to_home')).toHaveLength(0) + }) +}) diff --git a/src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts b/src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts new file mode 100644 index 00000000000..80959d3df48 --- /dev/null +++ b/src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts @@ -0,0 +1,50 @@ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it } from 'vitest' +import Database from '../../sqlite/sync-database' +import { OrchestrationDb } from './db' +import { resolveOrchestrationMigrationStartVersion } from './orchestration-schema-version-skew' + +describe('federation acknowledgment migration', () => { + let db: OrchestrationDb | undefined + let tempDir: string | undefined + + afterEach(() => { + db?.close() + if (tempDir) { + rmSync(tempDir, { recursive: true, force: true }) + } + }) + + it('adds a zeroed durable acknowledgment watermark to v26 dispatches', () => { + tempDir = mkdtempSync(join(tmpdir(), 'orca-federation-ack-migration-')) + const dbPath = join(tempDir, 'orchestration.db') + db = new OrchestrationDb(dbPath) + db.close() + db = undefined + + const oldDb = new Database(dbPath) + oldDb.exec('ALTER TABLE federated_dispatches DROP COLUMN to_home_acknowledged_sequence') + oldDb.pragma('user_version = 26') + expect(resolveOrchestrationMigrationStartVersion(oldDb, 26, 27)).toBe(26) + oldDb + .prepare( + `INSERT INTO federated_dispatches ( + dispatch_id, environment_id, environment_name, peer_fingerprint, + protocol_version, to_home_imported_sequence + ) VALUES ('ctx_migrated', 'env', 'worker', 'peer', 3, 2)` + ) + .run() + oldDb.close() + + db = new OrchestrationDb(dbPath) + const sqlite = (db as unknown as { db: Database.Database }).db + + expect(sqlite.pragma('user_version', { simple: true })).toBe(27) + expect(db.getFederatedDispatch('ctx_migrated')).toMatchObject({ + to_home_imported_sequence: 2, + to_home_acknowledged_sequence: 0 + }) + }) +}) diff --git a/src/main/runtime/orchestration/federation-lifecycle-settlement.test.ts b/src/main/runtime/orchestration/federation-lifecycle-settlement.test.ts new file mode 100644 index 00000000000..2d8cb540633 --- /dev/null +++ b/src/main/runtime/orchestration/federation-lifecycle-settlement.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, it } from 'vitest' +import { + areFederatedLifecycleSettlementsEqual, + type FederatedLifecycleSettlement +} from './federation-lifecycle-settlement' + +const rejected = (reason: string): FederatedLifecycleSettlement => ({ + action: 'rejected', + code: 'worker_report_rejected', + reason, + authority: 'run_home' +}) + +describe('federated lifecycle settlement equality', () => { + it('accepts exact duplicate rejection verdicts', () => { + expect(areFederatedLifecycleSettlementsEqual(rejected('same'), rejected('same'))).toBe(true) + }) + + it.each([ + [rejected('first'), rejected('second')], + [rejected('same'), { ...rejected('same'), code: 'different_code' }] + ])('distinguishes rejection verdicts with different details', (left, right) => { + expect(areFederatedLifecycleSettlementsEqual(left, right)).toBe(false) + }) + + it('distinguishes terminal outcomes', () => { + expect( + areFederatedLifecycleSettlementsEqual( + { action: 'completed', authority: 'run_home' }, + { action: 'failed', authority: 'run_home' } + ) + ).toBe(false) + }) +}) diff --git a/src/main/runtime/orchestration/federation-lifecycle-settlement.ts b/src/main/runtime/orchestration/federation-lifecycle-settlement.ts index 921db91b728..abd86fe099a 100644 --- a/src/main/runtime/orchestration/federation-lifecycle-settlement.ts +++ b/src/main/runtime/orchestration/federation-lifecycle-settlement.ts @@ -4,6 +4,17 @@ export type FederatedLifecycleSettlement = | { action: 'completed' | 'failed'; authority: 'run_home' } | { action: 'rejected'; code: string; reason: string; authority: 'run_home' } +export function areFederatedLifecycleSettlementsEqual( + left: FederatedLifecycleSettlement, + right: FederatedLifecycleSettlement +): boolean { + return ( + left.action === right.action && + (left.action !== 'rejected' || + (right.action === 'rejected' && left.code === right.code && left.reason === right.reason)) + ) +} + type Waiter = (settlement: FederatedLifecycleSettlement) => void const waitersByRuntime = new WeakMap>>() diff --git a/src/main/runtime/orchestration/federation-sync.test.ts b/src/main/runtime/orchestration/federation-sync.test.ts index 5470fe6416d..7a2c11b6f71 100644 --- a/src/main/runtime/orchestration/federation-sync.test.ts +++ b/src/main/runtime/orchestration/federation-sync.test.ts @@ -7,7 +7,7 @@ import { recordFederationAckCheckpoint, type FederationAckIdentity } from './federation-ack-checkpoints' -import { parseRelayedMessage } from './federation-sync' +import { parseRelayedMessage, syncFederatedDispatch } from './federation-sync' function createIdleSyncHarness() { let remoteRuntimeEpoch = 'remote_epoch_1' @@ -17,14 +17,23 @@ function createIdleSyncHarness() { environment_id: 'environment_windows', environment_name: 'windows', peer_fingerprint: 'windows_peer_fingerprint', - to_home_imported_sequence: 2 + remote_runtime_epoch: remoteRuntimeEpoch, + to_home_imported_sequence: 2, + to_home_acknowledged_sequence: 0 } const createDb = () => ({ getFederatedDispatch: () => federated, getDispatchContextById: () => ({ run_id: 'run_home', task_id: 'task_home' }), getWorkerDispatch: () => ({ state: 'ready' }), - listPendingFederationRelay: () => [] + listPendingFederationRelay: () => [], + recordFederatedHomeAcknowledgment: (params: { + remoteRuntimeEpoch: string + sequence: number + }) => { + federated.remote_runtime_epoch = params.remoteRuntimeEpoch + federated.to_home_acknowledged_sequence = params.sequence + } }) as never const runtime = new OrcaRuntimeService() runtime.setOrchestrationDb(createDb()) @@ -104,6 +113,125 @@ describe('federation relay parsing', () => { }) describe('federation relay acknowledgments', () => { + it('drains a terminal retry from the page after the first terminal report', async () => { + const pending = Array.from({ length: 51 }, (_, index) => { + const sequence = index + 1 + const terminal = sequence >= 50 + return { + dispatch_id: 'dispatch_remote', + direction: 'to_home' as const, + sequence, + message_id: `message_${sequence}`, + kind: terminal ? 'worker_done' : 'status', + payload: JSON.stringify({ + subject: terminal ? 'Done' : 'Progress', + body: terminal ? `Attempt ${sequence}` : `Update ${sequence}`, + type: terminal ? 'worker_done' : 'status', + ...(terminal + ? { + payload: JSON.stringify({ + taskId: 'task_home', + dispatchId: 'dispatch_remote', + outcome: 'succeeded' + }) + } + : {}) + }) + } + }) + const federated = { + environment_id: 'environment_windows', + environment_name: 'windows', + peer_fingerprint: 'windows_peer_fingerprint', + remote_runtime_epoch: 'remote_epoch_1', + protocol_version: 3, + to_home_imported_sequence: 0, + to_home_acknowledged_sequence: 0 + } + let pendingToWorker = [{ sequence: 1 }] + const runtime = new OrcaRuntimeService() + runtime.setOrchestrationDb({ + getFederatedDispatch: () => federated, + getDispatchContextById: () => ({ run_id: 'run_home', task_id: 'task_home' }), + importFederatedRelayItem: ({ + sequence, + message, + lifecycle + }: { + sequence: number + message: { to: string; type: 'status' | 'worker_done' } + lifecycle: + | { kind: 'worker_report'; outcome: 'succeeded' | 'failed' } + | { kind: 'none' | 'heartbeat' | 'rejected' } + }) => { + federated.to_home_imported_sequence = sequence + return { + message: { to_handle: message.to, type: message.type }, + duplicate: false, + ...(lifecycle.kind === 'worker_report' + ? { lifecycle: { action: 'settled', outcome: lifecycle.outcome } } + : {}) + } + }, + recordFederatedHomeAcknowledgment: ({ sequence }: { sequence: number }) => { + federated.to_home_acknowledged_sequence = sequence + }, + getWorkerDispatch: () => ({ state: 'ready' }), + listPendingFederationRelay: () => pendingToWorker, + acknowledgeFederationRelay: () => { + pendingToWorker = [] + } + } as never) + vi.spyOn(runtime, 'resolveOrchestrationWorkerServer').mockReturnValue({ + peerFingerprint: federated.peer_fingerprint + } as never) + vi.spyOn(runtime, 'notifyMessageArrived').mockImplementation(() => {}) + const remoteCall = vi + .spyOn(runtime, 'callOrchestrationWorkerServer') + .mockImplementation(async (_environmentId, method, params) => { + if (method === 'orchestration.federationPull') { + return { runtimeEpoch: 'remote_epoch_1', items: pending.slice(0, 50) } + } + if (method === 'orchestration.federationAck') { + const throughSequence = (params as { throughSequence: number }).throughSequence + pending.splice( + 0, + pending.findIndex((item) => item.sequence > throughSequence) === -1 + ? pending.length + : pending.findIndex((item) => item.sequence > throughSequence) + ) + return { acknowledgedThrough: throughSequence } + } + if (method === 'orchestration.federationImport') { + return { acknowledgedThrough: 1 } + } + throw new Error(`Unexpected method ${method}`) + }) + + const result = await syncFederatedDispatch(runtime, 'dispatch_remote') + + expect(result).toEqual({ imported: 51, acknowledgedThrough: 51 }) + expect(pending).toHaveLength(0) + expect(remoteCall.mock.calls.map(([, method]) => method)).toEqual([ + 'orchestration.federationPull', + 'orchestration.federationAck', + 'orchestration.federationImport', + 'orchestration.federationPull', + 'orchestration.federationAck' + ]) + expect( + remoteCall.mock.calls + .filter(([, method]) => method === 'orchestration.federationAck') + .map(([, , params]) => params) + ).toEqual([ + expect.objectContaining({ throughSequence: 50 }), + expect.objectContaining({ + throughSequence: 51, + settlements: [expect.objectContaining({ sequence: 51 })] + }) + ]) + }) + it('acknowledges only new progress until remote runtime identity changes', async () => { const { runtime, remoteCall, advanceCursor, restartRemote } = createIdleSyncHarness() const ackCalls = () => diff --git a/src/main/runtime/orchestration/federation-sync.ts b/src/main/runtime/orchestration/federation-sync.ts index 13ec9dce561..f73c3757582 100644 --- a/src/main/runtime/orchestration/federation-sync.ts +++ b/src/main/runtime/orchestration/federation-sync.ts @@ -13,8 +13,11 @@ import { getFederationAckedThrough, recordFederationAckCheckpoint } from './federation-ack-checkpoints' +import { parseFederatedWorkerReportPayload } from './federation-worker-report-payload' const MESSAGE_TYPE_SET = new Set(MESSAGE_TYPES) +const FEDERATION_PULL_PAGE_SIZE = 50 +const MAX_FEDERATION_PULL_PAGES_PER_SYNC = 6 function isMessageType(value: unknown): value is MessageType { return typeof value === 'string' && MESSAGE_TYPE_SET.has(value as MessageType) @@ -42,6 +45,14 @@ type RelayedMessage = { export async function syncFederatedDispatch( runtime: OrcaRuntimeService, dispatchId: string +): Promise<{ imported: number; acknowledgedThrough: number }> { + return syncFederatedDispatchPages(runtime, dispatchId, MAX_FEDERATION_PULL_PAGES_PER_SYNC) +} + +async function syncFederatedDispatchPages( + runtime: OrcaRuntimeService, + dispatchId: string, + remainingPages: number ): Promise<{ imported: number; acknowledgedThrough: number }> { const db = runtime.getOrchestrationDb() const federated = db.getFederatedDispatch(dispatchId) @@ -70,7 +81,7 @@ export async function syncFederatedDispatch( dispatchId, afterSequence: federated.to_home_imported_sequence, ...(supportsLifecycleSettlement ? { replayUnacknowledged: true } : {}), - limit: 50 + limit: FEDERATION_PULL_PAGE_SIZE }, 15_000 )) as { runtimeEpoch: string; items: PulledRelayItem[] } @@ -127,8 +138,14 @@ export async function syncFederatedDispatch( peerFingerprint: federated.peer_fingerprint, remoteRuntimeEpoch: pulled.runtimeEpoch } - if (cursor > getFederationAckedThrough(ackLease, ackIdentity)) { - await runtime.callOrchestrationWorkerServer( + const durableAcknowledgedThrough = + federated.remote_runtime_epoch === pulled.runtimeEpoch + ? (federated.to_home_acknowledged_sequence ?? 0) + : 0 + if ( + cursor > Math.max(getFederationAckedThrough(ackLease, ackIdentity), durableAcknowledgedThrough) + ) { + const delivered = (await runtime.callOrchestrationWorkerServer( federated.environment_id, 'orchestration.federationAck', { @@ -138,10 +155,20 @@ export async function syncFederatedDispatch( }, 15_000, { orchestrationRequestId: `relay_ack_${dispatchId}_${cursor}` } - ) + )) as { acknowledgedThrough: number } + const keepRelayEligible = + pulled.items.length === FEDERATION_PULL_PAGE_SIZE && remainingPages === 1 + const locallyAcknowledgedThrough = keepRelayEligible + ? Math.max(0, delivered.acknowledgedThrough - 1) + : delivered.acknowledgedThrough + db.recordFederatedHomeAcknowledgment({ + dispatchId, + remoteRuntimeEpoch: pulled.runtimeEpoch, + sequence: locallyAcknowledgedThrough + }) recordFederationAckCheckpoint(runtime, ackLease, { ...ackIdentity, - throughSequence: cursor + throughSequence: locallyAcknowledgedThrough }) } const toWorker = @@ -164,6 +191,13 @@ export async function syncFederatedDispatch( throughSequence: delivered.acknowledgedThrough }) } + if (pulled.items.length === FEDERATION_PULL_PAGE_SIZE && remainingPages > 1) { + const next = await syncFederatedDispatchPages(runtime, dispatchId, remainingPages - 1) + return { + imported: imported + next.imported, + acknowledgedThrough: next.acknowledgedThrough + } + } return { imported, acknowledgedThrough: cursor } } @@ -222,7 +256,7 @@ function parseFederatedLifecycle( } let payload try { - payload = parseWorkerReportPayload(message.payload) + payload = parseFederatedWorkerReportPayload(message.payload) } catch (error) { return { kind: 'rejected', @@ -256,38 +290,3 @@ function parseFederatedLifecycle( result } } - -function parseWorkerReportPayload(payload: string | null): { - taskId: string - dispatchId: string - outcome: WorkerReportOutcome - filesModified: string[] - reportPath: string | null -} { - let parsed: unknown - try { - parsed = payload ? JSON.parse(payload) : null - } catch { - parsed = null - } - if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) { - throw new OrchestrationError('invalid_argument', 'Federated worker report is invalid.') - } - const report = parsed as Record - if ( - typeof report.taskId !== 'string' || - typeof report.dispatchId !== 'string' || - (report.outcome !== 'succeeded' && report.outcome !== 'failed') - ) { - throw new OrchestrationError('invalid_argument', 'Federated worker report is incomplete.') - } - return { - taskId: report.taskId, - dispatchId: report.dispatchId, - outcome: report.outcome, - filesModified: Array.isArray(report.filesModified) - ? report.filesModified.filter((file): file is string => typeof file === 'string') - : [], - reportPath: typeof report.reportPath === 'string' ? report.reportPath : null - } -} diff --git a/src/main/runtime/orchestration/federation-terminal-recovery.test.ts b/src/main/runtime/orchestration/federation-terminal-recovery.test.ts new file mode 100644 index 00000000000..f35e6f8e645 --- /dev/null +++ b/src/main/runtime/orchestration/federation-terminal-recovery.test.ts @@ -0,0 +1,102 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { OrcaRuntimeService } from '../orca-runtime' + +describe('terminal federation acknowledgment recovery', () => { + afterEach(() => { + vi.useRealTimers() + }) + + it('bounds migration replay to one oldest-first recovery', async () => { + vi.useFakeTimers() + const candidates = Array.from({ length: 1_000 }, (_, index) => ({ + dispatchId: `dispatch_${index + 1}`, + rowId: index + 1 + })) + const runtime = new OrcaRuntimeService(null, undefined, { + orchestrationEnvironmentTransport: { + resolve: vi.fn(), + call: vi.fn() + } + }) + const sync = vi + .spyOn(runtime, 'syncOrchestrationFederatedDispatch') + .mockResolvedValue(undefined) + runtime.setOrchestrationDb({ + listActiveFederatedDispatches: () => [], + findNextTerminalFederatedDispatchPendingAcknowledgment: (afterRowId: number) => + candidates.find((candidate) => candidate.rowId > afterRowId) + } as never) + + await vi.waitFor(() => expect(sync).toHaveBeenCalledTimes(1)) + expect(sync.mock.calls).toEqual([['dispatch_1']]) + + await vi.advanceTimersByTimeAsync(3_000) + + expect(sync.mock.calls).toEqual([ + ['dispatch_1'], + ['dispatch_2'], + ['dispatch_3'], + ['dispatch_4'] + ]) + runtime.stopOrchestrationFederationRelay() + }) + + it('gives every unavailable terminal dispatch a turn before retrying', async () => { + vi.useFakeTimers() + const candidates = [1, 2, 3].map((rowId) => ({ + dispatchId: `dispatch_${rowId}`, + rowId + })) + const runtime = new OrcaRuntimeService(null, undefined, { + orchestrationEnvironmentTransport: { + resolve: vi.fn(), + call: vi.fn() + } + }) + const sync = vi + .spyOn(runtime, 'syncOrchestrationFederatedDispatch') + .mockRejectedValue(new Error('worker unavailable')) + runtime.setOrchestrationDb({ + listActiveFederatedDispatches: () => [], + findNextTerminalFederatedDispatchPendingAcknowledgment: (afterRowId: number) => + candidates.find((candidate) => candidate.rowId > afterRowId) + } as never) + + await vi.advanceTimersByTimeAsync(3_000) + + expect(sync.mock.calls).toEqual([ + ['dispatch_1'], + ['dispatch_2'], + ['dispatch_3'], + ['dispatch_1'] + ]) + runtime.stopOrchestrationFederationRelay() + }) + + it('does not restart recovery after relay shutdown', async () => { + vi.useFakeTimers() + let release!: () => void + const blocked = new Promise((resolve) => (release = resolve)) + const runtime = new OrcaRuntimeService(null, undefined, { + orchestrationEnvironmentTransport: { + resolve: vi.fn(), + call: vi.fn() + } + }) + const sync = vi.spyOn(runtime, 'syncOrchestrationFederatedDispatch').mockReturnValue(blocked) + runtime.setOrchestrationDb({ + listActiveFederatedDispatches: () => [], + findNextTerminalFederatedDispatchPendingAcknowledgment: () => ({ + dispatchId: 'dispatch_1', + rowId: 1 + }) + } as never) + expect(sync).toHaveBeenCalledTimes(1) + + runtime.stopOrchestrationFederationRelay() + release() + await vi.advanceTimersByTimeAsync(5_000) + + expect(sync).toHaveBeenCalledTimes(1) + }) +}) diff --git a/src/main/runtime/orchestration/federation-worker-report-payload.ts b/src/main/runtime/orchestration/federation-worker-report-payload.ts new file mode 100644 index 00000000000..7ddfc30638b --- /dev/null +++ b/src/main/runtime/orchestration/federation-worker-report-payload.ts @@ -0,0 +1,37 @@ +import type { WorkerReportOutcome } from './types' +import { OrchestrationError } from './orchestration-error' + +export function parseFederatedWorkerReportPayload(payload: string | null): { + taskId: string + dispatchId: string + outcome: WorkerReportOutcome + filesModified: string[] + reportPath: string | null +} { + let parsed: unknown + try { + parsed = payload ? JSON.parse(payload) : null + } catch { + parsed = null + } + if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) { + throw new OrchestrationError('invalid_argument', 'Federated worker report is invalid.') + } + const report = parsed as Record + if ( + typeof report.taskId !== 'string' || + typeof report.dispatchId !== 'string' || + (report.outcome !== 'succeeded' && report.outcome !== 'failed') + ) { + throw new OrchestrationError('invalid_argument', 'Federated worker report is incomplete.') + } + return { + taskId: report.taskId, + dispatchId: report.dispatchId, + outcome: report.outcome, + filesModified: Array.isArray(report.filesModified) + ? report.filesModified.filter((file): file is string => typeof file === 'string') + : [], + reportPath: typeof report.reportPath === 'string' ? report.reportPath : null + } +} diff --git a/src/main/runtime/orchestration/mutation-receipt-capacity.test.ts b/src/main/runtime/orchestration/mutation-receipt-capacity.test.ts index 6907d73ba81..327caf5f922 100644 --- a/src/main/runtime/orchestration/mutation-receipt-capacity.test.ts +++ b/src/main/runtime/orchestration/mutation-receipt-capacity.test.ts @@ -103,7 +103,7 @@ describe('mutation receipt capacity schema', () => { db = new OrchestrationDb(dbPath) const sqlite = sqliteFor(db) - expect(sqlite.pragma('user_version', { simple: true })).toBe(26) + expect(sqlite.pragma('user_version', { simple: true })).toBe(27) expect(sqlite.prepare('SELECT receipt_count FROM mutation_receipt_ledger').get()).toEqual({ receipt_count: 20 }) diff --git a/src/main/runtime/orchestration/orchestration-db-retention-pagination.test.ts b/src/main/runtime/orchestration/orchestration-db-retention-pagination.test.ts index 0b9c5ad7097..4b8095da873 100644 --- a/src/main/runtime/orchestration/orchestration-db-retention-pagination.test.ts +++ b/src/main/runtime/orchestration/orchestration-db-retention-pagination.test.ts @@ -240,7 +240,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => { db = new OrchestrationDb(dbPath) const sqlite = sqliteFor(db) - expect(sqlite.pragma('user_version', { simple: true })).toBe(26) + expect(sqlite.pragma('user_version', { simple: true })).toBe(27) expect(db.getDispatchContextById(dispatch.id)).toMatchObject({ assignee_handle: 'term_worker' }) expect(db.getTask(task.id)).toMatchObject({ created_by_pane_key: null, @@ -277,7 +277,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => { db.close() db = new OrchestrationDb(dbPath) - expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(26) + expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(27) expect(db.getDispatchContextById(dispatch.id)).toBeDefined() }) @@ -309,7 +309,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => { db = new OrchestrationDb(dbPath) const sqlite = sqliteFor(db) - expect(sqlite.pragma('user_version', { simple: true })).toBe(26) + expect(sqlite.pragma('user_version', { simple: true })).toBe(27) expect(db.getTask(task.id)).toMatchObject({ created_by_pane_key: 'tab_creator:leaf_creator', created_by_process_incarnation: 'pty_creator:incarnation-a', @@ -328,7 +328,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => { db.close() db = new OrchestrationDb(dbPath) - expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(26) + expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(27) expect(db.getTask(task.id)?.created_by_process_incarnation).toBe('pty_creator:incarnation-a') }) }) diff --git a/src/main/runtime/orchestration/orchestration-schema-version-skew.ts b/src/main/runtime/orchestration/orchestration-schema-version-skew.ts index 1a28d5e4d89..8fb3db36777 100644 --- a/src/main/runtime/orchestration/orchestration-schema-version-skew.ts +++ b/src/main/runtime/orchestration/orchestration-schema-version-skew.ts @@ -25,6 +25,10 @@ const POST_V6_COLUMNS = [ ['legacy_mail_receipts', 'principal_id'] ] as const +const VERSIONED_POST_V6_COLUMNS = [ + { version: 27, table: 'federated_dispatches', column: 'to_home_acknowledged_sequence' } +] as const + const POST_V6_INDEXES = [ 'idx_messages_run_sequence', 'idx_messages_delivery_contract', @@ -82,9 +86,13 @@ function hasConsistentLegacyAdoption(db: Database.Database): boolean { return true } -function hasCompletePostV6Schema(db: Database.Database): boolean { +function hasCompletePostV6Schema(db: Database.Database, storedVersion: number): boolean { return ( POST_V6_COLUMNS.every(([table, column]) => hasOrchestrationColumn(db, table, column)) && + VERSIONED_POST_V6_COLUMNS.every( + ({ version, table, column }) => + storedVersion < version || hasOrchestrationColumn(db, table, column) + ) && POST_V6_INDEXES.every((index) => hasOrchestrationIndex(db, index)) && messagesAllowQuestions(db) && hasConsistentLegacyAdoption(db) @@ -99,7 +107,7 @@ export function resolveOrchestrationMigrationStartVersion( if (storedVersion > schemaVersion) { return storedVersion } - if (hasCompletePostV6Schema(db)) { + if (hasCompletePostV6Schema(db, storedVersion)) { return storedVersion } // Why: version-skewed pre-Run databases can claim the post-v6 range while retaining v6 tables. diff --git a/src/main/runtime/orchestration/types.ts b/src/main/runtime/orchestration/types.ts index e399b13ce49..a965891580d 100644 --- a/src/main/runtime/orchestration/types.ts +++ b/src/main/runtime/orchestration/types.ts @@ -182,6 +182,7 @@ export type FederatedDispatchRow = { remote_worktree_id: string | null remote_terminal_handle: string | null to_home_imported_sequence: number + to_home_acknowledged_sequence: number created_at: string updated_at: string } diff --git a/src/main/runtime/rpc/methods/orchestration-federation-agent-launch.test.ts b/src/main/runtime/rpc/methods/orchestration-federation-agent-launch.test.ts index 486821231e1..3d65496c7de 100644 --- a/src/main/runtime/rpc/methods/orchestration-federation-agent-launch.test.ts +++ b/src/main/runtime/rpc/methods/orchestration-federation-agent-launch.test.ts @@ -59,7 +59,7 @@ describe('federated worker agent launch', () => { dispatchId: 'ctx_remote', taskId: 'task_remote', taskSpec: 'remote cursor worker', - protocolVersion: 1, + protocolVersion: 3, worktree: 'id:repo::remote-worktree', agent: 'cursor', model: 'gpt-5.3-codex', diff --git a/src/main/runtime/rpc/methods/orchestration-federation-folder-placement.test.ts b/src/main/runtime/rpc/methods/orchestration-federation-folder-placement.test.ts index 8806c92fe47..97814997250 100644 --- a/src/main/runtime/rpc/methods/orchestration-federation-folder-placement.test.ts +++ b/src/main/runtime/rpc/methods/orchestration-federation-folder-placement.test.ts @@ -30,7 +30,7 @@ describe('orchestration federated folder placement', () => { dispatchId: 'ctx_folder', taskId: 'task_folder', taskSpec: 'work in folder', - protocolVersion: 1, + protocolVersion: 3, worktree: 'new-top-level', repo: 'folder-repo', name: 'folder-worker', diff --git a/src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts b/src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts index 894462e5397..a54eca1fd84 100644 --- a/src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts +++ b/src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts @@ -1,6 +1,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { ORCHESTRATION_CONTRACT_VERSION, + ORCHESTRATION_FEDERATION_CONTROL_MAIL_RUNTIME_CAPABILITY, ORCHESTRATION_FEDERATION_LIFECYCLE_SETTLEMENT_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version' import type { RuntimeRpcResponse } from '../../../../shared/runtime-rpc-envelope' @@ -21,6 +22,8 @@ describe('orchestration federation lifecycle settlement', () => { let workerDispatcher: RpcDispatcher let workerCapabilities: string[] let failNextAckBeforeDelivery: boolean + let ackAttempts: number + let transport: OrchestrationEnvironmentTransport beforeEach(() => { homeDb = new OrchestrationDb(':memory:') @@ -30,7 +33,8 @@ describe('orchestration federation lifecycle settlement', () => { workerDispatcher = new RpcDispatcher({ runtime: workerRuntime, methods: ORCHESTRATION_METHODS }) workerCapabilities = [...(workerRuntime.getStatus().capabilities ?? [])] failNextAckBeforeDelivery = false - const transport: OrchestrationEnvironmentTransport = { + ackAttempts = 0 + transport = { resolve: () => ({ environmentId: 'environment_windows', name: 'windows', @@ -45,6 +49,9 @@ describe('orchestration federation lifecycle settlement', () => { _meta: { runtimeId: workerRuntime.getRuntimeId() } } } + if (method === 'orchestration.federationAck') { + ackAttempts += 1 + } if (method === 'orchestration.federationAck' && failNextAckBeforeDelivery) { failNextAckBeforeDelivery = false throw new Error('connection lost before acknowledgment') @@ -128,6 +135,22 @@ describe('orchestration federation lifecycle settlement', () => { return homeDb.createTask({ spec: 'Audit Windows behavior', runId: run.id }) } + function restartHomeRuntime(): void { + homeRuntime.stopOrchestrationFederationRelay() + homeRuntime = new OrcaRuntimeService(null, undefined, { + orchestrationEnvironmentTransport: transport + }) + homeRuntime.setOrchestrationDb(homeDb) + } + + function restartWorkerRuntime(): void { + workerRuntime.stopOrchestrationFederationRelay() + workerRuntime = new OrcaRuntimeService() + workerRuntime.setOrchestrationDb(workerDb) + workerDispatcher = new RpcDispatcher({ runtime: workerRuntime, methods: ORCHESTRATION_METHODS }) + configureWorkerRuntime() + } + async function sendRemoteCompletion(taskId: string, reportedTaskId: string, sync = true) { await homeDispatcher.dispatch(startRequest(taskId)) homeRuntime.stopOrchestrationFederationRelay() @@ -162,6 +185,51 @@ describe('orchestration federation lifecycle settlement', () => { return { sent, dispatch } } + function dispatchRemoteCompletion( + taskId: string, + dispatchId: string, + requestId: string, + signal?: AbortSignal, + outcome: 'succeeded' | 'failed' = 'succeeded' + ) { + const prompt = vi.mocked(workerRuntime.sendTerminalAgentPrompt).mock.calls[0]?.[1] ?? '' + const capability = prompt.match(/--dispatch-capability (dcap_[A-Za-z0-9_-]+)/)?.[1] + return workerDispatcher.dispatch( + { + id: `rpc_${requestId}`, + authToken: 'worker-local-token', + orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION, + orchestrationRequestId: requestId, + orchestrationCapability: capability, + method: 'orchestration.send', + params: { + from: 'term_windows_worker', + subject: 'Done', + type: 'worker_done', + payload: JSON.stringify({ taskId, dispatchId, outcome }) + } + }, + { signal } + ) + } + + function lifecycleResult(response: RuntimeRpcResponse): string { + if (!response.ok) { + return `error:${response.error.code}` + } + const result = response.result as { lifecycle?: { action?: string } } + return result.lifecycle?.action ?? 'missing' + } + + function legacyWorkerCapabilities(protocolVersion: 1 | 2): string[] { + return workerCapabilities.filter( + (capability) => + capability !== ORCHESTRATION_FEDERATION_LIFECYCLE_SETTLEMENT_RUNTIME_CAPABILITY && + (protocolVersion === 2 || + capability !== ORCHESTRATION_FEDERATION_CONTROL_MAIL_RUNTIME_CAPABILITY) + ) + } + it('waits for Run-home settlement when an older CLI omits the wait hint', async () => { const task = createHomeTask() @@ -197,45 +265,232 @@ describe('orchestration federation lifecycle settlement', () => { expect(homeDb.getTask(task.id)?.status).toBe('dispatched') }) - it('rejects before queueing when the negotiated protocol lacks settlement verdicts', async () => { - workerCapabilities = workerCapabilities.filter( - (capability) => - capability !== ORCHESTRATION_FEDERATION_LIFECYCLE_SETTLEMENT_RUNTIME_CAPABILITY - ) - const task = createHomeTask() + it.each([ + [1, 'succeeded'], + [1, 'failed'], + [2, 'succeeded'], + [2, 'failed'] + ] as const)( + 'runs a protocol v%s client through %s completion on a current worker server', + async (protocolVersion, outcome) => { + workerCapabilities = legacyWorkerCapabilities(protocolVersion) + const task = createHomeTask() - await homeDispatcher.dispatch(startRequest(task.id)) - homeRuntime.stopOrchestrationFederationRelay() - const dispatch = homeDb.getDispatchContext(task.id)! - const prompt = vi.mocked(workerRuntime.sendTerminalAgentPrompt).mock.calls[0]?.[1] ?? '' - const capability = prompt.match(/--dispatch-capability (dcap_[A-Za-z0-9_-]+)/)?.[1] + const started = await homeDispatcher.dispatch(startRequest(task.id)) + homeRuntime.stopOrchestrationFederationRelay() + expect(started).toMatchObject({ ok: true }) + const dispatch = homeDb.getDispatchContext(task.id)! + expect(workerDb.getRemoteDispatchAttachment(dispatch.id)?.protocol_version).toBe( + protocolVersion + ) - await expect( - workerDispatcher.dispatch({ - id: 'rpc_unsupported_worker_done', + const completed = (await dispatchRemoteCompletion( + task.id, + dispatch.id, + `protocol_${protocolVersion}_${outcome}_completion`, + undefined, + outcome + )) as RuntimeRpcResponse + await homeRuntime.syncOrchestrationFederatedDispatch(dispatch.id) + + expect(completed).toMatchObject({ + ok: true, + result: { + lifecycle: { + action: outcome === 'succeeded' ? 'completed' : 'failed', + authority: 'worker_server_legacy' + } + } + }) + expect({ + task: homeDb.getTask(task.id)?.status, + dispatch: homeDb.getDispatchContextById(dispatch.id)?.status, + attachment: workerDb.getRemoteDispatchAttachment(dispatch.id)?.state, + pending: workerDb.listPendingFederationRelay(dispatch.id, 'to_home').length, + worktreeCreates: vi.mocked(workerRuntime.createManagedWorktree).mock.calls.length, + promptWrites: vi.mocked(workerRuntime.sendTerminalAgentPrompt).mock.calls.length + }).toEqual({ + task: outcome === 'succeeded' ? 'completed' : 'failed', + dispatch: outcome === 'succeeded' ? 'completed' : 'failed', + attachment: outcome, + pending: 0, + worktreeCreates: 1, + promptWrites: 1 + }) + } + ) + + it.each([1, 2] as const)( + 'settles a protocol v%s report from a legacy worker server', + async (protocolVersion) => { + const legacyCapabilities = legacyWorkerCapabilities(protocolVersion) + let attachProtocol: number | undefined + let legacyItems: unknown[] = [] + let acknowledgment: Record | undefined + vi.spyOn(homeRuntime, 'callOrchestrationWorkerServer').mockImplementation( + async (_environmentId, method, params) => { + if (method === 'status.get') { + return { ...workerRuntime.getStatus(), capabilities: legacyCapabilities } + } + if (method === 'orchestration.federationAttachStart') { + const input = params as { dispatchId: string; protocolVersion: number } + attachProtocol = input.protocolVersion + return { + dispatchId: input.dispatchId, + state: 'ready', + runtimeEpoch: 'legacy_runtime_epoch', + worktreeId: 'repo::legacy-worktree', + terminalHandle: 'term_legacy_worker', + effects: [], + residualResources: [] + } + } + if (method === 'orchestration.federationPull') { + return { runtimeEpoch: 'legacy_runtime_epoch', items: legacyItems } + } + if (method === 'orchestration.federationAck') { + acknowledgment = params as Record + return { + acknowledgedThrough: (params as { throughSequence: number }).throughSequence + } + } + throw new Error(`Unexpected legacy worker method ${method}`) + } + ) + const task = createHomeTask() + + await homeDispatcher.dispatch(startRequest(task.id)) + homeRuntime.stopOrchestrationFederationRelay() + const dispatch = homeDb.getDispatchContext(task.id)! + legacyItems = [ + { + dispatch_id: dispatch.id, + direction: 'to_home', + sequence: 1, + message_id: `msg_legacy_protocol_${protocolVersion}`, + kind: 'worker_done', + payload: JSON.stringify({ + from: 'term_legacy_worker', + subject: 'Done', + body: 'Completed on an older worker server', + type: 'worker_done', + priority: 'normal', + threadId: null, + payload: JSON.stringify({ + taskId: task.id, + dispatchId: dispatch.id, + outcome: 'succeeded' + }) + }) + } + ] + + await homeRuntime.syncOrchestrationFederatedDispatch(dispatch.id) + + expect({ + attachProtocol, + task: homeDb.getTask(task.id)?.status, + dispatch: homeDb.getDispatchContextById(dispatch.id)?.status, + acknowledgment + }).toEqual({ + attachProtocol: protocolVersion, + task: 'completed', + dispatch: 'completed', + acknowledgment: { dispatchId: dispatch.id, throughSequence: 1 } + }) + } + ) + + it.each([1, 2] as const)( + 'retries a lost protocol v%s completion acknowledgment after Run-home restart', + async (protocolVersion) => { + workerCapabilities = legacyWorkerCapabilities(protocolVersion) + const task = createHomeTask() + await homeDispatcher.dispatch(startRequest(task.id)) + homeRuntime.stopOrchestrationFederationRelay() + const dispatch = homeDb.getDispatchContext(task.id)! + await dispatchRemoteCompletion( + task.id, + dispatch.id, + `protocol_${protocolVersion}_lost_ack_completion` + ) + failNextAckBeforeDelivery = true + + await expect(homeRuntime.syncOrchestrationFederatedDispatch(dispatch.id)).rejects.toThrow( + 'connection lost before acknowledgment' + ) + expect({ + task: homeDb.getTask(task.id)?.status, + acknowledged: homeDb.getFederatedDispatch(dispatch.id)?.to_home_acknowledged_sequence, + pending: workerDb.listPendingFederationRelay(dispatch.id, 'to_home').length + }).toEqual({ task: 'completed', acknowledged: 0, pending: 1 }) + + restartHomeRuntime() + await vi.waitFor(() => + expect(workerDb.listPendingFederationRelay(dispatch.id, 'to_home')).toHaveLength(0) + ) + + expect({ + acknowledged: homeDb.getFederatedDispatch(dispatch.id)?.to_home_acknowledged_sequence, + attachment: workerDb.getRemoteDispatchAttachment(dispatch.id)?.state, + pending: workerDb.listPendingFederationRelay(dispatch.id, 'to_home').length, + ackAttempts + }).toEqual({ acknowledged: 1, attachment: 'succeeded', pending: 0, ackAttempts: 2 }) + } + ) + + it.each([1, 2] as const)( + 'completes a persisted protocol v%s worker after its worker server updates', + async (protocolVersion) => { + const dispatchId = `ctx_persisted_protocol_${protocolVersion}` + const taskId = `task_persisted_protocol_${protocolVersion}` + workerDb.createRemoteDispatchAttachment({ + dispatchId, + taskId, + homePeerFingerprint: 'run-home-device-token', + protocolVersion, + runtimeEpoch: workerRuntime.getRuntimeId(), + mutationReceipt: { + callerFingerprint: 'run-home-device-token', + requestId: `persisted_protocol_${protocolVersion}_attach`, + method: 'orchestration.federationAttachStart', + payloadHash: `persisted_protocol_${protocolVersion}_payload` + } + }) + const capability = workerDb.prepareRemoteAttachmentAuthority({ + dispatchId, + paneKey: 'tab_worker:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb', + processIncarnation: 'windows_runtime:pty:1', + worktreeId: 'repo::persisted-worker', + terminalHandle: 'term_windows_worker', + setupState: 'completed', + effects: [] + }) + workerDb.markRemoteAttachmentReady(dispatchId) + restartWorkerRuntime() + + const completion = (await workerDispatcher.dispatch({ + id: `rpc_persisted_protocol_${protocolVersion}_completion`, authToken: 'worker-local-token', orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION, - orchestrationRequestId: 'unsupported_worker_done_request', + orchestrationRequestId: `persisted_protocol_${protocolVersion}_completion_request`, orchestrationCapability: capability, method: 'orchestration.send', params: { from: 'term_windows_worker', - subject: 'Done', + subject: 'Done after update', type: 'worker_done', - payload: JSON.stringify({ - taskId: task.id, - dispatchId: dispatch.id, - outcome: 'succeeded' - }) + payload: JSON.stringify({ taskId, dispatchId, outcome: 'succeeded' }) } - }) - ).resolves.toMatchObject({ - ok: false, - error: { code: 'capability_unsupported' } - }) - expect(workerDb.listPendingFederationRelay(dispatch.id, 'to_home')).toEqual([]) - expect(workerDb.getRemoteDispatchAttachment(dispatch.id)?.protocol_version).toBe(2) - }) + })) as RuntimeRpcResponse + + expect({ + completion: lifecycleResult(completion), + attachment: workerDb.getRemoteDispatchAttachment(dispatchId)?.state, + pending: workerDb.listPendingFederationRelay(dispatchId, 'to_home').length + }).toEqual({ completion: 'completed', attachment: 'succeeded', pending: 1 }) + } + ) it('does not settle an attachment from a verdict for non-lifecycle mail', async () => { const task = createHomeTask() @@ -313,6 +568,94 @@ describe('orchestration federation lifecycle settlement', () => { expect(workerDb.listPendingFederationRelay(dispatch.id, 'to_home')).toHaveLength(1) }) + it('rejects conflicting terminal outcomes without acknowledging either report', async () => { + const task = createHomeTask() + await homeDispatcher.dispatch(startRequest(task.id)) + homeRuntime.stopOrchestrationFederationRelay() + const dispatch = homeDb.getDispatchContext(task.id)! + const reports = (['succeeded', 'failed'] as const).map((outcome) => + workerDb.enqueueFederationRelay({ + dispatchId: dispatch.id, + direction: 'to_home', + kind: 'worker_done', + payload: JSON.stringify({ + payload: JSON.stringify({ taskId: task.id, dispatchId: dispatch.id, outcome }) + }) + }) + ) + + await expect( + workerDispatcher.dispatch({ + id: 'rpc_conflicting_settlements', + authToken: 'run-home-device-token', + orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION, + orchestrationRequestId: 'conflicting_settlements_request', + method: 'orchestration.federationAck', + params: { + dispatchId: dispatch.id, + throughSequence: reports[1].sequence, + settlements: [ + { + sequence: reports[0].sequence, + lifecycle: { action: 'completed', authority: 'run_home' } + }, + { + sequence: reports[1].sequence, + lifecycle: { action: 'failed', authority: 'run_home' } + } + ] + } + }) + ).resolves.toMatchObject({ ok: false, error: { code: 'request_mismatch' } }) + expect(workerDb.getRemoteDispatchAttachment(dispatch.id)?.state).toBe('ready') + expect(workerDb.listPendingFederationRelay(dispatch.id, 'to_home')).toHaveLength(2) + }) + + it('acknowledges preexisting same-outcome terminal reports atomically', async () => { + const task = createHomeTask() + await homeDispatcher.dispatch(startRequest(task.id)) + homeRuntime.stopOrchestrationFederationRelay() + const dispatch = homeDb.getDispatchContext(task.id)! + const reports = ['first', 'retry'].map((body) => + workerDb.enqueueFederationRelay({ + dispatchId: dispatch.id, + direction: 'to_home', + kind: 'worker_done', + payload: JSON.stringify({ + body, + payload: JSON.stringify({ + taskId: task.id, + dispatchId: dispatch.id, + outcome: 'succeeded' + }) + }) + }) + ) + + await expect( + workerDispatcher.dispatch({ + id: 'rpc_identical_settlements', + authToken: 'run-home-device-token', + orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION, + orchestrationRequestId: 'identical_settlements_request', + method: 'orchestration.federationAck', + params: { + dispatchId: dispatch.id, + throughSequence: reports[1].sequence, + settlements: reports.flatMap((report) => { + const settlement = { + sequence: report.sequence, + lifecycle: { action: 'completed' as const, authority: 'run_home' as const } + } + return [settlement, settlement] + }) + } + }) + ).resolves.toMatchObject({ ok: true, result: { acknowledgedThrough: reports[1].sequence } }) + expect(workerDb.getRemoteDispatchAttachment(dispatch.id)?.state).toBe('succeeded') + expect(workerDb.listPendingFederationRelay(dispatch.id, 'to_home')).toHaveLength(0) + }) + it('returns operation_unknown when Run-home settlement waiting is aborted', async () => { const task = createHomeTask() await homeDispatcher.dispatch(startRequest(task.id)) @@ -368,51 +711,96 @@ describe('orchestration federation lifecycle settlement', () => { expect(addEventListener).not.toHaveBeenCalled() }) - it('replays a committed settlement after the first acknowledgment is lost', async () => { + it('keeps terminal settlement replayable until the worker durably acknowledges it', async () => { const task = createHomeTask() await homeDispatcher.dispatch(startRequest(task.id)) homeRuntime.stopOrchestrationFederationRelay() const dispatch = homeDb.getDispatchContext(task.id)! - const prompt = vi.mocked(workerRuntime.sendTerminalAgentPrompt).mock.calls[0]?.[1] ?? '' - const capability = prompt.match(/--dispatch-capability (dcap_[A-Za-z0-9_-]+)/)?.[1] - const sent = workerDispatcher.dispatch({ - id: 'rpc_replayed_worker_done', - authToken: 'worker-local-token', - orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION, - orchestrationRequestId: 'replayed_worker_done_request', - orchestrationCapability: capability, - method: 'orchestration.send', - params: { - from: 'term_windows_worker', - subject: 'Done', - type: 'worker_done', - waitForLifecycleSettlement: true, - payload: JSON.stringify({ - taskId: task.id, - dispatchId: dispatch.id, - outcome: 'succeeded' - }) - } - }) + const controller = new AbortController() + const sent = dispatchRemoteCompletion( + task.id, + dispatch.id, + 'replayed_worker_done_request', + controller.signal + ) await vi.waitFor(() => expect(workerDb.listPendingFederationRelay(dispatch.id, 'to_home')).toHaveLength(1) ) failNextAckBeforeDelivery = true - await expect(homeRuntime.syncOrchestrationFederatedDispatch(dispatch.id)).rejects.toThrow( - 'connection lost before acknowledgment' + homeRuntime.ensureOrchestrationFederationRelay() + await vi.waitFor(() => expect(ackAttempts).toBe(1)) + const acknowledgedAfterLoss = homeDb.getFederatedDispatch( + dispatch.id + )?.to_home_acknowledged_sequence + restartHomeRuntime() + await vi.waitFor(() => + expect(workerDb.getRemoteDispatchAttachment(dispatch.id)?.state).toBe('succeeded') ) - expect(homeDb.getTask(task.id)?.status).toBe('completed') - await homeRuntime.syncOrchestrationFederatedDispatch(dispatch.id) + const acknowledgedAfterRetry = homeDb.getFederatedDispatch( + dispatch.id + )?.to_home_acknowledged_sequence + restartHomeRuntime() + homeRuntime.ensureOrchestrationFederationRelay() + await homeRuntime.syncOrchestrationFederation() - await expect(sent).resolves.toMatchObject({ - ok: true, - result: { lifecycle: { action: 'completed', authority: 'run_home' } } + const observed = { + homeTask: homeDb.getTask(task.id)?.status, + workerAttachment: workerDb.getRemoteDispatchAttachment(dispatch.id)?.state, + pendingWorkerRelay: workerDb.listPendingFederationRelay(dispatch.id, 'to_home').length, + acknowledgedAfterLoss, + acknowledgedAfterRetry, + ackAttempts + } + controller.abort() + const response = (await sent) as RuntimeRpcResponse + + expect({ ...observed, completion: lifecycleResult(response) }).toEqual({ + homeTask: 'completed', + workerAttachment: 'succeeded', + pendingWorkerRelay: 0, + acknowledgedAfterLoss: 0, + acknowledgedAfterRetry: 1, + ackAttempts: 2, + completion: 'completed' }) - expect(workerDb.getRemoteDispatchAttachment(dispatch.id)).toMatchObject({ - state: 'succeeded', - stage: 'worker_report_settled', - capability_hash: null + }) + + it('acknowledges duplicate identical terminal reports idempotently', async () => { + const task = createHomeTask() + await homeDispatcher.dispatch(startRequest(task.id)) + homeRuntime.stopOrchestrationFederationRelay() + const dispatch = homeDb.getDispatchContext(task.id)! + const controllers = [new AbortController(), new AbortController()] + const enqueue = vi.spyOn(workerDb, 'enqueueFederationRelay') + const sent = controllers.map((controller, index) => + dispatchRemoteCompletion( + task.id, + dispatch.id, + `duplicate_worker_done_${index + 1}`, + controller.signal + ) + ) + await vi.waitFor(() => expect(enqueue).toHaveBeenCalledTimes(2)) + await homeRuntime.syncOrchestrationFederation() + + const observed = { + homeTask: homeDb.getTask(task.id)?.status, + workerAttachment: workerDb.getRemoteDispatchAttachment(dispatch.id)?.state, + pendingWorkerRelay: workerDb.listPendingFederationRelay(dispatch.id, 'to_home').length, + ackAttempts + } + for (const controller of controllers) { + controller.abort() + } + const responses = (await Promise.all(sent)) as RuntimeRpcResponse[] + + expect({ ...observed, completions: responses.map(lifecycleResult) }).toEqual({ + homeTask: 'completed', + workerAttachment: 'succeeded', + pendingWorkerRelay: 0, + ackAttempts: 1, + completions: ['completed', 'completed'] }) }) diff --git a/src/main/runtime/rpc/methods/orchestration-federation-relay.ts b/src/main/runtime/rpc/methods/orchestration-federation-relay.ts index 848b711b851..235286fcbc8 100644 --- a/src/main/runtime/rpc/methods/orchestration-federation-relay.ts +++ b/src/main/runtime/rpc/methods/orchestration-federation-relay.ts @@ -3,6 +3,7 @@ import { ORCHESTRATION_FEDERATION_CONTROL_MAIL_PROTOCOL_VERSION } from '../../.. import { importFederatedControlMessage } from '../../orchestration/federation-control-message' import { OrchestrationError } from '../../orchestration/orchestration-error' import { + areFederatedLifecycleSettlementsEqual, publishFederatedLifecycleSettlement, type FederatedLifecycleSettlement } from '../../orchestration/federation-lifecycle-settlement' @@ -81,32 +82,52 @@ export const ORCHESTRATION_FEDERATION_RELAY_METHODS: RpcMethod[] = [ params: FederationAckParams, handler: (params, { runtime, authenticatedCallerFingerprint }) => { requireHomeAttachment(runtime, params.dispatchId, authenticatedCallerFingerprint) - const settlements = (params.settlements ?? []).filter( + const receivedSettlements = (params.settlements ?? []).filter( (settlement) => settlement.sequence <= params.throughSequence ) + const settlementsBySequence = new Map() + for (const settlement of receivedSettlements) { + const existing = settlementsBySequence.get(settlement.sequence) + if ( + existing && + !areFederatedLifecycleSettlementsEqual(existing.lifecycle, settlement.lifecycle) + ) { + throw new OrchestrationError( + 'request_mismatch', + `Federation acknowledgment for ${params.dispatchId} contains conflicting settlements.` + ) + } + settlementsBySequence.set(settlement.sequence, existing ?? settlement) + } + const settlements = [...settlementsBySequence.values()] const terminalSettlements = settlements.filter( (settlement) => settlement.lifecycle.action === 'completed' || settlement.lifecycle.action === 'failed' ) - if (terminalSettlements.length > 1) { + const terminalOutcomes = new Set( + terminalSettlements.map((settlement) => settlement.lifecycle.action) + ) + if (terminalOutcomes.size > 1) { throw new OrchestrationError( 'request_mismatch', `Federation acknowledgment for ${params.dispatchId} contains conflicting settlements.` ) } - const terminalSettlement = terminalSettlements[0] runtime.getOrchestrationDb().acknowledgeFederationRelay({ dispatchId: params.dispatchId, direction: 'to_home', throughSequence: params.throughSequence, - ...(!terminalSettlement + ...(settlements.length === 0 ? {} : { - settleRemoteReport: { - sequence: terminalSettlement.sequence, - outcome: - terminalSettlement.lifecycle.action === 'completed' ? 'succeeded' : 'failed' - } + settleRemoteReports: settlements.map((settlement) => ({ + sequence: settlement.sequence, + ...(settlement.lifecycle.action === 'completed' + ? { outcome: 'succeeded' as const } + : settlement.lifecycle.action === 'failed' + ? { outcome: 'failed' as const } + : {}) + })) }) }) for (const settlement of settlements) { diff --git a/src/main/runtime/rpc/methods/orchestration-federation.test.ts b/src/main/runtime/rpc/methods/orchestration-federation.test.ts index ca850587ff0..18db8750ff0 100644 --- a/src/main/runtime/rpc/methods/orchestration-federation.test.ts +++ b/src/main/runtime/rpc/methods/orchestration-federation.test.ts @@ -296,35 +296,25 @@ describe('orchestration federation', () => { expect(workerRuntime.sendTerminalAgentPrompt).not.toHaveBeenCalled() }) - it('rejects control mail before queueing when the worker lacks that capability', async () => { + it('starts a legacy federation worker through its negotiated protocol', async () => { workerCapabilities = workerCapabilities.filter( (capability) => capability !== ORCHESTRATION_FEDERATION_CONTROL_MAIL_RUNTIME_CAPABILITY ) const task = createHomeTask() const started = await homeDispatcher.dispatch(startRequest(task.id)) - expect(started).toMatchObject({ ok: true, result: { state: 'ready' } }) + expect(started).toMatchObject({ + ok: true, + result: { state: 'ready' } + }) const dispatch = homeDb.getDispatchContext(task.id)! - const sent = await homeDispatcher.dispatch({ - id: 'send-control-to-old-worker', - authToken: 'coordinator-token', - orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION, - orchestrationRequestId: 'send-control-to-old-worker-request', - method: 'orchestration.send', - params: { - from: 'term_coord', - to: `dispatch:${dispatch.id}`, - subject: 'Continue', - body: 'This worker cannot receive control mail yet.', - type: 'status' - } - }) - - expect(sent).toMatchObject({ - ok: false, - error: { code: 'capability_unsupported' } + expect(workerDb.getRemoteDispatchAttachment(dispatch.id)).toMatchObject({ + state: 'ready', + protocol_version: 1 }) expect(homeDb.listPendingFederationRelay(dispatch.id, 'to_worker')).toHaveLength(0) + expect(workerRuntime.createManagedWorktree).toHaveBeenCalledOnce() + expect(workerRuntime.sendTerminalAgentPrompt).toHaveBeenCalledOnce() }) it('durably relays remote completion into the home Run and acknowledges it', async () => { diff --git a/src/main/runtime/rpc/methods/orchestration.ts b/src/main/runtime/rpc/methods/orchestration.ts index f3dee4d2171..77dd8279ac9 100644 --- a/src/main/runtime/rpc/methods/orchestration.ts +++ b/src/main/runtime/rpc/methods/orchestration.ts @@ -451,16 +451,9 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ 'Remote worker_done requires outcome=succeeded|failed.' ) } - if ( - outcome && - remoteAttachment.protocol_version < - ORCHESTRATION_FEDERATION_LIFECYCLE_SETTLEMENT_PROTOCOL_VERSION - ) { - throw new OrchestrationError( - 'capability_unsupported', - 'The Run-home runtime cannot confirm worker_done settlement. Update it before retrying; no report was queued.' - ) - } + const supportsLifecycleSettlement = + remoteAttachment.protocol_version >= + ORCHESTRATION_FEDERATION_LIFECYCLE_SETTLEMENT_PROTOCOL_VERSION const relay = db.enqueueFederationRelay({ dispatchId: remoteAttachment.dispatch_id, direction: 'to_home', @@ -473,15 +466,27 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ priority: params.priority ?? 'normal', threadId: params.threadId ?? null, payload: params.payload ?? null - }) + }), + ...(!supportsLifecycleSettlement && outcome ? { settleRemoteOutcome: outcome } : {}) }) - const lifecycle = outcome - ? await waitForFederatedLifecycleSettlement(runtime, relay.dispatch_id, relay.sequence, { - timeoutMs: 30_000, - signal - }) - : undefined - if (outcome && !lifecycle) { + const lifecycle = + outcome && supportsLifecycleSettlement + ? await waitForFederatedLifecycleSettlement( + runtime, + relay.dispatch_id, + relay.sequence, + { + timeoutMs: 30_000, + signal + } + ) + : outcome + ? { + action: outcome === 'succeeded' ? ('completed' as const) : ('failed' as const), + authority: 'worker_server_legacy' as const + } + : undefined + if (outcome && supportsLifecycleSettlement && !lifecycle) { throw new OrchestrationError( 'operation_unknown', 'worker_done was queued, but the Run-home runtime did not confirm settlement. Verify the Task and Dispatch before retrying.'