From 39096a55a4926792a668d5b078942295adc4947d Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 02:01:01 -0600 Subject: [PATCH] fix(runner): answer lease renewal outside the PRP frame chain and name lease loss (#19419) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native runs in a managed sandbox talk to the server over one authenticated PRP connection. The runner (`paperclip-runnerd`) renews its connection lease on that connection. > - The server serializes every inbound PRP frame through one promise chain per connection. Before this change, `lease_renew` waited in that chain behind earlier events. An event whose business commit stalls (for example the commit of a tool event) therefore also stalls the lease reply. The runner then stops polling its provider (ACK backpressure) and exits with "active connection lease expired" when the lease runs out. > - When that happens, the server sees only a socket close. The run log and the terminal error say "transport loss", not "lease expired". > - A follow-up from #19407 asked for renewal to be handled outside the chain and for lease loss to be named. > - This pull request answers `lease_renew` (and `pong`) as soon as the frame is decrypted. It keeps in-order decryption and the order of state-changing frames. It also records why a lease was lost and reports it as `runner_connection_lease_expired`. > - The benefit is that a slow event commit can no longer starve lease renewal, and a lease loss is visible in the run log and error code. ## Linked Issues or Issue Description Refs #19407 **What happened?** On staging, native runs with ACPX providers (Grok, Kimi, and Pi with OpenRouter/DeepSeek) ran a long foreground shell command that printed nothing (a 15–20 minute `sleep 60` loop). Each run went silent soon after `tool.execution.started`. The command finished in the sandbox, but the run never got the completion and stayed `running` until someone cancelled it. It happened in 3 of 3 attempts. The same task on Codex kept reporting activity and succeeded. **What did you expect to happen?** A quiet tool call keeps its runner connection and its lease. If the connection or lease is lost, the run log and the error code say why. **Steps to reproduce** 1. Assign a native ACPX agent a task that runs a foreground shell loop with no output for longer than 2 minutes. 2. Observe that the run goes silent and does not finish after the command ends. **Root cause analysis** - The ACPX path emits no PRP frames during a silent tool call. The runner has no timer, and the agent sends no `tool_call_update`. Codex keeps the connection busy, because its model polls the running command. Silence is the only adapter-specific factor that I found. I found no ACPX code path in runnerd that blocks the main loop: polls use a 1 ms timeout, and synchronous sidecar requests run only for server commands. - The staging build (`de1143448b`) did not include #19407. That build had no server pings. With a 60-minute connection lease (the default for `runner-prp-coordinator` and `DurablePrpCodexTransport`), the runner renews only every 30 minutes. A fully silent connection that goes through the sandbox preview proxy therefore carried no traffic at all. A route that the proxy drops without a FIN then becomes half-open forever. #19407 added 20-second pings and a 90-second termination, which closes that window. - The lease hypothesis from #19407 is real but secondary. The 60-second lease applies only when the TTL is not set. Renewal waited behind any slow event commit, and the server could not report lease loss. This pull request fixes both. ## What Changed - `DurablePrpControlPlane.attachWireConnection` decrypts each secure frame when it arrives, in counter order. Frames that arrive before the channel is secure, or behind such a frame, still decrypt in the chain. - `lease_renew` and `pong` are handled at once, outside the per-connection chain. Admission against persisted authority is unchanged. Events, command results, and handoff frames keep their serialized order. - `#renewLease` persists the renewed expiry with `save()`, so an event write that is already staged is absorbed into the same write. Before, the renewal used `commit()`, which would invalidate that staged event and close a healthy connection when renewal runs concurrently. - New `connectionLeaseFailure` getter (exported type `DurablePrpConnectionLeaseFailure`). It records `connection_lease_expired` when an authenticated connection is refused or closes with an expired lease. A new authenticated connection clears it. A credential revoked by a warm handoff is not reported, because a replacement runner owns the run. - `connectRunnerPrpIngress` reads that reason after a close. It logs `state=reconnecting failure=runner_connection_lease_expired` in the run log. If reconnect grace runs out, it fails with that code instead of `runner_ingress_unavailable` or `runner_ingress_auth_failed`, and keeps the reconnect error as the cause. - `packages/paperclip-runner/docs/durable-recovery.md` documents the new behavior. ## Verification - `pnpm --filter @paperclipai/paperclip-runner exec vitest run src/control-plane` (220 passed). - New: "answers lease renewal while an earlier event commit is still in flight". It was red before the fix (no `lease_renewed` while `onCommittedEvent` waits), and it is green now. It also checks that the event later commits and ACKs, and that the renewed expiry survives. - New: "keeps a staged tool-progress write when a lease renewal lands mid-write". It holds a file-backed `tool.execution.progressed` write at `fsync`. It is red with the old `commit()` renewal (the event is dropped and the connection closes), and green with `save()`. - New: "reports an expired connection lease instead of closing silently". It was red before the fix and is green now. - `pnpm --filter @paperclipai/server exec vitest run src/realtime/runner-prp-outbound.test.ts src/realtime/runner-prp-ws.test.ts src/services/native-runtime/runner-prp-coordinator.test.ts` (all passed). - New: "names an expired runner connection lease in the run log and terminal error when reconnect is refused/unauthenticated". It was red before the fix and is green now. - `src/live/runnerd-codex-transport.test.ts` has the same result with and without this change with a locally built `paperclip-runnerd`: the same 48 tests fail locally on both, for environment reasons, and the other 207 pass. ## Risks - Lease renewal now runs concurrently with an in-flight event handler. The handler builds its state candidate at commit time, and `save()` absorbs a staged event write. A new regression test covers the file-backed race. External checkpoint persistence commits events synchronously, and the renewal reply still waits for the checkpoint flush. - Renewal still stays silent while a warm handoff or a terminal lifecycle command is pending. The runner treats any other control frame inside those fences as fatal, so this pull request does not change that behavior. - No protocol, schema, or contract-digest change. The runner is unchanged. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (shell, file edits, test runs) and a read-only exploration subagent on the same model. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- .../paperclip-runner/docs/durable-recovery.md | 13 ++ .../durable-prp-control-plane.test.ts | 147 +++++++++++++++ .../durable-prp-control-plane.ts | 176 +++++++++++++++--- packages/paperclip-runner/src/index.ts | 1 + .../src/realtime/runner-prp-outbound.test.ts | 63 +++++++ server/src/realtime/runner-prp-outbound.ts | 23 ++- 6 files changed, 399 insertions(+), 24 deletions(-) diff --git a/packages/paperclip-runner/docs/durable-recovery.md b/packages/paperclip-runner/docs/durable-recovery.md index 10b19629cb7..9375cd96af9 100644 --- a/packages/paperclip-runner/docs/durable-recovery.md +++ b/packages/paperclip-runner/docs/durable-recovery.md @@ -41,6 +41,19 @@ lease without replacing its process, provider, thread, turn, token, or epoch. Retries of the same observed expiry replay the persisted extension. Renewal never admits an expired, revoked, or differently bound credential. +The controller answers `lease_renew` (and `pong`) as soon as the frame is +decrypted, outside the per-connection chain that serializes events, command +results, and handoff frames. A slow business commit for an earlier event, such +as a tool event during a long silent tool call, therefore cannot starve +renewal. Frames still decrypt in arrival order, and the renewed expiry is +persisted (absorbing any event write already staged) before the reply is sent. +When an authenticated connection closes or is refused because its lease has +expired, the controller records that reason (`connectionLeaseFailure`). +Provider ingress reports it in the run log and, if reconnect grace runs out, +as the `runner_connection_lease_expired` failure code instead of a generic +transport loss. A credential revoked by a warm handoff is not reported, because +a replacement runner owns the run. + If a reply is lost, reconnect authentication may reconcile a later expiry only when this runner has an outstanding renewal on that same credential. Warm handoff receipts continue to bind exact expiry and renewal pauses during their diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts index 0e858fa1955..13aa10b34d1 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts @@ -1399,6 +1399,153 @@ it("renews one authenticated connection for three weeks without replacing its au } }, 30_000); +async function receiveSecureWithin(client: AuthenticatedClient, ms: number): Promise | "timeout" | null> { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + receiveSecure(client), + new Promise<"timeout">(resolveTimeout => { timer = setTimeout(() => resolveTimeout("timeout"), ms); }), + ]); + } finally { + if (timer) clearTimeout(timer); + } +} + +function credentialEvent(sourceSeq: number): Record { + const frame = semanticInputEvent(sourceSeq); + const event = frame.payload as Record; + event.sourceEventId = `slow-commit-${sourceSeq}`; + event.eventType = "auth.credentials.updated"; + event.payload = { schema: "paperclip.auth.credentials.updated.v1", provider: "codex", + bindingId: "harness-auth", revision: sourceSeq, detection: "credential_store" }; + return frame; +} + +it("answers lease renewal while an earlier event commit is still in flight", async () => { + const root = mkdtempSync(resolve(tmpdir(), "runner-lease-renewal-slow-commit-")); + let release!: () => void; + const barrier = new Promise(resolveCommit => { release = resolveCommit; }); + const commits = vi.fn(async () => barrier); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, + expectedRunnerVersion, expectedRunnerDigest, onCommittedEvent: commits }); + try { + await core.start(); + const client = (await authenticate(core, core.issueBootstrapTicket()))!; + const expiry = Number((client.welcome.payload as Record).connectionLeaseExpiresAtUnixMs); + // A tool event whose business commit stalls must not starve the lease. + sendSecure(client, credentialEvent(1)); + await vi.waitFor(() => expect(commits).toHaveBeenCalledTimes(1)); + sendSecure(client, renewalRequest(client, expiry)); + const reply = await receiveSecureWithin(client, 2_000); + expect(reply).not.toBe("timeout"); + expect(reply).toMatchObject({ kind: "lease_renewed", payload: { previousExpiresAtUnixMs: expiry } }); + const renewed = Number(((reply as Record).payload as Record).connectionLeaseExpiresAtUnixMs); + expect(renewed).toBeGreaterThanOrEqual(expiry); + expect(core.store.state.ackedSourceSeq).toBe(0); + // Ordering for state-changing frames is unchanged: the event still + // commits, ACKs, and the renewed lease survives that later commit. + release(); + expect(await receiveSecure(client)).toMatchObject({ kind: "ack", payload: { ackedSourceSeq: 1 } }); + sendSecure(client, credentialEvent(2)); + expect(await receiveSecure(client)).toMatchObject({ kind: "ack", payload: { ackedSourceSeq: 2 } }); + const saved = JSON.parse(readFileSync(core.store.path, "utf8")); + expect(saved.ackedSourceSeq).toBe(2); + expect(saved.leases[Object.keys(saved.leases)[0]!].expiresAtUnixMs).toBe(renewed); + client.socket.destroy(); + } finally { + release(); + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } +}); + +it("keeps a staged tool-progress write when a lease renewal lands mid-write", async () => { + const root = mkdtempSync(resolve(tmpdir(), "runner-lease-renewal-staged-write-")); + let release!: () => void; + const barrier = new Promise(resolveWrite => { release = resolveWrite; }); + let staged = false; + let hold = true; + const originalOpen = fsPromises.open; + const open = vi.spyOn(fsPromises, "open").mockImplementation(async (...args) => { + const file = await originalOpen(...args); + if (hold && String(args[0]).includes(".control-plane-state.json.")) { + hold = false; + const sync = file.sync.bind(file); + file.sync = async () => { + staged = true; + await barrier; + await sync(); + }; + } + return file; + }); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, + expectedRunnerVersion, expectedRunnerDigest }); + try { + await core.start(); + const client = (await authenticate(core, core.issueBootstrapTicket()))!; + const expiry = Number((client.welcome.payload as Record).connectionLeaseExpiresAtUnixMs); + const event = semanticInputEvent(); + const payload = event.payload as Record; + payload.eventType = "tool.execution.progressed"; + payload.priority = 2; + payload.payload = { schema: "paperclip.tool.execution.v1", executionId: "long-shell", + transport: "process", operation: "execute", status: "running", output: "", + outputBytes: 0, outputTruncated: false, outputDigest: null }; + expect(validatePrpEvent(payload)).toMatchObject({ ok: true }); + sendSecure(client, event); + await vi.waitFor(() => expect(staged).toBe(true)); + sendSecure(client, renewalRequest(client, expiry)); + const reply = await receiveSecureWithin(client, 2_000); + expect(reply).toMatchObject({ kind: "lease_renewed" }); + release(); + // The renewal absorbed the admitted event instead of invalidating it. + expect(await receiveSecure(client)).toMatchObject({ kind: "ack", payload: { ackedSourceSeq: 1 } }); + const saved = JSON.parse(readFileSync(core.store.path, "utf8")); + expect(saved.ackedSourceSeq).toBe(1); + expect(saved.leases[Object.keys(saved.leases)[0]!].expiresAtUnixMs).toBe( + ((reply as Record).payload as Record).connectionLeaseExpiresAtUnixMs); + client.socket.destroy(); + } finally { + release(); + open.mockRestore(); + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } +}); + +it("reports an expired connection lease instead of closing silently", async () => { + const root = mkdtempSync(resolve(tmpdir(), "runner-lease-expired-")); + let now = Date.now(); + const clock = vi.spyOn(Date, "now").mockImplementation(() => now); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, + expectedRunnerVersion, expectedRunnerDigest, connectionLeaseTtlMs: 60_000 }); + try { + await core.start(); + const client = (await authenticate(core, core.issueBootstrapTicket()))!; + const expiry = Number((client.welcome.payload as Record).connectionLeaseExpiresAtUnixMs); + expect(core.connectionLeaseFailure).toBeNull(); + now = expiry + 1; + sendSecure(client, renewalRequest(client, expiry)); + expect(await receiveSecureWithin(client, 2_000)).toBeNull(); + expect(core.connectionLeaseFailure).toMatchObject({ + code: "connection_lease_expired", + leaseId: client.welcome.connectionLeaseId, + expiresAtUnixMs: expiry, + }); + // A fresh authenticated connection clears the stale reason. + now = Date.now(); + const replacement = (await authenticate(core, core.issueBootstrapTicket()))!; + expect(core.connectionLeaseFailure).toBeNull(); + replacement.socket.destroy(); + client.socket.destroy(); + } finally { + clock.mockRestore(); + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } +}); + it("reports how long the authenticated Runner has been disconnected", async () => { const root = mkdtempSync(resolve(tmpdir(), "runner-disconnect-age-")); const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest }); diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts index cd29daad233..147267e0b82 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts @@ -1774,6 +1774,18 @@ class DurableCoreStore { } } +/** Why an authenticated runner connection lost its lease authority. */ +export interface DurablePrpConnectionLeaseFailure { + readonly code: "connection_lease_expired"; + readonly leaseId: string; + readonly expiresAtUnixMs: number; + readonly observedAtUnixMs: number; +} + +type SerializedFrame = + | { readonly raw: unknown } + | { readonly decrypted: Record }; + /** Reason supplied when a transport-neutral PRP peer closes. */ export interface TransportCloseReason { readonly code?: number; @@ -2004,6 +2016,7 @@ export class DurablePrpControlPlane { // next authentication. Diagnostic only: it never gates authority. #runnerDisconnectedAt: number | null = null; #connectionProcessing = new Map>(); + #connectionLeaseFailure: DurablePrpConnectionLeaseFailure | null = null; #pendingSemanticCalls = new Set(); #semanticCallbacksRetired = false; #semanticResultPersistenceFailed = false; @@ -2912,11 +2925,40 @@ export class DurablePrpControlPlane { attachWireConnection(wire: PrpWireConnection): PrpWireAttachment { let connection!: AuthorityConnection; let processing = Promise.resolve(); + // Frames received before the secure channel existed (or behind such a + // frame) are decrypted in order by the serialized chain. Once none are + // pending, frames are decrypted on arrival so that lease renewal can be + // answered without waiting for an earlier event's business commit. + let undecryptedFrames = 0; connection = new AuthorityConnection({ wire, onJson: (value) => { + let frame: SerializedFrame; + if (connection.secureChannel !== null && undecryptedFrames === 0) { + const decrypted = this.#decryptOnArrival(connection, value); + if (decrypted === null) return; + if (decrypted.kind === "lease_renew" || decrypted.kind === "pong") { + // Lease renewal is liveness, not a state-changing frame. It must + // never queue behind a slow event commit (a long silent tool call + // otherwise lets the runner's lease lapse). It still commits the + // renewed expiry durably before its reply, and its admission is + // checked against persisted authority exactly as before. + try { + this.#handleDecrypted(connection, decrypted); + } catch { + connection.close(); + } + return; + } + frame = { decrypted }; + } else { + undecryptedFrames += 1; + frame = { raw: value }; + } processing = processing - .then(() => this.#handleJson(connection, value)) + .then(() => "raw" in frame + ? this.#handleJson(connection, frame.raw).finally(() => { undecryptedFrames -= 1; }) + : this.#handleDecryptedFrame(connection, frame.decrypted)) .catch(() => connection.close()); const tail = processing; this.#connectionProcessing.set(connection, tail); @@ -2927,6 +2969,7 @@ export class DurablePrpControlPlane { void tail.then(release, release); }, onClose: () => { + this.#recordLeaseFailureIfLapsed(connection); this.#connections.delete(connection); if (connection.secureChannel !== null && this.activeRunnerConnectionCount() === 0) this.#runnerDisconnectedAt ??= Date.now(); @@ -2958,6 +3001,43 @@ export class DurablePrpControlPlane { connection.close(); return; } + await this.#handleDecryptedFrame(connection, envelope); + } + + /** Decrypt one secure frame in arrival order; null when it closed the connection. */ + #decryptOnArrival( + connection: AuthorityConnection, + wire: unknown, + ): Record | null { + try { + this.#store.assertWritable(); + return decryptSecureJson(connection.secureChannel!, wire); + } catch { + try { + this.#store.state.malformedFrames += 1; + this.#store.save(); + } finally { + connection.close(); + } + return null; + } + } + + /** Handle one decrypted renewal or pong immediately, outside the frame chain. */ + #handleDecrypted( + connection: AuthorityConnection, + envelope: Record, + ): void { + this.#store.assertWritable(); + if (!this.#admitEnvelopeVersion(connection, envelope)) return; + if (!this.#admitAuthenticatedFrame(connection)) return; + if (envelope.kind === "lease_renew") this.#renewLease(connection, envelope); + } + + #admitEnvelopeVersion( + connection: AuthorityConnection, + envelope: Record, + ): boolean { const envelopeVersion = envelope.version; const expectedVersion = connection.lease?.protocolVersion ?? @@ -2972,8 +3052,64 @@ export class DurablePrpControlPlane { : envelopeVersion !== expectedVersion) ) { connection.close(); - return; + return false; } + return true; + } + + /** + * Admit every post-handshake frame against persisted authority before + * dispatch, including lease_renew. Renewal cannot revive a revoked or + * expired credential or bypass changes to its persisted binding. An expired + * lease is recorded so the owner can report why the runner left. + */ + #admitAuthenticatedFrame(connection: AuthorityConnection): boolean { + if ( + connection.secureChannel === null || + connection.lease === null || + canonicalJson(this.#store.state.leases[connection.lease.credentialId]) !== + canonicalJson(connection.lease) || + connection.lease.revokedAt !== null || + connection.lease.expiresAtUnixMs <= Date.now() + ) { + this.#recordLeaseFailureIfLapsed(connection); + connection.close(); + return false; + } + return true; + } + + #recordLeaseFailureIfLapsed(connection: AuthorityConnection): void { + const lease = connection.lease; + if (lease === null || connection.secureChannel === null) return; + const now = Date.now(); + // Only expiry is reported. A revoked credential is the expected outcome of + // a warm handoff to a replacement runner, not a lost lease. + if (lease.expiresAtUnixMs > now) return; + this.#connectionLeaseFailure = { + code: "connection_lease_expired", + leaseId: lease.leaseId, + expiresAtUnixMs: lease.expiresAtUnixMs, + observedAtUnixMs: now, + }; + } + + /** + * Why the most recent authenticated runner connection lost its authority, + * or null while the current connection's lease is valid. A runner whose + * lease lapses exits without another authenticated frame, so the socket + * close alone would otherwise look like an ordinary transport loss. + */ + get connectionLeaseFailure(): DurablePrpConnectionLeaseFailure | null { + return this.#connectionLeaseFailure === null ? null : { ...this.#connectionLeaseFailure }; + } + + async #handleDecryptedFrame( + connection: AuthorityConnection, + envelope: Record, + ): Promise { + this.#store.assertWritable(); + if (!this.#admitEnvelopeVersion(connection, envelope)) return; const kind = envelope.kind; if (connection.secureChannel === null && kind === "auth_hello") { this.#authHello(connection, envelope); @@ -2983,20 +3119,9 @@ export class DurablePrpControlPlane { await this.#authResponse(connection, envelope); return; } - // Admit every post-handshake frame against persisted authority before - // dispatch, including lease_renew. Renewal cannot revive a revoked or - // expired credential or bypass changes to its persisted binding. - if ( - connection.secureChannel === null || - connection.lease === null || - canonicalJson(this.#store.state.leases[connection.lease.credentialId]) !== - canonicalJson(connection.lease) || - connection.lease.revokedAt !== null || - connection.lease.expiresAtUnixMs <= Date.now() - ) { - connection.close(); - return; - } + if (!this.#admitAuthenticatedFrame(connection)) return; + // After admission the lease is non-null; keep the narrowed reference. + if (connection.lease === null) return; if (kind === "lease_renew") { this.#renewLease(connection, envelope); return; @@ -3108,15 +3233,21 @@ export class DurablePrpControlPlane { // Repeating a request after a lost reply replays the persisted expiry. // It never extends a credential twice for the same observed generation. if (expectedExpiry === lease.expiresAtUnixMs) { - const candidate = structuredClone(this.#store.state); - const renewed = candidate.leases[lease.credentialId]!; - renewed.expiresAtUnixMs = Math.max( + const expiresAtUnixMs = Math.max( lease.expiresAtUnixMs, Date.now() + this.#connectionLeaseTtlMs, ); - renewed.expiresAt = new Date(renewed.expiresAtUnixMs).toISOString(); - candidate.lastLeaseExpiresAt = renewed.expiresAt; - this.#store.commit(candidate); + const renewed = { + ...structuredClone(lease), + expiresAtUnixMs, + expiresAt: new Date(expiresAtUnixMs).toISOString(), + }; + // Renewal runs outside the frame chain, so an event's staged write can + // be in flight. save() absorbs that admitted event into this write; an + // explicit commit() would invalidate it and close a healthy connection. + this.#store.state.leases[lease.credentialId] = renewed; + this.#store.state.lastLeaseExpiresAt = renewed.expiresAt; + this.#store.save(); connection.lease = this.#store.state.leases[lease.credentialId]!; } this.#sendJson(connection, @@ -3642,6 +3773,7 @@ export class DurablePrpControlPlane { } this.#store.state.connectionCount += 1; + this.#connectionLeaseFailure = null; this.#store.state.lastLeaseId = lease.leaseId; this.#store.state.lastLeaseExpiresAt = lease.expiresAt; diff --git a/packages/paperclip-runner/src/index.ts b/packages/paperclip-runner/src/index.ts index 5f40128c54a..5a68103f668 100644 --- a/packages/paperclip-runner/src/index.ts +++ b/packages/paperclip-runner/src/index.ts @@ -45,6 +45,7 @@ export { type StagedRunAttachment, type DurablePrpControlPlaneOptions, type DurablePrpCheckpoint, + type DurablePrpConnectionLeaseFailure, decodeDurablePrpCheckpoint, captureLegacyDurablePrpCheckpoint, inspectDurablePrpCheckpointHeader, diff --git a/server/src/realtime/runner-prp-outbound.test.ts b/server/src/realtime/runner-prp-outbound.test.ts index 06ab2f12a0a..86a702775f1 100644 --- a/server/src/realtime/runner-prp-outbound.test.ts +++ b/server/src/realtime/runner-prp-outbound.test.ts @@ -455,6 +455,69 @@ describe("runner provider-ingress WebSocket wire", () => { } }); + it.each(["refused", "unauthenticated"] as const)("names an expired runner connection lease in the run log and terminal error when reconnect is %s", async reconnect => { + vi.useFakeTimers(); + const first = new FakeSocket(); + const create = sockets.create; + let opens = 0; + sockets.create = () => { + opens += 1; + if (opens === 1) return openingSocket(first); + // An open socket whose PRP authentication never completes, as when the + // controller refuses the runner's lapsed credential. + if (reconnect === "unauthenticated") return openingSocket(); + const refused = new FakeSocket(); + refused.readyState = WebSocket.CONNECTING; + queueMicrotask(() => { + refused.readyState = WebSocket.CLOSED; + refused.emit("error", new Error("fixture runner exited")); + }); + return refused; + }; + const endpoint = { + kind: "authenticated_websocket" as const, + websocketUrl: "wss://preview.invalid/api/runner/v1/connect/run-lease-expired", + secretHeaders: [], + generation: "generation-lease-expired", + refresh: vi.fn(async () => endpoint), + close: vi.fn(async () => undefined), + }; + let leaseFailure: { code: string } | null = null; + const states: Array<[string, string | undefined]> = []; + const handle = connectRunnerPrpIngress({ + authority: { + attachWireConnection: vi.fn(() => { + const authenticated = opens === 1; + return { isAuthenticated: () => authenticated }; + }), + activeRunnerConnectionCount: () => 0, + get connectionLeaseFailure() { return leaseFailure; }, + } as unknown as DurablePrpControlPlane, + endpoint, + recoveryGraceMs: 10_000, + onStateChange: (state, code) => states.push([state, code]), + }); + const failure = vi.fn(); + void handle.failure.catch(failure); + try { + await vi.advanceTimersByTimeAsync(0); + await handle.ready; + // The runner's lease lapsed; it exits and the socket closes. + leaseFailure = { code: "connection_lease_expired" }; + first.close(1006); + await vi.advanceTimersByTimeAsync(0); + expect(states).toContainEqual(["reconnecting", "runner_connection_lease_expired"]); + await vi.advanceTimersByTimeAsync(20_000); + expect(failure).toHaveBeenCalledWith( + expect.objectContaining({ code: "runner_connection_lease_expired" }), + ); + } finally { + await handle.close(); + sockets.create = create; + vi.useRealTimers(); + } + }); + it("keeps a quiet but live runner attached through a long silent tool call", async () => { vi.useFakeTimers(); const live = new FakeSocket(); diff --git a/server/src/realtime/runner-prp-outbound.ts b/server/src/realtime/runner-prp-outbound.ts index 08db6258698..f563d2a3d67 100644 --- a/server/src/realtime/runner-prp-outbound.ts +++ b/server/src/realtime/runner-prp-outbound.ts @@ -21,7 +21,8 @@ const DEFAULT_LIVENESS_TIMEOUT_MS = 90_000; export type RunnerIngressFailureCode = | "runner_ingress_unavailable" - | "runner_ingress_auth_failed"; + | "runner_ingress_auth_failed" + | "runner_connection_lease_expired"; export class RunnerIngressConnectionError extends Error { constructor( @@ -354,6 +355,10 @@ export function connectRunnerPrpIngress(input: { const loop = (async () => { let attempt = 0; let recoveryDeadline = startupDeadline; + // A runner whose connection lease lapsed exits without another + // authenticated frame. Keep the authority's reason so the run log and the + // terminal error name it instead of a generic transport loss. + let leaseFailureCode: RunnerIngressFailureCode | null = null; while (!abort.signal.aborted) { try { input.onStateChange?.(attempt === 0 ? "connecting" : "reconnecting"); @@ -387,6 +392,7 @@ export function connectRunnerPrpIngress(input: { } input.onStateChange?.("authenticated"); attempt = 0; + leaseFailureCode = null; const stopLiveness = opened.wire.monitorLiveness({ intervalMs: input.livenessIntervalMs ?? DEFAULT_LIVENESS_INTERVAL_MS, timeoutMs: input.livenessTimeoutMs ?? DEFAULT_LIVENESS_TIMEOUT_MS, @@ -397,6 +403,11 @@ export function connectRunnerPrpIngress(input: { stopLiveness(); } recoveryDeadline = Date.now() + recoveryGraceMs; + const leaseFailure = input.authority.connectionLeaseFailure ?? null; + if (leaseFailure) { + leaseFailureCode = "runner_connection_lease_expired"; + input.onStateChange?.("reconnecting", leaseFailureCode); + } } catch (error) { // close() owns intentional shutdown. openEndpoint rejects its pending // handshake when the abort signal fires, but that cancellation is not @@ -431,8 +442,16 @@ export function connectRunnerPrpIngress(input: { } } if (Date.now() >= recoveryDeadline) { + // A lapsed lease is the root cause even when a later reconnect + // fails authentication; keep that failure as the cause. const terminal = - failure instanceof RunnerIngressConnectionError + leaseFailureCode !== null + ? new RunnerIngressConnectionError( + leaseFailureCode, + "The runner's connection lease expired before it was renewed.", + { cause: failure }, + ) + : failure instanceof RunnerIngressConnectionError ? failure : new RunnerIngressConnectionError( statusCode === 401 || statusCode === 403