From 7c255605350c00cc37ef2f605f94f25a7baffbb2 Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Thu, 13 Aug 2026 15:57:44 -0400 Subject: [PATCH] fix(orchestration): bound terminal federation recovery --- config/reliability-gates.jsonc | 22 +- src/main/index.ts | 19 +- src/main/runtime/orca-runtime.ts | 44 ++- src/main/runtime/orchestration/db.ts | 108 +++++- ...ederation-acknowledgment-migration.test.ts | 50 ++- .../federation-terminal-recovery-policy.ts | 30 ++ .../federation-terminal-recovery.test.ts | 350 +++++++++++++++--- .../mutation-receipt-capacity.test.ts | 2 +- ...hestration-db-retention-pagination.test.ts | 8 +- .../orchestration-schema-version-skew.ts | 6 +- src/main/runtime/orchestration/types.ts | 4 + src/main/runtime/rpc/errors.ts | 1 + src/shared/runtime-environment-store.ts | 9 +- 13 files changed, 573 insertions(+), 80 deletions(-) create mode 100644 src/main/runtime/orchestration/federation-terminal-recovery-policy.ts diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index a72b093f4b6..a3021aa7ba8 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -8401,16 +8401,17 @@ "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. 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.", + "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 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 fake-clock SQLite recovery oracle classifies proven permanent failures, preserves unknown outcomes, verifies capped durable backoff across restart, gives later rows a turn, skips duplicate ACKs, and fences disposal. Migration units upgrade v26 acknowledgments and v27 recovery state 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/pull/14105", "https://linear.app/stably/issue/STA-4020", "https://linear.app/stably/issue/STA-4021", - "https://linear.app/stably/issue/STA-4022" + "https://linear.app/stably/issue/STA-4022", + "https://linear.app/stably/issue/STA-4174" ], - "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.", + "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. Historical recovery must durably terminalize only proven permanent failures, keep transient and unknown outcomes retryable with capped persisted backoff, survive restart, and never let one row starve later eligible work. 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", @@ -8418,6 +8419,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-terminal-recovery.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", @@ -8441,6 +8443,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-terminal-recovery.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", @@ -8551,12 +8554,21 @@ "a rejected report replays its exact verdict without mutating durable rejection mail twice" ] }, + { + "file": "src/main/runtime/orchestration/federation-terminal-recovery.test.ts", + "assertions": [ + "proven permanent failures become durable terminal rows and are not retried after restart", + "transient and unknown failures use capped persisted backoff and converge when the worker returns", + "a permanently failing oldest row does not block a later recoverable row", + "duplicate acknowledgments and relay disposal schedule no further work" + ] + }, { "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" + "a v27 database gains zeroed terminal acknowledgment recovery state", + "version-aware migration preserves imported and acknowledged cursors" ] }, { diff --git a/src/main/index.ts b/src/main/index.ts index b4e0e1b5ae4..24a37e18085 100644 --- a/src/main/index.ts +++ b/src/main/index.ts @@ -77,7 +77,10 @@ import { type OrchestrationEnvironmentTransport } from './runtime/orchestration/environment-transport' import { callRuntimeEnvironment } from './ipc/runtime-environment-transport-routing' -import { resolveEnvironment } from '../shared/runtime-environment-store' +import { + resolveEnvironment, + RuntimeEnvironmentNotFoundError +} from '../shared/runtime-environment-store' import { getPreferredPairingOffer } from '../shared/runtime-environments' import { OrcaRuntimeRpcServer } from './runtime/runtime-rpc' import { @@ -88,6 +91,7 @@ import { resolveAdvertisedPairingEndpoint } from './runtime/pairing-endpoint' import { ServeReadinessPublisher } from './server/serve-readiness' import { reserveServeStdoutForReadiness } from './server/serve-stdout-boundary' import { DesktopRelayService } from './runtime/relay/desktop-relay-service' +import { OrchestrationError } from './runtime/orchestration/orchestration-error' import type { RelayBrokerStatus } from './runtime/relay/relay-session-broker' import { awaitRuntimeFileWatcherUnsubscribes } from './runtime/orca-runtime-files' import { clearRuntimeMetadataIfOwned } from './runtime/runtime-metadata' @@ -2488,7 +2492,18 @@ void app.whenReady().then(async () => { }) const orchestrationEnvironmentTransport: OrchestrationEnvironmentTransport = { resolve: (selector) => { - const environment = resolveEnvironment(app.getPath('userData'), selector) + let environment: ReturnType + try { + environment = resolveEnvironment(app.getPath('userData'), selector) + } catch (error) { + if (error instanceof RuntimeEnvironmentNotFoundError) { + throw new OrchestrationError( + 'environment_not_found', + `Saved worker environment ${selector} no longer exists.` + ) + } + throw error + } const pairing = getPreferredPairingOffer(environment) return { environmentId: environment.id, diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index daf9af49386..4d26d303018 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -164,6 +164,10 @@ import type { } from './orchestration/environment-transport' import { clearFederationAckCheckpoints } from './orchestration/federation-ack-checkpoints' import { syncFederatedDispatch } from './orchestration/federation-sync' +import { + classifyFederationTerminalRecoveryFailure, + FEDERATION_TERMINAL_RECOVERY_BASE_DELAY_MS +} from './orchestration/federation-terminal-recovery-policy' import { formatMessagePointer } from './orchestration/formatter' import { selectExactWorkerProviderSession } from './orchestration/worker-provider-session' import type { @@ -4783,25 +4787,57 @@ export class OrcaRuntimeService { private async recoverNextTerminalHistoryAcknowledgment(generation: number): Promise { const db = this.getOrchestrationDb() + const nowMs = Date.now() let historical = db.findNextTerminalFederatedDispatchPendingAcknowledgment( - this.orchestrationTerminalRecoveryRowId + this.orchestrationTerminalRecoveryRowId, + nowMs ) if (!historical && this.orchestrationTerminalRecoveryRowId > 0) { this.orchestrationTerminalRecoveryRowId = 0 - historical = db.findNextTerminalFederatedDispatchPendingAcknowledgment(0) + historical = db.findNextTerminalFederatedDispatchPendingAcknowledgment(0, nowMs) } if (!historical) { + const nextAt = db.getNextTerminalFederatedDispatchRecoveryAt() + if (nextAt !== undefined) { + this.scheduleTerminalHistoryRecovery(generation, Math.max(0, nextAt - nowMs)) + } return } this.orchestrationTerminalRecoveryRowId = historical.rowId - await this.syncOrchestrationFederatedDispatch(historical.dispatchId).catch(() => undefined) + const result = await this.syncOrchestrationFederatedDispatch(historical.dispatchId).then( + () => ({ ok: true as const }), + (error: unknown) => ({ ok: false as const, error }) + ) + if (generation !== this.orchestrationFederationRelayGeneration) { + return + } + if (result.ok) { + db.recordFederatedTerminalRecoverySuccess(historical.dispatchId, Date.now()) + } else { + db.recordFederatedTerminalRecoveryFailure({ + dispatchId: historical.dispatchId, + ...classifyFederationTerminalRecoveryFailure(result.error), + nowMs: Date.now() + }) + } + const nextAt = db.getNextTerminalFederatedDispatchRecoveryAt() + if (nextAt === undefined) { + return + } + this.scheduleTerminalHistoryRecovery( + generation, + Math.max(FEDERATION_TERMINAL_RECOVERY_BASE_DELAY_MS, nextAt - Date.now()) + ) + } + + private scheduleTerminalHistoryRecovery(generation: number, delayMs: number): void { if (generation !== this.orchestrationFederationRelayGeneration) { return } this.orchestrationTerminalHistoryRecoveryTimer = setTimeout(() => { this.orchestrationTerminalHistoryRecoveryTimer = null this.ensureTerminalHistoryRecovery() - }, 1_000) + }, delayMs) this.orchestrationTerminalHistoryRecoveryTimer.unref?.() } diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index 38b0b194c78..ed2ce9f9360 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -65,6 +65,7 @@ import { ensureMutationReceiptCapacity, migrateMutationReceiptCapacity } from './mutation-receipt-capacity' +import { getFederationTerminalRecoveryDelayMs } from './federation-terminal-recovery-policy' function parseWorkerTerminalPriorOwnerIds(value: string): string[] | null { try { @@ -287,8 +288,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, v27 durable federation acknowledgments. -const SCHEMA_VERSION = 27 +// 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, v28 durable terminal acknowledgment recovery. +const SCHEMA_VERSION = 28 function hardenOrchestrationDatabaseFiles(dbPath: (string & {}) | ':memory:'): void { if (dbPath === ':memory:' || process.platform === 'win32') { @@ -471,6 +472,11 @@ export class OrchestrationDb { remote_terminal_handle TEXT, to_home_imported_sequence INTEGER NOT NULL DEFAULT 0, to_home_acknowledged_sequence INTEGER NOT NULL DEFAULT 0, + terminal_ack_recovery_state TEXT NOT NULL DEFAULT 'pending' + CHECK(terminal_ack_recovery_state IN ('pending', 'retryable', 'terminal')), + terminal_ack_recovery_attempts INTEGER NOT NULL DEFAULT 0, + terminal_ack_recovery_next_at_ms INTEGER NOT NULL DEFAULT 0, + terminal_ack_recovery_error_code TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (datetime('now')) ); @@ -875,6 +881,11 @@ export class OrchestrationDb { remote_terminal_handle TEXT, to_home_imported_sequence INTEGER NOT NULL DEFAULT 0, to_home_acknowledged_sequence INTEGER NOT NULL DEFAULT 0, + terminal_ack_recovery_state TEXT NOT NULL DEFAULT 'pending' + CHECK(terminal_ack_recovery_state IN ('pending', 'retryable', 'terminal')), + terminal_ack_recovery_attempts INTEGER NOT NULL DEFAULT 0, + terminal_ack_recovery_next_at_ms INTEGER NOT NULL DEFAULT 0, + terminal_ack_recovery_error_code TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (datetime('now')) ); @@ -1002,6 +1013,28 @@ export class OrchestrationDb { 'ALTER TABLE federated_dispatches ADD COLUMN to_home_acknowledged_sequence INTEGER NOT NULL DEFAULT 0' ) } + if (current < 28) { + if (!this.hasColumn('federated_dispatches', 'terminal_ack_recovery_state')) { + this.db.exec( + "ALTER TABLE federated_dispatches ADD COLUMN terminal_ack_recovery_state TEXT NOT NULL DEFAULT 'pending' CHECK(terminal_ack_recovery_state IN ('pending', 'retryable', 'terminal'))" + ) + } + if (!this.hasColumn('federated_dispatches', 'terminal_ack_recovery_attempts')) { + this.db.exec( + 'ALTER TABLE federated_dispatches ADD COLUMN terminal_ack_recovery_attempts INTEGER NOT NULL DEFAULT 0' + ) + } + if (!this.hasColumn('federated_dispatches', 'terminal_ack_recovery_next_at_ms')) { + this.db.exec( + 'ALTER TABLE federated_dispatches ADD COLUMN terminal_ack_recovery_next_at_ms INTEGER NOT NULL DEFAULT 0' + ) + } + if (!this.hasColumn('federated_dispatches', 'terminal_ack_recovery_error_code')) { + this.db.exec( + 'ALTER TABLE federated_dispatches ADD COLUMN terminal_ack_recovery_error_code TEXT' + ) + } + } this.db.exec(` CREATE INDEX IF NOT EXISTS idx_dispatch_assignee_pane_leaf ON dispatch_contexts(${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL}) @@ -4564,7 +4597,8 @@ export class OrchestrationDb { } findNextTerminalFederatedDispatchPendingAcknowledgment( - afterRowId: number + afterRowId: number, + nowMs = Date.now() ): { dispatchId: string; rowId: number } | undefined { return this.db .prepare( @@ -4573,11 +4607,72 @@ export class OrchestrationDb { 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.terminal_ack_recovery_state != 'terminal' + AND fd.terminal_ack_recovery_next_at_ms <= ? AND fd.rowid > ? ORDER BY fd.rowid LIMIT 1` ) - .get(afterRowId) as { dispatchId: string; rowId: number } | undefined + .get(nowMs, afterRowId) as { dispatchId: string; rowId: number } | undefined + } + + getNextTerminalFederatedDispatchRecoveryAt(): number | undefined { + const row = this.db + .prepare( + `SELECT MIN(fd.terminal_ack_recovery_next_at_ms) AS nextAt + 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.terminal_ack_recovery_state != 'terminal'` + ) + .get() as { nextAt: number | null } + return row.nextAt ?? undefined + } + + recordFederatedTerminalRecoveryFailure(params: { + dispatchId: string + errorCode: string | null + terminal: boolean + nowMs: number + }): void { + const current = this.getFederatedDispatch(params.dispatchId) + if (!current || current.to_home_acknowledged_sequence >= current.to_home_imported_sequence) { + return + } + const attempts = current.terminal_ack_recovery_attempts + 1 + const nextAt = params.terminal + ? 0 + : params.nowMs + getFederationTerminalRecoveryDelayMs(attempts) + this.db + .prepare( + `UPDATE federated_dispatches + SET terminal_ack_recovery_state = ?, terminal_ack_recovery_attempts = ?, + terminal_ack_recovery_next_at_ms = ?, terminal_ack_recovery_error_code = ?, + updated_at = datetime('now') + WHERE dispatch_id = ? + AND to_home_acknowledged_sequence < to_home_imported_sequence` + ) + .run( + params.terminal ? 'terminal' : 'retryable', + attempts, + nextAt, + params.errorCode, + params.dispatchId + ) + } + + recordFederatedTerminalRecoverySuccess(dispatchId: string, nowMs: number): void { + this.db + .prepare( + `UPDATE federated_dispatches + SET terminal_ack_recovery_state = 'pending', terminal_ack_recovery_attempts = 0, + terminal_ack_recovery_next_at_ms = CASE + WHEN to_home_acknowledged_sequence < to_home_imported_sequence THEN ? ELSE 0 END, + terminal_ack_recovery_error_code = NULL, updated_at = datetime('now') + WHERE dispatch_id = ?` + ) + .run(nowMs + getFederationTerminalRecoveryDelayMs(1), dispatchId) } isFederatedDispatchRelayEligible(dispatchId: string): boolean { @@ -4592,6 +4687,7 @@ export class OrchestrationDb { wd.state IN ('starting', 'ready', 'stopping', 'start_unknown', 'stop_unknown') OR ( fd.to_home_acknowledged_sequence < fd.to_home_imported_sequence + AND fd.terminal_ack_recovery_state != 'terminal' ) )` ) @@ -5277,6 +5373,10 @@ export class OrchestrationDb { THEN MAX(to_home_acknowledged_sequence, ?) ELSE ? END, + terminal_ack_recovery_state = 'pending', + terminal_ack_recovery_attempts = 0, + terminal_ack_recovery_next_at_ms = 0, + terminal_ack_recovery_error_code = NULL, updated_at = datetime('now') WHERE dispatch_id = ?` ) diff --git a/src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts b/src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts index 80959d3df48..79b6c0cedc1 100644 --- a/src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts +++ b/src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts @@ -27,7 +27,7 @@ describe('federation acknowledgment migration', () => { 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) + expect(resolveOrchestrationMigrationStartVersion(oldDb, 26, 28)).toBe(26) oldDb .prepare( `INSERT INTO federated_dispatches ( @@ -41,10 +41,54 @@ describe('federation acknowledgment migration', () => { db = new OrchestrationDb(dbPath) const sqlite = (db as unknown as { db: Database.Database }).db - expect(sqlite.pragma('user_version', { simple: true })).toBe(27) + expect(sqlite.pragma('user_version', { simple: true })).toBe(28) expect(db.getFederatedDispatch('ctx_migrated')).toMatchObject({ to_home_imported_sequence: 2, - to_home_acknowledged_sequence: 0 + to_home_acknowledged_sequence: 0, + terminal_ack_recovery_state: 'pending', + terminal_ack_recovery_attempts: 0, + terminal_ack_recovery_next_at_ms: 0, + terminal_ack_recovery_error_code: null + }) + }) + + it('adds durable terminal recovery policy state to v27 dispatches', () => { + tempDir = mkdtempSync(join(tmpdir(), 'orca-federation-recovery-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 terminal_ack_recovery_error_code; + ALTER TABLE federated_dispatches DROP COLUMN terminal_ack_recovery_next_at_ms; + ALTER TABLE federated_dispatches DROP COLUMN terminal_ack_recovery_attempts; + ALTER TABLE federated_dispatches DROP COLUMN terminal_ack_recovery_state; + `) + oldDb.pragma('user_version = 27') + expect(resolveOrchestrationMigrationStartVersion(oldDb, 27, 28)).toBe(27) + oldDb + .prepare( + `INSERT INTO federated_dispatches ( + dispatch_id, environment_id, environment_name, peer_fingerprint, + protocol_version, to_home_imported_sequence, to_home_acknowledged_sequence + ) VALUES ('ctx_retry', 'env', 'worker', 'peer', 3, 2, 1)` + ) + .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(28) + expect(db.getFederatedDispatch('ctx_retry')).toMatchObject({ + to_home_imported_sequence: 2, + to_home_acknowledged_sequence: 1, + terminal_ack_recovery_state: 'pending', + terminal_ack_recovery_attempts: 0, + terminal_ack_recovery_next_at_ms: 0, + terminal_ack_recovery_error_code: null }) }) }) diff --git a/src/main/runtime/orchestration/federation-terminal-recovery-policy.ts b/src/main/runtime/orchestration/federation-terminal-recovery-policy.ts new file mode 100644 index 00000000000..32ed47cbf43 --- /dev/null +++ b/src/main/runtime/orchestration/federation-terminal-recovery-policy.ts @@ -0,0 +1,30 @@ +import { OrchestrationError } from './orchestration-error' + +export const FEDERATION_TERMINAL_RECOVERY_BASE_DELAY_MS = 1_000 +export const FEDERATION_TERMINAL_RECOVERY_MAX_DELAY_MS = 30_000 + +const TERMINAL_ERROR_CODES = new Set([ + 'dispatch_not_found', + 'environment_not_found', + 'peer_changed' +]) + +export type FederationTerminalRecoveryFailure = { + errorCode: string | null + terminal: boolean +} + +export function classifyFederationTerminalRecoveryFailure( + error: unknown +): FederationTerminalRecoveryFailure { + const errorCode = error instanceof OrchestrationError ? error.code : null + return { errorCode, terminal: errorCode !== null && TERMINAL_ERROR_CODES.has(errorCode) } +} + +export function getFederationTerminalRecoveryDelayMs(attempts: number): number { + const exponent = Math.max(0, Math.min(attempts - 1, 30)) + return Math.min( + FEDERATION_TERMINAL_RECOVERY_BASE_DELAY_MS * 2 ** exponent, + FEDERATION_TERMINAL_RECOVERY_MAX_DELAY_MS + ) +} diff --git a/src/main/runtime/orchestration/federation-terminal-recovery.test.ts b/src/main/runtime/orchestration/federation-terminal-recovery.test.ts index f35e6f8e645..3cff91d53cd 100644 --- a/src/main/runtime/orchestration/federation-terminal-recovery.test.ts +++ b/src/main/runtime/orchestration/federation-terminal-recovery.test.ts @@ -1,35 +1,308 @@ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' import { afterEach, describe, expect, it, vi } from 'vitest' import { OrcaRuntimeService } from '../orca-runtime' +import { OrchestrationDb } from './db' +import { OrchestrationError } from './orchestration-error' +import { classifyFederationTerminalRecoveryFailure } from './federation-terminal-recovery-policy' + +type HistoricalDispatch = { dispatchId: string; rowId: number } describe('terminal federation acknowledgment recovery', () => { + const databases: OrchestrationDb[] = [] + const runtimes: OrcaRuntimeService[] = [] + const tempDirs: string[] = [] + afterEach(() => { + for (const runtime of runtimes.splice(0)) { + runtime.stopOrchestrationFederationRelay() + } + for (const db of databases.splice(0)) { + db.close() + } + for (const dir of tempDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }) + } vi.useRealTimers() }) - it('bounds migration replay to one oldest-first recovery', async () => { - vi.useFakeTimers() + function createRuntime( + db: OrchestrationDb, + sync?: (dispatchId: string) => Promise + ): { runtime: OrcaRuntimeService; sync: ReturnType } { + const runtime = new OrcaRuntimeService(null, undefined, { + orchestrationEnvironmentTransport: { resolve: vi.fn(), call: vi.fn() } + }) + runtimes.push(runtime) + const syncSpy = vi + .spyOn(runtime, 'syncOrchestrationFederatedDispatch') + .mockImplementation(sync ?? (async () => {})) + runtime.setOrchestrationDb(db) + return { runtime, sync: syncSpy } + } + + function createDb(onDisk = false): { db: OrchestrationDb; dbPath?: string } { + if (!onDisk) { + const db = new OrchestrationDb(':memory:') + databases.push(db) + return { db } + } + const dir = mkdtempSync(join(tmpdir(), 'orca-federation-recovery-')) + const dbPath = join(dir, 'orchestration.db') + tempDirs.push(dir) + const db = new OrchestrationDb(dbPath) + databases.push(db) + return { db, dbPath } + } + + function createHistoricalDispatch(db: OrchestrationDb, suffix: string): string { + const run = db.createRun({ + objective: `recover ${suffix}`, + coordinatorHandle: `term_coord_${suffix}`, + coordinatorPaneKey: `tab_coord_${suffix}:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa` + }) + const task = db.createTask({ spec: `settle ${suffix}`, runId: run.id }) + const started = db.createStartingWorkerDispatch({ + taskId: task.id, + startOptions: {}, + runtimeEpoch: 'home_epoch', + federation: { + environmentId: `environment_${suffix}`, + environmentName: suffix, + peerFingerprint: `peer_${suffix}`, + protocolVersion: 3 + } + }) + db.recordWorkerStage({ + dispatchId: started.dispatch.id, + stage: 'completed', + state: 'succeeded' + }) + db.setFederatedHomeImportSequence(started.dispatch.id, 1) + return started.dispatch.id + } + + function terminalRecoveryState(db: OrchestrationDb, dispatchId: string) { + return db.getFederatedDispatch(dispatchId) as unknown as { + to_home_imported_sequence: number + to_home_acknowledged_sequence: number + terminal_ack_recovery_state?: string + terminal_ack_recovery_attempts?: number + terminal_ack_recovery_next_at_ms?: number + terminal_ack_recovery_error_code?: string | null + } + } + + it.each(['dispatch_not_found', 'environment_not_found', 'peer_changed'])( + 'classifies proven permanent error %s as terminal', + (code) => { + expect(classifyFederationTerminalRecoveryFailure(new OrchestrationError(code, code))).toEqual( + { errorCode: code, terminal: true } + ) + } + ) + + it.each(['capability_unsupported', 'operation_unknown', 'runtime_timeout'])( + 'keeps non-terminal error %s retryable', + (code) => { + expect(classifyFederationTerminalRecoveryFailure(new OrchestrationError(code, code))).toEqual( + { errorCode: code, terminal: false } + ) + } + ) + + it('durably terminalizes a permanent failure and does not retry it after restart', async () => { + vi.useFakeTimers({ now: 0 }) + const { db, dbPath } = createDb(true) + const dispatchId = createHistoricalDispatch(db, 'retired') + const attempts: number[] = [] + const { runtime } = createRuntime(db, async () => { + attempts.push(Date.now()) + throw new OrchestrationError('environment_not_found', 'The saved worker was decommissioned.') + }) + + runtime.ensureOrchestrationFederationRelay() + await vi.advanceTimersByTimeAsync(60_000) + + expect(attempts).toEqual([0]) + expect(db.findNextTerminalFederatedDispatchPendingAcknowledgment(0)).toBeUndefined() + expect(terminalRecoveryState(db, dispatchId)).toMatchObject({ + to_home_imported_sequence: 1, + to_home_acknowledged_sequence: 0, + terminal_ack_recovery_state: 'terminal', + terminal_ack_recovery_attempts: 1, + terminal_ack_recovery_error_code: 'environment_not_found' + }) + + runtime.stopOrchestrationFederationRelay() + db.close() + databases.splice(databases.indexOf(db), 1) + const restartedDb = new OrchestrationDb(dbPath!) + databases.push(restartedDb) + const { sync: restartedSync } = createRuntime(restartedDb) + + await vi.advanceTimersByTimeAsync(60_000) + + expect(restartedSync).not.toHaveBeenCalled() + expect(terminalRecoveryState(restartedDb, dispatchId).terminal_ack_recovery_state).toBe( + 'terminal' + ) + }) + + it('backs off transient failures durably and converges when the worker returns', async () => { + vi.useFakeTimers({ now: 0 }) + const { db } = createDb() + const dispatchId = createHistoricalDispatch(db, 'healing') + const attempts: number[] = [] + const { runtime } = createRuntime(db, async () => { + attempts.push(Date.now()) + if (attempts.length < 3) { + throw new Error('worker unavailable') + } + db.recordFederatedHomeAcknowledgment({ + dispatchId, + remoteRuntimeEpoch: 'worker_epoch', + sequence: 1 + }) + }) + + runtime.ensureOrchestrationFederationRelay() + await vi.advanceTimersByTimeAsync(3_000) + + expect(attempts).toEqual([0, 1_000, 3_000]) + expect(db.getFederatedDispatch(dispatchId)).toMatchObject({ + to_home_acknowledged_sequence: 1 + }) + expect(db.findNextTerminalFederatedDispatchPendingAcknowledgment(0)).toBeUndefined() + }) + + it('persists a transient retry deadline across app restart', async () => { + vi.useFakeTimers({ now: 0 }) + const { db, dbPath } = createDb(true) + createHistoricalDispatch(db, 'restart') + const { runtime, sync: firstSync } = createRuntime(db, async () => { + throw new Error('transport unavailable') + }) + + runtime.ensureOrchestrationFederationRelay() + await vi.advanceTimersByTimeAsync(500) + expect(firstSync).toHaveBeenCalledTimes(1) + runtime.stopOrchestrationFederationRelay() + db.close() + databases.splice(databases.indexOf(db), 1) + + const restartedDb = new OrchestrationDb(dbPath!) + databases.push(restartedDb) + const attempts: number[] = [] + createRuntime(restartedDb, async () => { + attempts.push(Date.now()) + throw new Error('transport unavailable') + }) + + await vi.advanceTimersByTimeAsync(499) + expect(attempts).toEqual([]) + await vi.advanceTimersByTimeAsync(1) + expect(attempts).toEqual([1_000]) + }) + + it('keeps unknown failures retryable with capped backoff', async () => { + vi.useFakeTimers({ now: 0 }) + const { db } = createDb() + const dispatchId = createHistoricalDispatch(db, 'ambiguous') + const attempts: number[] = [] + const { runtime } = createRuntime(db, async () => { + attempts.push(Date.now()) + throw new OrchestrationError('operation_unknown', 'The ACK outcome is unknown.') + }) + + runtime.ensureOrchestrationFederationRelay() + await vi.advanceTimersByTimeAsync(61_000) + + expect(attempts).toEqual([0, 1_000, 3_000, 7_000, 15_000, 31_000, 61_000]) + expect(db.getNextTerminalFederatedDispatchRecoveryAt()).toBe(91_000) + expect(db.findNextTerminalFederatedDispatchPendingAcknowledgment(0, 91_000)).toMatchObject({ + dispatchId + }) + expect(terminalRecoveryState(db, dispatchId)).toMatchObject({ + terminal_ack_recovery_state: 'retryable', + terminal_ack_recovery_attempts: 7, + terminal_ack_recovery_next_at_ms: 91_000 + }) + }) + + it('terminalizes a permanent oldest row without blocking a later recoverable row', async () => { + vi.useFakeTimers({ now: 0 }) + const { db } = createDb() + const permanentId = createHistoricalDispatch(db, 'oldest') + const recoverableId = createHistoricalDispatch(db, 'later') + const calls: { dispatchId: string; at: number }[] = [] + const { runtime } = createRuntime(db, async (dispatchId) => { + calls.push({ dispatchId, at: Date.now() }) + if (dispatchId === permanentId) { + throw new OrchestrationError('dispatch_not_found', 'The old worker discarded the dispatch.') + } + db.recordFederatedHomeAcknowledgment({ + dispatchId, + remoteRuntimeEpoch: 'worker_epoch', + sequence: 1 + }) + }) + + runtime.ensureOrchestrationFederationRelay() + await vi.advanceTimersByTimeAsync(1_000) + + expect(calls).toEqual([ + { dispatchId: permanentId, at: 0 }, + { dispatchId: recoverableId, at: 1_000 } + ]) + expect(db.findNextTerminalFederatedDispatchPendingAcknowledgment(0)).toBeUndefined() + }) + + it('skips an already acknowledged duplicate without scheduling work', async () => { + vi.useFakeTimers({ now: 0 }) + const { db } = createDb() + const dispatchId = createHistoricalDispatch(db, 'duplicate') + db.recordFederatedHomeAcknowledgment({ + dispatchId, + remoteRuntimeEpoch: 'worker_epoch', + sequence: 1 + }) + const { runtime, sync } = createRuntime(db) + + runtime.ensureOrchestrationFederationRelay() + await vi.advanceTimersByTimeAsync(60_000) + + expect(sync).not.toHaveBeenCalled() + }) + + it('keeps a thousand-row backlog to one timer and one in-flight sync', async () => { + vi.useFakeTimers({ now: 0 }) 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() - } + orchestrationEnvironmentTransport: { resolve: vi.fn(), call: vi.fn() } }) + runtimes.push(runtime) const sync = vi .spyOn(runtime, 'syncOrchestrationFederatedDispatch') .mockResolvedValue(undefined) runtime.setOrchestrationDb({ listActiveFederatedDispatches: () => [], findNextTerminalFederatedDispatchPendingAcknowledgment: (afterRowId: number) => - candidates.find((candidate) => candidate.rowId > afterRowId) + candidates.find((candidate) => candidate.rowId > afterRowId), + recordFederatedTerminalRecoverySuccess: (dispatchId: string) => { + candidates.splice( + candidates.findIndex((candidate) => candidate.dispatchId === dispatchId), + 1 + ) + }, + getNextTerminalFederatedDispatchRecoveryAt: () => + candidates.length > 0 ? Date.now() : undefined } 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([ @@ -38,65 +311,32 @@ describe('terminal federation acknowledgment recovery', () => { ['dispatch_3'], ['dispatch_4'] ]) - runtime.stopOrchestrationFederationRelay() + expect(vi.getTimerCount()).toBe(1) }) - it('gives every unavailable terminal dispatch a turn before retrying', async () => { - vi.useFakeTimers() - const candidates = [1, 2, 3].map((rowId) => ({ - dispatchId: `dispatch_${rowId}`, - rowId - })) + it('does not restart recovery after relay shutdown', async () => { + vi.useFakeTimers({ now: 0 }) + let release!: () => void + const blocked = new Promise((resolve) => (release = resolve)) + const candidates: HistoricalDispatch[] = [{ dispatchId: 'dispatch_1', rowId: 1 }] const runtime = new OrcaRuntimeService(null, undefined, { - orchestrationEnvironmentTransport: { - resolve: vi.fn(), - call: vi.fn() - } + orchestrationEnvironmentTransport: { resolve: vi.fn(), call: vi.fn() } }) - const sync = vi - .spyOn(runtime, 'syncOrchestrationFederatedDispatch') - .mockRejectedValue(new Error('worker unavailable')) + runtimes.push(runtime) + const sync = vi.spyOn(runtime, 'syncOrchestrationFederatedDispatch').mockReturnValue(blocked) 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) + expect(vi.getTimerCount()).toBe(0) runtime.stopOrchestrationFederationRelay() release() - await vi.advanceTimersByTimeAsync(5_000) + await vi.advanceTimersByTimeAsync(60_000) expect(sync).toHaveBeenCalledTimes(1) + expect(vi.getTimerCount()).toBe(0) }) }) diff --git a/src/main/runtime/orchestration/mutation-receipt-capacity.test.ts b/src/main/runtime/orchestration/mutation-receipt-capacity.test.ts index 327caf5f922..547117c1ed8 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(27) + expect(sqlite.pragma('user_version', { simple: true })).toBe(28) 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 4b8095da873..9a3afab1d81 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(27) + expect(sqlite.pragma('user_version', { simple: true })).toBe(28) 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(27) + expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(28) 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(27) + expect(sqlite.pragma('user_version', { simple: true })).toBe(28) 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(27) + expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(28) 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 8fb3db36777..a7eacf0162a 100644 --- a/src/main/runtime/orchestration/orchestration-schema-version-skew.ts +++ b/src/main/runtime/orchestration/orchestration-schema-version-skew.ts @@ -26,7 +26,11 @@ const POST_V6_COLUMNS = [ ] as const const VERSIONED_POST_V6_COLUMNS = [ - { version: 27, table: 'federated_dispatches', column: 'to_home_acknowledged_sequence' } + { version: 27, table: 'federated_dispatches', column: 'to_home_acknowledged_sequence' }, + { version: 28, table: 'federated_dispatches', column: 'terminal_ack_recovery_state' }, + { version: 28, table: 'federated_dispatches', column: 'terminal_ack_recovery_attempts' }, + { version: 28, table: 'federated_dispatches', column: 'terminal_ack_recovery_next_at_ms' }, + { version: 28, table: 'federated_dispatches', column: 'terminal_ack_recovery_error_code' } ] as const const POST_V6_INDEXES = [ diff --git a/src/main/runtime/orchestration/types.ts b/src/main/runtime/orchestration/types.ts index a965891580d..a058c238ad8 100644 --- a/src/main/runtime/orchestration/types.ts +++ b/src/main/runtime/orchestration/types.ts @@ -183,6 +183,10 @@ export type FederatedDispatchRow = { remote_terminal_handle: string | null to_home_imported_sequence: number to_home_acknowledged_sequence: number + terminal_ack_recovery_state: 'pending' | 'retryable' | 'terminal' + terminal_ack_recovery_attempts: number + terminal_ack_recovery_next_at_ms: number + terminal_ack_recovery_error_code: string | null created_at: string updated_at: string } diff --git a/src/main/runtime/rpc/errors.ts b/src/main/runtime/rpc/errors.ts index db6396b8e0d..b92b156a6a7 100644 --- a/src/main/runtime/rpc/errors.ts +++ b/src/main/runtime/rpc/errors.ts @@ -77,6 +77,7 @@ const STRUCTURED_RUNTIME_PASSTHROUGH_CODES: ReadonlySet = new Set([ 'source_changed', 'transcript_required', 'server_required', + 'environment_not_found', 'worktree_not_found_on_server', 'resource_server_mismatch', 'peer_changed', diff --git a/src/shared/runtime-environment-store.ts b/src/shared/runtime-environment-store.ts index e784dc80281..a55aa1ba985 100644 --- a/src/shared/runtime-environment-store.ts +++ b/src/shared/runtime-environment-store.ts @@ -32,6 +32,13 @@ export class RuntimeEnvironmentStoreError extends Error { } } +export class RuntimeEnvironmentNotFoundError extends RuntimeEnvironmentStoreError { + constructor(selector: string) { + super('invalid_argument', `Unknown environment: ${selector}`) + this.name = 'RuntimeEnvironmentNotFoundError' + } +} + export function getEnvironmentStorePath(userDataPath: string): string { return join(userDataPath, ENVIRONMENTS_FILE) } @@ -223,7 +230,7 @@ function resolveEnvironmentFromStore( `Environment name "${selector}" is ambiguous; use the environment id.` ) } - throw new RuntimeEnvironmentStoreError('invalid_argument', `Unknown environment: ${selector}`) + throw new RuntimeEnvironmentNotFoundError(selector) } function readEnvironmentStore(userDataPath: string): RuntimeEnvironmentStore {