PaperClipAI/packages/paperclip-runner/scripts/capability-issue-thread-server.mjs

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;

/**
 * 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;
  }
}

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<string, {
   *   session: unknown,
   *   surface: "issue" | "cleanroom",
   *   scenario: string,
   *   identity: unknown,
   *   workingDirectory: string,
   *   ownsWorkingDirectory: boolean,
   *   turns: number,
   *   createdAt: number,
   *   capabilityHash: Buffer,
   *   connection: { state: string, attempt: number },
   * }>}
   */
  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 projected = view(runner, entry);
    if (projected.mode !== "live" || projected.identity.agentLabel !== "Real Codex") {
      throw new RouteError(500, "live_mode_required", "The clean-room path refused a non-live view.");
    }
    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;
  }

  /** Retire a session: stop its work, clear its authority, drop its workspace. */
  async function retire(service, sessionId, reason) {
    const entry = sessions.get(sessionId);
    if (entry === undefined) return;
    sessions.delete(sessionId);
    await service.stop(sessionId, reason).catch(() => undefined);
    if (entry.ownsWorkingDirectory) {
      await rm(entry.workingDirectory, { recursive: true, force: true }).catch(() => undefined);
    }
  }

  /**
   * 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) {
    const open = [...sessions.values()].filter((entry) => entry.surface === "cleanroom");
    // 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,
        ...(options.requestedModel === undefined ? {} : { requestedModel: options.requestedModel }),
      });
    } 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,
      connection: { state: "connected", attempt: 0 },
    };
    sessions.set(session.id, entry);
    return entry;
  }

  function cleanRoomPayload(runner, entry) {
    return {
      sessionId: entry.session.id,
      surface: "cleanroom",
      identity: entry.identity,
      limits: { maxTurns: MAX_TURNS_PER_SESSION, maxMessageBytes: MAX_MESSAGE_BYTES },
      turns: entry.turns,
      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";

      /**
       * 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)
            : 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: true });
        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,
            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 }),
          });
        } 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 === "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 rotates the capability as well as the session: the cookie the
        // caller arrived with must stop working, here and on the replacement.
        const capability = mintCapability();
        const capabilityHash = sha256(capability);
        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);
          response.setHeader("set-cookie", capabilityCookie(request, "cleanroom", capability));
          send(response, 200, cleanRoomPayload(runner, replacement));
          return;
        }
        const next = await service.reset(entry.session.id);
        sessions.delete(entry.session.id);
        const resetEntry = {
          ...entry,
          session: next,
          turns: 0,
          capabilityHash,
          connection: { state: "connected", attempt: 0 },
        };
        sessions.set(next.id, resetEntry);
        response.setHeader("set-cookie", capabilityCookie(request, "issue", capability));
        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");
    },
  };
}