diff --git a/doc/DATABASE.md b/doc/DATABASE.md index adf08ad2d8..3f0a2ce57e 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -216,8 +216,14 @@ finalization to one company, issue, and run. The database rejects mixed-owner evidence even when every referenced ID exists. Native source identities on `heartbeat_run_events` are nullable so legacy events remain readable without rewriting historical rows. Per-run native source identifiers are unique, while -the existing legacy sequence behavior remains unchanged until an atomic event -allocator is introduced with the native writer. +the existing legacy sequence behavior remains unchanged. The hidden native +coordinator serializes on its bound `heartbeat_runs` row, allocates +`next_event_seq`, and commits a validated PRP event before the transport sends +its cumulative ACK. Byte-equivalent source retries return the existing cursor; +gaps and conflicting replays fail closed. Accepted structured results enter the +finalization ledger, whose retry time and owner lease are checked under a row +lock. None of these writes selects a runtime or changes a legacy run's execution +path. Issue `status_version` advances only when `status` changes. The JavaScript backup path includes user-defined functions and triggers so a restored database keeps diff --git a/doc/run-log-events.md b/doc/run-log-events.md index c7a55d80bd..ffaa8dce96 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -5,6 +5,24 @@ Run-log events write to the `heartbeat_run_events` table Paperclip Telemetry events, and they are not OpenTelemetry exports. A run-log event needs no operator endpoint. +## Native PRP Run-Log Events + +The hidden native coordinator writes each validated PRP event to the bound +run's existing event stream before it acknowledges the runner. The row keeps +the PRP `eventType`, source instance, source event ID, source sequence, protocol +schema version, and a SHA-256 digest of the canonical source envelope. Its +payload is `{ "prpEvent": }`. + +The writer locks the native `heartbeat_runs` row and allocates the existing +per-run `seq` cursor. A byte-equivalent retry reuses the first row; a changed +retry or source-sequence gap is rejected. Company, issue, agent, run, session, +and runner-source bindings must match the persisted native run. Bootstrap +tickets, reconnect leases, authentication proofs, encryption keys, and raw +credential material are never written to the run log. + +These records remain run-log events. They do not create an OpenTelemetry or +Paperclip Telemetry export, and legacy adapters do not use this writer. + ## Sandbox Startup Run-Log Event Paperclip writes one `run.startup.step` event to the run log for each bring-up diff --git a/packages/paperclip-runner/README.md b/packages/paperclip-runner/README.md index 9d3a5d83a5..58a76c39ca 100644 --- a/packages/paperclip-runner/README.md +++ b/packages/paperclip-runner/README.md @@ -2,7 +2,7 @@ This private workspace package contains the staged Paperclip Runner work. -The package currently exposes only the language-neutral PRP v1 TypeScript +The package currently exposes the language-neutral PRP v1 TypeScript contract, provider-neutral structured questions and responses, deterministic fixture validation/replay, structured-result normalization, and the session reducer oracle. It also contains a package-local Rust runner, scripted fake @@ -11,27 +11,35 @@ PRP transport. The transport authenticates and encrypts loopback WebSocket sessions, persists an ACK-driven outbox and command journal, and reconnects with a short-lived lease. The Rust runner now includes a Codex-only app-server provider bridge with durable thread resume, cancellation, structured questions, -and provider-neutral event normalization. No server code starts or invokes it. +and provider-neutral event normalization. The root surface now also exposes an +authenticated durable PRP authority for server-side use. It stores only +bootstrap and reconnect credential digests, validates immutable run identity on +every connection and event, and persists commands and cumulative event ACK +state across server restarts. The package also publishes the canonical semantic action declarations and their input and output schemas. Its package-local dispatcher projects only bound, run-authorized actions and emits redacted semantic receipts. It does not add application bindings, a server adapter, or production Paperclip behavior. The first and only installed provider is Codex. Dynamic semantic tools remain -undiscoverable because no production application binding or server authority -has landed. Catalog membership alone does not grant authority. See +undiscoverable unless the hidden server coordinator projects one of the five +same-task read bindings for an already persisted native Codex run. Catalog +membership alone does not grant authority, and no production adapter can create +or start such a run yet. See [`SEMANTIC_ACTIONS.md`](SEMANTIC_ACTIONS.md) for the catalog boundary. The package has two initial public surfaces: - `@paperclipai/paperclip-runner` contains runtime contracts, validation, - replay/reducer logic, the semantic catalog, and the authorization dispatcher. + replay/reducer logic, the semantic catalog, the authorization dispatcher, and + the Node-only durable server authority. - `@paperclipai/paperclip-runner/testing` adds Node-only fixture loading and a provider-neutral semantic conformance kit for deterministic test adapters. No SDK, browser, React, eval, live-console, lab, or provider-experiment entry -point is exported. The package remains private in this wave, and no production -adapter starts it yet. +point is exported. The package remains private in this wave. The server route +at `/api/runner/v1/connect/:runId` has no authority until the hidden coordinator +registers an exact existing run binding, and no production adapter starts it. Run the complete contract gate with: 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 new file mode 100644 index 0000000000..9b1ab8fedf --- /dev/null +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts @@ -0,0 +1,618 @@ +import { + createCipheriv, + createDecipheriv, + createHash, + createHmac, +} from "node:crypto"; +import { mkdtempSync, rmSync } from "node:fs"; +import { connect, type Socket } from "node:net"; +import { tmpdir } from "node:os"; +import { resolve } from "node:path"; + +import { describe, expect, it } from "vitest"; + +import { validatePrpEvent } from "../protocol/replay-contract.js"; +import { digestPaperclipSemanticContent } from "../semantic-tools/receipts.js"; +import { DurablePrpControlPlane } from "./durable-prp-control-plane.js"; +import type { DurableRecoveryIdentity } from "./prp-transport-types.js"; + +const identity: DurableRecoveryIdentity = { + runnerInstanceId: "runner-test-1", + environmentLeaseId: "environment-test-1", + runId: "00000000-0000-4000-8000-000000000001", + normalizedSessionId: "session-test-1", + turnId: "turn-test-1", + itemId: "item-test-1", +}; +const expectedRunnerVersion = "0.3.0"; +const expectedRunnerDigest = `sha256:${"a".repeat(64)}`; + +function domainDigest(domain: string, parts: readonly Buffer[]): Buffer { + const digest = createHash("sha256") + .update(domain) + .update(Buffer.from([0])); + for (const part of parts) { + const length = Buffer.alloc(8); + length.writeBigUInt64BE(BigInt(part.length)); + digest.update(length).update(part); + } + return digest.digest(); +} + +function domainHmac( + key: Buffer, + domain: string, + parts: readonly Buffer[], +): Buffer { + const digest = createHmac("sha256", key) + .update(domain) + .update(Buffer.from([0])); + for (const part of parts) { + const length = Buffer.alloc(8); + length.writeBigUInt64BE(BigInt(part.length)); + digest.update(length).update(part); + } + return digest.digest(); +} + +function canonicalJson(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(canonicalJson).join(",")}]`; + if (typeof value === "object" && value !== null) { + const record = value as Record; + return `{${Object.keys(record) + .sort() + .map((key) => `${JSON.stringify(key)}:${canonicalJson(record[key])}`) + .join(",")}}`; + } + return JSON.stringify(value); +} + +function credentialMaterial(token: string): { + credentialId: string; + authKey: Buffer; +} { + const bytes = Buffer.from(token); + const authKey = domainDigest("paperclip-runner-auth-key-v1", [bytes]); + return { + credentialId: `sha256:${domainDigest("paperclip-runner-credential-id-v1", [bytes]).toString("hex")}`, + authKey, + }; +} + +class ServerFrameReader { + #buffer = Buffer.alloc(0); + #frames: Array | null> = []; + #waiters: Array<(value: Record | null) => void> = []; + + constructor(socket: Socket) { + socket.on("data", (chunk: Buffer) => { + this.#buffer = Buffer.concat([this.#buffer, chunk]); + this.#drain(); + }); + socket.once("close", () => this.#publish(null)); + } + + next(): Promise | null> { + const queued = this.#frames.shift(); + if (queued !== undefined) return Promise.resolve(queued); + return new Promise((resolveFrame, rejectFrame) => { + const timer = setTimeout( + () => rejectFrame(new Error("Server WebSocket frame timed out.")), + 2_000, + ); + this.#waiters.push((value) => { + clearTimeout(timer); + resolveFrame(value); + }); + }); + } + + #publish(value: Record | null): void { + const waiter = this.#waiters.shift(); + if (waiter) waiter(value); + else this.#frames.push(value); + } + + #drain(): void { + while (this.#buffer.length >= 2) { + let length = this.#buffer[1]! & 0x7f; + let cursor = 2; + if (length === 126) { + if (this.#buffer.length < 4) return; + length = this.#buffer.readUInt16BE(2); + cursor = 4; + } else if (length === 127) { + if (this.#buffer.length < 10) return; + length = Number(this.#buffer.readBigUInt64BE(2)); + cursor = 10; + } + if (this.#buffer.length < cursor + length) return; + const payload = this.#buffer.subarray(cursor, cursor + length); + this.#buffer = this.#buffer.subarray(cursor + length); + this.#publish( + JSON.parse(payload.toString("utf8")) as Record, + ); + } + } +} + +async function upgradeSocket(url: string): Promise<{ + socket: Socket; + reader: ServerFrameReader; +}> { + const parsed = new URL(url); + const socket = await new Promise((resolveSocket, rejectSocket) => { + const candidate = connect(Number(parsed.port), parsed.hostname); + let response = Buffer.alloc(0); + const onData = (chunk: Buffer): void => { + response = Buffer.concat([response, chunk]); + const boundary = response.indexOf("\r\n\r\n"); + if (boundary < 0) return; + candidate.off("data", onData); + const status = Number( + response.toString("utf8").match(/^HTTP\/1\.1 (\d{3})/)?.[1], + ); + if (status !== 101) { + candidate.destroy(); + rejectSocket(new Error(`WebSocket upgrade returned ${String(status)}`)); + } else { + resolveSocket(candidate); + } + }; + candidate.once("error", rejectSocket); + candidate.once("connect", () => { + candidate.write( + [ + `GET ${parsed.pathname} HTTP/1.1`, + `Host: ${parsed.host}`, + "Upgrade: websocket", + "Connection: Upgrade", + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==", + "Sec-WebSocket-Version: 13", + "\r\n", + ].join("\r\n"), + ); + }); + candidate.on("data", onData); + }); + return { socket, reader: new ServerFrameReader(socket) }; +} + +function sendMaskedJson(socket: Socket, value: unknown): void { + const payload = Buffer.from(JSON.stringify(value)); + const mask = Buffer.from([0x11, 0x22, 0x33, 0x44]); + const header: number[] = [0x81]; + if (payload.length <= 125) { + header.push(0x80 | payload.length); + } else if (payload.length <= 0xffff) { + header.push(0x80 | 126, payload.length >>> 8, payload.length & 0xff); + } else { + throw new Error("Test client frame exceeds the supported size."); + } + const masked = Buffer.from(payload); + for (let index = 0; index < masked.length; index += 1) { + masked[index] = masked[index]! ^ mask[index % mask.length]!; + } + socket.write(Buffer.concat([Buffer.from(header), mask, masked])); +} + +interface AuthenticatedClient { + socket: Socket; + reader: ServerFrameReader; + authKey: Buffer; + sessionId: string; + sendCounter: bigint; + receiveCounter: bigint; + leaseToken: string | null; + welcome: Record; +} + +function authHello( + credentialId: string, + selectedIdentity: DurableRecoveryIdentity = identity, +): Record { + return { + protocol: "paperclip.runner", + version: 1, + kind: "auth_hello", + payload: { + credentialId, + clientNonce: "client-nonce-test", + protocolMin: 1, + protocolMax: 1, + ...selectedIdentity, + runnerVersion: expectedRunnerVersion, + runnerDigest: expectedRunnerDigest, + }, + }; +} + +async function authenticate( + controlPlane: DurablePrpControlPlane, + token: string, + selectedIdentity: DurableRecoveryIdentity = identity, +): Promise { + const { socket, reader } = await upgradeSocket(controlPlane.connectUrl); + const material = credentialMaterial(token); + sendMaskedJson(socket, authHello(material.credentialId, selectedIdentity)); + const challenge = await reader.next(); + if (challenge === null) return null; + const challengePayload = challenge.payload as Record; + const serverProof = challengePayload.serverProof; + if (typeof serverProof !== "string") throw new Error("Missing server proof."); + const canonicalPayload = { ...challengePayload }; + delete canonicalPayload.serverProof; + const canonicalChallenge = canonicalJson(canonicalPayload); + const expectedServerProof = domainHmac( + material.authKey, + "paperclip-runner-server-proof-v1", + [Buffer.from(canonicalChallenge)], + ).toString("hex"); + expect(serverProof).toBe(expectedServerProof); + const clientProof = domainHmac( + material.authKey, + "paperclip-runner-client-proof-v1", + [Buffer.from(canonicalChallenge), Buffer.from(serverProof)], + ).toString("hex"); + sendMaskedJson(socket, { + protocol: "paperclip.runner", + version: 1, + kind: "auth_response", + payload: { + credentialId: material.credentialId, + clientNonce: challengePayload.clientNonce, + serverNonce: challengePayload.serverNonce, + clientProof, + }, + }); + const binding = domainDigest("paperclip-runner-session-binding-v1", [ + Buffer.from(canonicalChallenge), + Buffer.from(serverProof), + Buffer.from(clientProof), + ]); + const client: AuthenticatedClient = { + socket, + reader, + authKey: material.authKey, + sessionId: `sha256:${binding.toString("hex")}`, + sendCounter: 0n, + receiveCounter: 0n, + leaseToken: null, + welcome: {}, + }; + const welcome = await receiveSecure(client); + if (welcome === null) return null; + expect(welcome.kind).toBe("welcome"); + client.welcome = welcome; + const leaseToken = (welcome.payload as Record) + .connectionLeaseToken; + client.leaseToken = typeof leaseToken === "string" ? leaseToken : null; + return client; +} + +function secureNonce(prefix: "P3C1" | "P3S1", counter: bigint): Buffer { + const nonce = Buffer.alloc(12); + nonce.write(prefix, 0, "ascii"); + nonce.writeBigUInt64BE(counter, 4); + return nonce; +} + +function secureAad( + client: AuthenticatedClient, + direction: "client_to_core" | "core_to_client", + counter: bigint, +): Buffer { + return Buffer.from( + `paperclip.runner.secure-frame.v1\0${client.sessionId}\0${direction}\0${counter}`, + ); +} + +function sendSecure( + client: AuthenticatedClient, + value: Record, +): void { + const binding = Buffer.from(client.sessionId.slice("sha256:".length), "hex"); + const key = domainHmac( + client.authKey, + "paperclip-runner-client-to-core-key-v1", + [binding], + ); + const counter = client.sendCounter; + const cipher = createCipheriv( + "aes-256-gcm", + key, + secureNonce("P3C1", counter), + ); + cipher.setAAD(secureAad(client, "client_to_core", counter)); + const encrypted = Buffer.concat([ + cipher.update(Buffer.from(JSON.stringify(value))), + cipher.final(), + cipher.getAuthTag(), + ]); + sendMaskedJson(client.socket, { + schema: "paperclip.runner.secure-frame.v1", + counter: Number(counter), + ciphertext: encrypted.toString("hex"), + }); + client.sendCounter += 1n; +} + +async function receiveSecure( + client: AuthenticatedClient, +): Promise | null> { + const frame = await client.reader.next(); + if (frame === null) return null; + const counter = BigInt(frame.counter as number); + expect(counter).toBe(client.receiveCounter); + const binding = Buffer.from(client.sessionId.slice("sha256:".length), "hex"); + const key = domainHmac( + client.authKey, + "paperclip-runner-core-to-client-key-v1", + [binding], + ); + const sealed = Buffer.from(String(frame.ciphertext), "hex"); + const decipher = createDecipheriv( + "aes-256-gcm", + key, + secureNonce("P3S1", counter), + ); + decipher.setAAD(secureAad(client, "core_to_client", counter)); + decipher.setAuthTag(sealed.subarray(-16)); + const value = JSON.parse( + Buffer.concat([ + decipher.update(sealed.subarray(0, -16)), + decipher.final(), + ]).toString("utf8"), + ) as Record; + client.receiveCounter += 1n; + return value; +} + +function semanticInputEvent(sourceSeq = 1): Record { + return { + protocol: "paperclip.runner", + version: 1, + kind: "event", + runnerInstanceId: identity.runnerInstanceId, + environmentLeaseId: identity.environmentLeaseId, + runId: identity.runId, + normalizedSessionId: identity.normalizedSessionId, + turnId: identity.turnId, + itemId: identity.itemId, + payload: { + sourceSeq, + sourceEventId: `semantic-event-${sourceSeq}`, + sourceInstanceId: identity.runnerInstanceId, + sourceKind: "runner", + runId: identity.runId, + normalizedSessionId: identity.normalizedSessionId, + turnId: identity.turnId, + itemId: identity.itemId, + eventType: "semantic_tool.input", + schema: "paperclip.prp.event.v1", + schemaVersion: 1, + priority: 0, + emittedAt: "2026-08-25T18:00:00.000Z", + payload: { + semantic_tool: { + schema: "paperclip.prp.semantic_tool.v1", + schemaVersion: 1, + phase: "input", + callId: "call-1", + operationId: "get_task_context", + correlation: { + runId: identity.runId, + normalizedSessionId: identity.normalizedSessionId, + turnId: identity.turnId, + itemId: identity.itemId, + }, + idempotencyKey: null, + content: { + digest: digestPaperclipSemanticContent({}), + redactionDisposition: "digest_only", + references: [], + }, + input: {}, + }, + }, + }, + }; +} + +describe.sequential("DurablePrpControlPlane", () => { + it("exchanges a one-use bootstrap for a run-bound reconnect lease", async () => { + const root = mkdtempSync(resolve(tmpdir(), "paperclip-prp-auth-")); + const controlPlane = new DurablePrpControlPlane({ + stateDirectory: root, + identity, + expectedRunnerVersion, + expectedRunnerDigest, + }); + try { + await controlPlane.start(); + const cancel = controlPlane.queueCommand( + "run.cancel", + { reason: "test" }, + "command-cancel-1", + ); + expect( + controlPlane.queueCommand( + "run.cancel", + { reason: "test" }, + "command-cancel-1", + ), + ).toEqual(cancel); + expect(() => + controlPlane.queueCommand( + "run.cancel", + { reason: "different" }, + "command-cancel-1", + ), + ).toThrow("command replay conflicts"); + expect(() => controlPlane.queueCommand("unknown.command")).toThrow( + "command is invalid", + ); + const ticket = controlPlane.issueBootstrapTicket(); + const first = await authenticate(controlPlane, ticket); + expect(first?.leaseToken).toEqual(expect.any(String)); + first?.socket.destroy(); + + const reused = await authenticate(controlPlane, ticket); + expect(reused).toBeNull(); + + const lease = await authenticate(controlPlane, first!.leaseToken!); + expect(lease).not.toBeNull(); + expect(lease?.leaseToken).toBeNull(); + lease?.socket.destroy(); + + const wrongRun = await authenticate( + controlPlane, + controlPlane.issueBootstrapTicket(), + { + ...identity, + runId: "00000000-0000-4000-8000-000000000999", + }, + ); + expect(wrongRun).toBeNull(); + } finally { + await controlPlane.stop(); + rmSync(root, { recursive: true, force: true }); + } + }); + + it("recovers one semantic call from the durable event after a coordinator restart", async () => { + const root = mkdtempSync(resolve(tmpdir(), "paperclip-prp-recovery-")); + let firstCalls = 0; + const first = new DurablePrpControlPlane({ + stateDirectory: root, + identity, + expectedRunnerVersion, + expectedRunnerDigest, + onSemanticToolInput: async () => { + firstCalls += 1; + return new Promise(() => undefined); + }, + }); + let leaseToken: string; + try { + await first.start(); + const client = await authenticate(first, first.issueBootstrapTicket()); + leaseToken = client!.leaseToken!; + const validation = validatePrpEvent(semanticInputEvent().payload); + expect(validation, JSON.stringify(validation)).toMatchObject({ + ok: true, + }); + sendSecure(client!, semanticInputEvent()); + const ack = await receiveSecure(client!); + expect(ack).toMatchObject({ kind: "ack" }); + expect(firstCalls).toBe(1); + client?.socket.destroy(); + } finally { + await first.stop(); + } + + let recoveredCalls = 0; + const recovered = new DurablePrpControlPlane({ + stateDirectory: root, + identity, + expectedRunnerVersion, + expectedRunnerDigest, + onSemanticToolInput: async (call) => { + recoveredCalls += 1; + return { + result: { ok: true, operationId: call.operationId }, + }; + }, + }); + try { + await recovered.start(); + const client = await authenticate(recovered, leaseToken!); + sendSecure(client!, semanticInputEvent()); + const outcomes = [ + await receiveSecure(client!), + await receiveSecure(client!), + ]; + const command = outcomes.find((outcome) => outcome?.kind === "command"); + expect(outcomes.some((outcome) => outcome?.kind === "ack")).toBe(true); + expect(command?.payload).toMatchObject({ + type: "semantic_tool.result", + payload: { + callId: "call-1", + operationId: "get_task_context", + result: { ok: true, operationId: "get_task_context" }, + isError: false, + }, + }); + expect(recoveredCalls).toBe(1); + + const tampered = semanticInputEvent(2); + const tamperedEvent = tampered.payload as Record; + const tamperedPayload = tamperedEvent.payload as Record; + const tamperedSemantic = tamperedPayload.semantic_tool as Record< + string, + unknown + >; + tamperedSemantic.content = { + ...(tamperedSemantic.content as Record), + digest: `sha256:${"0".repeat(64)}`, + }; + sendSecure(client!, tampered); + await expect(receiveSecure(client!)).resolves.toBeNull(); + client?.socket.destroy(); + } finally { + await recovered.stop(); + rmSync(root, { recursive: true, force: true }); + } + }); + + it("does not acknowledge an event before the caller commits it", async () => { + const root = mkdtempSync(resolve(tmpdir(), "paperclip-prp-commit-order-")); + const first = new DurablePrpControlPlane({ + stateDirectory: root, + identity, + expectedRunnerVersion, + expectedRunnerDigest, + onSemanticToolInput: async () => ({ result: { ok: true } }), + onCommittedEvent: async () => { + throw new Error("database commit failed"); + }, + }); + let leaseToken: string; + try { + await first.start(); + const client = await authenticate(first, first.issueBootstrapTicket()); + leaseToken = client!.leaseToken!; + sendSecure(client!, semanticInputEvent()); + await expect(receiveSecure(client!)).resolves.toBeNull(); + } finally { + await first.stop(); + } + + let committed = 0; + const recovered = new DurablePrpControlPlane({ + stateDirectory: root, + identity, + expectedRunnerVersion, + expectedRunnerDigest, + onSemanticToolInput: async () => ({ result: { ok: true } }), + onCommittedEvent: async () => { + committed += 1; + }, + }); + try { + await recovered.start(); + const client = await authenticate(recovered, leaseToken!); + expect(client?.welcome.payload).toMatchObject({ ackedSourceSeq: 0 }); + sendSecure(client!, semanticInputEvent()); + await expect(receiveSecure(client!)).resolves.toMatchObject({ + kind: "ack", + payload: { ackedSourceSeq: 1 }, + }); + expect(committed).toBe(1); + client?.socket.destroy(); + } finally { + await recovered.stop(); + rmSync(root, { recursive: true, force: true }); + } + }); +}); 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 new file mode 100644 index 0000000000..74ae7072be --- /dev/null +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts @@ -0,0 +1,1659 @@ +import { + createCipheriv, + createDecipheriv, + createHash, + createHmac, + randomUUID, + timingSafeEqual, +} from "node:crypto"; +import { + chmodSync, + closeSync, + constants, + fchmodSync, + fstatSync, + fsyncSync, + lstatSync, + mkdirSync, + openSync, + readFileSync, + renameSync, + unlinkSync, + writeFileSync, + type Stats, +} from "node:fs"; +import { createServer, type IncomingMessage, type Server } from "node:http"; +import { dirname, resolve } from "node:path"; +import type { Duplex } from "node:stream"; + +import { + validatePrpEvent, + type PrpEvent, +} from "../protocol/replay-contract.js"; +import { digestPaperclipSemanticContent } from "../semantic-tools/receipts.js"; +import { + type DurableRecoveryCommittedEvent, + type DurableRecoveryCoreCommand, + type DurableRecoveryIdentity, +} from "./prp-transport-types.js"; + +const protocol = "paperclip.runner"; +const protocolVersion = 1; +const secureFrameSchema = "paperclip.runner.secure-frame.v1"; +const websocketGuid = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; +const coreStateSchema = "paperclip.runner.durable.control-plane-state.v1"; +const maxFrameBytes = 1024 * 1024; +const maxCommandBytes = maxFrameBytes - 4 * 1024; +const maxCommands = 500; +const maxCommittedEventWindow = 64; +const maxStateBytes = 192 * 1024 * 1024; +const authChallengeTtlMs = 5_000; +const stableIdPattern = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,239}$/; +const runnerDigestPattern = /^sha256:[0-9a-f]{64}$/; +const commandTypes = new Set([ + "run.prepare", + "session.open", + "turn.start", + "turn.steer", + "turn.interrupt", + "turn.stop", + "request.resolve", + "interaction.receipt", + "semantic_tool.result", + "session.snapshot", + "session.close", + "session.budget.increase", + "session.destroy", + "run.cancel", + "runner.drain", + "runner.suspend", + "runner.shutdown", +]); + +interface BootstrapTicketRecord { + recordId: string; + credentialId: string; + authKeyDigest: string; + identity: DurableRecoveryIdentity; + runnerVersion: string; + runnerDigest: string; + expiresAt: string; + expiresAtUnixMs: number; + usedAt: string | null; +} + +interface ConnectionLeaseRecord { + recordId: string; + credentialId: string; + authKeyDigest: string; + leaseId: string; + identity: DurableRecoveryIdentity; + protocolVersion: number; + expiresAt: string; + expiresAtUnixMs: number; + revocationEpoch: number; + revokedAt: string | null; +} + +interface StoredCoreState { + schema: typeof coreStateSchema; + identity: DurableRecoveryIdentity; + tickets: Record; + leases: Record; + commands: DurableRecoveryCoreCommand[]; + committedEvents: DurableRecoveryCommittedEvent[]; + ackedSourceSeq: number; + connectionCount: number; + commandDeliveryCounts: Record; + replayDeliveries: number; + duplicateCommandResults: number; + freshBootstraps: number; + malformedFrames: number; + lastLeaseId: string | null; + lastLeaseExpiresAt: string | null; +} + +type PendingAuthorization = + | { + kind: "bootstrap"; + recordId: string; + credentialId: string; + authKey: Buffer; + identity: DurableRecoveryIdentity; + runnerVersion: string; + runnerDigest: string; + expiresAt: string; + expiresAtUnixMs: number; + recordSnapshot: string; + } + | { + kind: "lease"; + recordId: string; + credentialId: string; + authKey: Buffer; + identity: DurableRecoveryIdentity; + protocolVersion: number; + expiresAt: string; + expiresAtUnixMs: number; + leaseId: string; + revocationEpoch: number; + recordSnapshot: string; + }; + +type LiveAuthorization = + | { + kind: "bootstrap"; + authKey: Buffer; + ticket: BootstrapTicketRecord; + } + | { + kind: "lease"; + authKey: Buffer; + lease: ConnectionLeaseRecord; + }; + +interface PendingChallenge { + authorization: PendingAuthorization; + deadlineUnixMs: number; + canonicalChallenge: string; + serverProof: string; + clientNonce: string; + serverNonce: string; +} + +interface SecureChannel { + sendKey: Buffer; + receiveKey: Buffer; + sendCounter: bigint; + receiveCounter: bigint; + sessionId: string; +} + +export interface DurablePrpControlPlaneOptions { + stateDirectory: string; + identity: DurableRecoveryIdentity; + expectedRunnerVersion: string; + expectedRunnerDigest: string; + onSemanticToolInput?: (input: { + readonly callId: string; + readonly operationId: string; + readonly input: unknown; + /** Internal trace lineage for the canonical semantic_tool.input event. */ + readonly sourceEventId: string; + readonly sourceEventType: string; + readonly correlation: { + readonly runId: string; + readonly normalizedSessionId: string; + readonly turnId: string; + readonly itemId: string; + }; + }) => Promise<{ readonly result: unknown; readonly isError?: boolean }>; + /** Persist the canonical event before the runner receives its cumulative ACK. */ + onCommittedEvent?: (event: PrpEvent) => Promise; + connectionLeaseTtlMs?: number; +} + +function domainDigest(domain: string, parts: readonly Buffer[]): Buffer { + const digest = createHash("sha256") + .update(domain) + .update(Buffer.from([0])); + for (const part of parts) { + const length = Buffer.alloc(8); + length.writeBigUInt64BE(BigInt(part.length)); + digest.update(length).update(part); + } + return digest.digest(); +} + +function domainHmac( + key: Buffer, + domain: string, + parts: readonly Buffer[], +): Buffer { + const digest = createHmac("sha256", key) + .update(domain) + .update(Buffer.from([0])); + for (const part of parts) { + const length = Buffer.alloc(8); + length.writeBigUInt64BE(BigInt(part.length)); + digest.update(length).update(part); + } + return digest.digest(); +} + +function credentialMaterial(token: string): { + credentialId: string; + authKey: Buffer; +} { + const bytes = Buffer.from(token); + return { + credentialId: `sha256:${domainDigest("paperclip-runner-credential-id-v1", [bytes]).toString("hex")}`, + authKey: domainDigest("paperclip-runner-auth-key-v1", [bytes]), + }; +} + +function canonicalJson(value: unknown): string { + if (Array.isArray(value)) { + return `[${value.map(canonicalJson).join(",")}]`; + } + if (typeof value === "object" && value !== null) { + const object = value as Record; + return `{${Object.keys(object) + .sort() + .map((key) => `${JSON.stringify(key)}:${canonicalJson(object[key])}`) + .join(",")}}`; + } + return JSON.stringify(value) ?? "null"; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function isStoredCoreState( + value: unknown, + identity: DurableRecoveryIdentity, +): value is StoredCoreState { + if (!isRecord(value)) return false; + const commands = value.commands; + const events = value.committedEvents; + if ( + value.schema !== coreStateSchema || + canonicalJson(value.identity) !== canonicalJson(identity) || + !isRecord(value.tickets) || + !isRecord(value.leases) || + !Array.isArray(commands) || + commands.length > maxCommands || + !Array.isArray(events) || + events.length > maxCommittedEventWindow || + !Number.isSafeInteger(value.ackedSourceSeq) || + (value.ackedSourceSeq as number) < 0 || + !Number.isSafeInteger(value.connectionCount) || + (value.connectionCount as number) < 0 || + !isRecord(value.commandDeliveryCounts) + ) { + return false; + } + if ( + !commands.every( + (command, index) => + isRecord(command) && + command.schema === "paperclip.prp.command.v1" && + typeof command.commandId === "string" && + stableIdPattern.test(command.commandId) && + command.commandId.length <= 160 && + command.controllerSeq === index + 1 && + typeof command.type === "string" && + commandTypes.has(command.type) && + typeof command.issuedAt === "string" && + isRecord(command.payload) && + ["pending", "completed", "failed", "rejected"].includes( + String(command.status), + ) && + (command.result === null || isRecord(command.result)), + ) + ) { + return false; + } + if ( + !events.every( + (event) => + isRecord(event) && + Number.isSafeInteger(event.sourceSeq) && + (event.sourceSeq as number) > 0 && + typeof event.sourceEventId === "string" && + typeof event.eventType === "string" && + (event.priority === 0 || + event.priority === 1 || + event.priority === 2) && + isRecord(event.envelope) && + Number.isSafeInteger(event.deliveryCount) && + (event.deliveryCount as number) > 0 && + event.logicalEffectCount === 1, + ) + ) { + return false; + } + return [ + "replayDeliveries", + "duplicateCommandResults", + "freshBootstraps", + "malformedFrames", + ].every( + (field) => + Number.isSafeInteger(value[field]) && (value[field] as number) >= 0, + ); +} + +function authKeyFromDigest(digest: string): Buffer { + const hex = digest.match(/^sha256:([0-9a-f]{64})$/)?.[1]; + if (hex === undefined) + throw new Error("Stored transport authentication key is malformed."); + return Buffer.from(hex, "hex"); +} + +function proofMatches(expected: Buffer, supplied: unknown): boolean { + if (typeof supplied !== "string" || !/^[0-9a-f]{64}$/.test(supplied)) + return false; + return timingSafeEqual(expected, Buffer.from(supplied, "hex")); +} + +function createSecureChannel( + authKey: Buffer, + canonicalChallenge: string, + serverProof: string, + clientProof: string, +): SecureChannel { + const parts = [ + Buffer.from(canonicalChallenge), + Buffer.from(serverProof), + Buffer.from(clientProof), + ]; + const binding = domainDigest("paperclip-runner-session-binding-v1", parts); + return { + sendKey: domainHmac(authKey, "paperclip-runner-core-to-client-key-v1", [ + binding, + ]), + receiveKey: domainHmac(authKey, "paperclip-runner-client-to-core-key-v1", [ + binding, + ]), + sendCounter: 0n, + receiveCounter: 0n, + sessionId: `sha256:${binding.toString("hex")}`, + }; +} + +function secureNonce(prefix: "P3C1" | "P3S1", counter: bigint): Buffer { + const nonce = Buffer.alloc(12); + nonce.write(prefix, 0, "ascii"); + nonce.writeBigUInt64BE(counter, 4); + return nonce; +} + +function secureAad( + channel: SecureChannel, + direction: "client_to_core" | "core_to_client", + counter: bigint, +): Buffer { + return Buffer.from( + `${secureFrameSchema}\0${channel.sessionId}\0${direction}\0${counter}`, + ); +} + +function encryptSecureJson( + channel: SecureChannel, + value: unknown, +): Record { + const counter = channel.sendCounter; + const cipher = createCipheriv( + "aes-256-gcm", + channel.sendKey, + secureNonce("P3S1", counter), + ); + cipher.setAAD(secureAad(channel, "core_to_client", counter)); + const ciphertext = Buffer.concat([ + cipher.update(Buffer.from(JSON.stringify(value))), + cipher.final(), + cipher.getAuthTag(), + ]); + channel.sendCounter += 1n; + return { + schema: secureFrameSchema, + counter: Number(counter), + ciphertext: ciphertext.toString("hex"), + }; +} + +function decryptSecureJson( + channel: SecureChannel, + value: unknown, +): Record { + if (typeof value !== "object" || value === null) { + throw new Error("Secure frame must be an object."); + } + const frame = value as Record; + if ( + frame.schema !== secureFrameSchema || + typeof frame.counter !== "number" || + !Number.isSafeInteger(frame.counter) || + BigInt(frame.counter) !== channel.receiveCounter || + typeof frame.ciphertext !== "string" || + !/^[0-9a-f]+$/.test(frame.ciphertext) || + frame.ciphertext.length % 2 !== 0 + ) { + throw new Error("Secure frame metadata or counter is invalid."); + } + const sealed = Buffer.from(frame.ciphertext, "hex"); + if (sealed.length < 16) + throw new Error("Secure frame authentication tag is missing."); + const ciphertext = sealed.subarray(0, -16); + const tag = sealed.subarray(-16); + const counter = channel.receiveCounter; + const decipher = createDecipheriv( + "aes-256-gcm", + channel.receiveKey, + secureNonce("P3C1", counter), + ); + decipher.setAAD(secureAad(channel, "client_to_core", counter)); + decipher.setAuthTag(tag); + const plaintext = Buffer.concat([ + decipher.update(ciphertext), + decipher.final(), + ]); + channel.receiveCounter += 1n; + return JSON.parse(plaintext.toString("utf8")) as Record; +} + +function initialCoreState(identity: DurableRecoveryIdentity): StoredCoreState { + return { + schema: coreStateSchema, + identity, + tickets: {}, + leases: {}, + commands: [], + committedEvents: [], + ackedSourceSeq: 0, + connectionCount: 0, + commandDeliveryCounts: {}, + replayDeliveries: 0, + duplicateCommandResults: 0, + freshBootstraps: 0, + malformedFrames: 0, + lastLeaseId: null, + lastLeaseExpiresAt: null, + }; +} + +function isNodeError(error: unknown, code: string): boolean { + return error instanceof Error && "code" in error && error.code === code; +} + +function verifyPrivateDirectory(path: string): void { + const metadata = lstatSync(path); + if (metadata.isSymbolicLink() || !metadata.isDirectory()) { + throw new Error(`Private state directory is not a real directory: ${path}`); + } + if (process.platform !== "win32") { + if ((metadata.mode & 0o777) !== 0o700) { + throw new Error( + `Private state directory does not use mode 0700: ${path}`, + ); + } + if (process.geteuid !== undefined && metadata.uid !== process.geteuid()) { + throw new Error( + `Private state directory is not owned by the daemon user: ${path}`, + ); + } + } +} + +function verifyPrivateRegularFile(file: Stats, path: string): void { + if (!file.isFile()) { + throw new Error(`Private state path is not a regular file: ${path}`); + } + if (process.platform !== "win32") { + if ((file.mode & 0o777) !== 0o600) { + throw new Error(`Private state file does not use mode 0600: ${path}`); + } + if (process.geteuid !== undefined && file.uid !== process.geteuid()) { + throw new Error( + `Private state file is not owned by the daemon user: ${path}`, + ); + } + } +} + +function readPrivateFile(path: string): string | null { + let descriptor: number; + try { + descriptor = openSync( + path, + constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0), + ); + } catch (error) { + if (isNodeError(error, "ENOENT")) return null; + throw error; + } + try { + const metadata = fstatSync(descriptor); + verifyPrivateRegularFile(metadata, path); + if (metadata.size > maxStateBytes) { + throw new Error(`Private state file exceeds its size bound: ${path}`); + } + return readFileSync(descriptor, "utf8"); + } finally { + closeSync(descriptor); + } +} + +function syncParentDirectory(path: string): void { + if (process.platform === "win32") return; + const descriptor = openSync( + dirname(path), + constants.O_RDONLY | + (constants.O_DIRECTORY ?? 0) | + (constants.O_NOFOLLOW ?? 0), + ); + try { + fsyncSync(descriptor); + } finally { + closeSync(descriptor); + } +} + +function atomicPrivateWrite(path: string, contents: string): void { + const temporary = resolve( + dirname(path), + `.${path.split(/[\\/]/).at(-1)}.${randomUUID()}.tmp`, + ); + let descriptor: number | null = null; + let created = false; + try { + descriptor = openSync( + temporary, + constants.O_WRONLY | + constants.O_CREAT | + constants.O_EXCL | + (constants.O_NOFOLLOW ?? 0), + 0o600, + ); + created = true; + if (process.platform !== "win32") fchmodSync(descriptor, 0o600); + verifyPrivateRegularFile(fstatSync(descriptor), temporary); + writeFileSync(descriptor, contents, "utf8"); + fsyncSync(descriptor); + closeSync(descriptor); + descriptor = null; + renameSync(temporary, path); + created = false; + syncParentDirectory(path); + } finally { + if (descriptor !== null) closeSync(descriptor); + if (created) { + try { + unlinkSync(temporary); + } catch (error) { + if (!isNodeError(error, "ENOENT")) throw error; + } + } + } +} + +class DurableCoreStore { + readonly path: string; + #state: StoredCoreState; + + constructor(directory: string, identity: DurableRecoveryIdentity) { + try { + const metadata = lstatSync(directory); + if (metadata.isSymbolicLink() || !metadata.isDirectory()) { + throw new Error( + `Private state directory is not a real directory: ${directory}`, + ); + } + } catch (error) { + if (!isNodeError(error, "ENOENT")) throw error; + mkdirSync(directory, { recursive: true, mode: 0o700 }); + } + if (process.platform !== "win32") chmodSync(directory, 0o700); + verifyPrivateDirectory(directory); + this.path = resolve(directory, "control-plane-state.json"); + const stored = readPrivateFile(this.path); + if (stored !== null) { + const parsed = JSON.parse(stored) as unknown; + if (!isStoredCoreState(parsed, identity)) { + throw new Error( + "Control-plane state is invalid or does not match the requested PRP identity.", + ); + } + this.#state = parsed; + } else { + this.#state = initialCoreState(identity); + this.save(); + } + } + + get state(): StoredCoreState { + return this.#state; + } + + save(): void { + atomicPrivateWrite(this.path, `${JSON.stringify(this.#state, null, 2)}\n`); + } +} + +class PrpWebSocketConnection { + readonly socket: Duplex; + pendingChallenge: PendingChallenge | null = null; + secureChannel: SecureChannel | null = null; + lease: ConnectionLeaseRecord | null = null; + connectionId: string | null = null; + #buffer = Buffer.alloc(0); + #closed = false; + #onText: (text: string) => void | Promise; + #onClose: () => void; + #processing = Promise.resolve(); + + constructor( + socket: Duplex, + onText: (text: string) => void | Promise, + onClose: () => void, + ) { + this.socket = socket; + this.#onText = onText; + this.#onClose = onClose; + socket.on("data", (chunk: Buffer) => this.#consume(chunk)); + socket.on("close", () => { + if (!this.#closed) { + this.#closed = true; + this.#onClose(); + } + }); + socket.on("error", () => this.close()); + } + + acceptInitialData(data: Buffer): void { + if (data.length > 0) this.#consume(data); + } + + sendJson(value: unknown): void { + const wire = + this.secureChannel === null + ? value + : encryptSecureJson(this.secureChannel, value); + this.sendText(JSON.stringify(wire)); + } + + sendText(text: string): void { + if (this.#closed) { + return; + } + const payload = Buffer.from(text); + const header: number[] = [0x81]; + if (payload.length <= 125) { + header.push(payload.length); + } else if (payload.length <= 0xffff) { + header.push(126, (payload.length >>> 8) & 0xff, payload.length & 0xff); + } else { + const length = BigInt(payload.length); + header.push(127); + for (let shift = 56n; shift >= 0n; shift -= 8n) { + header.push(Number((length >> shift) & 0xffn)); + } + } + this.socket.write(Buffer.concat([Buffer.from(header), payload])); + } + + close(): void { + if (this.#closed) { + return; + } + this.#closed = true; + this.socket.destroy(); + this.#onClose(); + } + + #consume(chunk: Buffer): void { + this.#buffer = Buffer.concat([this.#buffer, chunk]); + while (this.#buffer.length >= 2) { + const first = this.#buffer[0]!; + const second = this.#buffer[1]!; + const opcode = first & 0x0f; + const masked = (second & 0x80) !== 0; + let length = second & 0x7f; + let cursor = 2; + if (length === 126) { + if (this.#buffer.length < 4) return; + length = this.#buffer.readUInt16BE(2); + cursor = 4; + } else if (length === 127) { + if (this.#buffer.length < 10) return; + const extended = this.#buffer.readBigUInt64BE(2); + if (extended > BigInt(maxFrameBytes)) { + this.close(); + return; + } + length = Number(extended); + cursor = 10; + } + if (length > maxFrameBytes || !masked) { + this.close(); + return; + } + if (this.#buffer.length < cursor + 4 + length) return; + const mask = this.#buffer.subarray(cursor, cursor + 4); + cursor += 4; + const payload = Buffer.from( + this.#buffer.subarray(cursor, cursor + length), + ); + this.#buffer = this.#buffer.subarray(cursor + length); + for (let index = 0; index < payload.length; index += 1) { + payload[index] = payload[index]! ^ mask[index % 4]!; + } + if (opcode === 0x1) { + const text = payload.toString("utf8"); + this.#processing = this.#processing + .then(() => this.#onText(text)) + .catch(() => this.close()); + } else if (opcode === 0x8) { + this.close(); + return; + } else if (opcode === 0x9) { + this.#sendControl(0x0a, payload); + } else if (opcode !== 0x0a) { + this.close(); + return; + } + } + } + + #sendControl(opcode: number, payload: Buffer): void { + if (payload.length > 125 || this.#closed) return; + this.socket.write( + Buffer.concat([Buffer.from([0x80 | opcode, payload.length]), payload]), + ); + } +} + +/** Authenticated, replay-safe PRP transport authority. Business operations are caller supplied. */ +export class DurablePrpControlPlane { + readonly #identity: DurableRecoveryIdentity; + readonly #store: DurableCoreStore; + #expectedRunnerVersion: string; + #expectedRunnerDigest: string; + #server: Server | null = null; + #connections = new Set(); + #pendingSemanticCalls = new Set(); + #port: number | null = null; + #onSemanticToolInput?: DurablePrpControlPlaneOptions["onSemanticToolInput"]; + #onCommittedEvent?: DurablePrpControlPlaneOptions["onCommittedEvent"]; + #connectionLeaseTtlMs: number; + + constructor(options: DurablePrpControlPlaneOptions) { + if ( + !Object.values(options.identity).every( + (value) => typeof value === "string" && stableIdPattern.test(value), + ) || + !stableIdPattern.test(options.expectedRunnerVersion) || + !runnerDigestPattern.test(options.expectedRunnerDigest) || + (options.connectionLeaseTtlMs !== undefined && + (!Number.isInteger(options.connectionLeaseTtlMs) || + options.connectionLeaseTtlMs < 60_000 || + options.connectionLeaseTtlMs > 24 * 60 * 60 * 1_000)) + ) { + throw new Error("Durable PRP control plane options are invalid."); + } + this.#identity = structuredClone(options.identity); + this.#store = new DurableCoreStore( + options.stateDirectory, + options.identity, + ); + this.#expectedRunnerVersion = options.expectedRunnerVersion; + this.#expectedRunnerDigest = options.expectedRunnerDigest; + this.#onSemanticToolInput = options.onSemanticToolInput; + this.#onCommittedEvent = options.onCommittedEvent; + this.#connectionLeaseTtlMs = options.connectionLeaseTtlMs ?? 60_000; + } + + get connectUrl(): string { + if (this.#port === null) { + throw new Error("Durable PRP control plane is not listening."); + } + return `ws://127.0.0.1:${this.#port}/durableRecovery/connect`; + } + + async start(port = 0): Promise { + if (this.#server !== null) { + throw new Error("Durable PRP control plane is already running."); + } + const server = createServer((_request, response) => { + response.writeHead(404).end(); + }); + this.#server = server; + server.on("upgrade", (request, socket, head) => + this.handleUpgrade(request, socket, "/durableRecovery/connect", head), + ); + await new Promise((resolveListen, rejectListen) => { + server.once("error", rejectListen); + server.listen(port, "127.0.0.1", () => { + server.off("error", rejectListen); + resolveListen(); + }); + }); + const address = server.address(); + if (address === null || typeof address === "string") { + throw new Error("Durable PRP control plane did not bind a TCP port."); + } + this.#port = address.port; + } + + async stop(): Promise { + for (const connection of this.#connections) { + connection.close(); + } + this.#connections.clear(); + const server = this.#server; + this.#server = null; + this.#port = null; + if (server !== null) { + await new Promise((resolveClose) => + server.close(() => resolveClose()), + ); + } + } + + /** Forces a resumable re-authentication after an immutable run attachment rotates. */ + disconnectActiveRunner(): void { + const connections = [...this.#connections]; + this.#connections.clear(); + for (const connection of connections) connection.close(); + } + + activeRunnerConnectionCount(): number { + return [...this.#connections].filter( + (connection) => connection.secureChannel !== null, + ).length; + } + + issueBootstrapTicket(ttlMs = 5_000): string { + if (!Number.isInteger(ttlMs) || ttlMs < 1_000 || ttlMs > 60_000) { + throw new Error("Durable PRP bootstrap TTL is invalid."); + } + this.#pruneCredentials(); + const ticket = `bootstrap_${randomUUID()}`; + const material = credentialMaterial(ticket); + const expiresAtUnixMs = Date.now() + ttlMs; + this.#store.state.tickets[material.credentialId] = { + recordId: `bootstrap_ticket_${randomUUID()}`, + credentialId: material.credentialId, + authKeyDigest: `sha256:${material.authKey.toString("hex")}`, + identity: structuredClone(this.#identity), + runnerVersion: this.#expectedRunnerVersion, + runnerDigest: this.#expectedRunnerDigest, + expiresAt: new Date(expiresAtUnixMs).toISOString(), + expiresAtUnixMs, + usedAt: null, + }; + this.#store.state.freshBootstraps += 1; + this.#store.save(); + return ticket; + } + + queueCommand( + type: string, + payload: Record = {}, + commandId?: string, + deliverImmediately = false, + ): DurableRecoveryCoreCommand { + if ( + !commandTypes.has(type) || + (commandId !== undefined && + (commandId.length > 160 || !stableIdPattern.test(commandId))) + ) { + throw new Error("Durable PRP command is invalid."); + } + if (commandId !== undefined) { + const existing = this.#store.state.commands.find( + (candidate) => candidate.commandId === commandId, + ); + if (existing !== undefined) { + if ( + existing.type !== type || + canonicalJson(existing.payload) !== canonicalJson(payload) + ) { + throw new Error( + "Durable PRP command replay conflicts with persisted state.", + ); + } + if (deliverImmediately && existing.status === "pending") { + for (const connection of this.#connections) { + if (connection.secureChannel !== null) + this.#sendNextCommand(connection); + } + } + return existing; + } + } + const controllerSeq = this.#store.state.commands.length + 1; + const command: DurableRecoveryCoreCommand = { + schema: "paperclip.prp.command.v1", + commandId: + commandId ?? `command_prp_${controllerSeq.toString().padStart(8, "0")}`, + controllerSeq, + type, + issuedAt: new Date().toISOString(), + payload, + status: "pending", + result: null, + }; + if ( + this.#store.state.commands.length >= maxCommands || + Buffer.byteLength(JSON.stringify(command)) > maxCommandBytes + ) { + throw new Error("Durable PRP command journal bound exceeded."); + } + this.#store.state.commands.push(command); + this.#store.save(); + if (deliverImmediately) { + for (const connection of this.#connections) { + if (connection.secureChannel !== null) { + this.#sendNextCommand(connection); + } + } + } + return command; + } + + /** Attach one HTTP upgrade to this run-bound authority. */ + handleUpgrade( + request: IncomingMessage, + socket: Duplex, + expectedPath = "/api/runner/v1/connect", + head: Buffer = Buffer.alloc(0), + ): void { + const requestPath = new URL(request.url ?? "/", "http://paperclip.invalid") + .pathname; + if (requestPath !== expectedPath) { + socket.write("HTTP/1.1 404 Not Found\r\nConnection: close\r\n\r\n"); + socket.destroy(); + return; + } + const websocketKey = request.headers["sec-websocket-key"]; + const decodedWebSocketKey = + typeof websocketKey === "string" + ? Buffer.from(websocketKey, "base64") + : Buffer.alloc(0); + if ( + request.method !== "GET" || + request.headers.upgrade?.toLowerCase() !== "websocket" || + request.headers["sec-websocket-version"] !== "13" || + typeof websocketKey !== "string" || + decodedWebSocketKey.length !== 16 || + decodedWebSocketKey.toString("base64") !== websocketKey + ) { + socket.write("HTTP/1.1 400 Bad Request\r\nConnection: close\r\n\r\n"); + socket.destroy(); + return; + } + const accept = createHash("sha1") + .update(`${websocketKey}${websocketGuid}`) + .digest("base64"); + socket.write( + [ + "HTTP/1.1 101 Switching Protocols", + "Upgrade: websocket", + "Connection: Upgrade", + `Sec-WebSocket-Accept: ${accept}`, + "\r\n", + ].join("\r\n"), + ); + let connection!: PrpWebSocketConnection; + connection = new PrpWebSocketConnection( + socket, + (text): Promise => this.#handleText(connection, text), + () => this.#connections.delete(connection), + ); + this.#connections.add(connection); + connection.acceptInitialData(head); + } + + async #handleText( + connection: PrpWebSocketConnection, + text: string, + ): Promise { + let envelope: Record; + try { + const wire = JSON.parse(text) as unknown; + envelope = + connection.secureChannel === null + ? (wire as Record) + : decryptSecureJson(connection.secureChannel, wire); + } catch { + this.#store.state.malformedFrames += 1; + this.#store.save(); + connection.close(); + return; + } + if ( + envelope.protocol !== protocol || + envelope.version !== protocolVersion + ) { + connection.close(); + return; + } + const kind = envelope.kind; + if (connection.secureChannel === null && kind === "auth_hello") { + this.#authHello(connection, envelope); + return; + } + if (connection.secureChannel === null && kind === "auth_response") { + this.#authResponse(connection, envelope); + return; + } + if ( + connection.secureChannel === null || + connection.lease === null || + connection.lease.revokedAt !== null || + connection.lease.expiresAtUnixMs <= Date.now() + ) { + connection.close(); + return; + } + if (kind === "event") { + await this.#event(connection, envelope); + return; + } + if (kind === "command_result") { + this.#commandResult(connection, envelope); + return; + } + if (kind !== "pong") { + connection.close(); + } + } + + #authorizeHello( + payload: Record, + ): PendingAuthorization | null { + this.#pruneCredentials(); + const credentialId = payload.credentialId; + if (typeof credentialId !== "string") return null; + const ticket = this.#store.state.tickets[credentialId]; + const lease = this.#store.state.leases[credentialId]; + const authorization: PendingAuthorization | null = + ticket !== undefined && + typeof ticket.recordId === "string" && + ticket.credentialId === credentialId && + ticket.usedAt === null && + ticket.expiresAtUnixMs > Date.now() + ? { + kind: "bootstrap", + recordId: ticket.recordId, + credentialId: ticket.credentialId, + authKey: authKeyFromDigest(ticket.authKeyDigest), + identity: structuredClone(ticket.identity), + runnerVersion: ticket.runnerVersion, + runnerDigest: ticket.runnerDigest, + expiresAt: ticket.expiresAt, + expiresAtUnixMs: ticket.expiresAtUnixMs, + recordSnapshot: canonicalJson(ticket), + } + : lease !== undefined && + typeof lease.recordId === "string" && + lease.credentialId === credentialId && + lease.revokedAt === null && + lease.expiresAtUnixMs > Date.now() + ? { + kind: "lease", + recordId: lease.recordId, + credentialId: lease.credentialId, + authKey: authKeyFromDigest(lease.authKeyDigest), + identity: structuredClone(lease.identity), + protocolVersion: lease.protocolVersion, + expiresAt: lease.expiresAt, + expiresAtUnixMs: lease.expiresAtUnixMs, + leaseId: lease.leaseId, + revocationEpoch: lease.revocationEpoch, + recordSnapshot: canonicalJson(lease), + } + : null; + if (authorization === null) return null; + const identity = authorization.identity; + if ( + payload.runnerInstanceId !== identity.runnerInstanceId || + payload.environmentLeaseId !== identity.environmentLeaseId || + payload.runId !== identity.runId || + payload.normalizedSessionId !== identity.normalizedSessionId || + payload.turnId !== identity.turnId || + payload.itemId !== identity.itemId || + payload.runnerVersion !== this.#expectedRunnerVersion || + payload.runnerDigest !== this.#expectedRunnerDigest || + payload.protocolMin !== 1 || + payload.protocolMax !== 1 || + (authorization.kind === "bootstrap" && + (authorization.runnerVersion !== this.#expectedRunnerVersion || + authorization.runnerDigest !== this.#expectedRunnerDigest)) || + (authorization.kind === "lease" && + authorization.protocolVersion !== protocolVersion) + ) { + return null; + } + return authorization; + } + + #pruneCredentials(): void { + const now = Date.now(); + for (const [credentialId, ticket] of Object.entries( + this.#store.state.tickets, + )) { + if (ticket.usedAt !== null || ticket.expiresAtUnixMs <= now) { + delete this.#store.state.tickets[credentialId]; + } + } + for (const [credentialId, lease] of Object.entries( + this.#store.state.leases, + )) { + if (lease.revokedAt !== null || lease.expiresAtUnixMs <= now) { + delete this.#store.state.leases[credentialId]; + } + } + } + + #reauthorizePendingChallenge( + pending: PendingChallenge, + now: number, + ): LiveAuthorization | null { + if (pending.deadlineUnixMs <= now) return null; + const expected = pending.authorization; + if (expected.kind === "bootstrap") { + const ticket = this.#store.state.tickets[expected.credentialId]; + if ( + ticket === undefined || + ticket.recordId !== expected.recordId || + ticket.credentialId !== expected.credentialId || + ticket.usedAt !== null || + ticket.expiresAtUnixMs <= now || + canonicalJson(ticket) !== expected.recordSnapshot + ) { + return null; + } + return { + kind: "bootstrap", + authKey: authKeyFromDigest(ticket.authKeyDigest), + ticket, + }; + } + + const lease = this.#store.state.leases[expected.credentialId]; + if ( + lease === undefined || + lease.recordId !== expected.recordId || + lease.credentialId !== expected.credentialId || + lease.revokedAt !== null || + lease.expiresAtUnixMs <= now || + canonicalJson(lease) !== expected.recordSnapshot + ) { + return null; + } + return { + kind: "lease", + authKey: authKeyFromDigest(lease.authKeyDigest), + lease, + }; + } + + #authHello( + connection: PrpWebSocketConnection, + envelope: Record, + ): void { + if (connection.pendingChallenge !== null) { + connection.close(); + return; + } + const payload = envelope.payload as Record | undefined; + if (payload === undefined || typeof payload.clientNonce !== "string") { + connection.close(); + return; + } + const authorization = this.#authorizeHello(payload); + if (authorization === null) { + connection.close(); + return; + } + const serverNonce = randomUUID(); + const challengePayload: Record = { + credentialId: authorization.credentialId, + credentialKind: authorization.kind, + clientNonce: payload.clientNonce, + serverNonce, + runnerInstanceId: payload.runnerInstanceId, + environmentLeaseId: payload.environmentLeaseId, + runId: payload.runId, + normalizedSessionId: payload.normalizedSessionId, + turnId: payload.turnId, + itemId: payload.itemId, + runnerVersion: payload.runnerVersion, + runnerDigest: payload.runnerDigest, + selectedVersion: protocolVersion, + credentialLeaseId: + authorization.kind === "lease" ? authorization.leaseId : null, + credentialExpiresAt: authorization.expiresAt, + credentialExpiresAtUnixMs: authorization.expiresAtUnixMs, + revocationEpoch: + authorization.kind === "lease" ? authorization.revocationEpoch : 0, + }; + const canonicalChallenge = canonicalJson(challengePayload); + const serverProof = domainHmac( + authorization.authKey, + "paperclip-runner-server-proof-v1", + [Buffer.from(canonicalChallenge)], + ).toString("hex"); + connection.pendingChallenge = { + authorization, + deadlineUnixMs: Math.min( + authorization.expiresAtUnixMs, + Date.now() + authChallengeTtlMs, + ), + canonicalChallenge, + serverProof, + clientNonce: payload.clientNonce, + serverNonce, + }; + connection.sendJson({ + protocol, + version: protocolVersion, + kind: "auth_challenge", + payload: { ...challengePayload, serverProof }, + }); + } + + #authResponse( + connection: PrpWebSocketConnection, + envelope: Record, + ): void { + const pending = connection.pendingChallenge; + const payload = envelope.payload as Record | undefined; + if ( + pending === null || + payload === undefined || + payload.credentialId !== pending.authorization.credentialId || + payload.clientNonce !== pending.clientNonce || + payload.serverNonce !== pending.serverNonce + ) { + connection.close(); + return; + } + // WebSocket callbacks run synchronously on the mock core's event loop. Re-reading, + // validating, consuming, minting, and persisting here forms one state mutation + // boundary, so another proof cannot interleave with bootstrap consumption. + const authorization = this.#reauthorizePendingChallenge( + pending, + Date.now(), + ); + if (authorization === null) { + connection.close(); + return; + } + const expectedClientProof = domainHmac( + authorization.authKey, + "paperclip-runner-client-proof-v1", + [ + Buffer.from(pending.canonicalChallenge), + Buffer.from(pending.serverProof), + ], + ); + if (!proofMatches(expectedClientProof, payload.clientProof)) { + connection.close(); + return; + } + const clientProof = expectedClientProof.toString("hex"); + let leaseToken: string | null = null; + let lease: ConnectionLeaseRecord; + if (authorization.kind === "bootstrap") { + authorization.ticket.usedAt = new Date().toISOString(); + leaseToken = `lease_${randomUUID()}`; + const material = credentialMaterial(leaseToken); + const expiresAtUnixMs = Date.now() + this.#connectionLeaseTtlMs; + lease = { + recordId: `connection_lease_record_${randomUUID()}`, + credentialId: material.credentialId, + authKeyDigest: `sha256:${material.authKey.toString("hex")}`, + leaseId: `connection_lease_${randomUUID()}`, + identity: structuredClone(this.#identity), + protocolVersion, + expiresAt: new Date(expiresAtUnixMs).toISOString(), + expiresAtUnixMs, + revocationEpoch: 0, + revokedAt: null, + }; + this.#store.state.leases[material.credentialId] = lease; + this.#store.save(); + } else { + lease = authorization.lease; + } + connection.pendingChallenge = null; + connection.lease = lease; + connection.connectionId = `connection_${this.#store.state.connectionCount + 1}`; + connection.secureChannel = createSecureChannel( + authorization.authKey, + pending.canonicalChallenge, + pending.serverProof, + clientProof, + ); + for (const active of this.#connections) { + if (active !== connection && active.secureChannel !== null) + active.close(); + } + this.#welcome(connection, leaseToken); + } + + #welcome( + connection: PrpWebSocketConnection, + leaseToken: string | null, + ): void { + const lease = connection.lease; + if (lease === null || connection.connectionId === null) { + connection.close(); + return; + } + + this.#store.state.connectionCount += 1; + this.#store.state.lastLeaseId = lease.leaseId; + this.#store.state.lastLeaseExpiresAt = lease.expiresAt; + + const pending = this.#nextPendingCommand(); + for (const command of pending) { + this.#store.state.commandDeliveryCounts[command.commandId] = + (this.#store.state.commandDeliveryCounts[command.commandId] ?? 0) + 1; + } + this.#store.save(); + connection.sendJson({ + protocol, + version: protocolVersion, + envelopeId: `welcome_${this.#store.state.connectionCount}`, + kind: "welcome", + runnerInstanceId: this.#identity.runnerInstanceId, + environmentLeaseId: this.#identity.environmentLeaseId, + runId: this.#identity.runId, + normalizedSessionId: this.#identity.normalizedSessionId, + turnId: this.#identity.turnId, + itemId: this.#identity.itemId, + connectionId: connection.connectionId, + connectionLeaseId: lease.leaseId, + sentAt: new Date().toISOString(), + payload: { + selectedVersion: 1, + heartbeatIntervalMs: 250, + connectionLeaseId: lease.leaseId, + ...(leaseToken === null ? {} : { connectionLeaseToken: leaseToken }), + connectionLeaseExpiresAt: lease.expiresAt, + connectionLeaseExpiresAtUnixMs: lease.expiresAtUnixMs, + connectionLeaseRevocationEpoch: lease.revocationEpoch, + leaseBinding: { + runnerInstanceId: this.#identity.runnerInstanceId, + environmentLeaseId: this.#identity.environmentLeaseId, + runId: this.#identity.runId, + normalizedSessionId: this.#identity.normalizedSessionId, + protocolVersion, + }, + maxFrameBytes, + maxBatchEvents: 100, + ackedSourceSeq: this.#store.state.ackedSourceSeq, + pendingCommands: pending.map(this.#wireCommand), + }, + }); + } + + #wireCommand( + command: DurableRecoveryCoreCommand, + ): Omit { + const { status: _status, result: _result, ...wire } = command; + return wire; + } + + #nextPendingCommand(): DurableRecoveryCoreCommand[] { + const command = this.#store.state.commands.find( + (candidate) => candidate.status === "pending", + ); + return command === undefined ? [] : [command]; + } + + #controlEnvelope( + connection: PrpWebSocketConnection, + envelopeId: string, + kind: string, + payload: Record, + ): Record { + if (connection.lease === null || connection.connectionId === null) { + throw new Error( + "Cannot send control data before transport authentication.", + ); + } + return { + protocol, + version: protocolVersion, + envelopeId, + kind, + runnerInstanceId: this.#identity.runnerInstanceId, + environmentLeaseId: this.#identity.environmentLeaseId, + runId: this.#identity.runId, + normalizedSessionId: this.#identity.normalizedSessionId, + turnId: this.#identity.turnId, + itemId: this.#identity.itemId, + connectionId: connection.connectionId, + connectionLeaseId: connection.lease.leaseId, + sentAt: new Date().toISOString(), + payload, + }; + } + + #sendNextCommand(connection: PrpWebSocketConnection): void { + const [command] = this.#nextPendingCommand(); + if (command === undefined) return; + this.#store.state.commandDeliveryCounts[command.commandId] = + (this.#store.state.commandDeliveryCounts[command.commandId] ?? 0) + 1; + this.#store.save(); + connection.sendJson( + this.#controlEnvelope( + connection, + `command_${command.commandId}_${this.#store.state.commandDeliveryCounts[command.commandId]}`, + "command", + this.#wireCommand(command), + ), + ); + } + + #commandResult( + connection: PrpWebSocketConnection, + envelope: Record, + ): void { + const result = envelope.payload as Record | undefined; + const commandId = result?.commandId; + if (result === undefined || typeof commandId !== "string") { + connection.close(); + return; + } + const command = this.#store.state.commands.find( + (candidate) => candidate.commandId === commandId, + ); + if (command === undefined) { + connection.close(); + return; + } + const status = result.status; + if ( + status !== "completed" && + status !== "failed" && + status !== "rejected" + ) { + connection.close(); + return; + } + if (command.status !== "pending") { + if (canonicalJson(command.result) !== canonicalJson(result)) { + connection.close(); + return; + } + this.#store.state.duplicateCommandResults += 1; + this.#store.save(); + this.#sendNextCommand(connection); + return; + } + command.status = status; + command.result = structuredClone(result); + this.#store.save(); + this.#sendNextCommand(connection); + } + + async #event( + connection: PrpWebSocketConnection, + envelope: Record, + ): Promise { + const validated = validatePrpEvent(envelope.payload); + if (!validated.ok) { + connection.close(); + return; + } + const event = validated.event; + const sourceSeq = event?.sourceSeq; + const sourceEventId = event?.sourceEventId; + const eventType = event?.eventType; + const priority = event?.priority; + if ( + typeof sourceSeq !== "number" || + typeof sourceEventId !== "string" || + typeof eventType !== "string" || + (priority !== 0 && priority !== 1 && priority !== 2) || + event?.sourceInstanceId !== this.#identity.runnerInstanceId || + event.runId !== this.#identity.runId || + event.normalizedSessionId !== this.#identity.normalizedSessionId || + event.turnId !== this.#identity.turnId || + event.itemId !== this.#identity.itemId + ) { + connection.close(); + return; + } + const semantic = (event.payload as Record | undefined) + ?.semantic_tool as Record | undefined; + const semanticCorrelation = semantic?.correlation as + Record | undefined; + const isSemanticInput = + eventType === "semantic_tool.input" || eventType === "mcp_app.tool_input"; + if ( + isSemanticInput && + (this.#onSemanticToolInput === undefined || + semantic?.phase !== "input" || + typeof semantic.callId !== "string" || + typeof semantic.operationId !== "string" || + !Object.prototype.hasOwnProperty.call(semantic, "input") || + typeof semantic.content !== "object" || + semantic.content === null || + (semantic.content as Record).digest !== + digestPaperclipSemanticContent(semantic.input) || + semanticCorrelation?.runId !== this.#identity.runId || + semanticCorrelation.normalizedSessionId !== + this.#identity.normalizedSessionId || + semanticCorrelation.turnId !== this.#identity.turnId || + semanticCorrelation.itemId !== this.#identity.itemId) + ) { + connection.close(); + return; + } + const existing = this.#store.state.committedEvents.find( + (candidate) => candidate.sourceEventId === sourceEventId, + ); + if (existing !== undefined) { + if (canonicalJson(existing.envelope) !== canonicalJson(envelope)) { + connection.close(); + return; + } + } else if (sourceSeq !== this.#store.state.ackedSourceSeq + 1) { + connection.close(); + return; + } + + // The caller's durable commit is the acknowledgement authority. A crash + // after that idempotent commit but before the local cursor save is safe: + // the runner replays the event, the caller observes a duplicate, and only + // then do we advance the cumulative cursor. Reversing this order can make + // an uncommitted event disappear from the runner outbox permanently. + try { + await this.#onCommittedEvent?.(event); + } catch { + connection.close(); + return; + } + + if (existing !== undefined) { + existing.deliveryCount += 1; + this.#store.state.replayDeliveries += 1; + } else { + this.#store.state.committedEvents.push({ + sourceSeq, + sourceEventId, + eventType, + priority, + envelope: structuredClone(envelope), + deliveryCount: 1, + logicalEffectCount: 1, + }); + if (this.#store.state.committedEvents.length > maxCommittedEventWindow) { + this.#store.state.committedEvents.splice( + 0, + this.#store.state.committedEvents.length - maxCommittedEventWindow, + ); + } + this.#store.state.ackedSourceSeq = sourceSeq; + } + this.#store.save(); + + if ( + isSemanticInput && + this.#onSemanticToolInput && + semantic !== undefined && + typeof semantic.callId === "string" && + typeof semantic.operationId === "string" + ) { + const call = { + callId: semantic.callId, + operationId: semantic.operationId, + input: semantic.input, + sourceEventId, + sourceEventType: eventType, + correlation: { + runId: this.#identity.runId, + normalizedSessionId: this.#identity.normalizedSessionId, + turnId: this.#identity.turnId, + itemId: + typeof event.itemId === "string" + ? event.itemId + : this.#identity.itemId, + }, + }; + const commandId = `command_tool_${createHash("sha256") + .update(`${this.#identity.runId}\0${call.callId}`) + .digest("hex") + .slice(0, 32)}`; + const alreadyQueued = this.#store.state.commands.some( + (command) => command.commandId === commandId, + ); + if (!alreadyQueued && !this.#pendingSemanticCalls.has(commandId)) { + this.#pendingSemanticCalls.add(commandId); + const queueResult = (result: unknown, isError: boolean): void => { + try { + this.queueCommand( + "semantic_tool.result", + { ...call, result, isError }, + commandId, + true, + ); + } catch { + // A result that cannot fit the bounded durable journal cannot be + // acknowledged as a usable tool response. Force a reconnect so + // the caller can recover or terminate the run explicitly. + this.disconnectActiveRunner(); + } + }; + void this.#onSemanticToolInput(call) + .then((outcome) => + queueResult(outcome.result, outcome.isError === true), + ) + .catch(() => + queueResult({ code: "semantic_tool_bridge_failed" }, true), + ) + .finally(() => this.#pendingSemanticCalls.delete(commandId)); + } + } + + connection.sendJson( + this.#controlEnvelope( + connection, + `ack_${this.#store.state.ackedSourceSeq}`, + "ack", + { + ackedSourceSeq: this.#store.state.ackedSourceSeq, + }, + ), + ); + } +} diff --git a/packages/paperclip-runner/src/control-plane/prp-transport-types.ts b/packages/paperclip-runner/src/control-plane/prp-transport-types.ts new file mode 100644 index 0000000000..6a9920573a --- /dev/null +++ b/packages/paperclip-runner/src/control-plane/prp-transport-types.ts @@ -0,0 +1,29 @@ +export interface DurableRecoveryIdentity { + runnerInstanceId: string; + environmentLeaseId: string; + runId: string; + normalizedSessionId: string; + turnId: string; + itemId: string; +} + +export interface DurableRecoveryCoreCommand { + schema: "paperclip.prp.command.v1"; + commandId: string; + controllerSeq: number; + type: string; + issuedAt: string; + payload: Record; + status: "pending" | "completed" | "failed" | "rejected"; + result: Record | null; +} + +export interface DurableRecoveryCommittedEvent { + sourceSeq: number; + sourceEventId: string; + eventType: string; + priority: 0 | 1 | 2; + envelope: Record; + deliveryCount: number; + logicalEffectCount: number; +} diff --git a/packages/paperclip-runner/src/index.ts b/packages/paperclip-runner/src/index.ts index c43eec2060..8c8f4d86db 100644 --- a/packages/paperclip-runner/src/index.ts +++ b/packages/paperclip-runner/src/index.ts @@ -1,6 +1,11 @@ export * from "./catalog/index.js"; export * from "./contracts/completion-result.js"; export * from "./contracts/question-set.js"; +export { + DurablePrpControlPlane, + type DurablePrpControlPlaneOptions, +} from "./control-plane/durable-prp-control-plane.js"; +export type { DurableRecoveryIdentity } from "./control-plane/prp-transport-types.js"; export * from "./protocol/replay-contract.js"; export * from "./protocol/result-normalization.js"; export * from "./reducer/session-reducer.js"; diff --git a/server/package.json b/server/package.json index c4bb50fc7a..747d52b622 100644 --- a/server/package.json +++ b/server/package.json @@ -35,12 +35,13 @@ "dev": "tsx src/index.ts", "dev:watch": "cross-env PAPERCLIP_MIGRATION_PROMPT=never PAPERCLIP_MIGRATION_AUTO_APPLY=true tsx ./scripts/dev-watch.ts", "prepare:ui-dist": "bash ../scripts/prepare-server-ui-dist.sh", - "build": "tsc && mkdir -p dist/onboarding-assets dist/built-ins dist/services/scripts && cp -R src/onboarding-assets/. dist/onboarding-assets/ && cp -R src/built-ins/. dist/built-ins/ && cp -R src/services/scripts/. dist/services/scripts/ && node scripts/write-build-stamp.mjs", + "build": "pnpm run prepare:runner-vendor && tsc && mkdir -p dist/onboarding-assets dist/built-ins dist/services/scripts dist/vendor/paperclip-runner && cp -R src/onboarding-assets/. dist/onboarding-assets/ && cp -R src/built-ins/. dist/built-ins/ && cp -R src/services/scripts/. dist/services/scripts/ && cp -R ../packages/paperclip-runner/dist/. dist/vendor/paperclip-runner/ && node scripts/write-build-stamp.mjs", "prepack": "pnpm run prepare:ui-dist && pnpm run build", "postpack": "rm -rf ui-dist", "clean": "rm -rf dist", "start": "node dist/index.js", - "typecheck": "pnpm --filter @paperclipai/plugin-sdk ensure-build-deps && tsc --noEmit" + "prepare:runner-vendor": "pnpm --filter @paperclipai/paperclip-runner build", + "typecheck": "pnpm run prepare:runner-vendor && pnpm --filter @paperclipai/plugin-sdk ensure-build-deps && tsc --noEmit" }, "dependencies": { "@aws-sdk/client-s3": "^3.1115.0", @@ -83,6 +84,7 @@ "zod": "^4.4.3" }, "devDependencies": { + "@paperclipai/paperclip-runner": "workspace:*", "@types/express": "^5.0.0", "@types/express-serve-static-core": "^5.1.3", "@types/jsdom": "^30.0.0", diff --git a/server/src/__tests__/server-package-build-script.test.ts b/server/src/__tests__/server-package-build-script.test.ts index 92b410ce70..f8f0d45a7c 100644 --- a/server/src/__tests__/server-package-build-script.test.ts +++ b/server/src/__tests__/server-package-build-script.test.ts @@ -2,7 +2,9 @@ import { readFileSync } from "node:fs"; import { fileURLToPath } from "node:url"; import { describe, expect, it } from "vitest"; -const packageJsonPath = fileURLToPath(new URL("../../package.json", import.meta.url)); +const packageJsonPath = fileURLToPath( + new URL("../../package.json", import.meta.url), +); describe("server package build script", () => { it("builds the compiled package entry during prepack", () => { @@ -10,7 +12,9 @@ describe("server package build script", () => { scripts?: Record; }; - expect(packageJson.scripts?.prepack).toBe("pnpm run prepare:ui-dist && pnpm run build"); + expect(packageJson.scripts?.prepack).toBe( + "pnpm run prepare:ui-dist && pnpm run build", + ); }); it("copies static runtime asset directories into dist", () => { @@ -19,8 +23,33 @@ describe("server package build script", () => { }; const buildScript = packageJson.scripts?.build ?? ""; - expect(buildScript).toContain("mkdir -p dist/onboarding-assets dist/built-ins"); - expect(buildScript).toContain("cp -R src/onboarding-assets/. dist/onboarding-assets/"); + expect(buildScript).toContain( + "mkdir -p dist/onboarding-assets dist/built-ins", + ); + expect(buildScript).toContain( + "cp -R src/onboarding-assets/. dist/onboarding-assets/", + ); expect(buildScript).toContain("cp -R src/built-ins/. dist/built-ins/"); }); + + it("vendors the private runner runtime without a production workspace dependency", () => { + const packageJson = JSON.parse(readFileSync(packageJsonPath, "utf8")) as { + scripts?: Record; + dependencies?: Record; + devDependencies?: Record; + }; + + expect( + packageJson.dependencies?.["@paperclipai/paperclip-runner"], + ).toBeUndefined(); + expect(packageJson.devDependencies?.["@paperclipai/paperclip-runner"]).toBe( + "workspace:*", + ); + expect(packageJson.scripts?.["prepare:runner-vendor"]).toBe( + "pnpm --filter @paperclipai/paperclip-runner build", + ); + expect(packageJson.scripts?.build).toContain( + "cp -R ../packages/paperclip-runner/dist/. dist/vendor/paperclip-runner/", + ); + }); }); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index c4242eb552..9477122b7f 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -114,6 +114,7 @@ const { }; const feedbackServiceFactoryMock = vi.fn(() => feedbackExportServiceMock); const fakeServer = { + on: vi.fn().mockReturnThis(), once: vi.fn().mockReturnThis(), off: vi.fn().mockReturnThis(), listen: vi.fn((_port: number, _host: string, callback?: () => void) => { diff --git a/server/src/index.ts b/server/src/index.ts index aa414603dc..e455de481b 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -40,6 +40,7 @@ import { } from "./services/managed-config.js"; import { setupEnvironmentCustomImageTerminalWebSocketServer } from "./realtime/environment-custom-image-terminal-ws.js"; import { setupLiveEventsWebSocketServer } from "./realtime/live-events-ws.js"; +import { setupRunnerPrpWebSocketServer } from "./realtime/runner-prp-ws.js"; import { cloudActorHeaderSourceFromHeaders, resolveCloudTenantActor } from "./middleware/auth.js"; import { feedbackService, @@ -890,6 +891,7 @@ export async function startServer(): Promise { process.env.PAPERCLIP_RUNTIME_API_CANDIDATES_JSON = JSON.stringify(runtimeApiCandidates); process.env.PAPERCLIP_API_URL = configuredApiUrl; + setupRunnerPrpWebSocketServer(server, { apiUrl: configuredApiUrl }); setupEnvironmentCustomImageTerminalWebSocketServer(server, db as any, { pluginWorkerManager, }); diff --git a/server/src/realtime/runner-prp-ws.test.ts b/server/src/realtime/runner-prp-ws.test.ts new file mode 100644 index 0000000000..5121f455c4 --- /dev/null +++ b/server/src/realtime/runner-prp-ws.test.ts @@ -0,0 +1,93 @@ +import { createServer } from "node:http"; +import { PassThrough } from "node:stream"; + +import type { DurablePrpControlPlane } from "@paperclipai/paperclip-runner"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +import { + registerRunnerPrpAuthority, + runnerPrpWebSocketInternals, + setupRunnerPrpWebSocketServer, +} from "./runner-prp-ws.js"; + +describe("runner PRP websocket route", () => { + afterEach(() => runnerPrpWebSocketInternals.resetForTests()); + + it("routes only the registered run and releases it by generation", async () => { + const server = createServer(); + setupRunnerPrpWebSocketServer(server, { apiUrl: "http://127.0.0.1:3210" }); + const handleUpgrade = vi.fn(); + const runId = "00000000-0000-4000-8000-000000000777"; + const registration = await registerRunnerPrpAuthority({ + companyId: "company-1", + runId, + authority: { handleUpgrade } as unknown as DurablePrpControlPlane, + }); + + expect(registration.connectUrl).toBe( + `ws://127.0.0.1:3210/api/runner/v1/connect/${runId}`, + ); + expect( + runnerPrpWebSocketInternals.activeRegistration({ + companyId: "company-1", + runId, + }), + ).toBe(true); + + const socket = new PassThrough(); + const request = { url: `/api/runner/v1/connect/${runId}`, headers: {} }; + server.emit("upgrade", request, socket, Buffer.alloc(0)); + expect(request).toMatchObject({ paperclipWebSocketHandled: true }); + expect(handleUpgrade).toHaveBeenCalledWith( + expect.objectContaining({ url: `/api/runner/v1/connect/${runId}` }), + socket, + `/api/runner/v1/connect/${runId}`, + expect.any(Buffer), + ); + + await registration.release(); + expect( + runnerPrpWebSocketInternals.activeRegistration({ + companyId: "company-1", + runId, + }), + ).toBe(false); + server.close(); + }); + + it.each([ + ["/api/runner/v1/connect/not-a-run", "400 Bad Request"], + [ + "/api/runner/v1/connect/00000000-0000-4000-8000-000000000778", + "404 Not Found", + ], + ])("fails closed for %s", (path, expectedStatus) => { + const server = createServer(); + setupRunnerPrpWebSocketServer(server, { apiUrl: "http://127.0.0.1:3211" }); + const socket = new PassThrough(); + const writes: Buffer[] = []; + socket.on("data", (chunk) => writes.push(Buffer.from(chunk))); + server.emit("upgrade", { url: path, headers: {} }, socket, Buffer.alloc(0)); + expect(Buffer.concat(writes).toString("utf8")).toContain(expectedStatus); + server.close(); + }); + + it("does not replace an active registration", async () => { + const server = createServer(); + setupRunnerPrpWebSocketServer(server, { apiUrl: "http://127.0.0.1:3212" }); + const runId = "00000000-0000-4000-8000-000000000779"; + const authority = { + handleUpgrade: vi.fn(), + } as unknown as DurablePrpControlPlane; + const first = await registerRunnerPrpAuthority({ + companyId: "company-1", + runId, + authority, + }); + await expect( + registerRunnerPrpAuthority({ companyId: "company-2", runId, authority }), + ).rejects.toThrow("runner_prp_authority_already_registered"); + await first.release(); + server.close(); + }); +}); diff --git a/server/src/realtime/runner-prp-ws.ts b/server/src/realtime/runner-prp-ws.ts new file mode 100644 index 0000000000..61676c2241 --- /dev/null +++ b/server/src/realtime/runner-prp-ws.ts @@ -0,0 +1,131 @@ +import type { IncomingMessage, Server } from "node:http"; +import type { Duplex } from "node:stream"; + +import type { DurablePrpControlPlane } from "../vendor/paperclip-runner/index.js"; + +import { logger } from "../middleware/logger.js"; + +const CONNECT_PATH_PREFIX = "/api/runner/v1/connect/"; +const UUID_PATTERN = + /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; + +interface RegisteredAuthority { + readonly companyId: string; + readonly authority: DurablePrpControlPlane; + readonly generation: symbol; +} + +interface RunnerPrpUpgradeRequest extends IncomingMessage { + paperclipWebSocketHandled?: boolean; +} + +const registrations = new Map(); +let loopbackOrigin: string | null = null; + +function rejectUpgrade( + socket: Duplex, + status: "400 Bad Request" | "404 Not Found", +): void { + if (socket.destroyed) return; + try { + socket.end( + `HTTP/1.1 ${status}\r\nConnection: close\r\nContent-Length: 0\r\n\r\n`, + ); + } catch (error) { + logger.warn( + { errorName: error instanceof Error ? error.name : typeof error }, + "failed to reject runner PRP websocket upgrade", + ); + socket.destroy(); + } +} + +export function setupRunnerPrpWebSocketServer( + server: Server, + options: { readonly apiUrl: string }, +): void { + const apiUrl = new URL(options.apiUrl); + if (!["http:", "https:"].includes(apiUrl.protocol)) { + throw new Error("runner_prp_websocket_api_url_invalid"); + } + apiUrl.protocol = apiUrl.protocol === "https:" ? "wss:" : "ws:"; + apiUrl.username = ""; + apiUrl.password = ""; + apiUrl.pathname = ""; + apiUrl.search = ""; + apiUrl.hash = ""; + loopbackOrigin = apiUrl.toString().replace(/\/$/, ""); + server.on( + "upgrade", + (request: IncomingMessage, socket: Duplex, head: Buffer) => { + const url = new URL(request.url ?? "/", "http://paperclip.invalid"); + if (!url.pathname.startsWith(CONNECT_PATH_PREFIX)) return; + + const ownedRequest = request as RunnerPrpUpgradeRequest; + if (ownedRequest.paperclipWebSocketHandled) return; + ownedRequest.paperclipWebSocketHandled = true; + socket.on("error", (error) => { + logger.warn( + { errorName: error.name }, + "runner PRP websocket upgrade socket failed", + ); + }); + + const runId = url.pathname.slice(CONNECT_PATH_PREFIX.length); + if (!UUID_PATTERN.test(runId)) { + rejectUpgrade(socket, "400 Bad Request"); + return; + } + const registration = registrations.get(runId); + if (!registration) { + rejectUpgrade(socket, "404 Not Found"); + return; + } + registration.authority.handleUpgrade(request, socket, url.pathname, head); + }, + ); +} + +export async function registerRunnerPrpAuthority(input: { + readonly companyId: string; + readonly runId: string; + readonly authority: DurablePrpControlPlane; +}): Promise<{ readonly connectUrl: string; release(): Promise }> { + if (loopbackOrigin === null) { + throw new Error("runner_prp_websocket_server_not_configured"); + } + if (!UUID_PATTERN.test(input.runId) || input.companyId.length === 0) { + throw new Error("runner_prp_authority_binding_invalid"); + } + if (registrations.has(input.runId)) { + throw new Error("runner_prp_authority_already_registered"); + } + const generation = Symbol(input.runId); + registrations.set(input.runId, { + companyId: input.companyId, + authority: input.authority, + generation, + }); + return { + connectUrl: `${loopbackOrigin}${CONNECT_PATH_PREFIX}${input.runId}`, + release: async () => { + if (registrations.get(input.runId)?.generation === generation) { + registrations.delete(input.runId); + } + }, + }; +} + +export const runnerPrpWebSocketInternals = { + connectPathPrefix: CONNECT_PATH_PREFIX, + activeRegistration(input: { + readonly companyId: string; + readonly runId: string; + }): boolean { + return registrations.get(input.runId)?.companyId === input.companyId; + }, + resetForTests(): void { + registrations.clear(); + loopbackOrigin = null; + }, +}; diff --git a/server/src/services/native-runtime/native-run-coordinator-store.ts b/server/src/services/native-runtime/native-run-coordinator-store.ts new file mode 100644 index 0000000000..24de0daba4 --- /dev/null +++ b/server/src/services/native-runtime/native-run-coordinator-store.ts @@ -0,0 +1,460 @@ +import { createHash } from "node:crypto"; + +import { and, desc, eq, lt } from "drizzle-orm"; + +import type { Db } from "@paperclipai/db"; +import { + heartbeatRunEvents, + heartbeatRuns, + nativeRunFinalizations, + nativeRunResults, +} from "@paperclipai/db"; +import { + type PrpEvent, + type PrpStructuredRunResult, + type PrpTerminalState, + validatePrpEvent, + validatePrpStructuredRunResult, +} from "../../vendor/paperclip-runner/index.js"; + +export interface NativeRunStoreBinding { + readonly companyId: string; + readonly issueId: string; + readonly runId: string; + readonly agentId: string; + readonly normalizedSessionId: string; + readonly runnerSourceInstanceId: string; + readonly completionContractId: string; + readonly completionContractSha256: string; +} + +export interface CompleteNativeRunInput { + readonly result: PrpStructuredRunResult; + readonly terminal: PrpTerminalState; + readonly turnId?: string; + readonly callerResultId?: string; + readonly callerDedupeKey?: string; +} + +function canonicalJson(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(canonicalJson).join(",")}]`; + if (typeof value === "object" && value !== null) { + const record = value as Record; + return `{${Object.keys(record) + .sort() + .map((key) => `${JSON.stringify(key)}:${canonicalJson(record[key])}`) + .join(",")}}`; + } + return JSON.stringify(value) ?? "null"; +} + +function sha256(value: unknown): string { + return `sha256:${createHash("sha256").update(canonicalJson(value)).digest("hex")}`; +} + +function assertTerminal(value: PrpTerminalState): void { + if ( + value.schema !== "paperclip.prp.terminal.v1" || + !["completed", "failed", "interrupted", "cancelled"].includes( + value.turnTerminalState, + ) || + !["succeeded", "failed", "cancelled"].includes(value.runTerminalState) || + !["done", "blocked", "needs_review", "yielded"].includes( + value.reportedWorkDisposition, + ) + ) { + throw new Error("native_terminal_schema_invalid"); + } +} + +/** Durable DB boundary used by the hidden PRP coordinator. */ +export class NativeRunCoordinatorStore { + readonly #db: Db; + readonly #binding: NativeRunStoreBinding; + + constructor(db: Db, binding: NativeRunStoreBinding) { + this.#db = db; + this.#binding = structuredClone(binding); + } + + async appendEvent(value: PrpEvent): Promise<{ + readonly disposition: "committed" | "duplicate"; + readonly cursor: number; + readonly highestContiguousSourceSeq: number; + }> { + const validated = validatePrpEvent(value); + if (!validated.ok) throw new Error("native_event_schema_invalid"); + const event = validated.event; + if ( + event.runId !== this.#binding.runId || + event.normalizedSessionId !== this.#binding.normalizedSessionId || + event.sourceKind !== "runner" || + event.sourceInstanceId !== this.#binding.runnerSourceInstanceId + ) { + throw new Error("native_event_binding_mismatch"); + } + const canonicalPayload = event as unknown as Record; + const payloadDigest = sha256(canonicalPayload); + + return this.#db.transaction(async (tx) => { + const [run] = await tx + .select() + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.id, this.#binding.runId), + eq(heartbeatRuns.companyId, this.#binding.companyId), + eq(heartbeatRuns.agentId, this.#binding.agentId), + eq(heartbeatRuns.nativeIssueId, this.#binding.issueId), + eq( + heartbeatRuns.nativeSessionId, + this.#binding.normalizedSessionId, + ), + eq(heartbeatRuns.runtimeMode, "native"), + ), + ) + .for("update") + .limit(1); + if (!run) throw new Error("native_event_run_not_authorized"); + + const [existing] = await tx + .select() + .from(heartbeatRunEvents) + .where( + and( + eq(heartbeatRunEvents.runId, this.#binding.runId), + eq(heartbeatRunEvents.sourceEventId, event.sourceEventId), + ), + ) + .limit(1); + if (existing) { + if ( + existing.sourcePayloadSha256 !== payloadDigest || + existing.sourceInstanceId !== event.sourceInstanceId || + existing.sourceSeq !== event.sourceSeq + ) { + throw new Error("native_event_replay_conflict"); + } + const [latest] = await tx + .select({ sourceSeq: heartbeatRunEvents.sourceSeq }) + .from(heartbeatRunEvents) + .where( + and( + eq(heartbeatRunEvents.runId, this.#binding.runId), + eq(heartbeatRunEvents.sourceInstanceId, event.sourceInstanceId), + ), + ) + .orderBy(desc(heartbeatRunEvents.sourceSeq)) + .limit(1); + return { + disposition: "duplicate" as const, + cursor: existing.seq, + highestContiguousSourceSeq: latest?.sourceSeq ?? event.sourceSeq, + }; + } + + const [previous] = await tx + .select({ sourceSeq: heartbeatRunEvents.sourceSeq }) + .from(heartbeatRunEvents) + .where( + and( + eq(heartbeatRunEvents.runId, this.#binding.runId), + eq(heartbeatRunEvents.sourceInstanceId, event.sourceInstanceId), + ), + ) + .orderBy(desc(heartbeatRunEvents.sourceSeq)) + .limit(1); + const expectedSourceSeq = (previous?.sourceSeq ?? 0) + 1; + if (event.sourceSeq !== expectedSourceSeq) { + throw new Error("native_event_source_gap"); + } + + const cursor = run.nextEventSeq; + const [inserted] = await tx + .insert(heartbeatRunEvents) + .values({ + companyId: this.#binding.companyId, + runId: this.#binding.runId, + agentId: this.#binding.agentId, + seq: cursor, + eventType: event.eventType, + stream: "system", + level: event.eventType.includes("failed") ? "error" : "info", + payload: { prpEvent: canonicalPayload }, + sourceInstanceId: event.sourceInstanceId, + sourceEventId: event.sourceEventId, + sourceSeq: event.sourceSeq, + sourcePayloadSha256: payloadDigest, + protocolSchemaVersion: event.schemaVersion, + }) + .returning({ seq: heartbeatRunEvents.seq }); + if (!inserted) throw new Error("native_event_not_persisted"); + await tx + .update(heartbeatRuns) + .set({ nextEventSeq: cursor + 1, updatedAt: new Date() }) + .where(eq(heartbeatRuns.id, this.#binding.runId)); + return { + disposition: "committed" as const, + cursor: inserted.seq, + highestContiguousSourceSeq: event.sourceSeq, + }; + }); + } + + async completeRun(input: CompleteNativeRunInput): Promise<{ + readonly disposition: "committed" | "duplicate"; + readonly resultId: string; + }> { + const validated = validatePrpStructuredRunResult(input.result); + if (!validated.ok) throw new Error("native_result_schema_invalid"); + assertTerminal(input.terminal); + if ( + validated.result.reportedWorkDisposition !== + input.terminal.reportedWorkDisposition + ) { + throw new Error("native_result_terminal_disposition_mismatch"); + } + const canonical = { + result: validated.result, + terminal: input.terminal, + turnId: input.turnId ?? null, + }; + const canonicalSha256 = sha256(canonical); + const serverFingerprint = sha256({ + runId: this.#binding.runId, + completionContractSha256: this.#binding.completionContractSha256, + canonicalSha256, + }); + + return this.#db.transaction(async (tx) => { + const [run] = await tx + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, this.#binding.runId)) + .for("update") + .limit(1); + if ( + !run || + run.companyId !== this.#binding.companyId || + run.agentId !== this.#binding.agentId || + run.nativeIssueId !== this.#binding.issueId || + run.runtimeMode !== "native" || + run.completionContractId !== this.#binding.completionContractId || + run.completionContractSha256 !== this.#binding.completionContractSha256 + ) { + throw new Error("native_result_run_not_authorized"); + } + + const [existing] = await tx + .select() + .from(nativeRunResults) + .where( + and( + eq(nativeRunResults.runId, this.#binding.runId), + eq(nativeRunResults.schemaStatus, "accepted"), + ), + ) + .limit(1); + if (existing) { + if (existing.canonicalSha256 !== canonicalSha256) { + throw new Error("native_result_replay_conflict"); + } + return { disposition: "duplicate" as const, resultId: existing.id }; + } + + const [inserted] = await tx + .insert(nativeRunResults) + .values({ + companyId: this.#binding.companyId, + issueId: this.#binding.issueId, + runId: this.#binding.runId, + turnId: input.turnId ?? null, + completionContractId: this.#binding.completionContractId, + callerResultId: input.callerResultId ?? null, + callerDedupeKey: input.callerDedupeKey ?? null, + serverFingerprint, + schemaStatus: "accepted", + resultJson: canonical as unknown as Record, + canonicalSha256, + }) + .returning({ id: nativeRunResults.id }); + if (!inserted) throw new Error("native_result_not_persisted"); + await tx + .insert(nativeRunFinalizations) + .values({ + runId: this.#binding.runId, + companyId: this.#binding.companyId, + issueId: this.#binding.issueId, + phase: "workspace_finalizing", + resultId: inserted.id, + }) + .onConflictDoUpdate({ + target: nativeRunFinalizations.runId, + set: { + phase: "workspace_finalizing", + resultId: inserted.id, + failureCode: null, + failureDetail: null, + nextAttemptAt: null, + updatedAt: new Date(), + }, + }); + await tx + .update(heartbeatRuns) + .set({ + nativePhase: "workspace_finalizing", + nativePhaseUpdatedAt: new Date(), + updatedAt: new Date(), + }) + .where(eq(heartbeatRuns.id, this.#binding.runId)); + return { disposition: "committed" as const, resultId: inserted.id }; + }); + } + + /** Rebuild the accepted result/finalization record from durable PRP events. */ + async reconcileTerminalEvent(event: PrpEvent): Promise<{ + readonly disposition: "committed" | "duplicate"; + readonly resultId: string; + } | null> { + if (event.eventType !== "run.terminal") return null; + if ( + event.runId !== this.#binding.runId || + event.normalizedSessionId !== this.#binding.normalizedSessionId || + event.sourceInstanceId !== this.#binding.runnerSourceInstanceId + ) { + throw new Error("native_terminal_binding_mismatch"); + } + const terminal = event.payload as PrpTerminalState; + assertTerminal(terminal); + const [proposed] = await this.#db + .select({ payload: heartbeatRunEvents.payload }) + .from(heartbeatRunEvents) + .where( + and( + eq(heartbeatRunEvents.runId, this.#binding.runId), + eq( + heartbeatRunEvents.sourceInstanceId, + this.#binding.runnerSourceInstanceId, + ), + eq(heartbeatRunEvents.eventType, "run.result.proposed"), + lt(heartbeatRunEvents.sourceSeq, event.sourceSeq), + ), + ) + .orderBy(desc(heartbeatRunEvents.sourceSeq)) + .limit(1); + const proposedEnvelope = ( + proposed?.payload as Record | undefined + )?.prpEvent as Record | undefined; + if (proposedEnvelope?.payload === undefined) { + throw new Error("native_terminal_result_missing"); + } + return this.completeRun({ + result: proposedEnvelope.payload as PrpStructuredRunResult, + terminal, + turnId: event.turnId, + callerDedupeKey: `prp-terminal:${event.sourceEventId}`, + }); + } + + async claimFinalization(input: { + readonly leaseOwner: string; + readonly leaseTtlMs?: number; + }): Promise<{ readonly attempt: number; readonly resultId: string }> { + const leaseTtlMs = input.leaseTtlMs ?? 30_000; + if ( + !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,239}$/.test(input.leaseOwner) || + !Number.isInteger(leaseTtlMs) || + leaseTtlMs < 1_000 || + leaseTtlMs > 300_000 + ) { + throw new Error("native_finalization_lease_invalid"); + } + return this.#db.transaction(async (tx) => { + const [row] = await tx + .select() + .from(nativeRunFinalizations) + .where( + and( + eq(nativeRunFinalizations.runId, this.#binding.runId), + eq(nativeRunFinalizations.companyId, this.#binding.companyId), + eq(nativeRunFinalizations.issueId, this.#binding.issueId), + ), + ) + .for("update") + .limit(1); + if (!row?.resultId || ["completed", "failed"].includes(row.phase)) { + throw new Error("native_finalization_not_claimable"); + } + if (row.nextAttemptAt && row.nextAttemptAt.getTime() > Date.now()) { + throw new Error("native_finalization_retry_not_due"); + } + if ( + row.leaseOwner && + row.leaseOwner !== input.leaseOwner && + row.leaseExpiresAt && + row.leaseExpiresAt.getTime() > Date.now() + ) { + throw new Error("native_finalization_lease_conflict"); + } + if ( + row.leaseOwner === input.leaseOwner && + row.leaseExpiresAt && + row.leaseExpiresAt.getTime() > Date.now() + ) { + await tx + .update(nativeRunFinalizations) + .set({ + leaseExpiresAt: new Date(Date.now() + leaseTtlMs), + updatedAt: new Date(), + }) + .where(eq(nativeRunFinalizations.runId, this.#binding.runId)); + return { attempt: row.attempt, resultId: row.resultId }; + } + const attempt = row.attempt + 1; + await tx + .update(nativeRunFinalizations) + .set({ + attempt, + leaseOwner: input.leaseOwner, + leaseExpiresAt: new Date(Date.now() + leaseTtlMs), + updatedAt: new Date(), + }) + .where(eq(nativeRunFinalizations.runId, this.#binding.runId)); + return { attempt, resultId: row.resultId }; + }); + } + + async markFinalizationRetry(input: { + readonly leaseOwner: string; + readonly failureCode: string; + readonly retryAfterMs: number; + }): Promise { + if ( + !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,239}$/.test(input.failureCode) || + !Number.isInteger(input.retryAfterMs) || + input.retryAfterMs < 1_000 || + input.retryAfterMs > 24 * 60 * 60 * 1_000 + ) { + throw new Error("native_finalization_retry_invalid"); + } + const [updated] = await this.#db + .update(nativeRunFinalizations) + .set({ + leaseOwner: null, + leaseExpiresAt: null, + failureCode: input.failureCode, + nextAttemptAt: new Date(Date.now() + input.retryAfterMs), + updatedAt: new Date(), + }) + .where( + and( + eq(nativeRunFinalizations.runId, this.#binding.runId), + eq(nativeRunFinalizations.companyId, this.#binding.companyId), + eq(nativeRunFinalizations.issueId, this.#binding.issueId), + eq(nativeRunFinalizations.leaseOwner, input.leaseOwner), + ), + ) + .returning({ runId: nativeRunFinalizations.runId }); + if (!updated) throw new Error("native_finalization_lease_lost"); + } +} diff --git a/server/src/services/native-runtime/runner-prp-coordinator.test.ts b/server/src/services/native-runtime/runner-prp-coordinator.test.ts new file mode 100644 index 0000000000..3b5349a27b --- /dev/null +++ b/server/src/services/native-runtime/runner-prp-coordinator.test.ts @@ -0,0 +1,421 @@ +import { randomUUID } from "node:crypto"; +import { mkdtempSync, rmSync } from "node:fs"; +import { createServer } from "node:http"; +import { tmpdir } from "node:os"; +import { resolve } from "node:path"; + +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; + +import { + agents, + companies, + completionContracts, + createDb, + heartbeatRunEvents, + heartbeatRuns, + issues, + nativeRunFinalizations, + nativeRunResults, +} from "@paperclipai/db"; +import type { + PrpEvent, + PrpStructuredRunResult, + PrpTerminalState, +} from "@paperclipai/paperclip-runner"; + +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "../../__tests__/helpers/embedded-postgres.js"; +import { + runnerPrpWebSocketInternals, + setupRunnerPrpWebSocketServer, +} from "../../realtime/runner-prp-ws.js"; +import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js"; +import { runnerPrpCoordinator } from "./runner-prp-coordinator.js"; +import { PaperclipRunnerSemanticAuthority } from "./runner-semantic-authority.js"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported + ? describe + : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping runner coordinator tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +interface SeededNativeRun { + companyId: string; + issueId: string; + agentId: string; + runId: string; + runnerInstanceId: string; + sessionId: string; + completionContractId: string; + completionContractSha256: string; +} + +const result: PrpStructuredRunResult = { + schema: "paperclip.run_result.v1", + reportedWorkDisposition: "done", + summary: "The hidden runner completed the bounded task.", + completionClaim: { + contractRevision: "contract-v1", + objectiveSatisfied: true, + criteria: [], + remainingWork: [], + }, + evidence: [], + verification: [{ commandOrCheck: "coordinator test", status: "passed" }], + attentionRequests: [], + artifacts: [], +}; + +const terminal: PrpTerminalState = { + schema: "paperclip.prp.terminal.v1", + turnTerminalState: "completed", + runTerminalState: "succeeded", + reportedWorkDisposition: "done", +}; + +function runnerEvent(seed: SeededNativeRun, sourceSeq = 1): PrpEvent { + return { + schema: "paperclip.prp.event.v1", + sourceEventId: `event-${sourceSeq}`, + sourceSeq, + sourceInstanceId: seed.runnerInstanceId, + sourceKind: "runner", + runId: seed.runId, + normalizedSessionId: seed.sessionId, + turnId: "turn-1", + itemId: "item-1", + eventType: "turn.started", + schemaVersion: 1, + priority: 1, + emittedAt: "2026-08-25T18:00:00.000Z", + payload: {}, + }; +} + +describeEmbeddedPostgres("hidden runner PRP coordinator", () => { + let db!: ReturnType; + let tempDb: Awaited< + ReturnType + > | null = null; + const scratchDirectories: string[] = []; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase( + "paperclip-runner-coordinator-", + ); + db = createDb(tempDb.connectionString); + }, 20_000); + + afterEach(async () => { + runnerPrpWebSocketInternals.resetForTests(); + await db.delete(nativeRunFinalizations); + await db.delete(nativeRunResults); + await db.delete(heartbeatRunEvents); + await db.update(issues).set({ executionRunId: null }); + await db.delete(heartbeatRuns); + await db.delete(completionContracts); + await db.delete(issues); + await db.delete(agents); + await db.delete(companies); + for (const directory of scratchDirectories.splice(0)) { + rmSync(directory, { recursive: true, force: true }); + } + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function seedNativeRun(): Promise { + const companyId = randomUUID(); + const issueId = randomUUID(); + const agentId = randomUUID(); + const runId = randomUUID(); + const runnerInstanceId = randomUUID(); + const sessionId = randomUUID(); + const completionContractId = randomUUID(); + const completionContractSha256 = `sha256:${"c".repeat(64)}`; + await db.insert(companies).values({ + id: companyId, + name: "Runner Test Company", + issuePrefix: `R${companyId.replaceAll("-", "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Codex runner", + role: "engineer", + status: "running", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + }); + await db.insert(issues).values({ + id: issueId, + companyId, + identifier: `RUN-${runId.slice(0, 8)}`, + title: "Exercise the hidden native coordinator", + description: "Verify transport and durable server boundaries.", + status: "in_progress", + priority: "medium", + workMode: "standard", + assigneeAgentId: agentId, + }); + await db.insert(completionContracts).values({ + id: completionContractId, + companyId, + issueId, + revision: 1, + schemaVersion: "paperclip.completion-contract.v1", + policyVersion: "policy-v1", + risk: "low", + completionAuthority: "runner", + incompleteCriteriaPolicy: "fail_closed", + contractJson: { criteria: [] }, + canonicalSha256: completionContractSha256, + createdByActorType: "system", + createdByActorId: "test", + }); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + status: "running", + runtimeMode: "native", + nativeIssueId: issueId, + runnerInstanceId, + nativeSessionId: sessionId, + driverKind: "codex", + driverVersion: "0.3.0", + completionContractId, + completionContractSha256, + nativePhase: "observed", + }); + await db + .update(issues) + .set({ executionRunId: runId }) + .where(eq(issues.id, issueId)); + return { + companyId, + issueId, + agentId, + runId, + runnerInstanceId, + sessionId, + completionContractId, + completionContractSha256, + }; + } + + function store(seed: SeededNativeRun): NativeRunCoordinatorStore { + return new NativeRunCoordinatorStore(db, { + companyId: seed.companyId, + issueId: seed.issueId, + runId: seed.runId, + agentId: seed.agentId, + normalizedSessionId: seed.sessionId, + runnerSourceInstanceId: seed.runnerInstanceId, + completionContractId: seed.completionContractId, + completionContractSha256: seed.completionContractSha256, + }); + } + + it("registers only an exact Codex native binding and exposes read-only tools", async () => { + const seed = await seedNativeRun(); + const server = createServer(); + setupRunnerPrpWebSocketServer(server, { apiUrl: "http://127.0.0.1:3213" }); + const stateRoot = mkdtempSync(resolve(tmpdir(), "paperclip-runner-state-")); + scratchDirectories.push(stateRoot); + const coordinator = runnerPrpCoordinator(db, { stateRoot }); + await expect( + coordinator.prepare({ + ...seed, + companyId: randomUUID(), + normalizedSessionId: seed.sessionId, + environmentLeaseId: "environment-lease-1", + turnId: "turn-1", + itemId: "item-1", + runnerVersion: "0.3.0", + runnerDigest: `sha256:${"a".repeat(64)}`, + }), + ).rejects.toThrow("runner_prp_run_not_authorized"); + const prepared = await coordinator.prepare({ + ...seed, + normalizedSessionId: seed.sessionId, + environmentLeaseId: "environment-lease-1", + turnId: "turn-1", + itemId: "item-1", + runnerVersion: "0.3.0", + runnerDigest: `sha256:${"a".repeat(64)}`, + }); + + expect(prepared.connectUrl).toBe( + `ws://127.0.0.1:3213/api/runner/v1/connect/${seed.runId}`, + ); + expect(prepared.bootstrapTicket).toMatch(/^bootstrap_/); + expect(prepared.semanticTools.map((tool) => tool.name)).toEqual([ + "get_task_context", + "get_task_history", + "list_documents", + "read_document", + "list_document_revisions", + ]); + expect( + runnerPrpWebSocketInternals.activeRegistration({ + companyId: seed.companyId, + runId: seed.runId, + }), + ).toBe(true); + await expect( + coordinator.prepare({ + ...seed, + normalizedSessionId: seed.sessionId, + environmentLeaseId: "environment-lease-1", + turnId: "turn-1", + itemId: "item-1", + runnerVersion: "0.3.0", + runnerDigest: `sha256:${"a".repeat(64)}`, + }), + ).rejects.toThrow("runner_prp_authority_already_registered"); + await prepared.release(); + expect( + runnerPrpWebSocketInternals.activeRegistration({ + companyId: seed.companyId, + runId: seed.runId, + }), + ).toBe(false); + server.close(); + }); + + it("rechecks task ownership and returns semantic receipts", async () => { + const seed = await seedNativeRun(); + const authority = new PaperclipRunnerSemanticAuthority(db, { + companyId: seed.companyId, + issueId: seed.issueId, + runId: seed.runId, + agentId: seed.agentId, + }); + const call = { + callId: "call-1", + operationId: "get_task_context", + correlation: { + runId: seed.runId, + normalizedSessionId: seed.sessionId, + turnId: "turn-1", + itemId: "item-1", + }, + input: {}, + }; + const allowed = await authority.dispatch(call); + expect(allowed).toMatchObject({ + ok: true, + operationId: "get_task_context", + value: { activeTask: { id: seed.issueId }, run: { id: seed.runId } }, + inputReceipt: { phase: "input" }, + resultReceipt: { phase: "result" }, + }); + + await db + .update(issues) + .set({ assigneeAgentId: null }) + .where(eq(issues.id, seed.issueId)); + const denied = await authority.dispatch({ ...call, callId: "call-2" }); + expect(denied).toMatchObject({ + ok: false, + error: { code: "task_ownership_denied", retryable: false }, + resultReceipt: { phase: "result" }, + }); + }); + + it("persists events and results idempotently and leases finalization", async () => { + const seed = await seedNativeRun(); + const nativeStore = store(seed); + const event = runnerEvent(seed); + await expect(nativeStore.appendEvent(event)).resolves.toMatchObject({ + disposition: "committed", + cursor: 1, + highestContiguousSourceSeq: 1, + }); + await expect(nativeStore.appendEvent(event)).resolves.toMatchObject({ + disposition: "duplicate", + cursor: 1, + }); + await expect( + nativeStore.appendEvent({ ...event, priority: 2 }), + ).rejects.toThrow("native_event_replay_conflict"); + await expect(nativeStore.appendEvent(runnerEvent(seed, 3))).rejects.toThrow( + "native_event_source_gap", + ); + + const resultEvent = { + ...runnerEvent(seed, 2), + eventType: "run.result.proposed", + payload: result, + } as PrpEvent; + const terminalEvent = { + ...runnerEvent(seed, 3), + eventType: "run.terminal", + payload: terminal, + } as PrpEvent; + await nativeStore.appendEvent(resultEvent); + await nativeStore.appendEvent(terminalEvent); + const firstResult = await nativeStore.reconcileTerminalEvent(terminalEvent); + if (!firstResult) + throw new Error("terminal reconciliation returned no result"); + expect(firstResult.disposition).toBe("committed"); + await expect(nativeStore.appendEvent(event)).resolves.toMatchObject({ + disposition: "duplicate", + highestContiguousSourceSeq: 3, + }); + await expect( + nativeStore.completeRun({ + result, + terminal, + turnId: "turn-1", + callerDedupeKey: "result-1", + }), + ).resolves.toEqual({ ...firstResult, disposition: "duplicate" }); + await expect( + nativeStore.completeRun({ + result: { ...result, summary: "Conflicting retry" }, + terminal, + turnId: "turn-1", + callerDedupeKey: "result-1", + }), + ).rejects.toThrow("native_result_replay_conflict"); + + await expect( + nativeStore.claimFinalization({ leaseOwner: "server-1" }), + ).resolves.toMatchObject({ attempt: 1, resultId: firstResult.resultId }); + await expect( + nativeStore.claimFinalization({ leaseOwner: "server-1" }), + ).resolves.toMatchObject({ attempt: 1, resultId: firstResult.resultId }); + await expect( + nativeStore.claimFinalization({ leaseOwner: "server-2" }), + ).rejects.toThrow("native_finalization_lease_conflict"); + await nativeStore.markFinalizationRetry({ + leaseOwner: "server-1", + failureCode: "workspace_busy", + retryAfterMs: 1_000, + }); + await expect( + nativeStore.claimFinalization({ leaseOwner: "server-2" }), + ).rejects.toThrow("native_finalization_retry_not_due"); + await db + .update(nativeRunFinalizations) + .set({ nextAttemptAt: new Date(0) }) + .where(eq(nativeRunFinalizations.runId, seed.runId)); + await expect( + nativeStore.claimFinalization({ leaseOwner: "server-2" }), + ).resolves.toMatchObject({ attempt: 2, resultId: firstResult.resultId }); + }); +}); diff --git a/server/src/services/native-runtime/runner-prp-coordinator.ts b/server/src/services/native-runtime/runner-prp-coordinator.ts new file mode 100644 index 0000000000..daabc6dd2a --- /dev/null +++ b/server/src/services/native-runtime/runner-prp-coordinator.ts @@ -0,0 +1,261 @@ +import { resolve } from "node:path"; + +import { and, eq } from "drizzle-orm"; + +import type { Db } from "@paperclipai/db"; +import { agents, heartbeatRuns, issues } from "@paperclipai/db"; +import { + DurablePrpControlPlane, + type PaperclipSemanticToolDefinition, + type PrpStructuredRunResult, + type PrpTerminalState, +} from "../../vendor/paperclip-runner/index.js"; + +import { registerRunnerPrpAuthority } from "../../realtime/runner-prp-ws.js"; +import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js"; +import { PaperclipRunnerSemanticAuthority } from "./runner-semantic-authority.js"; + +const UUID_PATTERN = + /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; +const STABLE_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,239}$/; +const RUNNER_DIGEST_PATTERN = /^sha256:[0-9a-f]{64}$/; + +export interface PrepareRunnerPrpSessionInput { + readonly companyId: string; + readonly issueId: string; + readonly runId: string; + readonly agentId: string; + readonly runnerInstanceId: string; + readonly environmentLeaseId: string; + readonly normalizedSessionId: string; + readonly turnId: string; + readonly itemId: string; + readonly runnerVersion: string; + readonly runnerDigest: string; + readonly bootstrapTtlMs?: number; + readonly connectionLeaseTtlMs?: number; +} + +export interface PreparedRunnerPrpSession { + readonly connectUrl: string; + /** One-use secret. Pass it only through the runner's protected bootstrap channel. */ + readonly bootstrapTicket: string; + readonly semanticTools: readonly PaperclipSemanticToolDefinition[]; + queueCommand( + type: string, + payload?: Record, + commandId?: string, + ): { readonly commandId: string; readonly controllerSeq: number }; + completeRun(input: { + readonly result: PrpStructuredRunResult; + readonly terminal: PrpTerminalState; + readonly turnId?: string; + readonly callerResultId?: string; + readonly callerDedupeKey?: string; + }): Promise<{ + readonly disposition: "committed" | "duplicate"; + readonly resultId: string; + }>; + release(): Promise; +} + +function clampDuration( + value: number | undefined, + fallback: number, + minimum: number, + maximum: number, +): number { + if (value === undefined) return fallback; + if (!Number.isInteger(value) || value < minimum || value > maximum) { + throw new Error("runner_prp_credential_ttl_invalid"); + } + return value; +} + +function validateInput(input: PrepareRunnerPrpSessionInput): void { + for (const id of [ + input.companyId, + input.issueId, + input.runId, + input.agentId, + input.runnerInstanceId, + input.normalizedSessionId, + ]) { + if (!UUID_PATTERN.test(id)) throw new Error("runner_prp_binding_invalid"); + } + for (const id of [input.environmentLeaseId, input.turnId, input.itemId]) { + if (!STABLE_ID_PATTERN.test(id)) + throw new Error("runner_prp_binding_invalid"); + } + if ( + !STABLE_ID_PATTERN.test(input.runnerVersion) || + !RUNNER_DIGEST_PATTERN.test(input.runnerDigest) + ) { + throw new Error("runner_prp_runner_identity_invalid"); + } +} + +/** + * Creates the hidden, run-bound PRP authority. This module does not select a + * runtime or start runnerd. The flagged adapter owns those actions later. + */ +export function runnerPrpCoordinator( + db: Db, + options: { + readonly stateRoot: string; + }, +) { + const stateRoot = resolve(options.stateRoot); + + return { + prepare: async ( + input: PrepareRunnerPrpSessionInput, + ): Promise => { + validateInput(input); + const bootstrapTtlMs = clampDuration( + input.bootstrapTtlMs, + 30_000, + 1_000, + 60_000, + ); + const connectionLeaseTtlMs = clampDuration( + input.connectionLeaseTtlMs, + 60 * 60 * 1_000, + 60_000, + 24 * 60 * 60 * 1_000, + ); + + const [binding] = await db + .select({ run: heartbeatRuns, issue: issues, agent: agents }) + .from(heartbeatRuns) + .innerJoin( + issues, + and( + eq(issues.id, heartbeatRuns.nativeIssueId), + eq(issues.companyId, heartbeatRuns.companyId), + ), + ) + .innerJoin( + agents, + and( + eq(agents.id, heartbeatRuns.agentId), + eq(agents.companyId, heartbeatRuns.companyId), + ), + ) + .where( + and( + eq(heartbeatRuns.id, input.runId), + eq(heartbeatRuns.companyId, input.companyId), + eq(heartbeatRuns.agentId, input.agentId), + eq(heartbeatRuns.nativeIssueId, input.issueId), + eq(heartbeatRuns.runnerInstanceId, input.runnerInstanceId), + eq(heartbeatRuns.nativeSessionId, input.normalizedSessionId), + eq(heartbeatRuns.runtimeMode, "native"), + ), + ) + .limit(1); + if ( + !binding || + !["queued", "running"].includes(binding.run.status) || + binding.run.driverKind !== "codex" || + !binding.run.completionContractId || + !binding.run.completionContractSha256 || + binding.issue.assigneeAgentId !== input.agentId || + binding.issue.executionRunId !== input.runId || + ["paused", "terminated", "pending_approval", "error"].includes( + binding.agent.status, + ) + ) { + throw new Error("runner_prp_run_not_authorized"); + } + + const semanticAuthority = new PaperclipRunnerSemanticAuthority(db, { + companyId: input.companyId, + issueId: input.issueId, + runId: input.runId, + agentId: input.agentId, + }); + const semanticTools = await semanticAuthority.listAlwaysAvailableTools(); + const nativeStore = new NativeRunCoordinatorStore(db, { + companyId: input.companyId, + issueId: input.issueId, + runId: input.runId, + agentId: input.agentId, + normalizedSessionId: input.normalizedSessionId, + runnerSourceInstanceId: input.runnerInstanceId, + completionContractId: binding.run.completionContractId, + completionContractSha256: binding.run.completionContractSha256, + }); + const authority = new DurablePrpControlPlane({ + stateDirectory: resolve(stateRoot, input.runId), + identity: { + runnerInstanceId: input.runnerInstanceId, + environmentLeaseId: input.environmentLeaseId, + runId: input.runId, + normalizedSessionId: input.normalizedSessionId, + turnId: input.turnId, + itemId: input.itemId, + }, + expectedRunnerVersion: input.runnerVersion, + expectedRunnerDigest: input.runnerDigest, + connectionLeaseTtlMs, + onCommittedEvent: async (event) => { + await nativeStore.appendEvent(event); + await nativeStore.reconcileTerminalEvent(event); + }, + onSemanticToolInput: async (call) => { + const result = await semanticAuthority.dispatch({ + callId: call.callId, + operationId: call.operationId, + correlation: call.correlation, + input: call.input, + }); + return { result, isError: !result.ok }; + }, + }); + + const registration = await registerRunnerPrpAuthority({ + companyId: input.companyId, + runId: input.runId, + authority, + }); + let bootstrapTicket: string; + try { + bootstrapTicket = authority.issueBootstrapTicket(bootstrapTtlMs); + } catch (error) { + authority.disconnectActiveRunner(); + await registration.release(); + throw error; + } + let released = false; + return { + connectUrl: registration.connectUrl, + bootstrapTicket, + semanticTools, + queueCommand: (type, payload = {}, commandId) => { + if (released) throw new Error("runner_prp_session_released"); + const command = authority.queueCommand( + type, + payload, + commandId, + true, + ); + return { + commandId: command.commandId, + controllerSeq: command.controllerSeq, + }; + }, + completeRun: (completeInput) => { + if (released) throw new Error("runner_prp_session_released"); + return nativeStore.completeRun(completeInput); + }, + release: async () => { + if (released) return; + released = true; + authority.disconnectActiveRunner(); + await registration.release(); + }, + }; + }, + }; +} diff --git a/server/src/services/native-runtime/runner-semantic-authority.ts b/server/src/services/native-runtime/runner-semantic-authority.ts new file mode 100644 index 0000000000..4af878c955 --- /dev/null +++ b/server/src/services/native-runtime/runner-semantic-authority.ts @@ -0,0 +1,351 @@ +import { and, desc, eq, isNull } from "drizzle-orm"; + +import type { Db } from "@paperclipai/db"; +import { + agents, + documentRevisions, + documents, + heartbeatRuns, + issueComments, + issueDocuments, + issues, +} from "@paperclipai/db"; +import { + PaperclipSemanticDispatcher, + type PaperclipJsonValue, + type PaperclipSemanticActionBinding, + type PaperclipSemanticActionId, + type PaperclipSemanticAuthorizationRecord, + type PaperclipSemanticRunContext, + type PaperclipSemanticToolCall, + type PaperclipSemanticToolDefinition, + type PaperclipSemanticToolResult, +} from "../../vendor/paperclip-runner/index.js"; + +export interface PaperclipRunnerSemanticBinding { + readonly companyId: string; + readonly issueId: string; + readonly runId: string; + readonly agentId: string; +} + +const READ_OPERATION_IDS = [ + "get_task_context", + "get_task_history", + "list_documents", + "read_document", + "list_document_revisions", +] as const satisfies readonly PaperclipSemanticActionId[]; + +type BoundContext = { + readonly run: typeof heartbeatRuns.$inferSelect; + readonly issue: typeof issues.$inferSelect; + readonly agent: typeof agents.$inferSelect; +}; + +function boundedLimit(value: unknown): number { + return typeof value === "number" && Number.isInteger(value) + ? Math.max(1, Math.min(value, 200)) + : 50; +} + +function requiredString(value: unknown): string { + if (typeof value !== "string" || value.length === 0 || value.length > 240) { + throw new Error("paperclip_runner_semantic_input_invalid"); + } + return value; +} + +function jsonValue(value: unknown): PaperclipJsonValue { + return JSON.parse(JSON.stringify(value)) as PaperclipJsonValue; +} + +function activeAgentStatus(status: string): "active" | "inactive" { + return ["paused", "terminated", "pending_approval", "error"].includes(status) + ? "inactive" + : "active"; +} + +/** + * Run-scoped semantic authority for the hidden native coordinator. + * This first server slice binds only same-task read operations. A catalog + * entry remains undiscoverable until a later PR adds its guarded binding. + */ +export class PaperclipRunnerSemanticAuthority { + readonly #db: Db; + readonly #binding: PaperclipRunnerSemanticBinding; + readonly #dispatcher: PaperclipSemanticDispatcher; + + constructor(db: Db, binding: PaperclipRunnerSemanticBinding) { + this.#db = db; + this.#binding = structuredClone(binding); + this.#dispatcher = new PaperclipSemanticDispatcher({ + contextProvider: (runId) => this.#context(runId), + bindings: READ_OPERATION_IDS.map((operationId) => + this.#readBinding(operationId), + ), + }); + } + + listAlwaysAvailableTools(): Promise< + readonly PaperclipSemanticToolDefinition[] + > { + return this.#dispatcher.listAlwaysAvailableTools(this.#binding.runId); + } + + dispatch( + call: Omit, + ): Promise { + return this.#dispatcher.dispatch({ ...call, runId: this.#binding.runId }); + } + + authorizationRecords(): readonly PaperclipSemanticAuthorizationRecord[] { + return this.#dispatcher.authorizationRecords(); + } + + #readBinding( + operationId: (typeof READ_OPERATION_IDS)[number], + ): PaperclipSemanticActionBinding { + return { + operationId, + execute: async (invocation) => { + const context = await this.#loadBoundContext(); + this.#assertActiveContext(context, true); + const input = invocation.input; + switch (operationId) { + case "get_task_context": + return { + value: jsonValue({ + company: { id: this.#binding.companyId }, + actor: { + id: context.agent.id, + name: context.agent.name, + role: context.agent.role, + title: context.agent.title, + capabilities: context.agent.capabilities, + }, + activeTask: { + id: context.issue.id, + identifier: context.issue.identifier, + title: context.issue.title, + description: context.issue.description, + status: context.issue.status, + statusVersion: context.issue.statusVersion, + priority: context.issue.priority, + workMode: context.issue.workMode, + parentId: context.issue.parentId, + projectId: context.issue.projectId, + goalId: context.issue.goalId, + }, + run: { + id: context.run.id, + status: context.run.status, + invocationSource: context.run.invocationSource, + }, + }), + references: [{ kind: "task", id: context.issue.id }], + }; + case "get_task_history": { + const rows = await this.#db + .select({ + id: issueComments.id, + body: issueComments.body, + authorAgentId: issueComments.authorAgentId, + authorUserId: issueComments.authorUserId, + createdAt: issueComments.createdAt, + }) + .from(issueComments) + .where( + and( + eq(issueComments.companyId, this.#binding.companyId), + eq(issueComments.issueId, this.#binding.issueId), + isNull(issueComments.deletedAt), + ), + ) + .orderBy(desc(issueComments.createdAt)) + .limit(boundedLimit(input.limit)); + return { + value: jsonValue({ comments: rows.reverse() }), + references: [{ kind: "task", id: context.issue.id }], + }; + } + case "list_documents": { + const rows = await this.#db + .select({ + key: issueDocuments.key, + id: documents.id, + title: documents.title, + format: documents.format, + latestRevisionId: documents.latestRevisionId, + latestRevisionNumber: documents.latestRevisionNumber, + lockedAt: documents.lockedAt, + updatedAt: documents.updatedAt, + }) + .from(issueDocuments) + .innerJoin(documents, eq(documents.id, issueDocuments.documentId)) + .where( + and( + eq(issueDocuments.companyId, this.#binding.companyId), + eq(issueDocuments.issueId, this.#binding.issueId), + eq(documents.companyId, this.#binding.companyId), + ), + ); + return { + value: jsonValue({ documents: rows }), + references: rows.map((row) => ({ + kind: "document_revision" as const, + id: row.latestRevisionId ?? row.id, + })), + }; + } + case "read_document": { + const key = requiredString(input.key); + const [row] = await this.#db + .select({ + key: issueDocuments.key, + id: documents.id, + title: documents.title, + format: documents.format, + body: documents.latestBody, + latestRevisionId: documents.latestRevisionId, + latestRevisionNumber: documents.latestRevisionNumber, + lockedAt: documents.lockedAt, + updatedAt: documents.updatedAt, + }) + .from(issueDocuments) + .innerJoin(documents, eq(documents.id, issueDocuments.documentId)) + .where( + and( + eq(issueDocuments.companyId, this.#binding.companyId), + eq(issueDocuments.issueId, this.#binding.issueId), + eq(issueDocuments.key, key), + eq(documents.companyId, this.#binding.companyId), + ), + ) + .limit(1); + if (!row) throw new Error("paperclip_runner_document_not_found"); + return { + value: jsonValue({ document: row }), + references: [ + { + kind: "document_revision", + id: row.latestRevisionId ?? row.id, + }, + ], + }; + } + case "list_document_revisions": { + const key = requiredString(input.key); + const rows = await this.#db + .select({ + id: documentRevisions.id, + revisionNumber: documentRevisions.revisionNumber, + title: documentRevisions.title, + format: documentRevisions.format, + body: documentRevisions.body, + changeSummary: documentRevisions.changeSummary, + createdByAgentId: documentRevisions.createdByAgentId, + createdByUserId: documentRevisions.createdByUserId, + createdAt: documentRevisions.createdAt, + }) + .from(issueDocuments) + .innerJoin( + documentRevisions, + eq(documentRevisions.documentId, issueDocuments.documentId), + ) + .where( + and( + eq(issueDocuments.companyId, this.#binding.companyId), + eq(issueDocuments.issueId, this.#binding.issueId), + eq(issueDocuments.key, key), + eq(documentRevisions.companyId, this.#binding.companyId), + ), + ) + .orderBy(desc(documentRevisions.revisionNumber)) + .limit(boundedLimit(input.limit)); + return { + value: jsonValue({ revisions: rows }), + references: rows.map((row) => ({ + kind: "document_revision" as const, + id: row.id, + })), + }; + } + } + }, + }; + } + + async #context(requestedRunId: string): Promise { + if (requestedRunId !== this.#binding.runId) { + throw new Error("paperclip_runner_semantic_run_mismatch"); + } + const context = await this.#loadBoundContext(); + this.#assertActiveContext(context, false); + return { + runId: context.run.id, + companyId: context.run.companyId, + actor: { + id: context.agent.id, + companyId: context.agent.companyId, + status: activeAgentStatus(context.agent.status), + role: context.agent.role, + claims: [], + }, + activeTask: { + id: context.issue.id, + companyId: context.issue.companyId, + assigneeActorId: context.issue.assigneeAgentId, + executionRunId: context.issue.executionRunId, + status: context.issue.status, + workMode: context.issue + .workMode as PaperclipSemanticRunContext["activeTask"]["workMode"], + }, + delegatedClaims: [], + }; + } + + async #loadBoundContext(): Promise { + const [row] = await this.#db + .select({ run: heartbeatRuns, issue: issues, agent: agents }) + .from(heartbeatRuns) + .innerJoin( + issues, + and( + eq(issues.id, heartbeatRuns.nativeIssueId), + eq(issues.companyId, heartbeatRuns.companyId), + ), + ) + .innerJoin( + agents, + and( + eq(agents.id, heartbeatRuns.agentId), + eq(agents.companyId, heartbeatRuns.companyId), + ), + ) + .where( + and( + eq(heartbeatRuns.id, this.#binding.runId), + eq(heartbeatRuns.companyId, this.#binding.companyId), + eq(heartbeatRuns.agentId, this.#binding.agentId), + eq(heartbeatRuns.nativeIssueId, this.#binding.issueId), + eq(heartbeatRuns.runtimeMode, "native"), + ), + ) + .limit(1); + if (!row) throw new Error("paperclip_runner_semantic_binding_not_found"); + return row; + } + + #assertActiveContext(context: BoundContext, requireOwnership: boolean): void { + if ( + !["queued", "running"].includes(context.run.status) || + activeAgentStatus(context.agent.status) !== "active" || + (requireOwnership && + (context.issue.assigneeAgentId !== this.#binding.agentId || + context.issue.executionRunId !== this.#binding.runId)) + ) { + throw new Error("paperclip_runner_semantic_binding_inactive"); + } + } +} diff --git a/server/src/vendor/paperclip-runner/index.ts b/server/src/vendor/paperclip-runner/index.ts new file mode 100644 index 0000000000..4789be99f5 --- /dev/null +++ b/server/src/vendor/paperclip-runner/index.ts @@ -0,0 +1,8 @@ +/** + * Development shim for the package-local runner runtime. + * + * The server build replaces this emitted module with the runner package's + * compiled `dist` tree so published server packages have no workspace runtime + * dependency. Keep server imports pointed at this relative boundary. + */ +export * from "@paperclipai/paperclip-runner"; diff --git a/server/vitest.config.ts b/server/vitest.config.ts index faa0c7b0e1..f87ac41efe 100644 --- a/server/vitest.config.ts +++ b/server/vitest.config.ts @@ -1,6 +1,18 @@ +import { fileURLToPath } from "node:url"; + import { defineConfig } from "vitest/config"; export default defineConfig({ + resolve: { + alias: [ + { + find: /^@paperclipai\/paperclip-runner$/, + replacement: fileURLToPath( + new URL("../packages/paperclip-runner/src/index.ts", import.meta.url), + ), + }, + ], + }, test: { environment: "node", include: ["src/**/*.test.ts"],