import { createHash, randomBytes, timingSafeEqual } from "node:crypto"; import { mkdir, mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { resolve } from "node:path"; /** * Package server for the Capability issue-thread UI and the Capability clean-room * chat. * * The browser posts intents here; this process owns the real runnerd and Codex * app-server pair, the mock ControlPlanePort, and every policy decision. Each * response is a server-projected issue-thread view, so the page never holds * state or policy authority (Capability UX contract §11). * * An interaction response is stored in the mock control plane before the * runner is resumed — that ordering is enforced inside `CapabilityLiveSession` * (§5 response authority path). * * Two surfaces share one session registry and one set of turn routes: * * - `issue` seeds the preset scenario the explorer selects; * - `cleanroom` seeds only a company, an agent, and a blank issue. * * Both run the same real runnerd + real Codex loop. Neither has a scripted or * replay path in this process, so there is nothing here that could quietly * substitute for a live turn. */ const ROUTE_PREFIX = "/api/capability/ui"; const CLEAN_ROOM_ROUTE = "cleanroom/session"; /** Bounded concurrency (revision 5 safety requirements). */ const MAX_CLEAN_ROOM_SESSIONS = 4; /** Bounded output: a clean room is a demo, not a long-lived agent. */ const MAX_TURNS_PER_SESSION = 24; const MAX_MESSAGE_BYTES = 8 * 1024; /** * Bounded frames per streamed turn. Past this the turn keeps running and still * settles with the authoritative payload; only the interim views stop, so a * chatty provider cannot turn one turn into unbounded socket writes. */ const MAX_TURN_STREAM_FRAMES = 600; const ACPX_QUALIFIED_MODELS = Object.freeze({ claude: "claude-sonnet-5", codex: "gpt-5.6-sol", }); /** * Per-browser session capability (track 7U). * * A session id used to be the only thing a route checked, so possession of * another browser's id authorized reading and mutating that browser's session. * Each surface now mints its own high-entropy capability, stores only its * SHA-256 with the session record, and compares in constant time on every read * and mutation. A valid id presented without its capability is answered `404`, * the same as an id that never existed: an unauthorized caller learns nothing * about which ids are live. * * The two surfaces use separate cookie names on purpose. They are separate * pages of one origin, so a single name would make opening the explorer revoke * the clean room (and the reverse) instead of keeping two independent tenants. */ const CAPABILITY_COOKIES = Object.freeze({ issue: "paperclip_capability_issue", cleanroom: "paperclip_capability_chat", }); const CAPABILITY_BYTES = 32; const CAPABILITY_MAX_AGE_SECONDS = 30 * 60; /** * A streamed turn fails with a code, never with the underlying text. Provider * and driver messages can quote prompts, paths, or protocol detail, and a frame * is browser surface, so the operator-facing copy is fixed here. */ const PUBLIC_TURN_ERROR_MESSAGE = "The turn could not be completed. Nothing further was sent to the provider."; function sha256(value) { return createHash("sha256").update(value, "utf8").digest(); } function mintCapability() { return randomBytes(CAPABILITY_BYTES).toString("base64url"); } function capabilityMatches(presented, expectedHash) { if (typeof presented !== "string" || presented.length < 32) return false; const digest = sha256(presented); // Length is compared first because `timingSafeEqual` throws on a mismatch; // both operands are fixed-width digests, so this never short-circuits on a // secret-dependent branch. return digest.length === expectedHash.length && timingSafeEqual(digest, expectedHash); } function parseCookies(request) { const header = request.headers.cookie; const source = typeof header === "string" ? header : ""; const cookies = new Map(); for (const segment of source.split(";")) { const separator = segment.indexOf("="); if (separator < 1) continue; cookies.set(segment.slice(0, separator).trim(), segment.slice(separator + 1).trim()); } return cookies; } function presentedCapability(request, surface) { return parseCookies(request).get(CAPABILITY_COOKIES[surface]) ?? ""; } function capabilityCookie(request, surface, value, maxAgeSeconds = CAPABILITY_MAX_AGE_SECONDS) { const forwardedProtocol = request.headers["x-forwarded-proto"]; const secure = String(Array.isArray(forwardedProtocol) ? forwardedProtocol[0] : forwardedProtocol ?? "") .toLowerCase() === "https"; return `${CAPABILITY_COOKIES[surface]}=${value}; Path=/; Max-Age=${maxAgeSeconds}; HttpOnly${ secure ? "; Secure" : "" }; SameSite=Strict`; } async function importDistRunnerModule(relativePath) { return import(new URL(`../dist/${relativePath}`, import.meta.url).href); } export async function loadCapabilityIssueThreadRunner(importModule = importDistRunnerModule) { const [cleanRoom, liveSession, turnStream, issueThread, fixtureState, liveConsole, devtools] = await Promise.all([ importModule("live/clean-room.js"), importModule("live/live-session.js"), importModule("live/turn-stream.js"), importModule("issue-thread/index.js"), importModule("mock-core/capability-control-plane-types.js"), importModule("mock-core/live-console-demo-server.js"), importModule("devtools/index.js"), ]); // This package-owned server intentionally depends on private demo/live modules. // Keep that dependency explicit instead of widening the package's public root. return { CAPABILITY_TURN_STREAM_HEADERS: turnStream.CAPABILITY_TURN_STREAM_HEADERS, CAPABILITY_TURN_STREAM_SCHEMA: turnStream.CAPABILITY_TURN_STREAM_SCHEMA, CapabilityLiveSessionService: liveSession.CapabilityLiveSessionService, InMemoryCapabilityLiveSessionStore: liveSession.InMemoryCapabilityLiveSessionStore, assertLiveConsoleLoopbackBindHost: liveConsole.assertLiveConsoleLoopbackBindHost, createCapabilityCleanRoomSessionInput: cleanRoom.createCapabilityCleanRoomSessionInput, createCapabilityFixtureState: fixtureState.createCapabilityFixtureState, encodeCapabilityTurnStreamFrame: turnStream.encodeCapabilityTurnStreamFrame, projectCapabilityIssueThread: issueThread.projectCapabilityIssueThread, toCapabilityPublicThreadView: issueThread.toCapabilityPublicThreadView, projectCapabilityDevtools: devtools.projectCapabilityDevtools, }; } function scratchRoot() { return process.env.PAPERCLIP_RUN_SCRATCH_DIR ?? process.env.PAPERCLIP_SCRATCH_DIR ?? tmpdir(); } async function createWorkingDirectory(root, prefix = "capability-issue-thread-") { // Managed previews can outlive the heartbeat that launched them. Paperclip // removes that heartbeat's scratch directory when the run ends, so ensure // the inherited parent still exists before every later session is minted. await mkdir(root, { recursive: true }); return mkdtemp(resolve(root, prefix)); } /** Mock-only fixture seed. Identifiers use the reserved `MCK-` prefix (§1.3). */ function issueThreadSeed(runner, scenario) { return runner.createCapabilityFixtureState({ epochMs: Date.UTC(2026, 7, 9, 9, 0, 0), company: { id: "company-1", name: "Mock Paperclip Company", issuePrefix: "MCK" }, actors: [ { id: "actor-1", companyId: "company-1", name: "Mock Engineer", role: "engineer", status: "active", budgetId: "budget-actor-1", capabilityGrants: [], }, ], tasks: [ { id: "task-31", companyId: "company-1", identifier: "MCK-31", title: "Wire the runner spike to the mock control plane", description: `Scenario ${scenario}. All records in this thread are mock records.`, status: "todo", priority: "high", workMode: "standard", parentId: null, assigneeActorId: "actor-1", checkoutRunId: null, executionRunId: null, startedAt: null, completedAt: null, }, ], }); } class RouteError extends Error { constructor(status, code, message) { super(message); this.status = status; this.code = code; } } function harnessConfiguration(source, fallbackModel) { const provider = source.provider === undefined ? "codex" : String(source.provider).trim(); if (provider !== "codex" && provider !== "opencode" && provider !== "claude_managed" && provider !== "aws_agentcore" && provider !== "acpx") { throw new RouteError(400, "invalid_provider", "Provider must be codex, opencode, claude_managed, aws_agentcore, or acpx."); } const rawModel = source.model === undefined ? fallbackModel : source.model; const model = rawModel === undefined || rawModel === null ? "" : String(rawModel).trim(); if (model.length > 256) throw new RouteError(400, "invalid_model", "Model is too long."); if (provider === "opencode" && (!model || !model.includes("/"))) { throw new RouteError(400, "invalid_model", "OpenCode requires a provider/model value."); } if (provider === "claude_managed" && model !== "claude-sonnet-5") { throw new RouteError(400, "invalid_model", "Claude Managed requires exact model claude-sonnet-5."); } if (provider === "aws_agentcore" && model !== "global.anthropic.claude-sonnet-4-6") { throw new RouteError(400, "invalid_model", "AWS AgentCore requires exact model global.anthropic.claude-sonnet-4-6."); } const acpxAgent = source.acpxAgent === undefined ? "codex" : String(source.acpxAgent).trim(); if (provider === "acpx") { if (!(acpxAgent in ACPX_QUALIFIED_MODELS)) { throw new RouteError(400, "invalid_acpx_agent", "ACPX agent must be claude or codex."); } if (model !== ACPX_QUALIFIED_MODELS[acpxAgent]) { throw new RouteError(400, "invalid_model", `The qualified ACPX ${acpxAgent} profile requires exact model ${ACPX_QUALIFIED_MODELS[acpxAgent]}.`); } } const requestedManagedProfileId = source.managedProfileId === undefined || source.managedProfileId === null ? "default" : String(source.managedProfileId).trim(); const configuredManagedProfileId = process.env.PAPERCLIP_CLAUDE_MANAGED_PROFILE_ID?.trim(); const managedProfileId = provider === "claude_managed" && requestedManagedProfileId === "default" && configuredManagedProfileId ? configuredManagedProfileId : requestedManagedProfileId; const maxSessionListCostUsd = source.maxSessionListCostUsd === undefined || source.maxSessionListCostUsd === null ? 1 : Number(source.maxSessionListCostUsd); if (provider === "claude_managed" && !managedProfileId) { throw new RouteError(400, "invalid_managed_profile", "Claude Managed requires a qualified profile ID."); } if (provider === "claude_managed" && (!Number.isFinite(maxSessionListCostUsd) || maxSessionListCostUsd <= 0)) { throw new RouteError(400, "invalid_spend_cap", "Claude Managed requires a positive session spend ceiling."); } const requestedAgentCoreProfileId = source.agentCoreProfileId === undefined || source.agentCoreProfileId === null ? "default" : String(source.agentCoreProfileId).trim(); const configuredAgentCoreProfileId = process.env.PAPERCLIP_AWS_AGENTCORE_PROFILE_ID?.trim(); const agentCoreProfileId = provider === "aws_agentcore" && requestedAgentCoreProfileId === "default" && configuredAgentCoreProfileId ? configuredAgentCoreProfileId : requestedAgentCoreProfileId; const maxEstimatedSessionCostUsd = source.maxEstimatedSessionCostUsd === undefined || source.maxEstimatedSessionCostUsd === null ? 1 : Number(source.maxEstimatedSessionCostUsd); if (provider === "aws_agentcore" && !agentCoreProfileId) { throw new RouteError(400, "invalid_agentcore_profile", "AWS AgentCore requires a qualified profile ID."); } if (provider === "aws_agentcore" && (!Number.isFinite(maxEstimatedSessionCostUsd) || maxEstimatedSessionCostUsd <= 0)) { throw new RouteError(400, "invalid_spend_cap", "AWS AgentCore requires a positive session spend ceiling."); } const suppliedLifecycle = source.lifecyclePolicy && typeof source.lifecyclePolicy === "object" ? source.lifecyclePolicy : source; const lifecycleMode = suppliedLifecycle.mode === undefined ? source.lifecycleMode === undefined ? "warm" : String(source.lifecycleMode).trim() : String(suppliedLifecycle.mode).trim(); if (lifecycleMode !== "per_turn" && lifecycleMode !== "warm") { throw new RouteError(400, "invalid_lifecycle_mode", "Execution mode must be per_turn or warm."); } let lifecyclePolicy; if (lifecycleMode === "per_turn") { lifecyclePolicy = { mode: "per_turn", idleTimeoutMs: null }; } else { const rawIdleTimeout = suppliedLifecycle.idleTimeoutMs === undefined ? source.idleTimeoutMs : suppliedLifecycle.idleTimeoutMs; const idleTimeoutMs = rawIdleTimeout === undefined ? 300_000 : Number(rawIdleTimeout); if (!Number.isSafeInteger(idleTimeoutMs) || idleTimeoutMs <= 0) { throw new RouteError(400, "invalid_idle_timeout", "Warm idle timeout must be a positive integer."); } lifecyclePolicy = { mode: "warm", idleTimeoutMs }; } return { provider, model: model || null, ...(provider === "acpx" ? { acpxAgent } : {}), ...(provider === "claude_managed" ? { managedProfileId, maxSessionListCostUsd } : {}), ...(provider === "aws_agentcore" ? { agentCoreProfileId, maxEstimatedSessionCostUsd } : {}), lifecyclePolicy, }; } function resolveManagedProfile(configuration) { const profileId = process.env.PAPERCLIP_CLAUDE_MANAGED_PROFILE_ID?.trim() || configuration.managedProfileId; if (profileId !== configuration.managedProfileId) { throw new RouteError(400, "managed_profile_not_found", "The selected Claude Managed profile is not configured on this Runner Lab server."); } const anthropicAgentId = process.env.ANTHROPIC_MANAGED_AGENT_ID?.trim(); const agentVersion = process.env.ANTHROPIC_MANAGED_AGENT_VERSION?.trim(); const environmentId = process.env.ANTHROPIC_MANAGED_ENVIRONMENT_ID?.trim(); const canonicalAgentVersion = agentVersion !== undefined && /^[1-9][0-9]*$/.test(agentVersion) && BigInt(agentVersion) <= 2_147_483_647n; if (!process.env.ANTHROPIC_API_KEY || !profileId || !anthropicAgentId || !canonicalAgentVersion || !environmentId) { throw new RouteError( 503, "managed_profile_unavailable", "The selected Claude Managed profile is not fully qualified on this Runner Lab server.", ); } return { profileId, anthropicAgentId, agentVersion, environmentId, betaVersion: "managed-agents-2026-04-01", maxSessionListCostUsd: configuration.maxSessionListCostUsd, }; } function resolveAgentCoreProfile(configuration) { const required = (name) => { const value = process.env[name]?.trim(); if (!value) throw new RouteError(503, "agentcore_profile_unavailable", `AWS AgentCore profile is missing ${name}. Run aws-agentcore:provision and aws-agentcore:lab.`); return value; }; const profileId = required("PAPERCLIP_AWS_AGENTCORE_PROFILE_ID"); if (profileId !== configuration.agentCoreProfileId) { throw new RouteError(400, "agentcore_profile_not_found", "The selected AWS AgentCore profile is not configured on this Runner Lab server."); } return { profileId, region: required("AWS_REGION"), accountId: required("PAPERCLIP_AWS_AGENTCORE_ACCOUNT_ID"), harnessArn: required("PAPERCLIP_AWS_AGENTCORE_HARNESS_ARN"), harnessVersion: required("PAPERCLIP_AWS_AGENTCORE_HARNESS_VERSION"), endpointArn: required("PAPERCLIP_AWS_AGENTCORE_ENDPOINT_ARN"), endpointQualifier: required("PAPERCLIP_AWS_AGENTCORE_ENDPOINT_QUALIFIER"), agentRuntimeArn: required("PAPERCLIP_AWS_AGENTCORE_RUNTIME_ARN"), memoryArn: required("PAPERCLIP_AWS_AGENTCORE_MEMORY_ARN"), memoryId: required("PAPERCLIP_AWS_AGENTCORE_MEMORY_ID"), invocationRoleArn: required("PAPERCLIP_AWS_AGENTCORE_INVOCATION_ROLE_ARN"), contextBucket: required("PAPERCLIP_AWS_AGENTCORE_CONTEXT_BUCKET"), contextPrefix: required("PAPERCLIP_AWS_AGENTCORE_CONTEXT_PREFIX"), contextKmsKeyArn: required("PAPERCLIP_AWS_AGENTCORE_CONTEXT_KMS_KEY_ARN"), qualificationRevision: required("PAPERCLIP_AWS_AGENTCORE_QUALIFICATION_REVISION"), eventExpiryDays: 90, maxEstimatedSessionCostUsd: configuration.maxEstimatedSessionCostUsd, maxIterations: 8, maxOutputTokens: 4096, timeoutSeconds: 300, }; } export const capabilityIssueThreadServerInternals = Object.freeze({ harnessConfiguration, resolveManagedProfile, resolveAgentCoreProfile, }); export function createCapabilityIssueThreadMiddleware(options = {}) { const load = options.loadRunner ?? loadCapabilityIssueThreadRunner; const workingDirectoryRoot = options.scratchRoot ?? scratchRoot(); let bootstrap = null; let bindHost = options.bindHost ?? "127.0.0.1"; /** * @type {Map} */ const sessions = new Map(); async function ready(requestedBindHost = bindHost) { bindHost = requestedBindHost; if (bootstrap !== null) return bootstrap; bootstrap = (async () => { const runner = await load(); runner.assertLiveConsoleLoopbackBindHost(bindHost); const workingDirectory = options.workingDirectory ?? (await createWorkingDirectory(workingDirectoryRoot)); const service = new runner.CapabilityLiveSessionService({ store: new runner.InMemoryCapabilityLiveSessionStore(), ...(options.transportFactory === undefined ? {} : { transportFactory: options.transportFactory }), }); return { runner, service, workingDirectory }; })(); return bootstrap; } /** * The published view. * * The projection is an internal shape; `toCapabilityPublicThreadView` is what the * browser is allowed to see. Every response path calls this — never the * projection directly — so interim frames, terminal payloads, and reconnect * replies cannot disagree about what is public (track 7U). */ function view(runner, entry) { const snapshot = entry.session.snapshot(); const projected = runner.projectCapabilityIssueThread({ snapshot, connection: entry.connection, mode: "live", fixtureProfile: entry.scenario, }); return runner.toCapabilityPublicThreadView(projected, { withheldValues: [snapshot.providerThreadId, snapshot.providerSessionId ?? ""], }); } /** * The clean room must never answer with a scripted or replayed turn. The * projection is the only thing the browser sees, so the guard sits on the way * out rather than on the way in. */ function liveView(runner, entry) { const snapshot = entry.session.snapshot(); const provider = snapshot.config.provider ?? "codex"; const expectedAgentLabel = provider === "claude_managed" ? "Claude Agent" : provider === "opencode" ? "Real OpenCode" : provider === "aws_agentcore" ? "Real AWS AgentCore" : provider === "acpx" ? `Real ${snapshot.config.acpxAgent === "claude" ? "Claude" : snapshot.config.acpxAgent === "codex" ? "Codex" : "Pi"} via ACPX` : "Real Codex"; const projected = view(runner, entry); if ( projected.mode !== "live" || projected.identity.agentLabel !== expectedAgentLabel ) { throw new RouteError( 500, "provider_identity_mismatch", "The clean-room path refused a view whose provider identity did not match its immutable session.", ); } return projected; } async function readBody(request) { const chunks = []; let total = 0; for await (const chunk of request) { total += chunk.length; if (total > MAX_MESSAGE_BYTES * 4) { throw new RouteError(413, "request_too_large", "Request body exceeds the server limit."); } chunks.push(chunk); } if (chunks.length === 0) return {}; try { return JSON.parse(Buffer.concat(chunks).toString("utf8")); } catch { return {}; } } function send(response, status, payload) { response.statusCode = status; response.setHeader("content-type", "application/json; charset=utf-8"); response.end(JSON.stringify(payload)); } async function createSession(runner, service, scenario, workingDirectory, capabilityHash) { const session = await service.create({ seed: issueThreadSeed(runner, scenario), workingDirectory, scenario: { id: scenario }, taskId: "task-31", actorId: "actor-1", companyId: "company-1", ...(options.requestedModel === undefined ? {} : { requestedModel: options.requestedModel }), }); const entry = { session, surface: "issue", scenario, identity: null, workingDirectory, ownsWorkingDirectory: false, turns: 0, createdAt: Date.now(), capabilityHash, connection: { state: "connected", attempt: 0 }, }; sessions.set(session.id, entry); return entry; } /** Archive a session resumably; destructive deletion is intentionally absent. */ async function retire(service, sessionId, reason) { const entry = sessions.get(sessionId); if (entry === undefined) return; await service.stop(sessionId, reason).catch(() => undefined); entry.connection = { state: "suspended", attempt: entry.connection.attempt }; } /** * Revokes every session already bound to the presented capability before a * replacement is minted, so a rotation deletes the old binding rather than * leaving a second reachable session behind it. */ async function revokeBoundSessions(service, surface, presented) { if (presented.length === 0) return; for (const [sessionId, entry] of [...sessions.entries()]) { if (entry.surface !== surface) continue; if (!capabilityMatches(presented, entry.capabilityHash)) continue; await retire(service, sessionId, "capability rotated"); } } async function createCleanRoomSession(runner, service, capabilityHash, configuration) { const open = [...sessions.values()].filter((entry) => entry.surface === "cleanroom" && entry.session.snapshot().status !== "suspended", ); // Bounded concurrency: the oldest clean room yields rather than refusing a // new board user, because an abandoned chat is the likelier tenant here. for (const stale of open.slice(0, Math.max(0, open.length - (MAX_CLEAN_ROOM_SESSIONS - 1)))) { await retire(service, stale.session.id, "clean-room capacity"); } const workingDirectory = await createWorkingDirectory( workingDirectoryRoot, "capability-clean-room-", ); const { identity, input } = runner.createCapabilityCleanRoomSessionInput({ workingDirectory }); let session; try { session = await service.create({ ...input, provider: configuration.provider, ...(configuration.provider === "claude_managed" ? { managedProfile: resolveManagedProfile(configuration) } : {}), ...(configuration.provider === "acpx" ? { acpxAgent: configuration.acpxAgent } : {}), ...(configuration.provider === "aws_agentcore" ? { agentCoreProfile: resolveAgentCoreProfile(configuration) } : {}), lifecyclePolicy: configuration.lifecyclePolicy, ...(configuration.model === null ? {} : { requestedModel: configuration.model }), }); } catch (error) { await rm(workingDirectory, { recursive: true, force: true }).catch(() => undefined); throw error; } const entry = { session, surface: "cleanroom", scenario: "clean-room", identity, workingDirectory, ownsWorkingDirectory: true, turns: 0, createdAt: Date.now(), capabilityHash, configuration, connection: { state: "connected", attempt: 0 }, }; sessions.set(session.id, entry); return entry; } function cleanRoomPayload(runner, entry) { const snapshot = entry.session.snapshot(); const configuration = { provider: snapshot.config.provider ?? "codex", model: snapshot.config.requestedModel ?? null, ...(snapshot.config.managedProfile === undefined ? {} : { managedProfileId: snapshot.config.managedProfile.profileId, maxSessionListCostUsd: snapshot.config.managedProfile.maxSessionListCostUsd, }), ...(snapshot.config.agentCoreProfile === undefined ? {} : { agentCoreProfileId: snapshot.config.agentCoreProfile.profileId, maxEstimatedSessionCostUsd: snapshot.config.agentCoreProfile.maxEstimatedSessionCostUsd, }), lifecyclePolicy: snapshot.config.lifecyclePolicy ?? { mode: "per_turn", idleTimeoutMs: null }, }; if ( entry.configuration !== undefined && (entry.configuration.provider !== configuration.provider || entry.configuration.model !== configuration.model || entry.configuration.managedProfileId !== configuration.managedProfileId || entry.configuration.maxSessionListCostUsd !== configuration.maxSessionListCostUsd || entry.configuration.agentCoreProfileId !== configuration.agentCoreProfileId || entry.configuration.maxEstimatedSessionCostUsd !== configuration.maxEstimatedSessionCostUsd || JSON.stringify(entry.configuration.lifecyclePolicy) !== JSON.stringify(configuration.lifecyclePolicy)) ) { throw new RouteError( 500, "provider_configuration_mismatch", "The clean-room path refused configuration that differed from its immutable session.", ); } return { sessionId: entry.session.id, surface: "cleanroom", identity: entry.identity, limits: { maxTurns: MAX_TURNS_PER_SESSION, maxMessageBytes: MAX_MESSAGE_BYTES }, turns: entry.turns, configuration, runtime: { providerSessionId: snapshot.providerSessionId ?? null, driverSessionId: snapshot.providerThreadId ?? null, runnerPid: snapshot.process?.runnerPid ?? null, providerPid: configuration.provider === "claude_managed" || configuration.provider === "aws_agentcore" ? null : snapshot.process?.providerPid ?? snapshot.process?.codexPid ?? null, sidecarPid: snapshot.process?.sidecarPid ?? null, agentPid: snapshot.process?.agentPid ?? null, providerVersion: snapshot.process?.providerVersion ?? null, agentServerVersion: snapshot.process?.agentServerVersion ?? null, agentRuntimeVersion: snapshot.process?.agentRuntimeVersion ?? null, acpProtocolVersion: snapshot.process?.acpProtocolVersion ?? null, executionKind: snapshot.process?.providerExecutionKind ?? (configuration.provider === "claude_managed" || configuration.provider === "aws_agentcore" ? "remote_service" : "local_process"), status: snapshot.status, }, view: liveView(runner, entry), }; } function payload(runner, entry) { return entry.surface === "cleanroom" ? cleanRoomPayload(runner, entry) : { sessionId: entry.session.id, surface: "issue", view: view(runner, entry) }; } /** The clean room keeps its live-only guard on every frame, not just the last. */ function frameView(runner, entry) { return entry.surface === "cleanroom" ? liveView(runner, entry) : view(runner, entry); } /** * Streams one turn as NDJSON frames (track 7Q). * * The turn used to be awaited whole and answered once, so the browser could * only ever reveal a finished reply. Now every provider delta, tool call, and * tool result the live session announces is projected and written while the * POST is still open. The final `settled` frame carries exactly the payload * the single JSON response used to carry, so the terminal projection — not * any interim frame — remains the authority. * * Interim frames are coalesced onto the next event-loop turn: a burst of * deltas that arrives in one tick becomes one frame, while deltas separated * by real provider I/O each get their own. That bounds the write rate without * a timer, and without inventing a cadence the provider did not have. */ async function streamTurn(runner, entry, message, request, response) { response.statusCode = 200; for (const [name, value] of Object.entries(runner.CAPABILITY_TURN_STREAM_HEADERS)) { response.setHeader(name, value); } // Headers before the first frame: a client that waits for them must not be // held until the provider speaks. response.flushHeaders(); let seq = 0; let frames = 0; let finished = false; let scheduled = null; let pendingReason = null; let pendingTurnId = null; const write = (frame) => { if (response.writableEnded || response.destroyed) return; seq += 1; response.write(runner.encodeCapabilityTurnStreamFrame({ ...frame, seq })); }; const flush = () => { scheduled = null; if (finished || pendingReason === null) return; if (frames >= MAX_TURN_STREAM_FRAMES) return; const reason = pendingReason; const turnId = pendingTurnId; pendingReason = null; pendingTurnId = null; let projected; try { projected = frameView(runner, entry); } catch { // A projection failure is reported by the terminal frame; dropping an // interim view must never abort a turn that is still running. return; } frames += 1; write({ schema: runner.CAPABILITY_TURN_STREAM_SCHEMA, type: "frame", reason, turnId, view: projected, }); }; const unsubscribe = entry.session.subscribe((event) => { if (finished || event.kind === "terminal" || event.kind === "error") return; pendingReason = event.kind === "delta" ? "delta" : event.reason === "stop_requested" ? "stop_requested" : "activity"; pendingTurnId = event.turnId; if (scheduled === null) scheduled = setImmediate(flush); }); // A browser that navigates away, reloads, or aborts the fetch mid-turn is a // stop: the provider turn is interrupted rather than left running against a // socket nobody is reading. const onDisconnect = () => { if (finished) return; void entry.session.interrupt("client disconnected").catch(() => undefined); }; request.on("aborted", onDisconnect); response.on("close", onDisconnect); // `sendMessage` records the user message and marks the session running // before its first await, so starting it and *then* flushing makes the // opening frame already show what was sent and a live composer. const turn = entry.session.sendMessage(message); pendingReason = "open"; pendingTurnId = null; flush(); try { await turn; finished = true; write({ schema: runner.CAPABILITY_TURN_STREAM_SCHEMA, type: "settled", payload: payload(runner, entry), }); } catch (error) { finished = true; options.onTurnError?.(error, entry.session.id, entry.session.snapshot()); // The code identifies the failure; the underlying message stays server // side because it can quote provider text (track 7U). write({ schema: runner.CAPABILITY_TURN_STREAM_SCHEMA, type: "error", error: error instanceof RouteError ? error.code : "turn_failed", message: PUBLIC_TURN_ERROR_MESSAGE, }); } finally { finished = true; if (scheduled !== null) clearImmediate(scheduled); unsubscribe(); request.off("aborted", onDisconnect); response.off("close", onDisconnect); response.end(); } } const middleware = async function capabilityIssueThreadMiddleware(request, response, next) { const url = new URL(request.url ?? "/", "http://capability.local"); if (!url.pathname.startsWith(`${ROUTE_PREFIX}/`)) { next(); return; } try { const { runner, service, workingDirectory } = await ready(); const route = url.pathname.slice(ROUTE_PREFIX.length + 1); const body = request.method === "POST" ? await readBody(request) : {}; const requestedId = typeof body.sessionId === "string" ? body.sessionId : url.searchParams.get("sessionId"); const scenario = typeof body.scenario === "string" ? body.scenario : url.searchParams.get("scenario") ?? "hb-baseline"; const requestedHarness = harnessConfiguration({ provider: body.provider ?? url.searchParams.get("provider") ?? undefined, model: body.model ?? url.searchParams.get("model") ?? undefined, acpxAgent: body.acpxAgent ?? url.searchParams.get("acpxAgent") ?? undefined, managedProfileId: body.managedProfileId ?? url.searchParams.get("managedProfileId") ?? undefined, maxSessionListCostUsd: body.maxSessionListCostUsd ?? url.searchParams.get("maxSessionListCostUsd") ?? undefined, agentCoreProfileId: body.agentCoreProfileId ?? url.searchParams.get("agentCoreProfileId") ?? undefined, maxEstimatedSessionCostUsd: body.maxEstimatedSessionCostUsd ?? url.searchParams.get("maxEstimatedSessionCostUsd") ?? undefined, lifecyclePolicy: body.lifecyclePolicy, lifecycleMode: body.lifecycleMode ?? url.searchParams.get("lifecycleMode") ?? undefined, idleTimeoutMs: body.idleTimeoutMs ?? url.searchParams.get("idleTimeoutMs") ?? undefined, }, options.requestedModel); /** * Resolves the caller's own session, or `undefined`. * * A session that exists but belongs to another capability is reported as * `denied` rather than as absent-so-make-a-new-one: the caller must not be * handed a fresh session under an id it does not own, and must not be able * to tell a live id from a dead one. */ const ownedSession = (surface) => { if (requestedId === null) return { state: "absent" }; const existing = sessions.get(requestedId); if (existing === undefined) return { state: "absent" }; if ( existing.surface !== surface || !capabilityMatches(presentedCapability(request, surface), existing.capabilityHash) ) { return { state: "denied" }; } return { state: "owned", entry: existing }; }; /** * Mints a session for this browser. * * `rotate` separates the two reasons a session gets created. Starting * something new — `New chat`, a scenario POST, a reset — rotates: the old * bindings are revoked first so the cookie the caller arrived with stops * working. Opening a page whose stored id is simply gone does not rotate; * it reuses the capability the browser already holds. Two tabs of one * surface would otherwise revoke each other's session on every load and * ping-pong, and rotating there protects nothing: the browser is the same * principal either way, and cross-browser denial rests on the binding, not * on how often the value changes. */ const mint = async (surface, { rotate }) => { const presented = presentedCapability(request, surface); const reuse = !rotate && presented.length >= 32; if (rotate) await revokeBoundSessions(service, surface, presented); const capability = reuse ? presented : mintCapability(); const capabilityHash = sha256(capability); const entry = surface === "cleanroom" ? await createCleanRoomSession(runner, service, capabilityHash, requestedHarness) : await createSession(runner, service, scenario, workingDirectory, capabilityHash); if (!reuse) { response.setHeader("set-cookie", capabilityCookie(request, surface, capability)); } return entry; }; if (route === CLEAN_ROOM_ROUTE && request.method === "GET") { // A stale id from localStorage opens a fresh room rather than a dead // end; a live id reconnects to the same durable chat; another browser's // id is refused. const owned = ownedSession("cleanroom"); if (owned.state === "denied") { send(response, 404, { error: "unknown_session" }); return; } const entry = owned.state === "owned" ? owned.entry : await mint("cleanroom", { rotate: false }); send(response, 200, cleanRoomPayload(runner, entry)); return; } if (route === CLEAN_ROOM_ROUTE && request.method === "POST") { // `New chat`: the caller's own room is retired and its capability // rotated before the next one is minted, so the prior authority is // cleared and the prior cookie stops working. const owned = ownedSession("cleanroom"); if (owned.state === "denied") { send(response, 404, { error: "unknown_session" }); return; } if (owned.state === "owned") { await retire(service, owned.entry.session.id, "new clean-room chat"); } const entry = await mint("cleanroom", { rotate: false }); send(response, 201, cleanRoomPayload(runner, entry)); return; } if (route === "session" && request.method === "GET") { const owned = ownedSession("issue"); if (owned.state === "denied") { send(response, 404, { error: "unknown_session" }); return; } const entry = owned.state === "owned" ? owned.entry : await mint("issue", { rotate: false }); send(response, 200, payload(runner, entry)); return; } if (route === "session" && request.method === "POST") { const entry = await mint("issue", { rotate: true }); send(response, 201, payload(runner, entry)); return; } // Shared session-scoped routes. The surface comes from the stored record, // and the capability is then checked against that record — so `message`, // `interrupt`, `reconnect`, `reset`, and `interaction` are all mediated, // not just the routes that create sessions. const located = requestedId === null ? undefined : sessions.get(requestedId); const entry = located !== undefined && capabilityMatches(presentedCapability(request, located.surface), located.capabilityHash) ? located : undefined; if (entry === undefined) { send(response, 404, { error: "unknown_session" }); return; } if (route === "devtools" && request.method === "GET") { send(response, 200, runner.projectCapabilityDevtools(entry.session.snapshot())); return; } if (route === "devtools/fork" && request.method === "POST") { const snapshot = entry.session.snapshot(); const revision = Number(body.revision); const seedState = JSON.parse(snapshot.config.seedState); const selected = (snapshot.stateHistory ?? []).find((item) => item.revision === revision) ?? (seedState.revision === revision ? { revision, state: snapshot.config.seedState } : undefined); if (selected === undefined) { throw new RouteError(404, "unknown_revision", "The requested state revision is not retained."); } const seed = JSON.parse(selected.state); seed.lifecycle = "stopped"; seed.activeRunId = null; for (const task of seed.tasks ?? []) { if (task.id !== snapshot.authority.taskId) continue; task.checkoutRunId = null; task.executionRunId = null; if (["in_progress", "done", "cancelled"].includes(task.status)) { task.status = "todo"; task.completedAt = null; } } const forkDirectory = await createWorkingDirectory( workingDirectoryRoot, "capability-devtools-fork-", ); const capability = mintCapability(); const capabilityHash = sha256(capability); let fork; try { fork = await service.create({ seed, workingDirectory: forkDirectory, provider: snapshot.config.provider ?? "codex", ...(snapshot.config.acpxAgent === undefined ? {} : { acpxAgent: snapshot.config.acpxAgent }), scenario: snapshot.config.scenario, capabilities: snapshot.config.capabilities, explicitClaims: snapshot.config.explicitClaims, companyId: snapshot.authority.companyId, actorId: snapshot.authority.actorId, taskId: snapshot.authority.taskId, turnTimeoutMs: snapshot.config.turnTimeoutMs, ...(snapshot.config.requestedModel === undefined ? {} : { requestedModel: snapshot.config.requestedModel }), ...(snapshot.config.managedProfile === undefined ? {} : { managedProfile: snapshot.config.managedProfile }), ...(snapshot.config.agentCoreProfile === undefined ? {} : { agentCoreProfile: snapshot.config.agentCoreProfile }), }); } catch (error) { await rm(forkDirectory, { recursive: true, force: true }).catch(() => undefined); throw error; } await retire(service, entry.session.id, `forked from revision ${revision}`); const forkEntry = { ...entry, session: fork, workingDirectory: forkDirectory, ownsWorkingDirectory: true, turns: 0, createdAt: Date.now(), capabilityHash, identity: entry.identity === null ? null : { ...entry.identity, token: randomBytes(4).toString("hex") }, connection: { state: "connected", attempt: 0 }, }; sessions.set(fork.id, forkEntry); response.setHeader("set-cookie", capabilityCookie(request, entry.surface, capability)); send(response, 201, payload(runner, forkEntry)); return; } if (route === "tool") { const invocation = await entry.session.invokeTool( String(body.operationId ?? ""), body.input ?? {}, ); const projected = payload(runner, entry); send(response, 200, { ...projected, toolResult: invocation.result, toolTurnId: projected.view.turns.at(-1)?.id ?? null, }); return; } else if (route === "message") { const message = String(body.message ?? ""); if (Buffer.byteLength(message, "utf8") > MAX_MESSAGE_BYTES) { throw new RouteError(413, "message_too_large", "Message exceeds the server limit."); } if (entry.turns >= MAX_TURNS_PER_SESSION) { throw new RouteError( 429, "turn_limit", `This chat reached its ${MAX_TURNS_PER_SESSION}-turn limit. Start a new chat to continue.`, ); } entry.turns += 1; // Every admission check has already answered with its own status code; // from here the response is a stream, so a failure is a framed error. await streamTurn(runner, entry, message, request, response); return; } else if (route === "interrupt") { await entry.session.interrupt("operator stopped the turn"); } else if (route === "managed-budget") { const requestedCap = body.maxSessionListCostUsd ?? body.maxEstimatedSessionCostUsd; const nextCap = Number(requestedCap); if (!Number.isFinite(nextCap) || nextCap <= 0) { throw new RouteError( 400, "invalid_spend_cap", "The new managed-session spend ceiling must be positive.", ); } await entry.session.increaseManagedSessionBudget(nextCap); if (entry.configuration?.provider === "claude_managed") { entry.configuration.maxSessionListCostUsd = nextCap; } if (entry.configuration?.provider === "aws_agentcore") { entry.configuration.maxEstimatedSessionCostUsd = nextCap; } } else if (route === "managed-session-delete") { if (body.confirm !== true) { throw new RouteError( 400, "confirmation_required", "Remote session deletion requires explicit confirmation.", ); } await entry.session.deleteManagedRemoteSession(); await retire( service, entry.session.id, "remote managed session explicitly deleted", ); send(response, 200, { deleted: true, sessionId: entry.session.id }); return; } else if (route === "reconnect") { entry.connection = { state: "reconnecting", attempt: entry.connection.attempt + 1 }; await entry.session.reconnect(); entry.connection = { state: "connected", attempt: 0 }; } else if (route === "reset") { // Reset archives the current session and starts another session for // the same browser principal. Keep the capability stable so the // archived session remains selectable and resumable from history. const capabilityHash = entry.capabilityHash; if (entry.surface === "cleanroom") { // A clean-room reset is a new tenant, not a rewound one: the seed is // blank either way, so restoring it would hand back the same mock // identities the board just saw. await retire(service, entry.session.id, "clean-room reset"); const replacement = await createCleanRoomSession( runner, service, capabilityHash, entry.configuration ?? { provider: "codex", model: null, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 300_000 }, }, ); send(response, 200, cleanRoomPayload(runner, replacement)); return; } const next = await service.reset(entry.session.id); const resetEntry = { ...entry, session: next, turns: 0, capabilityHash, connection: { state: "connected", attempt: 0 }, }; sessions.set(next.id, resetEntry); send(response, 200, payload(runner, resetEntry)); return; } else if (route === "interaction") { // The session stores the typed response in the mock control plane // before resuming the same provider thread. await entry.session.resolveInteraction({ interactionId: String(body.interactionId ?? ""), outcome: String(body.outcome ?? "answered"), result: body.result ?? null, }); } else { send(response, 404, { error: "unknown_route" }); return; } send(response, 200, payload(runner, entry)); } catch (error) { if (response.headersSent) { // A streamed turn reports its own failures as a framed error; there is // no status code left to send once the first frame is on the wire. if (!response.writableEnded) response.end(); return; } if (error instanceof RouteError) { send(response, error.status, { error: error.code, message: error.message }); return; } send(response, 500, { error: "capability_issue_thread_unavailable", message: String(error instanceof Error ? error.message : error), }); } }; middleware.close = async () => { if (bootstrap === null) return; const { service } = await bootstrap; for (const sessionId of [...sessions.keys()]) { await retire(service, sessionId, "server shutdown"); } }; middleware.prepare = ready; return middleware; } export function capabilityIssueThreadServerPlugin(options = {}) { async function mount(server, host) { const middleware = createCapabilityIssueThreadMiddleware({ ...options, bindHost: host }); await middleware.prepare(host); server.middlewares.use(middleware); server.httpServer?.once("close", () => void middleware.close()); } return { name: "paperclip-runner-capability-issue-thread-server", async configureServer(server) { const host = server.config.server.host; await mount(server, typeof host === "string" ? host : host === true ? "0.0.0.0" : "127.0.0.1"); }, async configurePreviewServer(server) { const host = server.config.preview.host; await mount(server, typeof host === "string" ? host : host === true ? "0.0.0.0" : "127.0.0.1"); }, }; }