diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index 899914339c76..c2bbf9628389 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -18698,6 +18698,131 @@ "Default config paths are shared with GNOME on a disposable hosted CI runner; nested mode refuses non-GitHub-Actions execution." ], "demotionRule": "Keep experimental on unexplained failures; retain exact bytes and participation checks without retries, skips, or longer deadlines." + }, + { + "id": "ssh-liveness.relay-owned-exit-certification", + "title": "Only the owning relay certifies that one of its PTYs exited", + "maturity": "experimental", + "protection": "partial", + "owner": "terminal-runtime", + "layer": "relay-handler-and-provider-service-composition", + "surfaces": [ + "SSH legacy worker terminal recovery", + "relay PTY lifecycle", + "orchestration Task reuse after a worker exits" + ], + "platforms": ["macos", "linux", "windows"], + "providers": ["ssh"], + "coveredPlatforms": ["macos"], + "coveredProviders": ["ssh"], + "coverageNotes": "Deterministic service-level composition of the real PtyHandler, the real SshPtyProvider, the real provider registry and runtime-controller readback, the real recovery candidate, and an in-memory orchestration DB. node-pty, the process-group kill and notification delivery are injected; the platform is pinned to linux inside the handler harness, so the run is macOS-hosted but platform-neutral. No live SSH server, real profile, kernel D-state process or app UI is exercised.", + "motivatingLinks": [ + "https://github.com/stablyai/orca/pull/19347", + "https://github.com/stablyai/orca/pull/16904" + ], + "invariant": "A PTY id is certified exited only by the relay that owned the record and watched the process end: node-pty's onExit, or a pid probe that answered ESRCH. Every other disappearance of a record - a shutdown that stopped waiting for an uninterruptible child, the torn-down-record sweep, an evicted or forgotten observation, an id this relay never held, a revive in flight, a record mid-teardown, and a host too old to answer - is unverifiable, and recovery defers instead of retiring the worker.", + "oracle": "The relay probe answers live/exited/unverifiable for each of those states directly, and binds the answer to one incarnation: admission under a reused id clears the prior observation, and only the incarnation that currently owns the id may record one, including a callback that arrives after teardown removed the record. The provider maps only live and exited, and maps a JSON-RPC -32601 rejection to null with no capability handshake. At the recovery seam, an observed exit converges to an exited resolution that releases the Task, while the shutdown-timeout case (with and without a sibling kill failure) and the old-relay case produce no resolution and a deferred dispatch, with a real orchestration DB refusing Task reuse.", + "commands": [ + "ORCA_BACKGROUND_LAUNCH=1 node_modules/.bin/vitest run --config config/vitest.config.ts src/relay/pty-handler-liveness-probe.test.ts src/relay/pty-handler-ssh-exit-certification.test.ts src/relay/pty-handler-ssh-recovery-convergence.test.ts src/main/providers/ssh-pty-liveness-probe.test.ts src/main/providers/ssh-pty-provider-liveness-readback.test.ts src/main/runtime/runtime-legacy-worker-terminal-exit-certification.test.ts" + ], + "testFiles": [ + "src/relay/pty-handler-liveness-probe.test.ts", + "src/relay/pty-handler-ssh-exit-certification.test.ts", + "src/relay/pty-handler-ssh-recovery-convergence.test.ts", + "src/main/providers/ssh-pty-liveness-probe.test.ts", + "src/main/providers/ssh-pty-provider-liveness-readback.test.ts", + "src/main/runtime/runtime-legacy-worker-terminal-exit-certification.test.ts" + ], + "assertionRefs": [ + { + "file": "src/relay/pty-handler-liveness-probe.test.ts", + "assertions": [ + "separates a real exit from a teardown removal in the same shutdown", + "certifies an exit node-pty reported after the shutdown wait had given up", + "stays unverifiable after the listing sweeps a torn-down record away", + "stays unverifiable while a revive holds the id, and answers for the new record after", + "does not certify a revived process with the exit of the one it replaced", + "does not let a superseded incarnation certify the id once both records are gone", + "evicts observations in the order it made them, not the order it was asked", + "forgets every observation when the relay restarts, even under the same mint epoch", + "certifies a long-lived resident whose owner entry was evicted", + "withholds a late observation once both the record and its owner entry are gone" + ] + }, + { + "file": "src/relay/pty-handler-ssh-recovery-convergence.test.ts", + "assertions": [ + "recovers a worker after the connected relay observed its physical exit while the client was absent", + "recovers a worker whose exit the relay observed after its shutdown wait gave up", + "does not retire a revived worker with the exit of the process it replaced", + "does not certify an unexited shutdown record; sibling kill failure=%s" + ] + }, + { + "file": "src/relay/pty-handler-ssh-exit-certification.test.ts", + "assertions": [ + "does not certify exit after shutdown bookkeeping removes an unexited PTY", + "keeps missing shutdown records unverifiable when another failed kill leaves the relay serving", + "publishes nothing that names the generation which minted its PTY ids", + "leaves a shutdown-removed id unverifiable for a client that infers from listings" + ] + }, + { + "file": "src/main/providers/ssh-pty-liveness-probe.test.ts", + "assertions": ["maps the owner verdict %s", "fails closed on %s"] + }, + { + "file": "src/main/providers/ssh-pty-provider-liveness-readback.test.ts", + "assertions": ["exposes the readback the liveness rule asks the owning provider for"] + }, + { + "file": "src/main/runtime/runtime-legacy-worker-terminal-exit-certification.test.ts", + "assertions": [ + "certifies an SSH exit the owning relay observed", + "defers an SSH candidate the owning relay will not certify", + "defers when the host is too old to know the readback, without a capability handshake" + ] + } + ], + "evidenceRuns": [ + { + "date": "2026-09-07", + "runner": "local", + "platform": "macos", + "command": "ORCA_BACKGROUND_LAUNCH=1 node_modules/.bin/vitest run --config config/vitest.config.ts src/relay/pty-handler-liveness-probe.test.ts src/relay/pty-handler-ssh-exit-certification.test.ts src/relay/pty-handler-ssh-recovery-convergence.test.ts src/main/providers/ssh-pty-liveness-probe.test.ts src/main/providers/ssh-pty-provider-liveness-readback.test.ts src/main/runtime/runtime-legacy-worker-terminal-exit-certification.test.ts", + "result": "passed", + "durationSeconds": 1.27, + "summary": "6 files, 58 tests passed. The convergence file fails 1/3 without the owner readback; thirteen mutations of the probe, its incarnation binding, the published capabilities, the bounded map and the forwarder each fail at least one of these tests." + } + ], + "runtimeBudget": { + "p95Seconds": 30, + "scope": "Six unit/service-level files; the ledger-eviction and owner-eviction cases each drive 4,096+ spawn-and-exit cycles through the real handler and reads the oldest entry before the overflow and dominates the run at roughly 0.8 s." + }, + "flakeHistory": { + "status": "not-started", + "evidence": "New with PR #19347. Deterministic: fake timers, injected node-pty, no real SSH, no network. No soak history yet." + }, + "redGreenEvidence": { + "status": "complete", + "evidence": "Red: the archived round-3 reproduction failed 1/3 at head 0be1813781 with an empty resolution list, and the round-2 reproduction failed 2/4 against the removed client-side epoch inference. The round-4 review reproduced three more at head 03af416c38 - a revived id certified from its predecessor's exit, an exit observed after disposal lost forever, and an old client's inference restored by a published mint epoch. Green at this head. Twelve independent mutations (record in the shared reap, dropped disposed check, dropped pending-revive check, absent-record-as-exited, provider unknown-as-exited, provider rejection-as-exited, read-reordered ledger eviction, dropped admission invalidation, dropped incarnation guard, incarnation guard weakened to an empty pool, recording moved back below the disposed guard, republished mint epoch) each fail a checked-in test." + }, + "performanceBudget": { + "required": false, + "evidence": "One exact-id RPC per candidate, replacing a full-inventory scan plus a second capability read; no polling or caching added." + }, + "promotionCriteria": [ + "Run a live SSH host journey where a supervised worker exits while the client is disconnected and the Task is reused after reconnection.", + "Cover Windows and Linux relay hosts, and an old-relay host binary that genuinely lacks pty.probeLiveness.", + "Collect 100 soak runs across 14 days with no unexplained flakes." + ], + "knownGaps": [ + "No live SSH server, real relay process restart, or kernel-uninterruptible child; the shutdown-timeout and restart cases are simulated at the handler seam.", + "The observed-exit ledger is in-memory and per relay process; a relay that restarts between the observation and the question answers unverifiable forever for that id, and only manual stop/abandon releases the worker.", + "A disposed record left in the pool is not reachable from any current teardown, so that guard, and the twice-superseded-id ordering built on top of it, are exercised by constructing the state rather than by a production path.", + "Local and daemon providers keep their existing readback; this gate makes no claim about them." + ], + "demotionRule": "Any failure that shows an exit certified without an observation is a P0 revert of the probe, not a test relaxation; unverifiable answers that merely defer stay experimental while the live SSH journey is missing." } ] } diff --git a/fix-reports/19347-round2.md b/fix-reports/19347-round2.md new file mode 100644 index 000000000000..1ed4d53441ca --- /dev/null +++ b/fix-reports/19347-round2.md @@ -0,0 +1,92 @@ +# PR #19347 round 2 — SSH exit certification moved to its owner + +## The defect + +`probeSshPtyLiveness` inferred "absent from `pty.listProcesses` and minted by this relay's +generation, therefore exited". The implication is false. The relay has removal paths that never +observe the process end: `disposePtyForRelayShutdown` waits `IMMEDIATE_PTY_EXIT_TIMEOUT_MS` for a +physical exit, catches the timeout, and deletes the record anyway; `disposePtys` settles with +`Promise.allSettled`, so another PTY's failed kill rejects the aggregate while that record is +already gone; `relay-grace-lifecycle` then defers shutdown and the relay keeps serving under the +same mint epoch. A newly connected provider asked, was told `false`, and recovery queued a live +worker as exited. + +Reproduced at `1e89522a9d` with the reviewer's test: 2 failed / 2 passed. + +## Census of record removals in `PtyHandler` + +Every removal funnels through `removePty`, which has three callers plus the shared reap. + +| Site | Path | Observed the process end? | +| --- | --- | --- | +| node-pty `onExit` handler | the process reported its own exit code | Yes | +| `reapPtyProvenExited` | pid probed, answered ESRCH from the owning host | Yes | +| `reapExitedPty(_, 'record-torn-down')` from `listProcesses` | tidies a record already disposed elsewhere | No | +| `disposePtyForRelayShutdown` | SIGKILL sent, exit wait timed out, record deleted regardless | No | + +The two "yes" rows now write the id into `observedPtyExitIds`. Nothing else does, and the write +sits at the line that made the observation rather than inside the shared reap, so the reap's +bookkeeping caller cannot inherit it. + +## What the relay answers + +`pty.probeLiveness` takes an exact id and answers from the handler's own state. + +- A live record: `live`, or `exited` if probing its pid reaps it. +- A record mid-teardown, or an id with a revive in flight: `unknown`. +- Any other id: `exited` only if it is in the ledger, otherwise `unknown`. + +Fail-closed by construction. A future removal path that forgets to record itself yields `unknown`, +which defers; it cannot yield a false death certificate. Eviction from the bounded ledger has the +same safe direction. + +Two gates were considered and dropped as dead code, each confirmed by a surviving mutation that +changed no observable behaviour. A mint-epoch comparison is subsumed because ledger keys are whole +ids that already embed the generation. A blanket "a shutdown has been fenced" refusal masked the +ledger in the only path where teardown runs, made that path's correctness untestable, and withheld +exits the relay really did observe. + +## Boundary + +**Invariant.** Only the execution host can distinguish a process that ended from bookkeeping that +stopped tracking it. A mint epoch proves who allocated an id; it says nothing about why that +owner's record disappeared. + +**Owner.** `PtyHandler`. It holds the records, the physical-exit path, the pid probe, the shutdown +fence, and pending revives. + +**Restores or compensates.** Restores. The previous version compensated at the client, rebuilding a +verdict from two projections that do not carry the missing evidence. The client now asks and +forwards; `SshPtyProvider.probePtyLiveness` maps three statuses to `true` / `false` / `null` and +does nothing else. + +**Alternative rejected.** Keeping the inference and excluding the shutdown case client-side. It +needs the client to know which removals were bookkeeping, which is precisely the knowledge only the +owner has, and the next removal path would silently reopen the hole. + +**Compatibility.** A relay too old to know the method answers JSON-RPC `-32601`, which arrives as a +rejection and maps to `null`, so no capability negotiation is needed and no discovery round trip is +spent. Old provider against a new relay is unaffected: the handler is additive. The `ptyIdMintEpoch` +capability and the shared id module stay because the relay still mints through them. + +## Mutation results + +Eight mutations against the relay verdict and the two suites that drive it. + +| Mutation | Result | +| --- | --- | +| Certify any id absent from the map | 9 failed | +| Record the shutdown teardown as an observed exit | 4 failed | +| Drop the node-pty `onExit` record | 2 failed | +| Drop the pid-probe record | 1 failed | +| Answer live instead of reaping a dead pid | 1 failed | +| Record the torn-down sweep as an observed exit | survives | +| Treat a disposed record as live | survives | +| Drop the pending-revive gate | survives | + +The three survivors guard states no current path constructs, all in the safe direction. A disposed +record is never left in the map: all three `disposeManagedPty` callers remove it in the same +synchronous block, so neither the torn-down sweep nor the disposed branch can be driven. An id in +the ledger cannot be revived, because revive requires `process.kill(pid, 0)` to succeed and a +recorded id's process is gone. The revive gate is kept anyway because that coupling lives in +another function. diff --git a/fix-reports/19347-round5.md b/fix-reports/19347-round5.md new file mode 100644 index 000000000000..3e898902bb3f --- /dev/null +++ b/fix-reports/19347-round5.md @@ -0,0 +1,186 @@ +# PR #19347 round 5 — exit evidence bound to the incarnation, mint epoch unpublished + +Round 4 found the recovery path and the removal census correct, and three narrow owner-evidence +bugs on top of them. All three are in `src/relay/pty-handler.ts`. F1 and F2 are one root cause. + +## The root cause: the ledger keyed evidence by id, and ids are reusable + +`observedPtyExitIds` answers "did this relay watch the process behind id X end". An id is not a +process. `pty.revive` re-creates a process under an id this relay may already have watched end, so a +single id names a sequence of incarnations, and an entry keyed by the id alone does not say which +one it describes. + +Two transitions were unguarded: + +- **Admission (F1, P0).** `wireAndStore` stored a new record without touching the ledger. While the + new record was in `this.ptys` the probe answered from it, so the stale entry was invisible. The + moment a shutdown that stopped waiting removed that record — the exact case this whole PR exists + for — the old entry became authoritative again and certified a process that never exited. +- **Observation (F2, P1).** The node-pty `onExit` callback returned at `if (managed.disposed)` + before recording. Disposal is bookkeeping: teardown gave up waiting and deleted the record. The + process ending afterwards is still this relay watching this process end, and dropping it left the + id unverifiable forever — the round-3 convergence bug in a narrower ordering that needs no + restart, eviction or revive. + +### The fix + +Admission is the single site where an id changes hands, so it is where ownership is established: + +```ts +// wireAndStore — the one this.ptys.set site +this.observedPtyExitIds.delete(managed.id) +this.ptyIdIncarnationOwners.set(managed.id, managed.incarnationId) +this.ptys.set(managed.id, managed) +``` + +Observation happens before the disposed guard, and only for the incarnation that still speaks for +the id: + +```ts +managed.pty.onExit(({ exitCode }) => { + managed.physicalExit?.markExited() + if (this.stillOwnsPtyId(managed)) { + this.recordObservedPtyExit(managed.id) + } + if (managed.disposed) { + return + } + ... +``` + +`stillOwnsPtyId` is `this.ptys.get(id) === managed || ptyIdIncarnationOwners.peek(id) === incarnationId`. +The redundant write further down the callback was removed; there is now exactly one write per +observation route (`onExit`, and `reapPtyProvenExited` for the ESRCH probe). + +### Why the incarnation map, and not the reviewer's two-clause guard + +The counterfactual in `run-counterfactuals.py` records when `!this.ptys.has(id) || this.ptys.get(id) === managed`. +That closes both reproduced cases and the reviewer said so explicitly — and also said it is not a +complete patch for every repeated-revive/late-callback ordering. It is not, and the gap is the same +class as F1: an empty pool is not evidence that this callback still speaks for the id. Sequence: + +1. incarnation A is admitted under `pty-7`; teardown removes it without observing an exit; +2. `pty.revive` admits incarnation B under `pty-7`; teardown removes B the same way; +3. A's `onExit` finally fires. `this.ptys.has('pty-7')` is false, so the empty-pool clause records — + and the id now belongs to B, whose process is still running. That is a false death certificate + with the admission invalidation in place. + +`this.ptys` can only answer the ownership question while the record is in it, and the observation +this ledger exists for is precisely the one that arrives after it left. So ownership needs to +outlive the record: `ptyIdIncarnationOwners` is a bounded map from id to the incarnation admitted +most recently under it, written at the same single admission site. `peek`, not `get`, so asking does +not renew an owner ahead of older ones. + +Fail-closed in both directions of forgetting. An unrecognised incarnation records nothing, so the id +stays unverifiable; and the owners map is bounded by the same `OBSERVED_PTY_EXIT_HISTORY` entry cap +as the ledger, so eviction can only withhold an exit, never invent one. The live-record disjunct +(`this.ptys.get(id) === managed`) keeps eviction from touching the normal path at all: a PTY that +outlives 4,096 admissions still certifies its own exit, because its record is in the pool when the +callback fires. + +Checked in as `does not certify a revived process with the exit of the one it replaced` (F1), +`certifies an exit node-pty reported after the shutdown wait had given up` (F2) and +`does not let a superseded incarnation certify the id once both records are gone` (the ordering +above), plus the reviewer's revived-id, foreign-epoch and failed-revive cases, in +`pty-handler-liveness-probe.test.ts`; and the two service-level compositions +(`recovers a worker whose exit the relay observed after its shutdown wait gave up`, +`does not retire a revived worker with the exit of the process it replaced`) in +`pty-handler-ssh-recovery-convergence.test.ts`. + +## F3 (P1) — the mint epoch is no longer published + +`pty.getCapabilities` no longer returns `ptyIdMintEpoch`, and the comment inviting list/epoch +inference is gone. Base `9fed61e5c246` published no such field, so this restores base behaviour +exactly; the epoch was added by this PR and read by nobody outside the relay. + +Mixed client and host versions are the normal state, and a host cannot fix an old client — it can +only decline to hand it the second half of an unsound inference. The round-two client's rule was +"id absent from `pty.listProcesses`, and its mint epoch equals the host's published epoch, therefore +exited". Removing the publication makes the second clause unsatisfiable, so that client answers +`null` and defers, which is its own fail-closed path. + +The oracle is checked in without shipping the old client: `pty-handler-ssh-exit-certification.test.ts` +reproduces the inference in ~10 lines against the real handler and asserts it reaches no verdict for +a shutdown-removed id, with and without a sibling kill failure, plus +`publishes nothing that names the generation which minted its PTY ids`, which compares every +published capability value against the epoch parsed from a real id rather than against one field +name. The reviewer's `run-mixed-version.py`, which does load the round-two client source, still runs +and passes unchanged. + +`toRelayPtyIdWithMintEpoch` and `parseRelayPtyMintEpoch` stay — ids are still minted with an epoch +so generations cannot collide — and `src/shared/relay-pty-mint-epoch.ts`'s comments no longer equate +same-epoch omission with an exit. Note the epoch is not secret: `authorityGeneration` in +`pty.inspectProcess` foreground evidence has carried it since before this PR and still does. The +change is that the relay no longer advertises a *contract* whose only purpose was list-absence +inference. + +## F4 (P2) — FIFO eviction is now pinned by a read before the overflow + +The eviction test filled 4,097 entries before its first read, so it never observed a read altering +the next eviction and the `BoundedMap.has()` → `get()` mutation survived the whole gate. +`evicts observations in the order it made them, not the order it was asked` fills exactly +`OBSERVED_PTY_EXIT_HISTORY`, probes the oldest entry, records one more exit and requires the oldest +to be `unverifiable`. The mutant keeps it and fails. + +## P3 — `unknown` → `unverifiable` + +`pty.probeLiveness` now answers `live | exited | unverifiable`, the execution-boundary vocabulary +(`docs/reference/ssh-execution-boundary.md`), which has no synonym for doubt. No released client +consumes this RPC, and the provider maps anything that is not `live` or `exited` to `null` anyway, +so the rename is safe in both directions across versions. Provider, forwarder and runtime +certification tests were renamed with it. + +## Mutations + +Twelve isolated mutations, each run against the six-file reliability command (56 tests). Every one +fails at least one checked-in test; the round-4 survivor is killed. + +| Mutation | Result | First test that fails | +| --- | --- | --- | +| Record the observation in the shared reap | 2/56 fail | separates a real exit from a teardown removal in the same shutdown | +| Drop the disposed guard in `probeLiveness` | 1/56 fails | stays unverifiable while a record is mid-teardown | +| Drop the pending-revive guard | 1/56 fails | stays unverifiable while a revive holds the id | +| Absent record → `exited` | 22/56 fail | stays unverifiable after a shutdown removes a record it never saw exit | +| Provider: unverifiable → `false` | 12/56 fail | maps the owner verdict unverifiable | +| Provider: rejection → `false` | 4/56 fail | fails closed on a relay too old to know the method | +| `BoundedMap.has()` reorders via `get()` | 1/56 fails | evicts observations in the order it made them, not the order it was asked | +| **Drop the admission invalidation** | 2/56 fail | does not certify a revived process with the exit of the one it replaced | +| **Drop the incarnation guard in `onExit`** | 1/56 fails | does not let a superseded incarnation certify the id once both records are gone | +| **Weaken that guard to the empty-pool form** | 1/56 fails | does not let a superseded incarnation certify the id once both records are gone | +| **Move the recording back below the disposed guard** | 2/56 fail | certifies an exit node-pty reported after the shutdown wait had given up | +| **Republish `ptyIdMintEpoch`** | 3/56 fail | publishes nothing that names the generation which minted its PTY ids | + +## Reviewer scripts + +All four pass at this head (run from copies whose evidence directory points at scratch, so the +round-4 archive is not overwritten): + +| Script | Result | Adaptation | +| --- | --- | --- | +| `run-owner-probes.py` | 29 passed, exit 0 | its appended assertions spell doubt `'unknown'`; replaced with `'unverifiable'` for the P3 rename | +| `run-recovery-edge-probes.py` | 7 passed, exit 0 | none | +| `run-mixed-version.py` | 5 passed, exit 0 | none; it still loads the round-two client from `1e89522a9d` | +| `run-counterfactuals.py` | 4 runs, exit 0 each | none needed, but its source patches are now no-ops or identity at this head, so its first two runs measure the shipped code rather than a counterfactual | +| `run-fifo-mutation-proof.py` | superseded by the `has-reorders` row above, which runs against the checked-in suite | | + +## Gates + +| Gate | Result | +| --- | --- | +| `src/relay src/main/providers src/main/runtime src/main/daemon src/main/ipc/pty` | 1,214 files passed, 8 skipped; 12,489 tests passed, 46 skipped | +| Six-file reliability command | 56 passed (was 47) | +| `tsc --noEmit -p config/tsconfig.node.json` / `config/tsconfig.tc.cli.json` | both pass | +| `node config/scripts/check-changed-code-quality.mjs` | 0 findings across 24 changed files (native, type-aware, React Doctor) | +| `node config/scripts/check-reliability-gates.mjs` | 110 gates pass | +| `node config/scripts/check-max-lines-ratchet.mjs` | 12 grandfathered suppressions, no new bypasses | + +## Known gaps + +- The twice-superseded-id ordering is built by constructing a disposed record and letting the + listing sweep it, the same synthetic state the existing suite documents; no current teardown + leaves a disposed record in the pool. +- `ptyIdIncarnationOwners` adds a second bounded map. Both caps are entry counts, not time or + retention guarantees: past 4,096 admissions an older id is forgotten and answers unverifiable, + which defers recovery rather than retiring a worker. +- The epoch removal is proven against the round-two client implementation, not against a public + release; no released client read that capability. diff --git a/src/main/providers/ssh-pty-liveness-probe.test.ts b/src/main/providers/ssh-pty-liveness-probe.test.ts new file mode 100644 index 000000000000..52a630a3176a --- /dev/null +++ b/src/main/providers/ssh-pty-liveness-probe.test.ts @@ -0,0 +1,50 @@ +import { describe, expect, it, vi } from 'vitest' +import { probeSshPtyLiveness } from './ssh-pty-liveness-probe' + +const PTY_ID = 'pty2:0f8f3a1e-1111-4111-8111-111111111111:7' + +function relay(answer: Error | Record | null) { + return vi.fn(async (method: string, params?: Record) => { + expect(method).toBe('pty.probeLiveness') + expect(params).toEqual({ id: PTY_ID }) + if (answer instanceof Error) { + throw answer + } + return answer + }) +} + +describe('SSH PTY liveness forwarder', () => { + it.each([ + ['live', true], + ['exited', false], + ['unverifiable', null] + ])('maps the owner verdict %s', async (status, expected) => { + const request = relay({ status }) + + await expect(probeSshPtyLiveness({ request, relayPtyId: PTY_ID })).resolves.toBe(expected) + // One exact-id question, so the client neither scans an inventory nor interprets one. + expect(request).toHaveBeenCalledTimes(1) + }) + + it.each([ + ['a relay too old to know the method', new Error('Method not found: pty.probeLiveness')], + ['a disposed multiplexer', new Error('Multiplexer disposed')], + ['a lost connection', new Error('SSH connection lost, reconnecting...')] + ])('fails closed on %s', async (_label, failure) => { + const request = relay(failure) + + await expect(probeSshPtyLiveness({ request, relayPtyId: PTY_ID })).resolves.toBeNull() + }) + + it.each([ + ['no answer at all', null], + ['an answer with no status', {}], + ['a status this client does not know', { status: 'probably_gone' }], + ['a non-string status', { status: true }] + ])('refuses to read %s as a verdict', async (_label, answer) => { + const request = relay(answer) + + await expect(probeSshPtyLiveness({ request, relayPtyId: PTY_ID })).resolves.toBeNull() + }) +}) diff --git a/src/main/providers/ssh-pty-liveness-probe.ts b/src/main/providers/ssh-pty-liveness-probe.ts new file mode 100644 index 000000000000..ff3c3700967d --- /dev/null +++ b/src/main/providers/ssh-pty-liveness-probe.ts @@ -0,0 +1,41 @@ +type RelayRequest = ( + method: string, + params?: Record, + options?: { timeoutMs?: number } +) => Promise + +const PROBE_TIMEOUT_MS = 10_000 + +/** + * Forwards the liveness question to the relay, which is the only party that can answer it. + * + * The client deliberately computes nothing here. A previous version inferred an exit from an id's + * absence in `pty.listProcesses` plus a matching mint epoch, and that is unsound: the relay also + * removes a record without observing the process end, so a shutdown that gave up waiting for an + * uninterruptible child produced a false death certificate (docs/reference/ssh-execution-boundary.md). + * + * Never throws, and fails closed. A relay too old to know the method answers JSON-RPC -32601, which + * arrives here as a rejection and maps to null, exactly like a timeout or a disposed multiplexer. + * Any status other than the two the owner certifies is unverifiable. + */ +export async function probeSshPtyLiveness(args: { + request: RelayRequest + relayPtyId: string +}): Promise { + try { + const answer = (await args.request( + 'pty.probeLiveness', + { id: args.relayPtyId }, + { timeoutMs: PROBE_TIMEOUT_MS } + )) as { status?: unknown } | null + if (answer?.status === 'live') { + return true + } + if (answer?.status === 'exited') { + return false + } + return null + } catch { + return null + } +} diff --git a/src/main/providers/ssh-pty-provider-liveness-readback.test.ts b/src/main/providers/ssh-pty-provider-liveness-readback.test.ts new file mode 100644 index 000000000000..84d4f6e1b96c --- /dev/null +++ b/src/main/providers/ssh-pty-provider-liveness-readback.test.ts @@ -0,0 +1,61 @@ +import { describe, expect, it, vi } from 'vitest' +import { SshPtyProvider } from './ssh-pty-provider' +import type { SshChannelMultiplexer } from '../ssh/ssh-channel-multiplexer' +import { toAppSshPtyId } from '../../shared/ssh-pty-id' +import { toRelayPtyIdWithMintEpoch } from '../../shared/relay-pty-mint-epoch' + +const CONNECTION_ID = 'conn-1' +const RELAY_EPOCH = '0f8f3a1e-1111-4111-8111-111111111111' +const RELAY_PTY_ID = toRelayPtyIdWithMintEpoch(RELAY_EPOCH, 42) +const APP_PTY_ID = toAppSshPtyId(CONNECTION_ID, RELAY_PTY_ID) + +/** + * The provider is the owner the liveness rule routes to, so this pins the contract seam itself: + * without `probePtyLiveness` on this class, `probePtyLivenessFromRuntimeController` answers null + * for every SSH id and no SSH worker can ever be certified exited. The verdict itself belongs to + * the relay; this class only carries the question to it. + */ +function makeProvider(status: 'live' | 'exited' | 'unverifiable') { + const request = vi.fn(async (method: string) => { + if (method === 'pty.probeLiveness') { + return { status } + } + throw new Error(`unexpected relay method ${method}`) + }) + const mux = { + request, + onNotification: () => () => {}, + onRequest: () => () => {} + } as unknown as SshChannelMultiplexer + return { provider: new SshPtyProvider(CONNECTION_ID, mux), request } +} + +describe('SshPtyProvider liveness readback', () => { + it('exposes the readback the liveness rule asks the owning provider for', () => { + const { provider } = makeProvider('unverifiable') + + expect(typeof provider.probePtyLiveness).toBe('function') + }) + + it('answers live from the relay for an id its own cache has never seen', async () => { + const { provider } = makeProvider('live') + + expect(provider.hasPty(APP_PTY_ID)).toBe(false) + await expect(provider.probePtyLiveness(APP_PTY_ID)).resolves.toBe(true) + }) + + it('certifies an exit the relay observed', async () => { + const { provider } = makeProvider('exited') + + await expect(provider.probePtyLiveness(APP_PTY_ID)).resolves.toBe(false) + }) + + it('refuses to answer for an id belonging to another SSH target', async () => { + const { provider, request } = makeProvider('exited') + + await expect( + provider.probePtyLiveness(toAppSshPtyId('other-conn', RELAY_PTY_ID)) + ).resolves.toBeNull() + expect(request).not.toHaveBeenCalled() + }) +}) diff --git a/src/main/providers/ssh-pty-provider.ts b/src/main/providers/ssh-pty-provider.ts index 76b5d9f85e48..1e3097ef0096 100644 --- a/src/main/providers/ssh-pty-provider.ts +++ b/src/main/providers/ssh-pty-provider.ts @@ -25,6 +25,7 @@ import { SshAgentSessionCapabilities } from './ssh-agent-session-capabilities' import type { PtyProcessInspection } from './pty-process-inspection' import { spawnWithTerminalRuntimeRepair, type TerminalRepairHook } from './ssh-pty-spawn-repair' import { createSshPtyProviderRpcOperations } from './ssh-pty-provider-rpc-operations' +import { probeSshPtyLiveness } from './ssh-pty-liveness-probe' // Why: sequential relay teardown calls share one absolute budget; convert to the mux-relative timeout only at dispatch. function relayTimeoutOptions(deadlineMs: number | undefined): { timeoutMs: number } | undefined { @@ -288,6 +289,21 @@ export class SshPtyProvider implements IPtyProvider { hasPty = (id: string): boolean => this.livePtyIds.has(id) + /** `hasPty` is this process's cache of what it has seen; only the relay may answer absent. */ + probePtyLiveness = async (id: string): Promise => { + let relayPtyId: string + try { + relayPtyId = this.toRelayPtyId(id) + } catch { + // The id names another SSH target, so this provider is not its owner and cannot answer. + return null + } + return await probeSshPtyLiveness({ + request: (method, params, options) => this.mux.request(method, params, options), + relayPtyId + }) + } + onData = (callback: SshPtyDataCallback): (() => void) => this.outputState.onData(callback) onRejectedData = (callback: SshPtyDataCallback): (() => void) => this.outputState.onRejectedData(callback) diff --git a/src/main/runtime/orca-runtime-preserved-branch-cleanup.ts b/src/main/runtime/orca-runtime-preserved-branch-cleanup.ts index d53994032a18..27054fdd7694 100644 --- a/src/main/runtime/orca-runtime-preserved-branch-cleanup.ts +++ b/src/main/runtime/orca-runtime-preserved-branch-cleanup.ts @@ -157,6 +157,7 @@ export class OrcaRuntimeWithPreservedBranchCleanup extends OrcaRuntimeWithTermin connectionId ), runMutation: (worktreeId, operation) => this.runWorktreeTerminalMutation(worktreeId, operation), + isPtyProvenAbsent: (ptyId) => this.isLeafPtyProvenAbsent(ptyId), getActivation: (worktreeId) => this.getLegacyWorkerRecoveryActivation(worktreeId), hasExactPersistedSurface: (candidate) => this.hasExactPersistedTerminalSurfaceIdentity(candidate), diff --git a/src/main/runtime/orca-runtime-test-scenario-builders.spec.ts b/src/main/runtime/orca-runtime-test-scenario-builders.spec.ts index 1f87d901c817..4634daf703e2 100644 --- a/src/main/runtime/orca-runtime-test-scenario-builders.spec.ts +++ b/src/main/runtime/orca-runtime-test-scenario-builders.spec.ts @@ -238,6 +238,8 @@ function makePostRevealWorkerRecoveryHarness( kill, getForegroundProcess: async () => null, hasPty, + // Absence is only an exit when the owning provider says so; these fixtures own no other PTY. + probePtyLiveness: async (candidate: string) => hasPty(candidate) === true, listProcesses: listProcesses ?? (async () => [ diff --git a/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-04.spec.ts b/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-04.spec.ts index d09b4c64ce81..5a2fddff53ff 100644 --- a/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-04.spec.ts +++ b/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-04.spec.ts @@ -79,6 +79,8 @@ describe('OrcaRuntimeService', () => { kill: vi.fn(() => true), getForegroundProcess: async () => null, hasPty: () => false, + // The owning provider's readback, not the empty listing, is what certifies the exit. + probePtyLiveness: async () => false, listProcesses: async () => [] }) const resolveLegacyWorkerTerminalRecovery = vi.fn() @@ -186,6 +188,8 @@ describe('OrcaRuntimeService', () => { kill: vi.fn(() => true), getForegroundProcess: async () => null, hasPty: () => false, + // The owning provider's readback, not the empty listing, is what certifies the exit. + probePtyLiveness: async () => false, listProcesses: async () => [] }) const resolveLegacyWorkerTerminalRecovery = vi.fn() @@ -319,6 +323,7 @@ describe('OrcaRuntimeService', () => { kill: vi.fn(() => true), getForegroundProcess: async () => null, hasPty: () => false, + probePtyLiveness: async (candidate: string) => candidate === 'pty-exited-two', listProcesses }) const revealTerminalSession = vi.fn() diff --git a/src/main/runtime/orchestration/db-lifecycle-state-space-settlement.test.ts b/src/main/runtime/orchestration/db-lifecycle-state-space-settlement.test.ts new file mode 100644 index 000000000000..50c89e92bc19 --- /dev/null +++ b/src/main/runtime/orchestration/db-lifecycle-state-space-settlement.test.ts @@ -0,0 +1,250 @@ +import { describe, expect, it } from 'vitest' +import { OrchestrationDb } from './db' + +const PANE = 'tab_w:11111111-1111-4111-8111-111111111111' + +/** + * A worker report arrives inside a federated relay import and inside the coordinator's message + * sweep, and a throw rolls both back — the relay cursor never advances again and the run is + * marked failed with its batch unread (#16904). So `settleWorkerReport` must answer with a + * settlement or a structured rejection, not an exception. + * + * Scope, so the green tick is not read for more than it covers. The search seeds one federated + * starting dispatch and applies the operations below breadth-first to depth five, memoizing on + * the (Dispatch status, Task status, worker state) triple alone. It therefore does not enumerate + * a second dispatch, an unsupervised dispatch, sibling settlement, or retry worker-start; it + * calls settlement directly rather than through the relay import or the coordinator sweep, which + * fixed examples in db-stopping-worker-report-settlement.test.ts cover; and states that differ + * only in authority, capability, epoch, failure reason, or observation facts collapse into one + * memo entry and are not expanded. Refused operations are swallowed, so a sequence explores only + * the states it actually reached. Depth five is where the triple set stopped growing when this + * was written; the test does not re-derive that each run. + */ +const OPERATIONS = [ + 'prepareAuthority', + 'markReady', + 'markStartUnknown', + 'failStart', + 'beginStop', + 'settleStop', + 'markStopUnknown', + 'resumeFedRelay', + 'abandon', + 'reconcileMissing', + 'fedStart:ready', + 'fedStart:start_unknown', + 'fedStart:failed', + 'fedStart:stopped', + 'fedStop', + 'failDispatch', + 'failDispatch:exited', + 'completeDispatch', + 'taskUpdate:pending', + 'taskUpdate:ready', + 'taskUpdate:blocked', + 'taskUpdate:dispatched', + 'taskUpdate:completed', + 'taskUpdate:failed', + 'mintCap', + 'revokeCap', + 'report:succeeded', + 'report:failed' +] as const + +type Probe = { + db: OrchestrationDb + taskId: string + dispatchId: string +} + +function replay(sequence: readonly string[]): Probe { + const db = new OrchestrationDb(':memory:') + const task = db.createTask({ spec: 'state space' }) + const { dispatch } = db.createStartingWorkerDispatch({ + taskId: task.id, + startOptions: {}, + creator: { kind: 'system' }, + maxDepth: 99, + federation: { + environmentId: 'e', + environmentName: 'n', + peerFingerprint: 'p', + protocolVersion: 12 + } + }) + const id = dispatch.id + const apply = (name: string): void => { + switch (name) { + case 'prepareAuthority': + db.prepareStartingWorkerAuthority({ + dispatchId: id, + handle: 'term_w', + paneKey: PANE, + processIncarnation: 'inc1', + worktreeId: 'wt', + effects: [], + setupState: 'not_configured' + }) + return + case 'markReady': + db.markWorkerDispatchReady(id) + return + case 'markStartUnknown': + db.markWorkerStartUnknown(id, 's', 'w') + return + case 'failStart': + db.failWorkerStart(id, 's', 'w') + return + case 'beginStop': + db.beginWorkerStop(id, 'ep') + return + case 'settleStop': + db.settleWorkerStop(id) + return + case 'markStopUnknown': + db.markWorkerStopUnknown(id, 'w') + return + case 'resumeFedRelay': + db.resumeFederatedWorkerForTerminalRelay(id) + return + case 'abandon': + db.abandonWorkerDispatch(id) + return + case 'reconcileMissing': + db.reconcileMissingWorkerTerminal(id, 'gone') + return + case 'fedStart:ready': + db.reconcileFederatedWorkerStart({ dispatchId: id, state: 'ready', stage: 's' }) + return + case 'fedStart:start_unknown': + db.reconcileFederatedWorkerStart({ dispatchId: id, state: 'start_unknown', stage: 's' }) + return + case 'fedStart:failed': + db.reconcileFederatedWorkerStart({ dispatchId: id, state: 'failed', stage: 's' }) + return + case 'fedStart:stopped': + db.reconcileFederatedWorkerStart({ dispatchId: id, state: 'stopped', stage: 's' }) + return + case 'fedStop': + db.reconcileFederatedWorkerStop(id) + return + case 'failDispatch': + db.failDispatch(id, 'x') + return + case 'failDispatch:exited': + db.failDispatch(id, 'x', { workerProcessExited: true }) + return + case 'completeDispatch': + db.completeDispatch(id) + return + case 'taskUpdate:pending': + db.updateTaskStatus(task.id, 'pending') + return + case 'taskUpdate:ready': + db.updateTaskStatus(task.id, 'ready') + return + case 'taskUpdate:blocked': + db.updateTaskStatus(task.id, 'blocked') + return + case 'taskUpdate:dispatched': + db.updateTaskStatus(task.id, 'dispatched') + return + case 'taskUpdate:completed': + db.updateTaskStatus(task.id, 'completed', 'r') + return + case 'taskUpdate:failed': + db.updateTaskStatus(task.id, 'failed', 'r') + return + case 'mintCap': + db.mintDispatchCapability({ dispatchId: id, paneKey: PANE, processIncarnation: 'inc2' }) + return + case 'revokeCap': + db.revokeDispatchCapability(id) + return + case 'report:succeeded': + db.settleWorkerReport({ + taskId: task.id, + dispatchId: id, + outcome: 'succeeded', + result: 'r' + }) + return + case 'report:failed': + db.settleWorkerReport({ taskId: task.id, dispatchId: id, outcome: 'failed', result: 'r' }) + } + } + for (const name of sequence) { + try { + apply(name) + } catch { + // A refused operation is a legal outcome; the state it did not reach is simply not explored. + } + } + return { db, taskId: task.id, dispatchId: id } +} + +function tripleOf(probe: Probe): string { + const dispatch = probe.db.getDispatchContextById(probe.dispatchId) + const task = probe.db.getTask(probe.taskId) + const worker = probe.db.getWorkerDispatch(probe.dispatchId) + return `${dispatch?.status}/${task?.status}/${worker?.state}` +} + +function reachableTriples(maxDepth: number): Map { + const seen = new Map() + const initial = replay([]) + seen.set(tripleOf(initial), []) + initial.db.close() + let frontier: string[][] = [[]] + for (let depth = 0; depth < maxDepth && frontier.length > 0; depth += 1) { + const next: string[][] = [] + for (const sequence of frontier) { + for (const operation of OPERATIONS) { + const probe = replay([...sequence, operation]) + const triple = tripleOf(probe) + probe.db.close() + if (seen.has(triple)) { + continue + } + seen.set(triple, [...sequence, operation]) + next.push([...sequence, operation]) + } + } + frontier = next + } + return seen +} + +describe('worker report settlement over the reachable lifecycle state space', () => { + it('answers every reachable state without throwing', () => { + // Depth five: see the scope note above. + const reachable = reachableTriples(5) + const throwing: string[] = [] + + for (const [triple, sequence] of reachable) { + for (const outcome of ['succeeded', 'failed'] as const) { + const probe = replay(sequence) + try { + const settlement = probe.db.settleWorkerReport({ + taskId: probe.taskId, + dispatchId: probe.dispatchId, + outcome, + result: 'r' + }) + expect(settlement.action === 'settled' || settlement.action === 'rejected').toBe(true) + } catch (error) { + throwing.push( + `${triple}/${outcome} after ${sequence.join(' -> ') || '(initial)'}: ${(error as Error).message}` + ) + } finally { + probe.db.close() + } + } + } + + expect(throwing).toEqual([]) + // Guards the search itself: a harness that stopped exploring would also report zero throws. + // Measured 71 at this operation set; the floor sits just under it so a real loss is visible. + expect(reachable.size).toBeGreaterThan(68) + }, 300_000) +}) diff --git a/src/main/runtime/orchestration/db-stopping-worker-report-settlement.test.ts b/src/main/runtime/orchestration/db-stopping-worker-report-settlement.test.ts new file mode 100644 index 000000000000..4ef775dd558a --- /dev/null +++ b/src/main/runtime/orchestration/db-stopping-worker-report-settlement.test.ts @@ -0,0 +1,290 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { OrchestrationDb } from './db' +import { Coordinator } from './coordinator' +import type { CoordinatorRuntime } from './coordinator-runtime-contract' +import { reconcileLifecycleMessage } from './lifecycle-reconciliation' +import { createRootDispatch } from './db/root-dispatch-test-fixture' +import type { MessagePriority, MessageType } from './types' + +const PANE_W = 'tab_w:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' + +describe('a Task whose supervised worker is stopping', () => { + let db: OrchestrationDb + beforeEach(() => { + db = new OrchestrationDb(':memory:') + }) + afterEach(() => db.close()) + + function localWorker() { + const task = db.createTask({ spec: 'local work' }) + const { dispatch } = db.createStartingWorkerDispatch({ + taskId: task.id, + startOptions: {}, + creator: { kind: 'system' }, + maxDepth: 9 + }) + db.prepareStartingWorkerAuthority({ + dispatchId: dispatch.id, + handle: 'term_w', + paneKey: PANE_W, + processIncarnation: 'inc1', + worktreeId: 'wt', + effects: [], + setupState: 'not_configured' + }) + db.markWorkerDispatchReady(dispatch.id) + return { task, dispatch } + } + + function federatedWorker() { + const task = db.createTask({ spec: 'remote work' }) + const { dispatch } = db.createStartingWorkerDispatch({ + taskId: task.id, + startOptions: {}, + creator: { kind: 'system' }, + maxDepth: 99, + federation: { + environmentId: 'env_remote', + environmentName: 'remote', + peerFingerprint: 'peer_fp', + protocolVersion: 12 + } + }) + db.reconcileFederatedWorkerStart({ + dispatchId: dispatch.id, + state: 'ready', + stage: 'input_accepted', + worktreeId: 'wt_remote', + terminalHandle: 'term_remote' + }) + return { task, dispatch } + } + + function relayItem( + taskId: string, + runId: string, + dispatchId: string, + sequence: number, + kind: 'status' | 'done', + outcome: 'succeeded' | 'failed' = 'succeeded' + ) { + return { + dispatchId, + sequence, + message: { + id: `msg_${sequence}`, + runId, + from: `dispatch:${dispatchId}`, + to: `run:${runId}`, + subject: kind === 'done' ? 'Worker done' : `Update ${sequence}`, + body: 'b', + type: 'status' as MessageType, + priority: 'normal' as MessagePriority + }, + lifecycle: + kind === 'done' + ? { kind: 'worker_report' as const, taskId, outcome, result: 'the real answer' } + : { kind: 'none' as const } + } + } + + describe('task-update', () => { + it('refuses to re-open the Task while the worker is stopping', () => { + const { task, dispatch } = localWorker() + db.beginWorkerStop(dispatch.id, 'epoch_home') + expect(db.getTask(task.id)?.status).toBe('blocked') + + expect(() => db.updateTaskStatus(task.id, 'dispatched')).toThrowError( + expect.objectContaining({ + code: 'task_not_startable', + data: { taskId: task.id, dispatchId: dispatch.id } + }) + ) + expect(db.getTask(task.id)?.status).toBe('blocked') + }) + + it('refuses to re-open the Task while the stop outcome is unknown', () => { + const { task, dispatch } = localWorker() + db.beginWorkerStop(dispatch.id, 'epoch_home') + db.markWorkerStopUnknown(dispatch.id, 'the execution host did not answer') + + expect(() => db.updateTaskStatus(task.id, 'dispatched')).toThrowError( + expect.objectContaining({ code: 'task_not_startable' }) + ) + expect(db.getTask(task.id)?.status).toBe('blocked') + }) + + it('control: still accepts dispatched for an active Dispatch with no supervised worker', () => { + const task = db.createTask({ spec: 'unsupervised work' }) + createRootDispatch(db, task.id, 'term_worker') + + expect(db.updateTaskStatus(task.id, 'dispatched')?.status).toBe('dispatched') + }) + + it('control: still accepts dispatched while the supervised worker is ready', () => { + const { task } = localWorker() + + expect(db.updateTaskStatus(task.id, 'dispatched')?.status).toBe('dispatched') + }) + }) + + describe('a worker report the lifecycle graph cannot settle', () => { + // Rows an older binary already wrote: it let task-update re-open the Task under a stopping + // worker, so a shipped database can hold this triple even though nothing can reach it now. + function wedgeTaskDispatchedUnderStoppingWorker(taskId: string, dispatchId: string): void { + db.beginWorkerStop(dispatchId, 'epoch_home') + db.db.prepare("UPDATE tasks SET status = 'dispatched' WHERE id = ?").run(taskId) + } + + it('rejects the federated report and still advances the relay cursor', () => { + const { task, dispatch } = federatedWorker() + wedgeTaskDispatchedUnderStoppingWorker(task.id, dispatch.id) + + const imported = db.importFederatedRelayItem( + relayItem(task.id, task.run_id, dispatch.id, 1, 'done') + ) + + expect(imported.lifecycle).toMatchObject({ + action: 'rejected', + code: 'worker_not_settleable' + }) + expect(db.getFederatedDispatch(dispatch.id)?.to_home_imported_sequence).toBe(1) + expect(db.getMessageById('msg_1')).toBeDefined() + // The stream is not wedged behind the report it could not apply. + expect( + db.importFederatedRelayItem(relayItem(task.id, task.run_id, dispatch.id, 2, 'status')) + .message.id + ).toBe('msg_2') + expect(db.getFederatedDispatch(dispatch.id)?.to_home_imported_sequence).toBe(2) + }) + + it('rejects the local report instead of throwing out of reconciliation', () => { + const { task, dispatch } = localWorker() + wedgeTaskDispatchedUnderStoppingWorker(task.id, dispatch.id) + const msg = workerDoneMessage(task.id, task.run_id, dispatch.id) + + expect(reconcileLifecycleMessage(db, msg)).toMatchObject({ + action: 'rejected', + code: 'worker_not_settleable' + }) + expect(db.getWorkerDispatch(dispatch.id)?.state).toBe('stopping') + }) + + it('lets the coordinator loop finish its batch and mark the report read', async () => { + const { task, dispatch } = localWorker() + wedgeTaskDispatchedUnderStoppingWorker(task.id, dispatch.id) + workerDoneMessage(task.id, task.run_id, dispatch.id) + const coordinator = new Coordinator(db, stubCoordinatorRuntime(), { + spec: 'stop-wedge', + coordinatorHandle: `run:${task.run_id}`, + pollIntervalMs: 0, + onLog: (line) => { + if (line.includes('rejected')) { + coordinator.stop() + } + } + }) + + const run = await coordinator.runFromExistingRun(task.run_id) + + expect(run.failedTasks).toEqual([]) + expect(db.getUnreadMessages(`run:${task.run_id}`)).toEqual([]) + }) + + // The graph accepts stopping -> failed, so this report is the other half of the guard: it + // must settle from the state the worker is actually in. A hardcoded `ready` precondition + // throws here and wedges the very relay batch the success case proves is safe. + it('settles a failed report the graph does accept from a stopping worker', () => { + const { task, dispatch } = federatedWorker() + wedgeTaskDispatchedUnderStoppingWorker(task.id, dispatch.id) + + const imported = db.importFederatedRelayItem( + relayItem(task.id, task.run_id, dispatch.id, 1, 'done', 'failed') + ) + + expect(imported.lifecycle).toMatchObject({ action: 'settled', outcome: 'failed' }) + expect(db.getFederatedDispatch(dispatch.id)?.to_home_imported_sequence).toBe(1) + expect(db.getWorkerDispatch(dispatch.id)?.state).toBe('failed') + expect(db.getTask(task.id)?.status).toBe('failed') + expect(db.getDispatchContextById(dispatch.id)?.status).toBe('failed') + }) + + it('control: a ready worker still settles its Task', () => { + const { task, dispatch } = localWorker() + + expect( + db.settleWorkerReport({ + taskId: task.id, + dispatchId: dispatch.id, + outcome: 'succeeded', + result: 'done' + }) + ).toMatchObject({ action: 'settled', outcome: 'succeeded' }) + expect(db.getTask(task.id)?.status).toBe('completed') + expect(db.getWorkerDispatch(dispatch.id)?.state).toBe('succeeded') + }) + }) + + describe('operator escape', () => { + it('accepts a re-issued worker-stop and reaches an honest stop_unknown outcome', () => { + const { task, dispatch } = localWorker() + db.beginWorkerStop(dispatch.id, 'epoch_dead_runtime') + + // The runtime that owned the first stop died mid-flight; the re-issue is the way out. + const reissued = db.beginWorkerStop(dispatch.id, 'epoch_new_runtime') + expect(reissued).toMatchObject({ disposition: 'stopping' }) + expect(db.getWorkerDispatch(dispatch.id)?.runtime_epoch).toBe('epoch_new_runtime') + + db.markWorkerStopUnknown(dispatch.id, 'the execution host did not answer') + expect(db.abandonWorkerDispatch(dispatch.id)).toMatchObject({ disposition: 'abandoned' }) + expect(db.getTask(task.id)?.status).toBe('blocked') + }) + + it('refuses a re-issue from the runtime whose own stop is still in flight', () => { + const { dispatch } = localWorker() + db.beginWorkerStop(dispatch.id, 'epoch_this_runtime') + + // The terminal is closing and its exit event has not landed yet. Letting this second pass + // record stop_unknown would make the exit read as a crash instead of this stop succeeding. + expect(() => db.beginWorkerStop(dispatch.id, 'epoch_this_runtime')).toThrowError( + /cannot stop from stopping/ + ) + + // The row is still the one the exit path claims a clean stop from: stopping, same epoch. + expect(db.getWorkerDispatch(dispatch.id)).toMatchObject({ + state: 'stopping', + runtime_epoch: 'epoch_this_runtime' + }) + expect(db.settleWorkerStop(dispatch.id).state).toBe('stopped') + expect(db.getDispatchContextById(dispatch.id)).toMatchObject({ + status: 'failed', + last_failure: 'stopped' + }) + }) + }) + + function workerDoneMessage(taskId: string, runId: string, dispatchId: string) { + return db.insertMessage({ + runId, + from: 'term_w', + to: `run:${runId}`, + subject: 'Worker done', + body: 'finished the work', + type: 'worker_done', + priority: 'normal', + senderPaneKey: PANE_W, + payload: JSON.stringify({ taskId, dispatchId, outcome: 'succeeded' }) + }) + } +}) + +/** The coordinator's message sweep touches none of these; a dispatched Task creates no worker. */ +function stubCoordinatorRuntime(): CoordinatorRuntime { + return { + sendTerminalAgentPrompt: async () => ({}), + listTerminals: async () => ({ terminals: [] }), + createTerminal: async () => ({ handle: 'term_unused', worktreeId: 'wt' }), + waitForTerminal: async (handle: string) => ({ handle, condition: 'idle' }), + probeWorktreeDrift: async () => null + } +} diff --git a/src/main/runtime/orchestration/db/dispatch-context/worker-report-settlement.ts b/src/main/runtime/orchestration/db/dispatch-context/worker-report-settlement.ts index 3584a2e59a65..1cb0526b90cb 100644 --- a/src/main/runtime/orchestration/db/dispatch-context/worker-report-settlement.ts +++ b/src/main/runtime/orchestration/db/dispatch-context/worker-report-settlement.ts @@ -3,7 +3,7 @@ import type { OrchestrationDb } from '../orchestration-db' import { AGENT_PROMPT_STALLED_ERROR } from '../../../agent-prompt-submission-verification' import { settleActiveDispatchesForTask } from './dispatch-completion' import { getActiveDispatchForTask } from './task-dispatch-reconciliation' -import { transitionLifecycleWithDb } from '../lifecycle-transition' +import { isLegalLifecycleTransition, transitionLifecycleWithDb } from '../lifecycle-transition' import { runLifecycleWriteTransaction } from '../lifecycle-write-transaction-runner' type WorkerReportObservation = { @@ -159,6 +159,24 @@ export function settleWorkerReportInTransaction( ) .all(params.taskId, params.dispatchId) as { id: string }[] + const settledWorkerState = params.outcome === 'succeeded' ? 'succeeded' : 'failed' + // Why: this settlement rides a federated relay import and the coordinator's message sweep, and + // both roll the whole batch back on a throw — a relay that can never advance its cursor and a + // run marked failed with its messages left unread (#16904). A worker whose own state the + // lifecycle graph will not settle is a rejection, which those callers already record and move on. + if ( + reportingWorker && + !settledByUnobservedPrompt && + !reconnectingStart && + !isLegalLifecycleTransition('worker', reportingWorker.state, settledWorkerState) + ) { + return { + action: 'rejected', + code: 'worker_not_settleable', + reason: `Dispatch ${params.dispatchId} worker is ${reportingWorker.state}; it cannot settle as ${settledWorkerState}.` + } + } + this.db.exec('SAVEPOINT settle_worker_report') let dispatchUpdate: { changes: number } let taskUpdate: { changes: number } @@ -250,11 +268,12 @@ export function settleWorkerReportInTransaction( }) } else if (reportingWorker) { transitionLifecycleWithDb(this.db, { + // The guard above already proved the graph accepts this edge from the state the worker is + // actually in; a hardcoded `from` would instead throw for every other live state. entity: 'worker', id: params.dispatchId, - // A start_unknown success report reconnects through 'ready' above; only failure settles here. - from: params.outcome === 'succeeded' ? 'ready' : ['ready', 'start_unknown'], - to: params.outcome === 'succeeded' ? 'succeeded' : 'failed', + from: reportingWorker.state, + to: settledWorkerState, projection: { stage: 'settled', updated_at: new Date().toISOString() } }) } diff --git a/src/main/runtime/orchestration/db/lifecycle-transition.ts b/src/main/runtime/orchestration/db/lifecycle-transition.ts index 6fcc40f19131..1d0b8ec6a22a 100644 --- a/src/main/runtime/orchestration/db/lifecycle-transition.ts +++ b/src/main/runtime/orchestration/db/lifecycle-transition.ts @@ -117,6 +117,15 @@ const PROJECTION_COLUMNS = new Set([ 'runtime_epoch' ]) +/** The lifecycle graph asked without writing, for callers that must reject rather than throw. */ +export function isLegalLifecycleTransition( + entity: LifecycleEntity, + from: string, + to: string +): boolean { + return (LEGAL_TRANSITIONS[entity][from] ?? []).includes(to) +} + export function transitionLifecycle( this: OrchestrationDb, params: LifecycleTransitionParams @@ -151,14 +160,16 @@ export function transitionLifecycleWithDb( { entity: params.entity, id: params.id, state: current.state } ) } - const legal = LEGAL_TRANSITIONS[params.entity][current.state] ?? [] const promptReportCorrection = params.correction === 'unobserved_prompt_report' && current.state === 'failed' && ((params.entity === 'task' && params.to === 'completed') || (params.entity === 'dispatch' && params.to === 'completed') || (params.entity === 'worker' && params.to === 'succeeded')) - if (!legal.includes(params.to) && !promptReportCorrection) { + if ( + !isLegalLifecycleTransition(params.entity, current.state, params.to) && + !promptReportCorrection + ) { throw new OrchestrationError( 'lifecycle_conflict', `${params.entity} ${params.id} cannot transition from ${current.state} to ${params.to}.`, diff --git a/src/main/runtime/orchestration/db/tasks/task-status-transition.ts b/src/main/runtime/orchestration/db/tasks/task-status-transition.ts index e1de8dc5b17a..dcddc781b6d8 100644 --- a/src/main/runtime/orchestration/db/tasks/task-status-transition.ts +++ b/src/main/runtime/orchestration/db/tasks/task-status-transition.ts @@ -35,18 +35,24 @@ export function updateTaskStatus( ORDER BY rowid DESC LIMIT 1` ) .get(id) as { id: string } | undefined - const activeWorker = terminalStatus - ? (this.db - .prepare( - `SELECT active.id + // Why: a supervised worker owns its Task for as long as it is alive. Every status this + // function lets past the active-Dispatch check must clear the same worker check, or the Task + // re-opens under a worker whose own lifecycle can no longer settle it (#16904 relay wedge). + // A no-op re-assert of `dispatched` re-opens nothing and stays legal. + const reopensUnderWorker = requiresActiveDispatch && task.status !== 'dispatched' + const activeWorker = + terminalStatus || reopensUnderWorker + ? (this.db + .prepare( + `SELECT active.id FROM dispatch_contexts active JOIN worker_dispatches worker ON worker.dispatch_id = active.id WHERE active.task_id = ? AND active.status IN ('pending', 'dispatched') AND worker.state NOT IN ('failed', 'succeeded', 'stopped', 'abandoned') ORDER BY active.rowid DESC LIMIT 1` - ) - .get(id) as { id: string } | undefined) - : undefined + ) + .get(id) as { id: string } | undefined) + : undefined if (activeWorker) { throw new OrchestrationError( 'task_not_startable', diff --git a/src/main/runtime/orchestration/db/worker-dispatch/worker-dispatch-stop.ts b/src/main/runtime/orchestration/db/worker-dispatch/worker-dispatch-stop.ts index 8dbda2030c59..fd4ee773bb4e 100644 --- a/src/main/runtime/orchestration/db/worker-dispatch/worker-dispatch-stop.ts +++ b/src/main/runtime/orchestration/db/worker-dispatch/worker-dispatch-stop.ts @@ -59,7 +59,19 @@ export function beginWorkerStop( this.db.exec('COMMIT') return { disposition: 'already_settled', worker, dispatch } } - if (!['ready', 'start_unknown'].includes(worker.state)) { + // Why `stopping` under a DIFFERENT epoch is accepted: a stop whose runtime died mid-flight + // leaves the row here forever, and refusing the re-issue was the only operator escape + // (#16904). Re-running the stop earns the honest outcome — settled, or `stop_unknown`, from + // which the worker can be abandoned. It never asserts an exit the runtime did not observe. + // + // Why the epoch and not just the state: this runtime's own `stopping` row means its stop is + // still in flight, and a second pass would record `stop_unknown` over it. The exit event that + // follows claims a clean stop only from `stopping` under its own epoch + // (failActiveDispatchOnExit), so it would then read the operator's stop as a crash and + // escalate it. Same predicate as that reader, so both agree on whose stop this is. + const stopStrandedByAnotherRuntime = + worker.state === 'stopping' && worker.runtime_epoch !== runtimeEpoch + if (!['ready', 'start_unknown'].includes(worker.state) && !stopStrandedByAnotherRuntime) { throw new OrchestrationError( 'dispatch_inactive', `Dispatch ${dispatchId} cannot stop from ${worker.state}.` diff --git a/src/main/runtime/orchestration/lifecycle-reconciliation.ts b/src/main/runtime/orchestration/lifecycle-reconciliation.ts index ff6d96491fef..5873bb540793 100644 --- a/src/main/runtime/orchestration/lifecycle-reconciliation.ts +++ b/src/main/runtime/orchestration/lifecycle-reconciliation.ts @@ -51,6 +51,7 @@ export type LifecycleRejectionCode = | 'task_dispatch_mismatch' | 'inactive_dispatch' | 'stale_dispatch' + | 'worker_not_settleable' export type LifecycleRejectionResult = { action: 'rejected' diff --git a/src/main/runtime/orchestration/types.ts b/src/main/runtime/orchestration/types.ts index 00005443006d..3179d38b121c 100644 --- a/src/main/runtime/orchestration/types.ts +++ b/src/main/runtime/orchestration/types.ts @@ -33,6 +33,7 @@ export type WorkerReportSettlement = | 'task_dispatch_mismatch' | 'inactive_dispatch' | 'stale_dispatch' + | 'worker_not_settleable' reason: string } diff --git a/src/main/runtime/rpc/methods/orchestration/worker/worker-stop.ts b/src/main/runtime/rpc/methods/orchestration/worker/worker-stop.ts index 605d8c52d4a8..643323cf62e1 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/worker-stop.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/worker-stop.ts @@ -245,8 +245,9 @@ export const ORCHESTRATION_WORKER_STOP_METHODS: RpcMethod[] = [ const activeStopByRuntime = new WeakMap>>() -/** Two callers stopping one Dispatch: the second reached `beginWorkerStop` after the first moved - * the row to `stopping` and got `dispatch_inactive` instead of the first caller's receipt. */ +/** Two callers stopping one Dispatch: coalesced so only one of them closes the terminal. Both are + * in this runtime and so carry one epoch, which `beginWorkerStop` refuses a second time anyway; + * the epoch it does accept belongs to a row a dead runtime stranded, and no caller here holds one. */ function dedupeWorkerStop( runtime: OrcaRuntimeService, dispatchId: string, diff --git a/src/main/runtime/runtime-legacy-worker-terminal-exit-certification.test.ts b/src/main/runtime/runtime-legacy-worker-terminal-exit-certification.test.ts new file mode 100644 index 000000000000..315d872b9f8e --- /dev/null +++ b/src/main/runtime/runtime-legacy-worker-terminal-exit-certification.test.ts @@ -0,0 +1,211 @@ +import { describe, expect, it, vi } from 'vitest' +import { reconcileLegacyWorkerCandidate } from './runtime-legacy-worker-terminal-recovery-candidate' +import type { + LegacyWorkerRecoveryCandidate, + LegacyWorkerRecoveryInventory, + LegacyWorkerRecoveryPorts, + LegacyWorkerRecoveryResolution +} from './runtime-legacy-worker-terminal-recovery-types' +import { probeSshPtyLiveness } from '../providers/ssh-pty-liveness-probe' +import { toAppSshPtyId, toRelaySshPtyId } from '../../shared/ssh-pty-id' +import { toRelayPtyIdWithMintEpoch } from '../../shared/relay-pty-mint-epoch' + +const CONNECTION_ID = 'conn-1' +const RELAY_EPOCH = '0f8f3a1e-1111-4111-8111-111111111111' +// A real SSH id, so the population this suite claims to cover is the one it drives. +const PTY_ID = toAppSshPtyId(CONNECTION_ID, toRelayPtyIdWithMintEpoch(RELAY_EPOCH, 42)) +const OTHER_PTY_ID = toAppSshPtyId(CONNECTION_ID, toRelayPtyIdWithMintEpoch(RELAY_EPOCH, 43)) + +const candidate = { + dispatchId: 'dispatch_1', + dispatchStatus: 'dispatched', + contractVersion: 3, + taskId: 'task_1', + worktreeId: 'repo::worktree', + terminalHandle: 'term_worker', + paneKey: 'tab_a:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb', + tabId: 'tab_a', + leafId: 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb', + processIncarnation: `${PTY_ID}:inc-1`, + ptyId: PTY_ID, + incarnationId: 'inc-1' +} as unknown as LegacyWorkerRecoveryCandidate + +function inventory(options: { + ptyIds: string[] + identity?: { handle: string; incarnationId: string } +}): LegacyWorkerRecoveryInventory { + return { + livePtyIds: new Set(options.ptyIds), + allLivePtyIds: new Set(options.ptyIds), + terminalIdentityByPtyId: new Map(options.identity ? [[PTY_ID, options.identity]] : []), + queriedHostIds: new Set(['ssh:conn-1']) + } as unknown as LegacyWorkerRecoveryInventory +} + +const listingWithoutThePty = inventory({ ptyIds: [OTHER_PTY_ID] }) +const listingWithThePty = inventory({ + ptyIds: [PTY_ID], + identity: { handle: 'term_worker', incarnationId: 'inc-1' } +}) + +function reconcile(options: { + inventory: LegacyWorkerRecoveryInventory + isPtyProvenAbsent: (ptyId: string) => Promise + refreshInventory?: LegacyWorkerRecoveryPorts['refreshInventory'] +}) { + const pendingResolutions: LegacyWorkerRecoveryResolution[] = [] + const deferredDispatchIds = new Set() + const onPtyExit = vi.fn() + const ports = { + refreshInventory: options.refreshInventory ?? vi.fn(async () => options.inventory), + runMutation: async (_worktreeId: string, operation: () => Promise) => operation(), + isPtyProvenAbsent: vi.fn(options.isPtyProvenAbsent), + hasExactPersistedSurface: () => true, + hasExactSurface: () => true, + adopt: vi.fn(), + getRendererEpoch: () => 1, + reveal: vi.fn(), + onPtyExit, + getActivation: () => ({}) + } as unknown as LegacyWorkerRecoveryPorts + + return reconcileLegacyWorkerCandidate({ + controller: { hasReceipt: () => true, setReceipt: vi.fn(), deleteReceipt: vi.fn() } as never, + ports, + options: { connectionId: 'conn-1' }, + candidate, + workspace: { scope: { connectionId: 'conn-1' }, resolved: { id: 'repo::worktree' } } as never, + resolvedWorktrees: [], + inventory: options.inventory, + deferredDispatchIds, + pendingResolutions + }).then(() => ({ pendingResolutions, deferredDispatchIds, ports, onPtyExit })) +} + +describe('legacy worker recovery: certifying that a worker PTY exited', () => { + it('defers when a listing omits the PTY and no owner proved it absent', async () => { + const { pendingResolutions, deferredDispatchIds, ports } = await reconcile({ + inventory: listingWithoutThePty, + isPtyProvenAbsent: async () => false + }) + + // Abandoning the worker here would dispatch the same Task twice while its agent still runs. + expect(pendingResolutions).toEqual([]) + expect([...deferredDispatchIds]).toEqual(['dispatch_1']) + expect(ports.isPtyProvenAbsent).toHaveBeenCalledWith(PTY_ID) + }) + + it('certifies the exit once the owning provider answers absent', async () => { + const { pendingResolutions, deferredDispatchIds } = await reconcile({ + inventory: listingWithoutThePty, + isPtyProvenAbsent: async () => true + }) + + expect(pendingResolutions).toEqual([{ candidate, resolution: 'exited' }]) + expect(deferredDispatchIds.size).toBe(0) + }) + + it('defers when the pre-adoption refresh loses the PTY without an absence proof', async () => { + // A relay that restarts between the plan's inventory and the adoption re-list omits every id + // the previous one minted; adopting or retiring on that omission is the same unsound step. + const refreshInventory = vi.fn().mockResolvedValue(listingWithoutThePty) + const { pendingResolutions, deferredDispatchIds, ports, onPtyExit } = await reconcile({ + inventory: listingWithThePty, + isPtyProvenAbsent: async () => false, + refreshInventory: refreshInventory as never + }) + + expect(pendingResolutions).toEqual([]) + expect([...deferredDispatchIds]).toEqual(['dispatch_1']) + expect(ports.adopt).not.toHaveBeenCalled() + expect(onPtyExit).not.toHaveBeenCalled() + }) + + it('defers when the PTY leaves the listing after adoption without an absence proof', async () => { + const refreshInventory = vi + .fn() + .mockResolvedValueOnce(listingWithThePty) + .mockResolvedValueOnce(listingWithoutThePty) + const { pendingResolutions, deferredDispatchIds, onPtyExit } = await reconcile({ + inventory: listingWithThePty, + isPtyProvenAbsent: async () => false, + refreshInventory: refreshInventory as never + }) + + expect(pendingResolutions).toEqual([]) + expect([...deferredDispatchIds]).toEqual(['dispatch_1']) + expect(onPtyExit).not.toHaveBeenCalled() + }) + + it('control: still certifies the exit when the listed PTY carries another process', async () => { + const { pendingResolutions, ports } = await reconcile({ + inventory: inventory({ + ptyIds: [PTY_ID], + identity: { handle: 'term_worker', incarnationId: 'inc-2' } + }), + isPtyProvenAbsent: async () => false + }) + + // A different incarnation on the id is a positive observation, not an omission. + expect(pendingResolutions).toEqual([{ candidate, resolution: 'exited' }]) + expect(ports.isPtyProvenAbsent).not.toHaveBeenCalled() + }) + + // The rule `isLeafPtyProvenAbsent` applies: the owning provider's readback is proven absence + // only when it answers false, and the SSH provider's readback is the relay itself. + function provenAbsentViaRelay(request: Parameters[0]['request']) { + return async (appPtyId: string): Promise => + (await probeSshPtyLiveness({ + request, + relayPtyId: toRelaySshPtyId(CONNECTION_ID, appPtyId) + })) === false + } + + it('certifies an SSH exit the owning relay observed', async () => { + const { pendingResolutions, deferredDispatchIds } = await reconcile({ + inventory: listingWithoutThePty, + isPtyProvenAbsent: provenAbsentViaRelay(async () => ({ status: 'exited' })) + }) + + expect(pendingResolutions).toEqual([{ candidate, resolution: 'exited' }]) + expect(deferredDispatchIds.size).toBe(0) + }) + + it('defers an SSH candidate the owning relay will not certify', async () => { + const { pendingResolutions, deferredDispatchIds } = await reconcile({ + inventory: listingWithoutThePty, + isPtyProvenAbsent: provenAbsentViaRelay(async () => ({ status: 'unverifiable' })) + }) + + expect(pendingResolutions).toEqual([]) + expect([...deferredDispatchIds]).toEqual(['dispatch_1']) + }) + + it('defers when the host is too old to know the readback, without a capability handshake', async () => { + // A relay predating `pty.probeLiveness` answers JSON-RPC -32601, which arrives as a rejection. + // The old client behaviour is preserved by deferring, not by asking the host what it supports. + const request = vi.fn(async () => { + throw new Error('Method not found: pty.probeLiveness') + }) + const { pendingResolutions, deferredDispatchIds } = await reconcile({ + inventory: listingWithoutThePty, + isPtyProvenAbsent: provenAbsentViaRelay(request) + }) + + expect(pendingResolutions).toEqual([]) + expect([...deferredDispatchIds]).toEqual(['dispatch_1']) + expect(request).toHaveBeenCalledTimes(1) + }) + + it('control: still adopts a PTY the listing names with the recorded identity', async () => { + const { pendingResolutions, deferredDispatchIds, ports } = await reconcile({ + inventory: listingWithThePty, + isPtyProvenAbsent: async () => true + }) + + expect(pendingResolutions).toEqual([{ candidate, resolution: 'adopted' }]) + expect(deferredDispatchIds.size).toBe(0) + expect(ports.isPtyProvenAbsent).not.toHaveBeenCalled() + }) +}) diff --git a/src/main/runtime/runtime-legacy-worker-terminal-recovery-candidate.ts b/src/main/runtime/runtime-legacy-worker-terminal-recovery-candidate.ts index c25ca8deab69..8fb60e4c8531 100644 --- a/src/main/runtime/runtime-legacy-worker-terminal-recovery-candidate.ts +++ b/src/main/runtime/runtime-legacy-worker-terminal-recovery-candidate.ts @@ -8,6 +8,20 @@ import type { LegacyWorkerRecoveryWorkspace } from './runtime-legacy-worker-terminal-recovery-types' +/** + * A listing that does not name the PTY is not an observation of its exit. `listProcesses` returns + * the host's CURRENT session map, so a restarted relay omits every id the previous one minted and + * a daemon session this process never attached was never in the cache at all — neither shell + * necessarily died. Only the owning provider's readback may certify absence; everything else is + * unverifiable and defers to the next sweep (docs/reference/ssh-execution-boundary.md). + */ +async function resolvePtyAbsentFromListing( + ports: LegacyWorkerRecoveryPorts, + ptyId: string +): Promise<'exited' | 'unverifiable'> { + return (await ports.isPtyProvenAbsent(ptyId)) ? 'exited' : 'unverifiable' +} + export async function reconcileLegacyWorkerCandidate(args: { controller: RuntimeLegacyWorkerTerminalRecoveryController ports: LegacyWorkerRecoveryPorts @@ -21,7 +35,11 @@ export async function reconcileLegacyWorkerCandidate(args: { }): Promise { const { controller, ports, options, candidate, workspace, resolvedWorktrees } = args if (!args.inventory.livePtyIds.has(candidate.ptyId)) { - args.pendingResolutions.push({ candidate, resolution: 'exited' }) + if ((await resolvePtyAbsentFromListing(ports, candidate.ptyId)) === 'exited') { + args.pendingResolutions.push({ candidate, resolution: 'exited' }) + return + } + args.deferredDispatchIds.add(candidate.dispatchId) return } const controllerIdentity = args.inventory.terminalIdentityByPtyId.get(candidate.ptyId) @@ -47,7 +65,7 @@ export async function reconcileLegacyWorkerCandidate(args: { return 'unverifiable' } if (!preAdoptionInventory.livePtyIds.has(candidate.ptyId)) { - return 'exited' + return await resolvePtyAbsentFromListing(ports, candidate.ptyId) } const preAdoptionIdentity = preAdoptionInventory.terminalIdentityByPtyId.get(candidate.ptyId) if (!preAdoptionIdentity) { @@ -133,6 +151,10 @@ export async function reconcileLegacyWorkerCandidate(args: { } if (!finalInventory.livePtyIds.has(candidate.ptyId)) { controller.deleteReceipt(candidate.paneKey) + if ((await resolvePtyAbsentFromListing(ports, candidate.ptyId)) !== 'exited') { + args.deferredDispatchIds.add(candidate.dispatchId) + return + } ports.onPtyExit(candidate) args.pendingResolutions.push({ candidate, resolution: 'exited' }) return diff --git a/src/main/runtime/runtime-legacy-worker-terminal-recovery-types.ts b/src/main/runtime/runtime-legacy-worker-terminal-recovery-types.ts index c65afd739523..373db21bf56c 100644 --- a/src/main/runtime/runtime-legacy-worker-terminal-recovery-types.ts +++ b/src/main/runtime/runtime-legacy-worker-terminal-recovery-types.ts @@ -49,6 +49,8 @@ export type LegacyWorkerRecoveryPorts = { ) => Promise /** Serializes the pre-adoption liveness probe and the adoption itself against other terminal mutations. */ runMutation: (worktreeId: string, operation: () => Promise) => Promise + /** True only when the owning provider answered that the PTY is absent; doubt answers false. */ + isPtyProvenAbsent: (ptyId: string) => Promise getActivation: (worktreeId: string) => { activeTabId?: string; activeGroupId?: string } hasExactPersistedSurface: (candidate: LegacyWorkerRecoveryCandidate) => boolean hasExactSurface: (candidate: LegacyWorkerRecoveryCandidate) => boolean diff --git a/src/relay/pty-handler-liveness-probe.test.ts b/src/relay/pty-handler-liveness-probe.test.ts new file mode 100644 index 000000000000..3ece338bea6b --- /dev/null +++ b/src/relay/pty-handler-liveness-probe.test.ts @@ -0,0 +1,423 @@ +// Which removals of a PTY record observed the process end, and which are bookkeeping. Only the +// relay knows, so only the relay may answer, and it certifies an exit from an observation alone. +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const { mockPtySpawn, mockPtyInstance, mockCreateShellPromptReadinessProbe } = vi.hoisted(() => ({ + mockPtySpawn: vi.fn(), + mockCreateShellPromptReadinessProbe: vi.fn(), + mockPtyInstance: { + pid: process.pid, + process: 'zsh', + onData: vi.fn(), + onExit: vi.fn(), + write: vi.fn(), + resize: vi.fn(), + kill: vi.fn(), + clear: vi.fn(), + pause: vi.fn(), + resume: vi.fn() + } +})) + +vi.mock('node-pty', () => ({ spawn: mockPtySpawn })) + +vi.mock('../main/pty/posix-pty-process-groups', () => ({ + forceKillPosixPtyProcessGroups: vi.fn((_pid: number, fallback: () => void) => fallback()) +})) + +vi.mock('../main/shell-prompt-readiness-probe', () => ({ + createShellPromptReadinessProbe: mockCreateShellPromptReadinessProbe +})) + +import { OBSERVED_PTY_EXIT_HISTORY, type PtyHandler } from './pty-handler' +import { + beginPtyHandlerTest, + createMockDispatcher, + createPtyRequestHelpers, + createTestPtyHandler, + endPtyHandlerTest, + testPtyId +} from './pty-handler-test-harness' +import type { MockDispatcher } from './pty-handler-test-harness' + +const UNREACHABLE_PID = 424_242 + +describe('PtyHandler.probeLiveness', () => { + let dispatcher: MockDispatcher + let handler: PtyHandler + let originalPlatform: PropertyDescriptor | undefined + + const { spawnPty } = createPtyRequestHelpers(() => dispatcher) + + beforeEach(() => { + ;({ dispatcher, handler, originalPlatform } = beginPtyHandlerTest({ + mockPtySpawn, + mockPtyInstance, + mockCreateShellPromptReadinessProbe + })) + }) + + afterEach(async () => { + await endPtyHandlerTest(handler, originalPlatform) + }) + + async function probe(id: string, on: MockDispatcher = dispatcher): Promise { + const answer = (await on.callRequest('pty.probeLiveness', { id })) as { status: string } + return answer.status + } + + /** Fire the exit node-pty reported for the most recently spawned PTY. */ + function reportExitOfLatestPty(): void { + const onExit = mockPtyInstance.onExit.mock.calls.at(-1)?.[0] as (event: { + exitCode: number + }) => void + onExit({ exitCode: 0 }) + } + + /** + * No current teardown leaves a disposed record in the pool — each disposer removes it in the + * same synchronous step — so the state is built here rather than left to a future one. The + * probe's disposed guard and the listing's torn-down sweep both exist for exactly that teardown, + * and both must keep answering unverifiable when it arrives. + */ + function tearDownRecordWithoutRemovingIt(id: string): void { + const pool = (handler as unknown as { ptys: Map }).ptys + const managed = pool.get(id) + if (!managed) { + throw new Error(`no record for ${id}`) + } + managed.disposed = true + } + + /** Tear a record down and let the listing sweep it away; nothing watched the process end. */ + async function forgetRecordWithoutObservingItsExit(id: string): Promise { + tearDownRecordWithoutRemovingIt(id) + await dispatcher.callRequest('pty.listProcesses', { includeForegroundProcessEvidence: false }) + } + + /** The state `pty.revive` re-creates a process from, for an id this relay did not mint. */ + function serializedState(id: string): string { + return JSON.stringify([{ id, pid: process.pid, cols: 80, rows: 24, cwd: process.cwd() }]) + } + + /** A shutdown whose sibling refuses to die: the aggregate fails and the relay keeps serving. */ + async function failedShutdownThatKeepsTheRelayServing(): Promise { + const failedKill = vi.fn<() => void>(() => { + throw new Error('host refused kill') + }) + mockPtySpawn.mockReturnValueOnce({ ...mockPtyInstance, kill: failedKill }) + await spawnPty() + const disposal = handler.dispose().catch((error: Error) => error) + await vi.advanceTimersByTimeAsync(8_001) + expect(await disposal).toMatchObject({ message: 'host refused kill' }) + // Let the fixture be cleaned up now that the refusal has been asserted. + failedKill.mockImplementation(() => {}) + } + + describe('observed exits', () => { + it('answers live while the relay owns a running record', async () => { + const { id } = await spawnPty() + + expect(await probe(id)).toBe('live') + }) + + it('certifies the exit node-pty reported', async () => { + const { id } = await spawnPty() + const onExit = mockPtyInstance.onExit.mock.calls.at(-1)?.[0] as (event: { + exitCode: number + signal?: number + }) => void + onExit({ exitCode: 0 }) + + expect(await probe(id)).toBe('exited') + }) + + it('certifies an exit it proves by probing the pid, and keeps certifying it', async () => { + mockPtySpawn.mockReturnValueOnce({ ...mockPtyInstance, pid: UNREACHABLE_PID }) + const { id } = await spawnPty() + + // The shell ended without node-pty reporting it; the pid answers ESRCH. + expect(await probe(id)).toBe('exited') + // The first probe reaped the record, so the second has only the observation to go on — and + // the recovery sweep that drives this asks again on every pass. + expect(await probe(id)).toBe('exited') + }) + + it('certifies an exit node-pty reported after the shutdown wait had given up', async () => { + const { id } = await spawnPty() + const exiting = mockPtyInstance.onExit.mock.calls.at(-1)?.[0] as (event: { + exitCode: number + }) => void + + await failedShutdownThatKeepsTheRelayServing() + + // The record was removed on a timeout, which watched nothing. + expect(await probe(id)).toBe('unverifiable') + exiting({ exitCode: 0 }) + // Disposal is bookkeeping; the callback that follows it is still this relay watching this + // process end, and losing it is what left these workers unverifiable forever. + expect(await probe(id)).toBe('exited') + }) + }) + + describe('bookkeeping removals and states that observed nothing', () => { + it('stays unverifiable after a shutdown removes a record it never saw exit', async () => { + const { id } = await spawnPty() + const disposal = handler.dispose() + await vi.advanceTimersByTimeAsync(8_001) + await disposal + + expect(handler.activePtyCount).toBe(0) + expect(await probe(id)).toBe('unverifiable') + }) + + it('stays unverifiable for an id this relay never minted', async () => { + expect(await probe(testPtyId(99))).toBe('unverifiable') + }) + + it('stays unverifiable for an id another relay generation minted', async () => { + const { id } = await spawnPty() + reportExitOfLatestPty() + expect(await probe(id)).toBe('exited') + + // Ledger keys are the whole ids of records this relay held, so no number of exits it has + // watched can put another generation's id in it. + expect(await probe('pty2:some-other-generation:1')).toBe('unverifiable') + }) + + it.each(['pty-7', '', 'not-a-pty-id'])( + 'stays unverifiable for the id shape %s, which names no generation', + async (id) => { + expect(await probe(id)).toBe('unverifiable') + } + ) + + it('separates a real exit from a teardown removal in the same shutdown', async () => { + const exiting = await spawnPty() + reportExitOfLatestPty() + const tornDown = await spawnPty() + + const disposal = handler.dispose() + await vi.advanceTimersByTimeAsync(8_001) + await disposal + + // Both records are gone from the map; only one of them was ever watched ending. + expect(await probe(exiting.id)).toBe('exited') + expect(await probe(tornDown.id)).toBe('unverifiable') + }) + + it('keeps a record a shutdown could not kill live, rather than retiring it on paper', async () => { + mockPtySpawn.mockReturnValueOnce({ + ...mockPtyInstance, + kill: vi.fn(() => { + throw new Error('host refused kill') + }) + }) + const { id } = await spawnPty() + + const disposal = handler.dispose().catch((error: Error) => error) + await vi.advanceTimersByTimeAsync(8_001) + await disposal + + expect(await probe(id)).toBe('live') + }) + + it('stays unverifiable while a record is mid-teardown', async () => { + const { id } = await spawnPty() + tearDownRecordWithoutRemovingIt(id) + + // The pid is this test process, so without the guard the record reads as a live PTY. + expect(await probe(id)).toBe('unverifiable') + }) + + it('stays unverifiable after the listing sweeps a torn-down record away', async () => { + const { id } = await spawnPty() + tearDownRecordWithoutRemovingIt(id) + + await dispatcher.callRequest('pty.listProcesses', { includeForegroundProcessEvidence: false }) + + expect(handler.activePtyCount).toBe(0) + expect(await probe(id)).toBe('unverifiable') + }) + + it('never certifies an exit from a worktree removal it could not prove', async () => { + const { id } = await spawnPty({ cwd: process.cwd() }) + + // The removal asks for the kill; nothing reports the process ending, so the record stays. + const removal = handler.shutdownForWorktreePath(process.cwd()).catch((error: Error) => error) + await vi.advanceTimersByTimeAsync(8_001) + await removal + + expect(await probe(id)).toBe('live') + }) + + it('stays unverifiable while a revive holds the id, and answers for the new record after', async () => { + const { id } = await spawnPty({ cwd: process.cwd() }) + const state = (await dispatcher.callRequest('pty.serialize', { ids: [id] })) as string + reportExitOfLatestPty() + expect(await probe(id)).toBe('exited') + + // The revive is about to put a different process on this id, so the observation it would + // otherwise be certified from describes a process that no longer occupies it. + const revived = dispatcher.callRequest('pty.revive', { state }) + expect(await probe(id)).toBe('unverifiable') + await revived + + expect(await probe(id)).toBe('live') + }) + + it('evicts observations in the order it made them, not the order it was asked', async () => { + let oldest = '' + let newest = '' + for (let index = 0; index < OBSERVED_PTY_EXIT_HISTORY; index++) { + const { id } = await spawnPty() + reportExitOfLatestPty() + oldest ||= id + newest = id + } + + // Asking about the oldest entry at the cap must not renew it: retention is observation + // order, so the next observation still pushes exactly this one out. + expect(await probe(oldest)).toBe('exited') + expect(await probe(newest)).toBe('exited') + const overflowing = await spawnPty() + reportExitOfLatestPty() + + expect(await probe(overflowing.id)).toBe('exited') + // Evicted, not remembered as absent: a forgotten observation is unverifiable, never exited. + expect(await probe(oldest)).toBe('unverifiable') + }) + + it('forgets every observation when the relay restarts, even under the same mint epoch', async () => { + const { id } = await spawnPty() + reportExitOfLatestPty() + expect(await probe(id)).toBe('exited') + + // The ledger is per process by design. A relay that restarts between the observation and the + // question has nothing to certify from, which is the fail-closed direction. + const restartedDispatcher = createMockDispatcher() + const restarted = createTestPtyHandler(restartedDispatcher) + try { + expect(await probe(id, restartedDispatcher)).toBe('unverifiable') + } finally { + await restarted.dispose({ waitForPhysicalExit: false }).catch(() => {}) + } + }) + }) + + describe('incarnations sharing one id, which `pty.revive` re-creates a process under', () => { + it.each(['pty-7', 'pty2:previous-generation:99'])( + 'certifies the exit of the process it revived under %s, and only for itself', + async (id) => { + await dispatcher.callRequest('pty.revive', { state: serializedState(id) }) + expect(await probe(id)).toBe('live') + + reportExitOfLatestPty() + + expect(await probe(id)).toBe('exited') + // Reviving is what put the id in this relay's hands; whatever its shape, it names nothing + // on a relay that never held it. + const otherDispatcher = createMockDispatcher() + const other = createTestPtyHandler(otherDispatcher) + try { + expect(await probe(id, otherDispatcher)).toBe('unverifiable') + } finally { + await other.dispose({ waitForPhysicalExit: false }).catch(() => {}) + } + } + ) + + it('stays unverifiable for an id whose revive never created a record', async () => { + mockPtySpawn.mockImplementationOnce(() => { + throw new Error('spawn refused') + }) + + await expect( + dispatcher.callRequest('pty.revive', { state: serializedState('pty-77') }) + ).rejects.toThrow('spawn refused') + + expect(await probe('pty-77')).toBe('unverifiable') + }) + + it('does not certify a revived process with the exit of the one it replaced', async () => { + const id = 'pty-7' + await dispatcher.callRequest('pty.revive', { state: serializedState(id) }) + reportExitOfLatestPty() + expect(await probe(id)).toBe('exited') + + // Admitting a second incarnation clears that observation: it described the process this one + // replaced, so the id starts again with nothing to certify from. + await dispatcher.callRequest('pty.revive', { state: serializedState(id) }) + expect(await probe(id)).toBe('live') + + await failedShutdownThatKeepsTheRelayServing() + + // The revived record left the pool on a timeout, and its process is still running. + expect(() => process.kill(process.pid, 0)).not.toThrow() + expect(await probe(id)).toBe('unverifiable') + }) + + it('does not let a superseded incarnation certify the id once both records are gone', async () => { + const id = 'pty-7' + await dispatcher.callRequest('pty.revive', { state: serializedState(id) }) + const supersededExit = mockPtyInstance.onExit.mock.calls.at(-1)?.[0] as (event: { + exitCode: number + }) => void + await forgetRecordWithoutObservingItsExit(id) + await dispatcher.callRequest('pty.revive', { state: serializedState(id) }) + await forgetRecordWithoutObservingItsExit(id) + + // The first process really did end. The id belongs to a later one that nothing watched end, + // so an empty pool is not permission for this callback to answer for it. + supersededExit({ exitCode: 0 }) + + expect(await probe(id)).toBe('unverifiable') + }) + }) + + describe('owner entries evicted from the bounded map while a record outlives them', () => { + async function churnOwnersPastTheCap(): Promise { + for (let index = 0; index < OBSERVED_PTY_EXIT_HISTORY; index++) { + await spawnPty() + reportExitOfLatestPty() + } + } + + function ownerEntry(id: string): string | undefined { + return ( + handler as unknown as { ptyIdIncarnationOwners: { peek(key: string): string | undefined } } + ).ptyIdIncarnationOwners.peek(id) + } + + it('certifies a long-lived resident whose owner entry was evicted', async () => { + const { id } = await spawnPty() + const exit = mockPtyInstance.onExit.mock.calls.at(-1)?.[0] as (event: { + exitCode: number + }) => void + await churnOwnersPastTheCap() + expect(ownerEntry(id)).toBeUndefined() + expect(await probe(id)).toBe('live') + + // The record is still in the pool, so it speaks for the id without the owner map: a PTY + // that outlives 4,096 admissions must still certify its own exit. + exit({ exitCode: 0 }) + + expect(await probe(id)).toBe('exited') + }) + + it('withholds a late observation once both the record and its owner entry are gone', async () => { + const { id } = await spawnPty() + const exit = mockPtyInstance.onExit.mock.calls.at(-1)?.[0] as (event: { + exitCode: number + }) => void + await forgetRecordWithoutObservingItsExit(id) + await churnOwnersPastTheCap() + expect(ownerEntry(id)).toBeUndefined() + + // Forgetting fails closed: with no owner to compare against, nothing may record. + exit({ exitCode: 0 }) + + expect(await probe(id)).toBe('unverifiable') + }) + }) +}) diff --git a/src/relay/pty-handler-ssh-exit-certification.test.ts b/src/relay/pty-handler-ssh-exit-certification.test.ts new file mode 100644 index 000000000000..dfbdfc423524 --- /dev/null +++ b/src/relay/pty-handler-ssh-exit-certification.test.ts @@ -0,0 +1,179 @@ +// A shutdown timeout removes bookkeeping without proving the process exited. +import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest' + +const { mockPtySpawn, mockPtyInstance, mockCreateShellPromptReadinessProbe } = vi.hoisted(() => ({ + mockPtySpawn: vi.fn(), + mockCreateShellPromptReadinessProbe: vi.fn(), + mockPtyInstance: { + pid: process.pid, + process: 'zsh', + onData: vi.fn(), + onExit: vi.fn(), + write: vi.fn(), + resize: vi.fn(), + kill: vi.fn(), + clear: vi.fn(), + pause: vi.fn(), + resume: vi.fn() + } +})) + +vi.mock('node-pty', () => ({ spawn: mockPtySpawn })) + +vi.mock('../main/pty/posix-pty-process-groups', () => ({ + forceKillPosixPtyProcessGroups: vi.fn((_pid: number, fallback: () => void) => fallback()) +})) + +vi.mock('../main/shell-prompt-readiness-probe', () => ({ + createShellPromptReadinessProbe: mockCreateShellPromptReadinessProbe +})) + +import type { PtyHandler } from './pty-handler' +import { SshPtyProvider } from '../main/providers/ssh-pty-provider' +import { toAppSshPtyId } from '../shared/ssh-pty-id' +import { reconcileLegacyWorkerCandidate } from '../main/runtime/runtime-legacy-worker-terminal-recovery-candidate' +import type { LegacyWorkerRecoveryResolution } from '../main/runtime/runtime-legacy-worker-terminal-recovery-types' +import { + beginPtyHandlerTest, + createPtyRequestHelpers, + endPtyHandlerTest +} from './pty-handler-test-harness' +import type { MockDispatcher } from './pty-handler-test-harness' +import { parseRelayPtyMintEpoch } from '../shared/relay-pty-mint-epoch' + +describe('SSH exit certification across relay shutdown', () => { + let dispatcher: MockDispatcher + let handler: PtyHandler + let originalPlatform: PropertyDescriptor | undefined + + const { spawnPty } = createPtyRequestHelpers(() => dispatcher) + + beforeEach(() => { + ;({ dispatcher, handler, originalPlatform } = beginPtyHandlerTest({ + mockPtySpawn, + mockPtyInstance, + mockCreateShellPromptReadinessProbe + })) + }) + + afterEach(async () => { + await endPtyHandlerTest(handler, originalPlatform) + }) + + async function capabilities(): Promise> { + return (await dispatcher.callRequest('pty.getCapabilities', {})) as Record + } + + /** + * The rule an older client applied, reproduced here rather than shipped: an id missing from + * `pty.listProcesses` was an exit whenever its mint epoch matched one the host published. Mixed + * client and host versions are the normal state (docs/reference/remote-wire-compatibility.md), + * so what this host publishes has to keep that client unable to reach a verdict — a host cannot + * fix an old client, only decline to hand it the second half of the inference. + */ + async function inferLivenessFromListingAndEpoch(relayPtyId: string): Promise { + const listed = (await dispatcher.callRequest('pty.listProcesses', { + includeForegroundProcessEvidence: false + })) as { id?: unknown }[] + if (listed.some((session) => session.id === relayPtyId)) { + return true + } + const mintEpoch = parseRelayPtyMintEpoch(relayPtyId) + if (!mintEpoch) { + return null + } + return Object.values(await capabilities()).includes(mintEpoch) ? false : null + } + + it('publishes nothing that names the generation which minted its PTY ids', async () => { + const { id } = await spawnPty() + + const mintEpoch = parseRelayPtyMintEpoch(id) + expect(mintEpoch).toBeTruthy() + expect(Object.values(await capabilities())).not.toContain(mintEpoch) + }) + + it('leaves a shutdown-removed id unverifiable for a client that infers from listings', async () => { + const { id } = await spawnPty() + const disposal = handler.dispose() + await vi.advanceTimersByTimeAsync(8_001) + await disposal + + expect(await inferLivenessFromListingAndEpoch(id)).toBeNull() + }) + + it('keeps that inference unreachable when a failed kill leaves the relay serving', async () => { + const { id } = await spawnPty() + const failedKill = vi.fn<() => void>(() => { + throw new Error('host refused kill') + }) + mockPtySpawn.mockReturnValueOnce({ ...mockPtyInstance, kill: failedKill }) + await spawnPty() + const disposal = handler.dispose().catch((error: Error) => error) + await vi.advanceTimersByTimeAsync(8_001) + expect(await disposal).toMatchObject({ message: 'host refused kill' }) + failedKill.mockImplementation(() => {}) + + expect(() => process.kill(process.pid, 0)).not.toThrow() + expect(await inferLivenessFromListingAndEpoch(id)).toBeNull() + }) + it('does not certify exit after shutdown bookkeeping removes an unexited PTY', async () => { + const { id } = await spawnPty() + const provider = new SshPtyProvider('review-target', { + request: (method: string, params: Record) => + dispatcher.callRequest(method, params), + onNotification: () => () => {}, + onRequest: () => () => {} + } as never) + const appId = toAppSshPtyId('review-target', id) + expect(await provider.probePtyLiveness(appId)).toBe(true) + const disposal = handler.dispose() + await vi.advanceTimersByTimeAsync(8_001) + await disposal + expect(dispatcher._notifications.filter((event) => event.method === 'pty.exit')).toEqual([]) + expect(mockPtyInstance.kill).toHaveBeenCalled() + expect(() => process.kill(process.pid, 0)).not.toThrow() + expect(handler.activePtyCount).toBe(0) + expect(await provider.probePtyLiveness(appId)).toBeNull() + }) + + it('keeps missing shutdown records unverifiable when another failed kill leaves the relay serving', async () => { + const { id } = await spawnPty() + const failedKill = vi.fn<() => void>(() => { + throw new Error('host refused kill') + }) + mockPtySpawn.mockReturnValueOnce({ ...mockPtyInstance, kill: failedKill }) + await spawnPty() + const disposal = handler.dispose().catch((error: Error) => error) + await vi.advanceTimersByTimeAsync(8_001) + expect(await disposal).toMatchObject({ message: 'host refused kill' }) + expect(handler.activePtyCount).toBe(1) + const provider = new SshPtyProvider('review-target', { + request: (method: string, params: Record) => + dispatcher.callRequest(method, params), + onNotification: () => () => {}, + onRequest: () => () => {} + } as never) + expect(() => process.kill(process.pid, 0)).not.toThrow() + expect(dispatcher._notifications.filter((event) => event.method === 'pty.exit')).toEqual([]) + // Permit cleanup of the second fixture after the assertion. + failedKill.mockImplementation(() => {}) + const appId = toAppSshPtyId('review-target', id) + const verdict = await provider.probePtyLiveness(appId) + const pendingResolutions: LegacyWorkerRecoveryResolution[] = [] + const deferredDispatchIds = new Set() + await reconcileLegacyWorkerCandidate({ + controller: {} as never, + ports: { isPtyProvenAbsent: async () => verdict === false } as never, + options: {}, + candidate: { dispatchId: 'review-worker', ptyId: appId } as never, + workspace: {} as never, + resolvedWorktrees: [], + inventory: { livePtyIds: new Set() } as never, + pendingResolutions, + deferredDispatchIds + }) + expect(pendingResolutions).toEqual([]) + expect([...deferredDispatchIds]).toEqual(['review-worker']) + }) +}) diff --git a/src/relay/pty-handler-ssh-recovery-convergence.test.ts b/src/relay/pty-handler-ssh-recovery-convergence.test.ts new file mode 100644 index 000000000000..4cf750cde8bd --- /dev/null +++ b/src/relay/pty-handler-ssh-recovery-convergence.test.ts @@ -0,0 +1,209 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +const { mockPtySpawn, mockPtyInstance, mockCreateShellPromptReadinessProbe } = vi.hoisted(() => ({ + mockPtySpawn: vi.fn(), + mockCreateShellPromptReadinessProbe: vi.fn(), + mockPtyInstance: { + pid: process.pid, + process: 'zsh', + onData: vi.fn(), + onExit: vi.fn(), + write: vi.fn(), + resize: vi.fn(), + kill: vi.fn(), + clear: vi.fn(), + pause: vi.fn(), + resume: vi.fn() + } +})) +vi.mock('node-pty', () => ({ spawn: mockPtySpawn })) +vi.mock('../main/pty/posix-pty-process-groups', () => ({ + forceKillPosixPtyProcessGroups: vi.fn((_pid: number, fallback: () => void) => fallback()) +})) +vi.mock('../main/shell-prompt-readiness-probe', () => ({ + createShellPromptReadinessProbe: mockCreateShellPromptReadinessProbe +})) +import { + beginPtyHandlerTest, + endPtyHandlerTest, + type MockDispatcher +} from './pty-handler-test-harness' +import type { PtyHandler } from './pty-handler' +import { SshPtyProvider } from '../main/providers/ssh-pty-provider' +import { toAppSshPtyId } from '../shared/ssh-pty-id' +import { sshProviders } from '../main/ipc/pty/provider/registry' +import { probePtyLivenessFromRuntimeController } from '../main/ipc/pty/runtime/operations' +import { reconcileLegacyWorkerCandidate } from '../main/runtime/runtime-legacy-worker-terminal-recovery-candidate' +import type { LegacyWorkerRecoveryResolution } from '../main/runtime/runtime-legacy-worker-terminal-recovery-types' +import { OrchestrationDb } from '../main/runtime/orchestration/db' +const connection = 'review-convergence' +const deps = { getLocalPtyProviderStartupPromise: () => undefined } as never + +describe('SSH recovery convergence', () => { + let handler: PtyHandler + let dispatcher: MockDispatcher + let originalPlatform: PropertyDescriptor | undefined + let provider: SshPtyProvider + let request: ReturnType + beforeEach(() => { + ;({ handler, dispatcher, originalPlatform } = beginPtyHandlerTest({ + mockPtySpawn, + mockPtyInstance, + mockCreateShellPromptReadinessProbe + })) + request = vi.fn((method: string, params: Record) => + dispatcher.callRequest(method, params) + ) + provider = new SshPtyProvider(connection, { + request, + onNotification: () => () => {}, + onRequest: () => () => {} + } as never) + sshProviders.set(connection, provider) + }) + afterEach(async () => { + sshProviders.delete(connection) + provider.dispose() + await endPtyHandlerTest(handler, originalPlatform) + }) + async function candidateResult(ptyId: string, dispatchId = 'review-worker') { + const pendingResolutions: LegacyWorkerRecoveryResolution[] = [] + const deferredDispatchIds = new Set() + await reconcileLegacyWorkerCandidate({ + controller: {} as never, + ports: { + isPtyProvenAbsent: async () => + (await probePtyLivenessFromRuntimeController(deps, ptyId)) === false + } as never, + options: { connectionId: connection }, + candidate: { dispatchId, ptyId } as never, + workspace: {} as never, + resolvedWorktrees: [], + inventory: { livePtyIds: new Set() } as never, + pendingResolutions, + deferredDispatchIds + }) + return { pendingResolutions, deferredDispatchIds } + } + it.each([false, true])( + 'does not certify an unexited shutdown record; sibling kill failure=%s', + async (sibling) => { + const { id } = (await dispatcher.callRequest('pty.spawn', {})) as { id: string } + const failedKill = vi.fn<() => void>(() => { + throw new Error('host refused kill') + }) + if (sibling) { + mockPtySpawn.mockReturnValueOnce({ ...mockPtyInstance, kill: failedKill }) + await dispatcher.callRequest('pty.spawn', {}) + } + const disposal = handler.dispose().catch((error: Error) => error) + await vi.advanceTimersByTimeAsync(8001) + const outcome = await disposal + if (sibling) { + expect(outcome).toMatchObject({ message: 'host refused kill' }) + failedKill.mockImplementation(() => {}) + } + expect(() => process.kill(process.pid, 0)).not.toThrow() + expect(dispatcher._notifications.filter((n) => n.method === 'pty.exit')).toEqual([]) + const appId = toAppSshPtyId(connection, id) + expect(await probePtyLivenessFromRuntimeController(deps, appId)).toBeNull() + const result = await candidateResult(appId) + expect(result.pendingResolutions).toEqual([]) + expect([...result.deferredDispatchIds]).toEqual(['review-worker']) + } + ) + it('recovers a worker after the connected relay observed its physical exit while the client was absent', async () => { + const { id } = (await dispatcher.callRequest('pty.spawn', {})) as { id: string } + const exit = mockPtyInstance.onExit.mock.calls.at(-1)![0] as (event: { + exitCode: number + }) => void + exit({ exitCode: 0 }) + expect(handler.activePtyCount).toBe(0) + expect(dispatcher._notifications.some((n) => n.method === 'pty.exit')).toBe(true) + expect( + await dispatcher.callRequest('pty.listProcesses', { includeForegroundProcessEvidence: false }) + ).toEqual([]) + const appId = toAppSshPtyId(connection, id) + const db = new OrchestrationDb(':memory:') + try { + const task = db.createTask({ spec: 'recover terminated SSH worker' }) + const start = { + taskId: task.id, + startOptions: {}, + creator: { kind: 'system' as const }, + maxDepth: 9 + } + const { dispatch } = db.createStartingWorkerDispatch(start) + db.markWorkerDispatchReady(dispatch.id) + const rounds: Awaited>[] = [] + for (let sweep = 0; sweep < 3; sweep++) { + rounds.push(await candidateResult(appId, dispatch.id)) + } + expect(db.getWorkerDispatch(dispatch.id)?.state).toBe('ready') + expect(() => db.updateTaskStatus(task.id, 'ready')).toThrow() + expect(() => db.createStartingWorkerDispatch(start)).toThrow() + // The reachable owner needs to answer this instead of silently deferring every sweep. + expect(rounds.at(-1)?.pendingResolutions).toEqual([ + { candidate: { dispatchId: dispatch.id, ptyId: appId }, resolution: 'exited' } + ]) + } finally { + db.close() + } + }) + it('recovers a worker whose exit the relay observed after its shutdown wait gave up', async () => { + const { id } = (await dispatcher.callRequest('pty.spawn', {})) as { id: string } + const exit = mockPtyInstance.onExit.mock.calls.at(-1)![0] as (event: { + exitCode: number + }) => void + const failedKill = vi.fn<() => void>(() => { + throw new Error('host refused kill') + }) + mockPtySpawn.mockReturnValueOnce({ ...mockPtyInstance, kill: failedKill }) + await dispatcher.callRequest('pty.spawn', {}) + const disposal = handler.dispose().catch((error: Error) => error) + await vi.advanceTimersByTimeAsync(8001) + expect(await disposal).toMatchObject({ message: 'host refused kill' }) + failedKill.mockImplementation(() => {}) + const appId = toAppSshPtyId(connection, id) + expect(await probePtyLivenessFromRuntimeController(deps, appId)).toBeNull() + + exit({ exitCode: 0 }) + + // The owner watched this process end, after the timeout that removed its record. Recovery has + // to see that, or the sweep defers this worker forever. + const result = await candidateResult(appId) + expect(result.pendingResolutions).toEqual([ + { candidate: { dispatchId: 'review-worker', ptyId: appId }, resolution: 'exited' } + ]) + }) + it('does not retire a revived worker with the exit of the process it replaced', async () => { + const id = 'pty-7' + const state = JSON.stringify([{ id, pid: process.pid, cols: 80, rows: 24, cwd: process.cwd() }]) + await dispatcher.callRequest('pty.revive', { state }) + const supersededExit = mockPtyInstance.onExit.mock.calls.at(-1)![0] as (event: { + exitCode: number + }) => void + supersededExit({ exitCode: 0 }) + await dispatcher.callRequest('pty.revive', { state }) + const appId = toAppSshPtyId(connection, id) + expect(await probePtyLivenessFromRuntimeController(deps, appId)).toBe(true) + const failedKill = vi.fn<() => void>(() => { + throw new Error('host refused kill') + }) + mockPtySpawn.mockReturnValueOnce({ ...mockPtyInstance, kill: failedKill }) + await dispatcher.callRequest('pty.spawn', {}) + const exitsBefore = dispatcher._notifications.filter((n) => n.method === 'pty.exit').length + const disposal = handler.dispose().catch((error: Error) => error) + await vi.advanceTimersByTimeAsync(8001) + expect(await disposal).toMatchObject({ message: 'host refused kill' }) + failedKill.mockImplementation(() => {}) + + // Nothing watched the revived process end, and its pid is this live test process. + expect(() => process.kill(process.pid, 0)).not.toThrow() + expect(dispatcher._notifications.filter((n) => n.method === 'pty.exit')).toHaveLength( + exitsBefore + ) + const result = await candidateResult(appId) + expect(result.pendingResolutions).toEqual([]) + expect([...result.deferredDispatchIds]).toEqual(['review-worker']) + }) +}) diff --git a/src/relay/pty-handler.ts b/src/relay/pty-handler.ts index cdf436bca2ab..9a98a2176c71 100644 --- a/src/relay/pty-handler.ts +++ b/src/relay/pty-handler.ts @@ -34,8 +34,10 @@ import { resolveRelaySpawnCwd, type RelaySpawnCwdResolution } from './pty-spawn-cwd' +import { BoundedMap } from '../shared/bounded-map' import { PhysicalExitTracker } from '../shared/physical-exit-tracker' import { PTY_ATTACH_PROVEN_EXITED_MARKER } from '../shared/pty-attach-absence-evidence' +import { toRelayPtyIdWithMintEpoch } from '../shared/relay-pty-mint-epoch' import { SHELL_READY_MARKER_PREFIX } from '../main/shell-ready-marker-scanner' import { createShellStartupOutputScanState, @@ -353,6 +355,9 @@ const STARTUP_COMMAND_SHELL_READY_FALLBACK_MS = 1500 const RENDERER_SHELL_READY_RETENTION_MS = 15_000 const PTY_FORCE_KILL_RETRY_DELAY_MS = 250 const PTY_FORCE_KILL_MAX_ATTEMPTS = 2 + +/** Cap on remembered observed exits. Eviction answers unverifiable, never a false exit. */ +export const OBSERVED_PTY_EXIT_HISTORY = 4_096 const ALLOWED_SIGNALS = new Set([ 'SIGINT', 'SIGTERM', @@ -533,6 +538,40 @@ export class PtyHandler { private interactiveOutputCharsByPty = new Map() private pendingSpawnCount = 0 private pendingReviveIds = new Set() + /** + * Ids this relay retired having OBSERVED the process end: node-pty's `onExit`, or a pid probe + * that answered ESRCH. Every other removal is bookkeeping — a shutdown that gave up waiting for + * an uninterruptible child still deletes the record — so absence from `this.ptys` cannot be read + * as an exit. This is the positive half, recorded where the observation happens, and a liveness + * probe certifies an exit from nothing else. + * + * Only ids this relay itself held can enter it, so "this relay never had that id" is the same + * answer as "never watched it end": unverifiable. Deliberately in memory and per process — a + * relay that restarts between the observation and the question forgets, and forgetting is the + * fail-closed direction, where persisting it would have to survive the ambiguity of the crash + * that lost the records. Bounded for the same reason: an evicted id answers unverifiable + * (docs/reference/ssh-execution-boundary.md). + * + * An entry describes ONE incarnation. `pty.revive` re-creates a process under an id this relay + * already watched end, so admission clears the id's entry and only the incarnation that currently + * owns the id may write one; see {@link ptyIdIncarnationOwners}. + */ + private readonly observedPtyExitIds = new BoundedMap({ + maxEntries: OBSERVED_PTY_EXIT_HISTORY + }) + /** + * The incarnation that most recently took each id. `this.ptys` answers that question only while + * the record is still in the pool, and the observation this ledger needs can arrive after it left + * — a shutdown that stopped waiting removes the record, and the process ends afterwards. Without + * this, that late callback could not be told apart from one belonging to a process a `pty.revive` + * has since replaced under the same id, and would certify the replacement's exit. + * + * Bounded like the ledger, and forgetting fails closed: an unrecognized incarnation records + * nothing, so the id stays unverifiable. + */ + private readonly ptyIdIncarnationOwners = new BoundedMap({ + maxEntries: OBSERVED_PTY_EXIT_HISTORY + }) private creationFenced = false private pendingCreationDrainResolvers = new Set<() => void>() private worktreeRemovalCoordinator: RelayPtyWorktreeRemovalCoordinator | null = null @@ -765,6 +804,23 @@ export class PtyHandler { } } + /** Called only where the process end was observed; see {@link observedPtyExitIds}. */ + private recordObservedPtyExit(id: string): void { + this.observedPtyExitIds.set(id, true) + } + + /** + * Whether this record still speaks for its id. An exit observed for a superseded incarnation is + * evidence about the process that ended, never about the one occupying the id now. `peek` rather + * than `get` so asking does not extend an owner's retention past older ones. + */ + private stillOwnsPtyId(managed: ManagedPty): boolean { + return ( + this.ptys.get(managed.id) === managed || + this.ptyIdIncarnationOwners.peek(managed.id) === managed.incarnationId + ) + } + // Why: the sole removal path, so the three exit routes can't drift on who announces an empty pool. private removePty(id: string): void { this.ptys.delete(id) @@ -941,6 +997,11 @@ export class PtyHandler { /** Wire onData/onExit listeners for a managed PTY and store it. */ private wireAndStore(managed: ManagedPty): void { managed.physicalExit = new PhysicalExitTracker() + // Why here: this is the single admission site, so it is where an id changes hands. A new + // incarnation inherits no exit evidence — the observation described the process it replaced — + // and becomes the only one whose exit may be recorded for this id. + this.observedPtyExitIds.delete(managed.id) + this.ptyIdIncarnationOwners.set(managed.id, managed.incarnationId) this.ptys.set(managed.id, managed) // Why: a PTY joining the pool under this paneKey means the surface exists again (reopened pane // or revive), so a prior retirement no longer describes anything and must not mute its hooks. @@ -1007,6 +1068,12 @@ export class PtyHandler { }) managed.pty.onExit(({ exitCode }: { exitCode: number }) => { managed.physicalExit?.markExited() + // Why before the disposed guard: this callback IS the observation. Teardown that stopped + // waiting removed the record and marked it disposed, but the process ending afterwards is + // still this relay watching it end, and dropping it left the worker unverifiable forever. + if (this.stillOwnsPtyId(managed)) { + this.recordObservedPtyExit(managed.id) + } if (managed.disposed) { return } @@ -1093,7 +1160,11 @@ export class PtyHandler { // Additive capability: clients may request the no-process-table inventory // projection and consume fenced inspect evidence on this host. foregroundProcessEvidenceVersion: 1 + // Deliberately publishes no mint epoch. Pairing one with `pty.listProcesses` is how a client + // used to read an id's absence as an exit, and absence is also what a shutdown that stopped + // waiting for an uninterruptible child leaves behind. `pty.probeLiveness` is the only answer. })) + this.dispatcher.onRequest('pty.probeLiveness', async (params) => this.probeLiveness(params)) this.dispatcher.onRequest('pty.listProcesses', (params) => this.listProcesses(params)) this.dispatcher.onRequest('pty.getDefaultShell', async () => resolveDefaultShell()) this.dispatcher.onRequest('pty.serialize', (p) => this.serialize(p)) @@ -1863,7 +1934,7 @@ export class PtyHandler { const shell = resolvedShellOverride || requestedEnvShell || resolveDefaultShell() let id: string do { - id = `pty2:${encodeURIComponent(this.ptyIdMintEpoch)}:${this.nextId++}` + id = toRelayPtyIdWithMintEpoch(this.ptyIdMintEpoch, this.nextId++) } while (this.ptys.has(id) || this.pendingReviveIds.has(id)) // Why: augmenter values override renderer env so remote paths and hook coords win over local userData. @@ -2464,6 +2535,9 @@ export class PtyHandler { if (!managed.pty.pid || isProcessAlive(managed.pty.pid)) { return false } + // Recorded here rather than inside the shared reap: this is the line that observed the pid was + // gone. The reap's other caller is tidying a torn-down record and watched nothing. + this.recordObservedPtyExit(managed.id) this.reapExitedPty(managed, 'exited') return true } @@ -2600,6 +2674,40 @@ export class PtyHandler { } } + /** + * Exact-id liveness, answered from this relay's own lifecycle state. The client cannot compute + * this: absence from an inventory means "no record", and only the owner knows whether a record + * left because the process ended or because teardown stopped waiting for it + * (docs/reference/ssh-execution-boundary.md). + * + * `exited` requires an observation — a live record whose pid probes absent, or an id in + * {@link observedPtyExitIds}. Everything else is `unverifiable`, the execution boundary's word + * for doubt: an id this relay never held or never saw end, a revive in flight, and a record + * mid-teardown. + */ + private async probeLiveness( + params: Record + ): Promise<{ status: 'live' | 'exited' | 'unverifiable' }> { + const id = typeof params.id === 'string' ? params.id : '' + const managed = this.ptys.get(id) + if (managed) { + if (managed.disposed) { + return { status: 'unverifiable' } + } + return { status: this.reapPtyProvenExited(managed) ? 'exited' : 'live' } + } + if (this.pendingReviveIds.has(id)) { + return { status: 'unverifiable' } + } + // The ledger is the only gate, deliberately. A blanket "shutdown has started" refusal would + // also mask a wrong entry in it, since teardown only ever runs behind that fence — and it + // would withhold exits this relay really did observe. Keys are whole ids of records this relay + // held: ids minted with this generation's epoch, plus any id `pty.revive` re-created a process + // under. An id another generation minted and this one never revived is therefore absent, and + // absent is unverifiable. + return { status: this.observedPtyExitIds.has(id) ? 'exited' : 'unverifiable' } + } + private async hasChildProcesses(params: Record): Promise { const id = params.id as string const managed = this.ptys.get(id) diff --git a/src/shared/relay-pty-mint-epoch.ts b/src/shared/relay-pty-mint-epoch.ts new file mode 100644 index 000000000000..77b140c169cd --- /dev/null +++ b/src/shared/relay-pty-mint-epoch.ts @@ -0,0 +1,33 @@ +/** + * A relay PTY id carries the mint epoch of the relay process that allocated it, so ids from + * different generations of the same relay cannot collide. The shape lives here so the relay that + * mints it and anything that reads it back cannot drift. + * + * The epoch certifies nothing on its own, and this relay no longer publishes it. Pairing it with + * `pty.listProcesses` used to read an id's absence from a same-epoch listing as an exit, which is + * unsound: the relay also removes a record it never watched end, so a shutdown that gave up waiting + * for an uninterruptible child produced that same absence. Only the owner's exact-id + * `pty.probeLiveness` answers whether a process ended (docs/reference/ssh-execution-boundary.md). + */ +const MINT_EPOCH_PTY_ID_PREFIX = 'pty2:' + +export function toRelayPtyIdWithMintEpoch(mintEpoch: string, sequence: number): string { + return `${MINT_EPOCH_PTY_ID_PREFIX}${encodeURIComponent(mintEpoch)}:${sequence}` +} + +/** Null for a legacy `pty-N` id, which names no generation. */ +export function parseRelayPtyMintEpoch(relayPtyId: string): string | null { + if (!relayPtyId.startsWith(MINT_EPOCH_PTY_ID_PREFIX)) { + return null + } + const remainder = relayPtyId.slice(MINT_EPOCH_PTY_ID_PREFIX.length) + const sequenceSeparator = remainder.lastIndexOf(':') + if (sequenceSeparator <= 0) { + return null + } + try { + return decodeURIComponent(remainder.slice(0, sequenceSeparator)) || null + } catch { + return null + } +}