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