diff --git a/server/src/__tests__/workspace-runtime-exposure-reservation.test.ts b/server/src/__tests__/workspace-runtime-exposure-reservation.test.ts
new file mode 100644
index 0000000000..a01d12c8d3
--- /dev/null
+++ b/server/src/__tests__/workspace-runtime-exposure-reservation.test.ts
@@ -0,0 +1,446 @@
+/**
+ * PAP-17419 end-to-end regression: a leased HTTPS app/HMR pair is never handed
+ * to another execution workspace.
+ *
+ * The pure mediation rules are covered in
+ * `services/runtime-exposure/port-reservation.test.ts`. This file drives the
+ * *wired* path against a real database, so the persisted-lease query, the
+ * allocator, the ownership gate, and the startup reconciliation sweep are all
+ * exercised as they actually run — the layer where PAP-17251 failed.
+ *
+ * Reproduces the finding: workspace `b7ce28b4` held `42001/52001` under an open
+ * lease. Its backend was stopped and its exposure torn down, so the row read
+ * `stopped` with `exposure.state = "removed"` and an emptied `listeners` array,
+ * and nothing on the allocation path could still see the reservation. The
+ * unrelated workspace `PAP-16986-add-posthog-mcp` was then handed the pair.
+ *
+ * ## Host safety
+ *
+ * The broker is a fake, so no Tailscale Serve state is ever read or mutated.
+ * Allocation is driven by an injected `isPortAvailable` that models a synthetic
+ * host, and the only ports it will ever return are the `425xx`/`525xx` pairs
+ * named below — deliberately far from a canary lane on `42000`/`42001`, which
+ * this suite must not disturb. Guests do bind those two pairs on loopback, so
+ * the readiness and exposure lifecycle is exercised for real; they are reaped in
+ * `afterEach`. `PAPERCLIP_HOME` is redirected to a temp dir so the local-service
+ * registry never touches the real instance on this host.
+ */
+import { randomUUID } from "node:crypto";
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest";
+import {
+  activityLog,
+  companies,
+  createDb,
+  executionWorkspaces,
+  projectWorkspaces,
+  projects,
+  workspaceRuntimeServices,
+  type Db,
+} from "@paperclipai/db";
+import { eq } from "drizzle-orm";
+
+import type { BrokerClient, BrokerListenerRequest } from "../services/runtime-exposure/broker-client.js";
+import {
+  reconcilePersistedRuntimeServicesOnStartup,
+  resetRuntimeServicesForTests,
+  setWorkspaceRuntimeExposureDepsForTests,
+  startRuntimeServicesForWorkspaceControl,
+} from "../services/workspace-runtime.js";
+import {
+  getEmbeddedPostgresTestSupport,
+  startEmbeddedPostgresTestDatabase,
+} from "./helpers/embedded-postgres.js";
+
+/** The pair lane B holds under an open lease. */
+const LEASED_APP_PORT = 42_501;
+const LEASED_HMR_PORT = 52_501;
+/** The next pair a correct allocator must relocate to. */
+const NEXT_APP_PORT = 42_502;
+const NEXT_HMR_PORT = 52_502;
+
+const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
+
+/**
+ * A synthetic host on which only the two dedicated pairs above are usable.
+ *
+ * Crucially, the leased pair reads FREE — no listener, because lane B is
+ * stopped. That is the exact condition under which the old allocator handed it
+ * away, so a fixture that reported it busy would not reproduce the bug at all.
+ */
+async function isPortAvailableOnSyntheticHost(port: number): Promise<boolean> {
+  return port === LEASED_APP_PORT || port === LEASED_HMR_PORT
+    || port === NEXT_APP_PORT || port === NEXT_HMR_PORT;
+}
+
+function createFakeBroker() {
+  const calls: string[] = [];
+  const reservedByHandle = new Map<string, BrokerListenerRequest[]>();
+  const exposedByRuntimeId = new Map<string, BrokerListenerRequest[]>();
+  /** Mappings attributed to a runtime this server did not reserve. */
+  const foreignListeners: Array<{ runtimeId: string; port: number; purpose: "app" | "vite_hmr" }> = [];
+
+  const broker: BrokerClient = {
+    async reserve(runtimeId, requested) {
+      calls.push(`reserve:${requested.map((listener) => listener.port).join(",")}`);
+      const handle = `handle-${runtimeId}`;
+      reservedByHandle.set(handle, requested);
+      return { handle, reservedPorts: requested.map((listener) => listener.port) };
+    },
+    async expose(runtimeId, handle) {
+      calls.push("expose");
+      const requested = reservedByHandle.get(handle) ?? [];
+      exposedByRuntimeId.set(runtimeId, requested);
+      return { handle, publicPorts: requested.map((listener) => listener.port) };
+    },
+    async remove(runtimeId, handle) {
+      calls.push(`remove:${runtimeId}`);
+      const listeners = exposedByRuntimeId.get(runtimeId) ?? reservedByHandle.get(handle) ?? [];
+      exposedByRuntimeId.delete(runtimeId);
+      reservedByHandle.delete(handle);
+      return { removedPorts: listeners.map((listener) => listener.port) };
+    },
+    async list() {
+      return [
+        ...foreignListeners,
+        ...[...exposedByRuntimeId.entries()].flatMap(([runtimeId, listeners]) =>
+          listeners.map((listener) => ({ runtimeId, port: listener.port, purpose: listener.purpose })),
+        ),
+      ];
+    },
+  };
+
+  return { broker, calls, foreignListeners };
+}
+
+function installDeps(broker: BrokerClient, overrides?: {
+  isPortAvailable?: (port: number) => Promise<boolean>;
+}) {
+  setWorkspaceRuntimeExposureDepsForTests({
+    broker,
+    isPortAvailable: overrides?.isPortAvailable ?? isPortAvailableOnSyntheticHost,
+    isBrokerAvailable: async () => true,
+    resolveHostname: async () => "runner.tail123.ts.net",
+    probeHealth: async () => true,
+    now: () => new Date().toISOString(),
+    // Loopback-bind diagnosis is covered by `workspace-runtime-exposure.test.ts`
+    // against real guests; stubbing it here keeps these cases about ownership.
+    diagnoseListenerBinds: async () => null,
+  });
+}
+
+const DECLARED_EXPOSE = {
+  type: "tailscale_https",
+  hostname: "auto",
+  publicPort: "same",
+  includePaperclipViteHmr: true,
+  failurePolicy: "fail_closed",
+} as const;
+
+/**
+ * A backend that binds its allocated app port and the HMR companion on
+ * loopback, matching the real managed lane. Readiness then has a real listener
+ * to probe, so the lifecycle runs to `ready` exactly as in production.
+ */
+const GUEST_COMMAND =
+  "node -e \"const http=require('node:http');const p=Number(process.env.PORT);"
+  + "for(const q of [p,p+10000])http.createServer((_,r)=>{r.statusCode=200;r.end('ok')}).listen(q,'127.0.0.1');"
+  + "setInterval(()=>{},1000)\"";
+
+(embeddedPostgresSupport.supported ? describe : describe.skip)(
+  "PAP-17419 leased exposure pair reservation",
+  () => {
+    let db: Db;
+    let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+    let previousHttpsMode: string | undefined;
+    let previousPaperclipHome: string | undefined;
+    let previousInstanceId: string | undefined;
+    /**
+     * An EMPTY workspace root, never the repo checkout.
+     *
+     * Pointing `cwd` at the real repo makes the start path do git and
+     * dependency-provisioning work on a large tree, which is slow enough to
+     * blow the test timeout and has nothing to do with what is under test.
+     */
+    let workspaceRoot: string;
+    let paperclipHome: string;
+
+    beforeAll(async () => {
+      tempDb = await startEmbeddedPostgresTestDatabase("pap17419-reservation-");
+      db = createDb(tempDb.connectionString);
+      previousHttpsMode = process.env.PAPERCLIP_MANAGED_RUNTIME_HTTPS;
+      process.env.PAPERCLIP_MANAGED_RUNTIME_HTTPS = "auto";
+    }, 60_000);
+
+    beforeEach(async () => {
+      workspaceRoot = await fs.mkdtemp(path.join(os.tmpdir(), "pap17419-workspace-"));
+      paperclipHome = await fs.mkdtemp(path.join(os.tmpdir(), "pap17419-home-"));
+      // Redirect the local-service registry into a throwaway instance. Without
+      // this the suite would write runtime-service records into the real
+      // Paperclip instance on this host and could confuse a live server.
+      previousPaperclipHome = process.env.PAPERCLIP_HOME;
+      previousInstanceId = process.env.PAPERCLIP_INSTANCE_ID;
+      process.env.PAPERCLIP_HOME = paperclipHome;
+      process.env.PAPERCLIP_INSTANCE_ID = `pap17419-${randomUUID()}`;
+    });
+
+    afterAll(async () => {
+      if (previousHttpsMode === undefined) delete process.env.PAPERCLIP_MANAGED_RUNTIME_HTTPS;
+      else process.env.PAPERCLIP_MANAGED_RUNTIME_HTTPS = previousHttpsMode;
+      await tempDb?.cleanup();
+    });
+
+    afterEach(async () => {
+      // Terminate first, while the suite's fake broker is still installed.
+      await resetRuntimeServicesForTests({ terminateProcesses: true });
+      if (previousPaperclipHome === undefined) delete process.env.PAPERCLIP_HOME;
+      else process.env.PAPERCLIP_HOME = previousPaperclipHome;
+      if (previousInstanceId === undefined) delete process.env.PAPERCLIP_INSTANCE_ID;
+      else process.env.PAPERCLIP_INSTANCE_ID = previousInstanceId;
+      await fs.rm(workspaceRoot, { recursive: true, force: true });
+      await fs.rm(paperclipHome, { recursive: true, force: true });
+      await db.delete(workspaceRuntimeServices);
+      await db.delete(executionWorkspaces);
+      await db.delete(projectWorkspaces);
+      await db.delete(projects);
+      // Drift reporting writes activity rows, which hold a company FK.
+      await db.delete(activityLog);
+      await db.delete(companies);
+    });
+
+    /**
+     * Seed the incident's two workspaces: lane B with an open lease and a
+     * stopped, torn-down runtime row on the leased pair, and the unrelated
+     * workspace that will ask for a pair next.
+     */
+    async function seedIncident(options?: { leaseStatus?: string; leasedRowStatus?: string }) {
+      const companyId = randomUUID();
+      const projectId = randomUUID();
+      const projectWorkspaceId = randomUUID();
+      const leasedWorkspaceId = randomUUID();
+      const otherWorkspaceId = randomUUID();
+      const leasedRuntimeId = randomUUID();
+
+      await db.insert(companies).values({
+        id: companyId,
+        name: "Paperclip",
+        issuePrefix: `Q${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
+        requireBoardApprovalForNewAgents: false,
+      });
+      await db.insert(projects).values({ id: projectId, companyId, name: "Paperclip App", status: "in_progress" });
+      await db.insert(projectWorkspaces).values({
+        id: projectWorkspaceId,
+        companyId,
+        projectId,
+        name: "Primary",
+        cwd: workspaceRoot,
+        isPrimary: true,
+      });
+      for (const [id, name] of [[leasedWorkspaceId, "lane-b"], [otherWorkspaceId, "posthog-mcp"]] as const) {
+        await db.insert(executionWorkspaces).values({
+          id,
+          companyId,
+          projectId,
+          projectWorkspaceId,
+          mode: "isolated_workspace",
+          strategyType: "git_worktree",
+          name,
+          // Lane B's lease is open unless a case closes it.
+          status: id === leasedWorkspaceId ? (options?.leaseStatus ?? "active") : "active",
+          closedAt: id === leasedWorkspaceId && options?.leaseStatus === "archived" ? new Date() : null,
+          cwd: workspaceRoot,
+          baseRef: "HEAD",
+          branchName: name,
+          providerType: "git_worktree",
+        });
+      }
+
+      // Lane B as the incident left it: stopped, torn down, `removed` — and an
+      // EMPTY listeners array, which is what `deprovisionExposure` writes. The
+      // `port` column is the only surviving record of the leased pair.
+      await db.insert(workspaceRuntimeServices).values({
+        id: leasedRuntimeId,
+        companyId,
+        projectId,
+        projectWorkspaceId,
+        executionWorkspaceId: leasedWorkspaceId,
+        scopeType: "execution_workspace",
+        scopeId: leasedWorkspaceId,
+        serviceName: "preview",
+        status: options?.leasedRowStatus ?? "stopped",
+        lifecycle: "shared",
+        command: GUEST_COMMAND,
+        cwd: workspaceRoot,
+        port: LEASED_APP_PORT,
+        provider: "local_process",
+        exposure: {
+          provider: "tailscale_https",
+          state: "removed",
+          publicUrl: null,
+          hostname: null,
+          listeners: [],
+          brokerRef: null,
+          lastError: null,
+          updatedAt: new Date().toISOString(),
+        },
+        stoppedAt: new Date(),
+      });
+
+      return { companyId, projectId, projectWorkspaceId, leasedWorkspaceId, otherWorkspaceId, leasedRuntimeId };
+    }
+
+    function startInput(seed: Awaited<ReturnType<typeof seedIncident>>, executionWorkspaceId: string) {
+      return {
+        invocationId: "pap-17419-reservation",
+        actor: { id: null, name: "Paperclip", companyId: seed.companyId },
+        issue: null,
+        db,
+        workspace: {
+          baseCwd: workspaceRoot,
+          source: "project_primary" as const,
+          projectId: seed.projectId,
+          workspaceId: seed.projectWorkspaceId,
+          repoUrl: null,
+          repoRef: null,
+          strategy: "project_primary" as const,
+          cwd: workspaceRoot,
+          branchName: "test",
+          worktreePath: null,
+          warnings: [],
+          created: false,
+        },
+        executionWorkspaceId,
+        config: {
+          workspaceRuntime: {
+            services: [{
+              name: "preview",
+              command: GUEST_COMMAND,
+              port: { type: "auto", envKey: "PORT" },
+              readiness: { type: "http", urlTemplate: "http://127.0.0.1:{{port}}", timeoutSec: 10, intervalMs: 50 },
+              expose: DECLARED_EXPOSE,
+            }],
+          },
+        },
+        adapterEnv: {},
+      };
+    }
+
+    it("regression 1: does not hand a stopped-but-leased pair to another workspace", async () => {
+      const seed = await seedIncident();
+      const { broker, calls } = createFakeBroker();
+      installDeps(broker);
+
+      const [runtime] = await startRuntimeServicesForWorkspaceControl(
+        startInput(seed, seed.otherWorkspaceId),
+      );
+
+      // The incident outcome would be LEASED_APP_PORT here.
+      expect(runtime.port).toBe(NEXT_APP_PORT);
+      expect(runtime.port).not.toBe(LEASED_APP_PORT);
+      expect(runtime.url).toBe(`https://runner.tail123.ts.net:${NEXT_APP_PORT}`);
+      // The broker was never even asked to reserve the leased pair.
+      expect(calls.filter((call) => call.includes(String(LEASED_APP_PORT)))).toEqual([]);
+      expect(calls[0]).toBe(`reserve:${NEXT_APP_PORT},${NEXT_HMR_PORT}`);
+    }, 30_000);
+
+    it("regression 1: fails closed rather than reusing the leased pair when it is the only one left", async () => {
+      const seed = await seedIncident();
+      const { broker, calls } = createFakeBroker();
+      // A host on which the leased pair is the only free pair in the range.
+      installDeps(broker, {
+        isPortAvailable: async (port) => port === LEASED_APP_PORT || port === LEASED_HMR_PORT,
+      });
+
+      await expect(startRuntimeServicesForWorkspaceControl(
+        startInput(seed, seed.otherWorkspaceId),
+      )).rejects.toThrow(/no free app\/HMR port pair available/);
+
+      expect(calls).toEqual([]);
+    }, 30_000);
+
+    it("regression 1: lets the leaseholder's own workspace take its pair back", async () => {
+      const seed = await seedIncident();
+      const { broker } = createFakeBroker();
+      installDeps(broker);
+
+      const [runtime] = await startRuntimeServicesForWorkspaceControl(
+        startInput(seed, seed.leasedWorkspaceId),
+      );
+
+      // Same workspace, so the preserved-port preference still applies.
+      expect(runtime.port).toBe(LEASED_APP_PORT);
+    }, 30_000);
+
+    it("regression 3: denies a pair whose Serve mapping belongs to another runtime", async () => {
+      const seed = await seedIncident();
+      const { broker, foreignListeners } = createFakeBroker();
+      // The pair the allocator would otherwise relocate to is already published
+      // by a runtime this server has no row for — the unattributable case.
+      foreignListeners.push({ runtimeId: "runtime-not-ours", port: NEXT_APP_PORT, purpose: "app" });
+      installDeps(broker);
+
+      await expect(startRuntimeServicesForWorkspaceControl(
+        startInput(seed, seed.otherWorkspaceId),
+      )).rejects.toThrow(/no free app\/HMR port pair available|HTTPS exposure allocation denied/);
+    }, 30_000);
+
+    it("regression 5: release makes the pair reusable by a different workspace", async () => {
+      const seed = await seedIncident({ leaseStatus: "archived" });
+      const { broker } = createFakeBroker();
+      installDeps(broker);
+
+      const [runtime] = await startRuntimeServicesForWorkspaceControl(
+        startInput(seed, seed.otherWorkspaceId),
+      );
+
+      // Lane B's lease is released, so its pair is genuinely free again and the
+      // ascending scan hands out the lowest one.
+      expect(runtime.port).toBe(LEASED_APP_PORT);
+    }, 30_000);
+
+    it("regression 3: reconciliation surfaces a removed row whose pair is mapped elsewhere, and adopts nothing", async () => {
+      const seed = await seedIncident();
+      const { broker, foreignListeners, calls } = createFakeBroker();
+      // Lane B reads `stopped`/`removed`, but the host still publishes its pair
+      // for someone else. Exactly the falsely-attributed state from PAP-17251.
+      foreignListeners.push({ runtimeId: "runtime-not-ours", port: LEASED_APP_PORT, purpose: "app" });
+      installDeps(broker);
+
+      const result = await reconcilePersistedRuntimeServicesOnStartup(db);
+
+      expect(result.exposureReservationDrift).toHaveLength(1);
+      expect(result.exposureReservationDrift[0]).toMatchObject({
+        runtimeServiceId: seed.leasedRuntimeId,
+        port: LEASED_APP_PORT,
+        reason: "serve_mapping",
+        owner: { executionWorkspaceId: seed.leasedWorkspaceId },
+        conflictingOwner: { runtimeServiceId: "runtime-not-ours" },
+      });
+      // The occupying mapping is left strictly alone: no removal, no adoption.
+      expect(calls.filter((call) => call.startsWith("remove"))).toEqual([]);
+
+      const [row] = await db
+        .select()
+        .from(workspaceRuntimeServices)
+        .where(eq(workspaceRuntimeServices.id, seed.leasedRuntimeId));
+      expect(row!.status).toBe("stopped");
+      expect(row!.port).toBe(LEASED_APP_PORT);
+    }, 30_000);
+
+    it("regression 4: concurrent starts in different workspaces never share a pair", async () => {
+      const seed = await seedIncident({ leaseStatus: "archived" });
+      const { broker } = createFakeBroker();
+      installDeps(broker);
+
+      const [first, second] = await Promise.all([
+        startRuntimeServicesForWorkspaceControl(startInput(seed, seed.otherWorkspaceId)),
+        startRuntimeServicesForWorkspaceControl(startInput(seed, seed.leasedWorkspaceId)),
+      ]);
+
+      const ports = [first[0]?.port, second[0]?.port].sort();
+      expect(ports).toEqual([LEASED_APP_PORT, NEXT_APP_PORT]);
+    }, 30_000);
+  },
+);
diff --git a/server/src/services/runtime-exposure/port-pair.ts b/server/src/services/runtime-exposure/port-pair.ts
index 244224c9f7..dc274994ab 100644
--- a/server/src/services/runtime-exposure/port-pair.ts
+++ b/server/src/services/runtime-exposure/port-pair.ts
@@ -47,6 +47,17 @@ export interface AllocateExposurePortPairInput {
    * the caller special-case it.
    */
   preferredAppPort?: number | null;
+  /**
+   * Atomically take the pair for this allocation, or refuse it. Returning false
+   * makes the scan move on as if the pair were busy.
+   *
+   * Without it, two concurrent allocators observe identical reservations and
+   * identical probe results and both walk away with the lowest free pair —
+   * neither has bound anything yet, so nothing downstream can tell them apart.
+   * The claim must cover both ports together: a half-claimed pair is the
+   * orphaned-HMR-companion failure this allocator exists to prevent.
+   */
+  claimPair?: (pair: ExposurePortPair) => boolean;
 }
 
 /**
@@ -68,7 +79,11 @@ export async function allocateExposurePortPair(
     // Probe the app port first; short-circuit before probing the companion.
     if (!(await input.isPortAvailable(appPort))) return null;
     if (!(await input.isPortAvailable(hmrPort))) return null;
-    return { appPort, hmrPort };
+    const pair = { appPort, hmrPort };
+    // Claim last: the probes are the cheap filter, and claiming a pair we then
+    // reject would leak a hold on it for the claim's whole TTL.
+    if (input.claimPair && !input.claimPair(pair)) return null;
+    return pair;
   };
 
   if (input.preferredAppPort != null) {
diff --git a/server/src/services/runtime-exposure/port-reservation.test.ts b/server/src/services/runtime-exposure/port-reservation.test.ts
new file mode 100644
index 0000000000..1ee65a2116
--- /dev/null
+++ b/server/src/services/runtime-exposure/port-reservation.test.ts
@@ -0,0 +1,551 @@
+/**
+ * PAP-17419 regression coverage for central exposure reservation/ownership
+ * mediation.
+ *
+ * Each `describe` below is one of the five regression requirements on the
+ * issue, named so a reviewer can map test → requirement without reading the
+ * bodies. The scenario ports and workspace IDs mirror the PAP-17251 finding:
+ * lane B held `42001/52001` under an open lease while the unrelated workspace
+ * `PAP-16986-add-posthog-mcp` took the pair.
+ */
+import { describe, expect, it } from "vitest";
+
+import {
+  buildExposureReservationLedger,
+  collectRowExposurePorts,
+  describeExposurePortConflict,
+  describeExposureReservationDrift,
+  ExposurePortOwnershipConflictError,
+  ExposurePortPairClaims,
+  findExposurePairConflict,
+  findExposureReservationDrift,
+  isExposureAdoptionPermitted,
+  type ExposureOwnerIdentity,
+  type PersistedExposureRowSnapshot,
+} from "./port-reservation.js";
+import { allocateExposurePortPair } from "./port-pair.js";
+
+/** The stopped-but-leased lane from the finding. */
+const LEASED_WORKSPACE = "b7ce28b4-7921-4e23-8be2-8b42193bf4ad";
+/** The unrelated workspace that took its pair. */
+const OTHER_WORKSPACE = "9f2c1d40-1111-4222-8333-444455556666";
+const LEASED_ISSUE = "17251aaa-1111-4222-8333-444455556666";
+
+function identity(overrides: Partial<ExposureOwnerIdentity> = {}): ExposureOwnerIdentity {
+  return {
+    runtimeServiceId: null,
+    executionWorkspaceId: null,
+    projectWorkspaceId: null,
+    issueId: null,
+    ...overrides,
+  };
+}
+
+/**
+ * A lane that has been stopped AND torn down.
+ *
+ * `deprovisionExposure` writes a fresh `removed` status whose `listeners` array
+ * is empty, so the `port` column is the only surviving record of which pair the
+ * lease owns. Reproducing that faithfully is the whole point of the fixture.
+ */
+function stoppedTornDownRow(overrides: Partial<PersistedExposureRowSnapshot> = {}): PersistedExposureRowSnapshot {
+  return {
+    id: "runtime-lane-b",
+    status: "stopped",
+    port: 42001,
+    exposure: { state: "removed", listeners: [] },
+    executionWorkspaceId: LEASED_WORKSPACE,
+    projectWorkspaceId: null,
+    issueId: LEASED_ISSUE,
+    ...overrides,
+  };
+}
+
+describe("collectRowExposurePorts", () => {
+  it("derives the pair from the port column when teardown erased the listeners", () => {
+    expect([...collectRowExposurePorts(stoppedTornDownRow())]).toEqual([42001, 52001]);
+  });
+
+  it("ignores a legacy pinned port outside the dedicated range", () => {
+    const row = stoppedTornDownRow({ port: 45439 });
+    expect([...collectRowExposurePorts(row)]).toEqual([]);
+  });
+
+  it("unions declared listeners with the port-derived pair", () => {
+    const row = stoppedTornDownRow({
+      exposure: { state: "ready", listeners: [{ targetPort: 42001 }, { targetPort: 52001 }] },
+    });
+    expect([...collectRowExposurePorts(row)].sort()).toEqual([42001, 52001]);
+  });
+});
+
+describe("regression 1: stopped-but-leased pair reuse is denied", () => {
+  it("reserves a torn-down lane's pair while its lease is still open", () => {
+    const ledger = buildExposureReservationLedger({
+      persistedRows: [stoppedTornDownRow()],
+      activeExecutionWorkspaceIds: [LEASED_WORKSPACE],
+    });
+
+    expect(ledger.reservedPorts.has(42001)).toBe(true);
+    expect(ledger.reservedPorts.has(52001)).toBe(true);
+    expect(ledger.reservationByPort.get(42001)).toMatchObject({
+      source: "leased_workspace",
+      owner: { executionWorkspaceId: LEASED_WORKSPACE, issueId: LEASED_ISSUE },
+    });
+  });
+
+  it("denies an unrelated workspace the leased pair, naming the holder", () => {
+    const ledger = buildExposureReservationLedger({
+      persistedRows: [stoppedTornDownRow()],
+      activeExecutionWorkspaceIds: [LEASED_WORKSPACE],
+    });
+
+    const conflict = findExposurePairConflict({
+      pair: { appPort: 42001, hmrPort: 52001 },
+      claimant: identity({ executionWorkspaceId: OTHER_WORKSPACE, runtimeServiceId: "runtime-posthog" }),
+      ledger,
+    });
+
+    expect(conflict).toMatchObject({ code: "reserved_by_other_workspace", port: 42001 });
+    const message = describeExposurePortConflict(conflict!);
+    expect(message).toContain("42001");
+    expect(message).toContain(LEASED_WORKSPACE);
+    expect(message).toContain(LEASED_ISSUE);
+    // The operator must be told the lane looks stopped on purpose, or the
+    // report reads as a stale-row bug rather than a live reservation.
+    expect(message).toContain("lease is still open");
+  });
+
+  it("lets the leaseholder's own lane reclaim its pair on restart", () => {
+    const ledger = buildExposureReservationLedger({
+      persistedRows: [stoppedTornDownRow()],
+      activeExecutionWorkspaceIds: [LEASED_WORKSPACE],
+    });
+
+    expect(findExposurePairConflict({
+      pair: { appPort: 42001, hmrPort: 52001 },
+      claimant: identity({ executionWorkspaceId: LEASED_WORKSPACE, runtimeServiceId: "runtime-lane-b-restart" }),
+      ledger,
+    })).toBeNull();
+  });
+
+  it("keeps the HMR companion reserved even when only the app port is claimed", () => {
+    const ledger = buildExposureReservationLedger({
+      persistedRows: [stoppedTornDownRow()],
+      activeExecutionWorkspaceIds: [LEASED_WORKSPACE],
+    });
+
+    // A different pair whose app port is free but whose companion is the
+    // lease's HMR port can never happen with the fixed offset, so assert the
+    // companion is independently defended instead.
+    expect(findExposurePairConflict({
+      pair: { appPort: 52001, hmrPort: 62001 },
+      claimant: identity({ executionWorkspaceId: OTHER_WORKSPACE }),
+      ledger,
+    })).toMatchObject({ code: "reserved_by_other_workspace", port: 52001 });
+  });
+
+  it("still reserves nothing for a row with no lease and no exposure", () => {
+    const ledger = buildExposureReservationLedger({
+      persistedRows: [stoppedTornDownRow()],
+      activeExecutionWorkspaceIds: [],
+    });
+    expect(ledger.reservedPorts.size).toBe(0);
+  });
+});
+
+describe("regression 2: cross-execution-workspace process adoption is denied", () => {
+  it("refuses adoption when the execution workspace IDs differ", () => {
+    expect(isExposureAdoptionPermitted(
+      identity({ executionWorkspaceId: LEASED_WORKSPACE }),
+      identity({ executionWorkspaceId: OTHER_WORKSPACE }),
+    )).toBe(false);
+  });
+
+  it("refuses adoption when either side's workspace is unknown", () => {
+    expect(isExposureAdoptionPermitted(
+      identity({ executionWorkspaceId: null, runtimeServiceId: "runtime-unknown" }),
+      identity({ executionWorkspaceId: OTHER_WORKSPACE }),
+    )).toBe(false);
+    expect(isExposureAdoptionPermitted(
+      identity({ executionWorkspaceId: LEASED_WORKSPACE }),
+      identity({ executionWorkspaceId: null }),
+    )).toBe(false);
+    // Two nulls are not "the same workspace"; they are two unknowns.
+    expect(isExposureAdoptionPermitted(identity(), identity())).toBe(false);
+  });
+
+  it("permits the same workspace and the same runtime service", () => {
+    expect(isExposureAdoptionPermitted(
+      identity({ executionWorkspaceId: LEASED_WORKSPACE }),
+      identity({ executionWorkspaceId: LEASED_WORKSPACE }),
+    )).toBe(true);
+    expect(isExposureAdoptionPermitted(
+      identity({ runtimeServiceId: "runtime-lane-b" }),
+      identity({ runtimeServiceId: "runtime-lane-b" }),
+    )).toBe(true);
+  });
+
+  it("denies a pair whose live listener belongs to another workspace", () => {
+    const conflict = findExposurePairConflict({
+      pair: { appPort: 42001, hmrPort: 52001 },
+      claimant: identity({ executionWorkspaceId: OTHER_WORKSPACE }),
+      ledger: buildExposureReservationLedger({}),
+      listenerOwners: new Map([[42001, identity({ executionWorkspaceId: LEASED_WORKSPACE, issueId: LEASED_ISSUE })]]),
+    });
+
+    expect(conflict).toMatchObject({ code: "listener_owned_by_other_workspace", port: 42001 });
+    expect(describeExposurePortConflict(conflict!)).toContain(LEASED_WORKSPACE);
+  });
+
+  it("denies a live listener it cannot attribute rather than assuming it is ours", () => {
+    const conflict = findExposurePairConflict({
+      pair: { appPort: 42001, hmrPort: 52001 },
+      claimant: identity({ executionWorkspaceId: OTHER_WORKSPACE }),
+      ledger: buildExposureReservationLedger({}),
+      listenerOwners: new Map([[52001, null]]),
+    });
+
+    expect(conflict).toMatchObject({ code: "listener_owned_by_other_workspace", port: 52001 });
+    expect(describeExposurePortConflict(conflict!)).toContain("an unidentified owner");
+  });
+
+  it("throws a terminal, attributed error on the allocation path", () => {
+    const conflict = findExposurePairConflict({
+      pair: { appPort: 42001, hmrPort: 52001 },
+      claimant: identity({ executionWorkspaceId: OTHER_WORKSPACE }),
+      ledger: buildExposureReservationLedger({
+        persistedRows: [stoppedTornDownRow()],
+        activeExecutionWorkspaceIds: [LEASED_WORKSPACE],
+      }),
+    });
+    const error = new ExposurePortOwnershipConflictError(conflict!);
+
+    expect(error.code).toBe("exposure_port_ownership_conflict");
+    expect(error.message).toContain("HTTPS exposure allocation denied");
+    expect(error.message).toContain(LEASED_WORKSPACE);
+  });
+});
+
+describe("regression 3: Serve mapping ownership mismatch is visible and fails closed", () => {
+  it("denies a pair published by another workspace's Serve mapping", () => {
+    const rows: PersistedExposureRowSnapshot[] = [
+      {
+        id: "runtime-posthog",
+        status: "running",
+        port: 42001,
+        exposure: { state: "ready", listeners: [{ targetPort: 42001 }, { targetPort: 52001 }] },
+        executionWorkspaceId: OTHER_WORKSPACE,
+        projectWorkspaceId: null,
+        issueId: null,
+      },
+    ];
+    const ledger = buildExposureReservationLedger({
+      persistedRows: rows,
+      brokerMappings: [{ runtimeId: "runtime-posthog", port: 42001 }],
+    });
+
+    expect(ledger.reservationByPort.get(42001)).toMatchObject({
+      source: "broker_mapping",
+      owner: { executionWorkspaceId: OTHER_WORKSPACE },
+    });
+
+    const conflict = findExposurePairConflict({
+      pair: { appPort: 42001, hmrPort: 52001 },
+      claimant: identity({ executionWorkspaceId: LEASED_WORKSPACE }),
+      ledger,
+      serveMappingOwners: new Map([[42001, identity({ executionWorkspaceId: OTHER_WORKSPACE })]]),
+    });
+    expect(conflict).toMatchObject({ port: 42001 });
+    expect(describeExposurePortConflict(conflict!)).toContain(OTHER_WORKSPACE);
+  });
+
+  it("reports a serve-mapping mismatch when the ledger alone would allow it", () => {
+    const conflict = findExposurePairConflict({
+      pair: { appPort: 42005, hmrPort: 52005 },
+      claimant: identity({ executionWorkspaceId: LEASED_WORKSPACE }),
+      ledger: buildExposureReservationLedger({}),
+      serveMappingOwners: new Map([[52005, identity({ executionWorkspaceId: OTHER_WORKSPACE })]]),
+    });
+
+    expect(conflict).toMatchObject({ code: "serve_mapping_owned_by_other_workspace", port: 52005 });
+    expect(describeExposurePortConflict(conflict!)).toContain("Tailscale Serve mapping");
+  });
+
+  it("surfaces a stopped/removed row whose reserved port is mapped to another runtime", () => {
+    const drift = findExposureReservationDrift({
+      persistedRows: [
+        stoppedTornDownRow(),
+        {
+          id: "runtime-posthog",
+          status: "running",
+          port: null,
+          exposure: null,
+          executionWorkspaceId: OTHER_WORKSPACE,
+          projectWorkspaceId: null,
+          issueId: null,
+        },
+      ],
+      brokerMappings: [{ runtimeId: "runtime-posthog", port: 42001 }],
+    });
+
+    expect(drift).toHaveLength(1);
+    expect(drift[0]).toMatchObject({
+      runtimeServiceId: "runtime-lane-b",
+      port: 42001,
+      reason: "serve_mapping",
+      conflictingOwner: { executionWorkspaceId: OTHER_WORKSPACE },
+    });
+    const description = describeExposureReservationDrift(drift[0]!);
+    expect(description).toContain("recorded stopped/removed");
+    expect(description).toContain(OTHER_WORKSPACE);
+  });
+
+  it("surfaces a stopped/removed row whose reserved port still has a live listener", () => {
+    const drift = findExposureReservationDrift({
+      persistedRows: [stoppedTornDownRow()],
+      livePorts: new Set([52001]),
+    });
+
+    expect(drift).toMatchObject([{ port: 52001, reason: "live_listener", conflictingOwner: null }]);
+  });
+
+  it("does not call a lane's own live listener drift", () => {
+    const drift = findExposureReservationDrift({
+      persistedRows: [stoppedTornDownRow()],
+      livePorts: new Set([42001]),
+      listenerOwners: new Map([[42001, identity({ runtimeServiceId: "runtime-lane-b" })]]),
+    });
+
+    expect(drift).toEqual([]);
+  });
+
+  it("does not report a healthy running lane as drift", () => {
+    const drift = findExposureReservationDrift({
+      persistedRows: [{
+        id: "runtime-lane-a",
+        status: "running",
+        port: 42000,
+        exposure: { state: "ready", listeners: [{ targetPort: 42000 }, { targetPort: 52000 }] },
+        executionWorkspaceId: LEASED_WORKSPACE,
+        projectWorkspaceId: null,
+        issueId: null,
+      }],
+      livePorts: new Set([42000, 52000]),
+      brokerMappings: [{ runtimeId: "runtime-lane-a", port: 42000 }],
+    });
+
+    expect(drift).toEqual([]);
+  });
+});
+
+describe("regression 4: concurrent allocators return unique pairs", () => {
+  it("claims both ports of a pair or neither", () => {
+    const claims = new ExposurePortPairClaims();
+    expect(claims.claim({ appPort: 42000, hmrPort: 52000 }, 0)).toBe(true);
+    // Overlapping on either port is refused, and the refusal must not leave a
+    // partial hold behind.
+    expect(claims.claim({ appPort: 42000, hmrPort: 52000 }, 0)).toBe(false);
+    expect(claims.claim({ appPort: 42001, hmrPort: 52000 }, 0)).toBe(false);
+    expect([...claims.activePorts(0)].sort()).toEqual([42000, 52000]);
+  });
+
+  it("expires a claim after its TTL so a crashed start cannot burn the range", () => {
+    const claims = new ExposurePortPairClaims(1_000);
+    expect(claims.claim({ appPort: 42000, hmrPort: 52000 }, 0)).toBe(true);
+    expect(claims.claim({ appPort: 42000, hmrPort: 52000 }, 500)).toBe(false);
+    expect(claims.claim({ appPort: 42000, hmrPort: 52000 }, 1_001)).toBe(true);
+  });
+
+  it("hands two concurrent allocators distinct pairs", async () => {
+    const claims = new ExposurePortPairClaims();
+    const allocate = () => allocateExposurePortPair({
+      isPortAvailable: async () => true,
+      claimPair: (pair) => claims.claim(pair),
+    });
+
+    const [first, second, third] = await Promise.all([allocate(), allocate(), allocate()]);
+
+    expect(first).toEqual({ appPort: 42000, hmrPort: 52000 });
+    expect(second).toEqual({ appPort: 42001, hmrPort: 52001 });
+    expect(third).toEqual({ appPort: 42002, hmrPort: 52002 });
+  });
+
+  it("gives concurrent allocators distinct pairs even when both prefer the same port", async () => {
+    const claims = new ExposurePortPairClaims();
+    const allocate = () => allocateExposurePortPair({
+      isPortAvailable: async () => true,
+      preferredAppPort: 42007,
+      claimPair: (pair) => claims.claim(pair),
+    });
+
+    const [first, second] = await Promise.all([allocate(), allocate()]);
+
+    expect(first.appPort).toBe(42007);
+    expect(second.appPort).not.toBe(42007);
+    expect(second.hmrPort).not.toBe(first.hmrPort);
+  });
+
+  it("does not claim a pair it rejects for availability", async () => {
+    const claims = new ExposurePortPairClaims();
+    await allocateExposurePortPair({
+      isPortAvailable: async (port) => port !== 42000 && port !== 52000,
+      claimPair: (pair) => claims.claim(pair),
+    });
+
+    // 42000 was probed and refused; holding it would waste the port for a full TTL.
+    expect(claims.activePorts()).toEqual(new Set([42001, 52001]));
+  });
+
+  it("folds live claims into a ledger so a second allocator sees them", () => {
+    const claims = new ExposurePortPairClaims();
+    claims.claim({ appPort: 42000, hmrPort: 52000 }, 0);
+    const ledger = buildExposureReservationLedger({ inFlightClaimedPorts: claims.activePorts(0) });
+
+    expect(findExposurePairConflict({
+      pair: { appPort: 42000, hmrPort: 52000 },
+      claimant: identity({ executionWorkspaceId: OTHER_WORKSPACE }),
+      ledger,
+    })).toMatchObject({ code: "reserved_by_other_workspace" });
+  });
+});
+
+describe("regression 5: release/teardown makes the pair reusable, sparing manual mappings", () => {
+  it("frees the pair once the lease is released", () => {
+    const row = stoppedTornDownRow();
+    const leased = buildExposureReservationLedger({
+      persistedRows: [row],
+      activeExecutionWorkspaceIds: [LEASED_WORKSPACE],
+    });
+    expect(leased.reservedPorts.has(42001)).toBe(true);
+
+    // Lease released: the workspace is archived/closed, so it no longer appears.
+    const released = buildExposureReservationLedger({
+      persistedRows: [row],
+      activeExecutionWorkspaceIds: [],
+    });
+
+    expect(released.reservedPorts.has(42001)).toBe(false);
+    expect(findExposurePairConflict({
+      pair: { appPort: 42001, hmrPort: 52001 },
+      claimant: identity({ executionWorkspaceId: OTHER_WORKSPACE }),
+      ledger: released,
+    })).toBeNull();
+  });
+
+  it("releases an in-process claim on teardown", () => {
+    const claims = new ExposurePortPairClaims();
+    claims.claim({ appPort: 42001, hmrPort: 52001 }, 0);
+    claims.release({ appPort: 42001, hmrPort: 52001 });
+
+    expect(claims.activePorts(0).size).toBe(0);
+    expect(claims.claim({ appPort: 42001, hmrPort: 52001 }, 0)).toBe(true);
+  });
+
+  it("keeps quarantined ports out of circulation after release", () => {
+    const ledger = buildExposureReservationLedger({
+      persistedRows: [stoppedTornDownRow()],
+      activeExecutionWorkspaceIds: [],
+      quarantinedPorts: [42001],
+    });
+
+    const conflict = findExposurePairConflict({
+      pair: { appPort: 42001, hmrPort: 52001 },
+      claimant: identity({ executionWorkspaceId: OTHER_WORKSPACE }),
+      ledger,
+    });
+    expect(conflict).toMatchObject({ code: "quarantined_port" });
+    expect(describeExposurePortConflict(conflict!)).toContain("quarantined");
+  });
+
+  it("never reserves or reports a manual/unknown mapping the broker does not own", () => {
+    // The broker's `list()` structurally cannot return unknown/manual entries,
+    // so a manual mapping on 42010 reaches neither the ledger nor drift. Assert
+    // the absence explicitly: a release path that started reserving or
+    // reporting them would be the PAP-17285 regression coming back.
+    const ledger = buildExposureReservationLedger({
+      persistedRows: [stoppedTornDownRow()],
+      activeExecutionWorkspaceIds: [],
+      brokerMappings: [],
+    });
+    expect(ledger.reservedPorts.has(42010)).toBe(false);
+
+    expect(findExposureReservationDrift({
+      persistedRows: [stoppedTornDownRow({ port: 42010 })],
+      brokerMappings: [],
+    })).toEqual([]);
+  });
+
+  it("lets a fresh workspace take the released pair end to end", async () => {
+    const claims = new ExposurePortPairClaims();
+    const ledger = buildExposureReservationLedger({
+      persistedRows: [stoppedTornDownRow()],
+      activeExecutionWorkspaceIds: [],
+      inFlightClaimedPorts: claims.activePorts(),
+    });
+
+    const pair = await allocateExposurePortPair({
+      isPortAvailable: async (port) => port >= 42001,
+      reserved: ledger.reservedPorts,
+      preferredAppPort: 42001,
+      claimPair: (candidate) => claims.claim(candidate),
+    });
+
+    expect(pair).toEqual({ appPort: 42001, hmrPort: 52001 });
+    expect(findExposurePairConflict({
+      pair,
+      claimant: identity({ executionWorkspaceId: OTHER_WORKSPACE }),
+      ledger,
+    })).toBeNull();
+  });
+});
+
+describe("reservation precedence", () => {
+  it("reports the strongest available evidence for a contested port", () => {
+    const ledger = buildExposureReservationLedger({
+      persistedRows: [stoppedTornDownRow()],
+      activeExecutionWorkspaceIds: [LEASED_WORKSPACE],
+      inMemoryRuntimes: [{
+        runtimeServiceId: "runtime-lane-b",
+        executionWorkspaceId: LEASED_WORKSPACE,
+        projectWorkspaceId: null,
+        issueId: LEASED_ISSUE,
+        ports: [42001],
+      }],
+      brokerMappings: [{ runtimeId: "runtime-lane-b", port: 42001 }],
+    });
+
+    // Running in this process beats a Serve mapping, which beats the lease.
+    expect(ledger.reservationByPort.get(42001)?.source).toBe("in_memory_runtime");
+    expect(ledger.reservationByPort.get(52001)?.source).toBe("leased_workspace");
+  });
+
+  it("quarantine outranks every other claim", () => {
+    const ledger = buildExposureReservationLedger({
+      persistedRows: [stoppedTornDownRow()],
+      activeExecutionWorkspaceIds: [LEASED_WORKSPACE],
+      quarantinedPorts: [42001],
+    });
+
+    expect(ledger.reservationByPort.get(42001)?.source).toBe("quarantine");
+    // Even the leaseholder cannot take a quarantined port back.
+    expect(findExposurePairConflict({
+      pair: { appPort: 42001, hmrPort: 52001 },
+      claimant: identity({ executionWorkspaceId: LEASED_WORKSPACE }),
+      ledger,
+    })).toMatchObject({ code: "quarantined_port" });
+  });
+
+  it("names an unresolvable broker runtime rather than reporting no holder", () => {
+    const ledger = buildExposureReservationLedger({
+      brokerMappings: [{ runtimeId: "runtime-vanished", port: 42003 }],
+    });
+
+    const conflict = findExposurePairConflict({
+      pair: { appPort: 42003, hmrPort: 52003 },
+      claimant: identity({ executionWorkspaceId: OTHER_WORKSPACE }),
+      ledger,
+    });
+    expect(describeExposurePortConflict(conflict!)).toContain("runtime-vanished");
+  });
+});
diff --git a/server/src/services/runtime-exposure/port-reservation.ts b/server/src/services/runtime-exposure/port-reservation.ts
new file mode 100644
index 0000000000..5fbb5659c2
--- /dev/null
+++ b/server/src/services/runtime-exposure/port-reservation.ts
@@ -0,0 +1,539 @@
+/**
+ * Central reservation + ownership mediation for `tailscale_https` app/HMR port
+ * pairs (PAP-17419, from the PAP-17251 finding).
+ *
+ * ## Why this module exists
+ *
+ * Before it, "is this pair free?" was answered by three unrelated views that
+ * never had to agree:
+ *
+ *  1. the allocator's liveness probe — is anything *listening* right now,
+ *  2. `collectReservedExposurePorts` — persisted rows, but only those whose
+ *     `exposure.state` was not yet `removed`,
+ *  3. the broker's Serve mapping — consulted nowhere on the allocation path.
+ *
+ * A workspace can hold an exclusive lease on a pair while satisfying none of
+ * those: stop the backend and the listener is gone; tear the exposure down and
+ * the row goes `stopped` with `exposure.state = "removed"` **and an emptied
+ * `listeners` array**, so the row stops contributing anything to the reserved
+ * set even though its `port` column still names the pair the lease owns. That
+ * is exactly how the pair leased to workspace `b7ce28b4` (`42001/52001`) was
+ * handed to the unrelated workspace `PAP-16986-add-posthog-mcp`, while
+ * Paperclip went on reporting the leased lane `stopped` and its exposure
+ * `removed`.
+ *
+ * ## The mediation this module centralizes
+ *
+ * - **Reservation follows the lease, not the process.** An execution workspace
+ *   whose lease is still open reserves its pair through stop, teardown, and
+ *   `removed` — until the lease is explicitly released. {@link
+ *   buildExposureReservationLedger} derives the pair from the row's `port`
+ *   column precisely because teardown erases `exposure.listeners`.
+ * - **Complete mediation on allocation.** {@link findExposurePairConflict}
+ *   checks the persisted ledger *and* the live listener owner *and* the Serve
+ *   mapping owner. Any of the three disagreeing fails the allocation closed,
+ *   naming the conflicting workspace/issue.
+ * - **No adoption across execution workspaces, ever.** {@link
+ *   isExposureAdoptionPermitted} is the single predicate for "may this claimant
+ *   take over that holder's port", and it answers no whenever the two execution
+ *   workspace IDs are not the identical, non-null value.
+ * - **Drift is surfaced, not silently reused.** {@link
+ *   findExposureReservationDrift} reports a `stopped`/`removed` row whose
+ *   reserved ports are live or mapped by someone else, so reconciliation can
+ *   report a false-`stopped` lane instead of recycling it.
+ *
+ * Pure functions and one in-process claim registry: no sockets, no DB, no
+ * broker. Every caller injects its own view, which is what makes all five
+ * PAP-17419 regression cases testable without touching host Serve state.
+ */
+import { deriveViteHmrPort, isRuntimeExposureAppPort } from "@paperclipai/shared";
+
+/**
+ * Who holds (or wants) a port pair. Every field is optional evidence — a
+ * conflict report is only as specific as the views that produced it — but
+ * `executionWorkspaceId` is the identity that governs adoption.
+ */
+export interface ExposureOwnerIdentity {
+  runtimeServiceId: string | null;
+  executionWorkspaceId: string | null;
+  projectWorkspaceId: string | null;
+  issueId: string | null;
+}
+
+/** Where a reservation came from, ordered by strength of evidence below. */
+export type ExposureReservationSource =
+  /** This process is running the runtime that owns the pair. */
+  | "in_memory_runtime"
+  /** The broker reports a live Paperclip-owned Serve mapping on the port. */
+  | "broker_mapping"
+  /** A persisted row still claims an exposure that is not `removed`. */
+  | "persisted_exposure"
+  /** A stopped/removed row whose execution-workspace lease is still open. */
+  | "leased_workspace"
+  /** Quarantined after an ambiguous cleanup; never reusable by anyone. */
+  | "quarantine";
+
+/**
+ * Precedence when several views claim one port. Strongest first: the report
+ * should name the most concrete evidence available, so an operator reading
+ * "held by the live Serve mapping for workspace X" is not told "reserved by a
+ * lease" when both are true.
+ */
+const RESERVATION_SOURCE_RANK: Record<ExposureReservationSource, number> = {
+  quarantine: 0,
+  in_memory_runtime: 1,
+  broker_mapping: 2,
+  persisted_exposure: 3,
+  leased_workspace: 4,
+};
+
+export interface ExposurePortReservation {
+  port: number;
+  source: ExposureReservationSource;
+  owner: ExposureOwnerIdentity;
+}
+
+export interface ExposureReservationLedger {
+  /** Every port currently spoken for, whoever holds it. */
+  reservedPorts: ReadonlySet<number>;
+  /** Strongest-evidence reservation per port. */
+  reservationByPort: ReadonlyMap<number, ExposurePortReservation>;
+}
+
+/**
+ * The subset of a `workspace_runtime_services` row this module reasons about.
+ * Deliberately structural rather than the Drizzle row type: startup
+ * reconciliation, allocation, and tests all project into it from different
+ * queries.
+ */
+export interface PersistedExposureRowSnapshot {
+  id: string;
+  status: string;
+  executionWorkspaceId: string | null;
+  projectWorkspaceId: string | null;
+  issueId: string | null;
+  /** App port column; survives teardown and is the lease's durable claim. */
+  port: number | null;
+  exposure: {
+    state: string;
+    listeners: ReadonlyArray<{ targetPort: number }>;
+  } | null;
+}
+
+/** A runtime this process is currently running with an exposure attached. */
+export interface InMemoryExposureSnapshot {
+  runtimeServiceId: string;
+  executionWorkspaceId: string | null;
+  projectWorkspaceId: string | null;
+  issueId: string | null;
+  ports: ReadonlyArray<number>;
+}
+
+/** One Paperclip-owned listener as reported by the broker's `list()`. */
+export interface BrokerMappingSnapshot {
+  runtimeId: string;
+  port: number;
+}
+
+export interface BuildExposureReservationLedgerInput {
+  persistedRows?: Iterable<PersistedExposureRowSnapshot>;
+  inMemoryRuntimes?: Iterable<InMemoryExposureSnapshot>;
+  brokerMappings?: Iterable<BrokerMappingSnapshot>;
+  quarantinedPorts?: Iterable<number>;
+  /**
+   * Execution workspaces whose exclusive lease is still open. A row belonging
+   * to one of these reserves its pair regardless of process or exposure state;
+   * that is the whole fix for "stopped-but-leased pair reuse".
+   */
+  activeExecutionWorkspaceIds?: Iterable<string>;
+  /** In-process pair claims held by starts that have not bound a listener yet. */
+  inFlightClaimedPorts?: Iterable<number>;
+}
+
+function emptyIdentity(): ExposureOwnerIdentity {
+  return {
+    runtimeServiceId: null,
+    executionWorkspaceId: null,
+    projectWorkspaceId: null,
+    issueId: null,
+  };
+}
+
+/**
+ * Every port a row's pair covers.
+ *
+ * The union of the exposure's declared listeners and the pair derived from the
+ * `port` column matters: `deprovisionExposure` replaces the status with a fresh
+ * `removed` one whose `listeners` is `[]`, so a torn-down row would otherwise
+ * contribute nothing at all. The `port` column is what still names the pair the
+ * lease paid for.
+ */
+export function collectRowExposurePorts(row: PersistedExposureRowSnapshot): Set<number> {
+  const ports = new Set<number>();
+  for (const listener of row.exposure?.listeners ?? []) {
+    if (Number.isInteger(listener.targetPort)) ports.add(listener.targetPort);
+  }
+  if (row.port !== null && isRuntimeExposureAppPort(row.port)) {
+    ports.add(row.port);
+    ports.add(deriveViteHmrPort(row.port));
+  }
+  return ports;
+}
+
+/** True when the row's own exposure state still claims the mapping. */
+function rowClaimsLiveExposure(row: PersistedExposureRowSnapshot): boolean {
+  return Boolean(row.exposure) && row.exposure!.state !== "removed";
+}
+
+function rowIdentity(row: PersistedExposureRowSnapshot): ExposureOwnerIdentity {
+  return {
+    runtimeServiceId: row.id,
+    executionWorkspaceId: row.executionWorkspaceId,
+    projectWorkspaceId: row.projectWorkspaceId,
+    issueId: row.issueId,
+  };
+}
+
+/**
+ * Merge every reservation view into one ledger. Later, weaker evidence never
+ * overwrites a stronger claim already recorded for the same port.
+ */
+export function buildExposureReservationLedger(
+  input: BuildExposureReservationLedgerInput,
+): ExposureReservationLedger {
+  const reservationByPort = new Map<number, ExposurePortReservation>();
+
+  const record = (port: number, source: ExposureReservationSource, owner: ExposureOwnerIdentity) => {
+    if (!Number.isInteger(port)) return;
+    const existing = reservationByPort.get(port);
+    if (existing && RESERVATION_SOURCE_RANK[existing.source] <= RESERVATION_SOURCE_RANK[source]) return;
+    reservationByPort.set(port, { port, source, owner });
+  };
+
+  const activeLeases = new Set(input.activeExecutionWorkspaceIds ?? []);
+
+  for (const port of input.quarantinedPorts ?? []) record(port, "quarantine", emptyIdentity());
+  // An in-flight claim has no identity yet by construction — the claiming start
+  // has not persisted a row. It still must block a concurrent allocator.
+  for (const port of input.inFlightClaimedPorts ?? []) record(port, "in_memory_runtime", emptyIdentity());
+
+  for (const runtime of input.inMemoryRuntimes ?? []) {
+    const owner: ExposureOwnerIdentity = {
+      runtimeServiceId: runtime.runtimeServiceId,
+      executionWorkspaceId: runtime.executionWorkspaceId,
+      projectWorkspaceId: runtime.projectWorkspaceId,
+      issueId: runtime.issueId,
+    };
+    for (const port of runtime.ports) record(port, "in_memory_runtime", owner);
+  }
+
+  // Broker mappings identify a runtime, not a workspace; resolve the workspace
+  // from the persisted rows when we can so a conflict can name it.
+  const rows = [...(input.persistedRows ?? [])];
+  const identityByRuntimeId = new Map(rows.map((row) => [row.id, rowIdentity(row)]));
+  for (const mapping of input.brokerMappings ?? []) {
+    record(
+      mapping.port,
+      "broker_mapping",
+      identityByRuntimeId.get(mapping.runtimeId)
+        ?? { ...emptyIdentity(), runtimeServiceId: mapping.runtimeId },
+    );
+  }
+
+  for (const row of rows) {
+    const leaseOpen = Boolean(row.executionWorkspaceId) && activeLeases.has(row.executionWorkspaceId!);
+    const claimsExposure = rowClaimsLiveExposure(row);
+    if (!leaseOpen && !claimsExposure) continue;
+    const source: ExposureReservationSource = claimsExposure ? "persisted_exposure" : "leased_workspace";
+    for (const port of collectRowExposurePorts(row)) record(port, source, rowIdentity(row));
+  }
+
+  return { reservationByPort, reservedPorts: new Set(reservationByPort.keys()) };
+}
+
+/**
+ * May `claimant` take over a port currently attributed to `holder`?
+ *
+ * The rule is deliberately narrow, because the failure this prevents is a
+ * managed start silently adopting another issue's live service and then
+ * generating security evidence attributed to the wrong workspace:
+ *
+ *  - the same runtime service reclaiming its own port is fine (restart);
+ *  - otherwise both sides must name the *same, non-null* execution workspace.
+ *
+ * An unknown identity on either side is never adopted. "We could not tell whose
+ * it is" is a reason to fail closed, not a reason to assume it is ours.
+ */
+export function isExposureAdoptionPermitted(
+  holder: ExposureOwnerIdentity | null,
+  claimant: ExposureOwnerIdentity,
+): boolean {
+  if (!holder) return true;
+  if (
+    holder.runtimeServiceId !== null
+    && claimant.runtimeServiceId !== null
+    && holder.runtimeServiceId === claimant.runtimeServiceId
+  ) {
+    return true;
+  }
+  if (holder.executionWorkspaceId === null || claimant.executionWorkspaceId === null) return false;
+  return holder.executionWorkspaceId === claimant.executionWorkspaceId;
+}
+
+export type ExposurePortConflictCode =
+  | "quarantined_port"
+  | "reserved_by_other_workspace"
+  | "listener_owned_by_other_workspace"
+  | "serve_mapping_owned_by_other_workspace";
+
+export interface ExposurePortConflict {
+  code: ExposurePortConflictCode;
+  port: number;
+  claimant: ExposureOwnerIdentity;
+  holder: ExposureOwnerIdentity | null;
+  source: ExposureReservationSource | null;
+}
+
+export interface FindExposurePairConflictInput {
+  pair: { appPort: number; hmrPort: number };
+  claimant: ExposureOwnerIdentity;
+  ledger: ExposureReservationLedger;
+  /**
+   * Identity behind the process actually listening on each port, when one could
+   * be resolved. `null` for a port with a live listener whose owner is unknown —
+   * which is itself a conflict, since an unattributable listener must never be
+   * adopted.
+   */
+  listenerOwners?: ReadonlyMap<number, ExposureOwnerIdentity | null>;
+  /** Identity behind each live Serve mapping, keyed by port. */
+  serveMappingOwners?: ReadonlyMap<number, ExposureOwnerIdentity | null>;
+}
+
+/**
+ * Complete mediation for one candidate pair: the first conflict across the
+ * persisted ledger, the live listeners, and the Serve mappings — or null when
+ * all three agree the claimant may have it.
+ */
+export function findExposurePairConflict(
+  input: FindExposurePairConflictInput,
+): ExposurePortConflict | null {
+  const ports = [input.pair.appPort, input.pair.hmrPort];
+
+  for (const port of ports) {
+    const reservation = input.ledger.reservationByPort.get(port);
+    if (!reservation) continue;
+    if (reservation.source === "quarantine") {
+      return { code: "quarantined_port", port, claimant: input.claimant, holder: null, source: "quarantine" };
+    }
+    if (!isExposureAdoptionPermitted(reservation.owner, input.claimant)) {
+      return {
+        code: "reserved_by_other_workspace",
+        port,
+        claimant: input.claimant,
+        holder: reservation.owner,
+        source: reservation.source,
+      };
+    }
+  }
+
+  // Presence in these maps means the host has something on the port. A `null`
+  // value is therefore NOT "no holder" — it is "a holder we could not name",
+  // which is the one case that must never be adopted. `isExposureAdoptionPermitted`
+  // is only consulted once we have an identity to compare.
+  for (const port of ports) {
+    if (!input.listenerOwners?.has(port)) continue;
+    const owner = input.listenerOwners.get(port) ?? null;
+    if (owner === null || !isExposureAdoptionPermitted(owner, input.claimant)) {
+      return {
+        code: "listener_owned_by_other_workspace",
+        port,
+        claimant: input.claimant,
+        holder: owner,
+        source: null,
+      };
+    }
+  }
+
+  for (const port of ports) {
+    if (!input.serveMappingOwners?.has(port)) continue;
+    const owner = input.serveMappingOwners.get(port) ?? null;
+    if (owner === null || !isExposureAdoptionPermitted(owner, input.claimant)) {
+      return {
+        code: "serve_mapping_owned_by_other_workspace",
+        port,
+        claimant: input.claimant,
+        holder: owner,
+        source: null,
+      };
+    }
+  }
+
+  return null;
+}
+
+function describeIdentity(identity: ExposureOwnerIdentity | null): string {
+  if (!identity) return "an unidentified owner";
+  const parts: string[] = [];
+  if (identity.executionWorkspaceId) parts.push(`execution workspace ${identity.executionWorkspaceId}`);
+  if (identity.issueId) parts.push(`issue ${identity.issueId}`);
+  if (identity.runtimeServiceId) parts.push(`runtime service ${identity.runtimeServiceId}`);
+  if (identity.projectWorkspaceId && parts.length === 0) {
+    parts.push(`project workspace ${identity.projectWorkspaceId}`);
+  }
+  return parts.length > 0 ? parts.join(", ") : "an unidentified owner";
+}
+
+const CONFLICT_REASONS: Record<ExposurePortConflictCode, string> = {
+  quarantined_port: "is quarantined after an ambiguous exposure cleanup and can never be reused",
+  reserved_by_other_workspace: "is reserved by",
+  listener_owned_by_other_workspace: "has a live loopback listener owned by",
+  serve_mapping_owned_by_other_workspace: "is published by a Tailscale Serve mapping owned by",
+};
+
+/**
+ * Operator-facing conflict text. Always names the port and the conflicting
+ * workspace/issue identity: a bare "port unavailable" is what made the original
+ * incident take a manual `tailscale serve status` read to attribute.
+ */
+export function describeExposurePortConflict(conflict: ExposurePortConflict): string {
+  const subject = `port ${conflict.port}`;
+  if (conflict.code === "quarantined_port") {
+    return `${subject} ${CONFLICT_REASONS.quarantined_port}`;
+  }
+  const suffix = conflict.source === "leased_workspace"
+    ? " (the lane is stopped, but its lease is still open)"
+    : "";
+  return `${subject} ${CONFLICT_REASONS[conflict.code]} ${describeIdentity(conflict.holder)}${suffix}`;
+}
+
+/** Thrown on the allocation/start path so the failure is terminal and attributed. */
+export class ExposurePortOwnershipConflictError extends Error {
+  readonly code = "exposure_port_ownership_conflict" as const;
+
+  constructor(readonly conflict: ExposurePortConflict) {
+    super(`HTTPS exposure allocation denied: ${describeExposurePortConflict(conflict)}`);
+    this.name = "ExposurePortOwnershipConflictError";
+  }
+}
+
+export interface ExposureReservationDrift {
+  /** The row Paperclip believes is stopped/removed. */
+  runtimeServiceId: string;
+  owner: ExposureOwnerIdentity;
+  port: number;
+  reason: "live_listener" | "serve_mapping";
+  /** Who actually holds it, when that could be attributed. */
+  conflictingOwner: ExposureOwnerIdentity | null;
+}
+
+export interface FindExposureReservationDriftInput {
+  persistedRows: Iterable<PersistedExposureRowSnapshot>;
+  /** Ports with a live loopback listener right now. */
+  livePorts?: ReadonlySet<number>;
+  listenerOwners?: ReadonlyMap<number, ExposureOwnerIdentity | null>;
+  brokerMappings?: Iterable<BrokerMappingSnapshot>;
+}
+
+/**
+ * Rows whose state says "nothing here" while the host says otherwise.
+ *
+ * Reconciliation calls this so a lane reported `stopped` with its exposure
+ * `removed` — but whose reserved ports are live, or mapped to some other
+ * runtime — becomes a visible, attributable finding instead of a pair that
+ * quietly returns to the free list. A port still held by the row's *own*
+ * runtime is not drift; that is just a lane the sweep has yet to adopt.
+ */
+export function findExposureReservationDrift(
+  input: FindExposureReservationDriftInput,
+): ExposureReservationDrift[] {
+  const rows = [...input.persistedRows];
+  const identityByRuntimeId = new Map(rows.map((row) => [row.id, rowIdentity(row)]));
+  const mappingOwnerByPort = new Map<number, ExposureOwnerIdentity>();
+  for (const mapping of input.brokerMappings ?? []) {
+    mappingOwnerByPort.set(
+      mapping.port,
+      identityByRuntimeId.get(mapping.runtimeId)
+        ?? { ...emptyIdentity(), runtimeServiceId: mapping.runtimeId },
+    );
+  }
+
+  const drift: ExposureReservationDrift[] = [];
+  for (const row of rows) {
+    const dormant = row.status === "stopped" || row.status === "failed" || !rowClaimsLiveExposure(row);
+    if (!dormant) continue;
+    const owner = rowIdentity(row);
+    for (const port of collectRowExposurePorts(row)) {
+      const mappingOwner = mappingOwnerByPort.get(port) ?? null;
+      if (mappingOwner && mappingOwner.runtimeServiceId !== row.id) {
+        drift.push({ runtimeServiceId: row.id, owner, port, reason: "serve_mapping", conflictingOwner: mappingOwner });
+        continue;
+      }
+      if (!input.livePorts?.has(port)) continue;
+      const listenerOwner = input.listenerOwners?.get(port) ?? null;
+      if (listenerOwner && listenerOwner.runtimeServiceId === row.id) continue;
+      drift.push({ runtimeServiceId: row.id, owner, port, reason: "live_listener", conflictingOwner: listenerOwner });
+    }
+  }
+  return drift;
+}
+
+/** Human-readable drift line for reconciliation logs and operator evidence. */
+export function describeExposureReservationDrift(entry: ExposureReservationDrift): string {
+  const held = entry.reason === "serve_mapping"
+    ? "is still published by a Tailscale Serve mapping"
+    : "still has a live loopback listener";
+  return (
+    `runtime service ${entry.runtimeServiceId} (${describeIdentity(entry.owner)}) is recorded stopped/removed, `
+    + `but its reserved port ${entry.port} ${held} owned by ${describeIdentity(entry.conflictingOwner)}`
+  );
+}
+
+/**
+ * In-process, pair-atomic claims for starts that have not bound a listener yet.
+ *
+ * Two concurrent allocators see identical persisted state and identical probe
+ * results, so without a claim they both walk away with the lowest free pair.
+ * Claiming is all-or-nothing across the two ports: a half-claimed pair would
+ * let one start own the app port and another the HMR port, which is the
+ * orphaned-companion failure the pairing exists to prevent.
+ */
+export class ExposurePortPairClaims {
+  private readonly heldUntilByPort = new Map<number, number>();
+
+  constructor(private readonly ttlMs: number = 120_000) {}
+
+  private isHeld(port: number, nowMs: number): boolean {
+    const heldUntil = this.heldUntilByPort.get(port);
+    if (heldUntil === undefined) return false;
+    if (heldUntil > nowMs) return true;
+    this.heldUntilByPort.delete(port);
+    return false;
+  }
+
+  /** Claim both ports, or neither. Returns false when either is already held. */
+  claim(pair: { appPort: number; hmrPort: number }, nowMs: number = Date.now()): boolean {
+    if (this.isHeld(pair.appPort, nowMs) || this.isHeld(pair.hmrPort, nowMs)) return false;
+    this.heldUntilByPort.set(pair.appPort, nowMs + this.ttlMs);
+    this.heldUntilByPort.set(pair.hmrPort, nowMs + this.ttlMs);
+    return true;
+  }
+
+  /** Release on lease release/teardown, or when a start reaches a terminal state. */
+  release(pair: { appPort: number; hmrPort: number }): void {
+    this.heldUntilByPort.delete(pair.appPort);
+    this.heldUntilByPort.delete(pair.hmrPort);
+  }
+
+  /** Ports still claimed, for folding into an allocator's reserved set. */
+  activePorts(nowMs: number = Date.now()): Set<number> {
+    const active = new Set<number>();
+    for (const port of [...this.heldUntilByPort.keys()]) {
+      if (this.isHeld(port, nowMs)) active.add(port);
+    }
+    return active;
+  }
+
+  clear(): void {
+    this.heldUntilByPort.clear();
+  }
+}
diff --git a/server/src/services/workspace-runtime-exposure.test.ts b/server/src/services/workspace-runtime-exposure.test.ts
index a0c2431c67..10c4f12a26 100644
--- a/server/src/services/workspace-runtime-exposure.test.ts
+++ b/server/src/services/workspace-runtime-exposure.test.ts
@@ -1,4 +1,5 @@
 import fs from "node:fs/promises";
+import net from "node:net";
 import os from "node:os";
 import path from "node:path";
 
@@ -141,18 +142,42 @@ function createBroker() {
   return { broker, calls };
 }
 
+/**
+ * A real loopback bind probe, matching production's `isLoopbackPortAvailable`.
+ *
+ * The default used to be `async () => true`, which claimed every port in the
+ * dedicated range was free. On a developer or canary host that already runs a
+ * managed lane on `42000`, that lie made the suite allocate a port something
+ * else was listening on — so the spawned guest could not bind, and every
+ * assertion downstream failed for a reason that had nothing to do with the
+ * behaviour under test. It also meant the suite exercised allocation against a
+ * range it never actually checked (PAP-17419; PAP-17255 started this with
+ * "make exposure fixture honor occupied ports").
+ */
+async function isLoopbackPortFree(port: number): Promise<boolean> {
+  return await new Promise<boolean>((resolve) => {
+    const probe = net.createServer();
+    probe.unref();
+    probe.once("error", () => resolve(false));
+    probe.listen(port, "127.0.0.1", () => {
+      probe.close(() => resolve(true));
+    });
+  });
+}
+
 function installDeps(overrides: {
   broker: BrokerClient;
   probeHealth?: () => Promise<boolean>;
   isBrokerAvailable?: () => Promise<boolean>;
   isPortAvailable?: (port: number) => Promise<boolean>;
   diagnoseListenerBinds?: (ports: number[]) => Promise<string | null>;
+  resolveHostname?: () => Promise<string>;
 }) {
   setWorkspaceRuntimeExposureDepsForTests({
     broker: overrides.broker,
-    isPortAvailable: overrides.isPortAvailable ?? (async () => true),
+    isPortAvailable: overrides.isPortAvailable ?? isLoopbackPortFree,
     isBrokerAvailable: overrides.isBrokerAvailable ?? (async () => true),
-    resolveHostname: async () => "runner.tail123.ts.net",
+    resolveHostname: overrides.resolveHostname ?? (async () => "runner.tail123.ts.net"),
     probeHealth: overrides.probeHealth ?? (async () => true),
     now: () => "2026-08-11T00:00:00.000Z",
     // The real /proc-backed diagnosis, not a fake: the fake broker below always
@@ -372,6 +397,48 @@ describe("automatic tailscale_https default for managed worktree runtimes", () =
     expect(second.port).toBe(firstPort);
     expect(second.url).toBe(`https://runner.tail123.ts.net:${firstPort}`);
   }, 25_000);
+
+  it("releases the in-flight pair claim when hostname resolution fails, so a retry storm cannot exhaust the range", async () => {
+    // The reservation keeps the winning pair's in-process claim on purpose, for
+    // the caller to release on stop/teardown. A hostname failure throws before any
+    // runtime record exists, so that teardown can never run for this pair — and
+    // removing only the broker reservation would leave the claim held. Each retry
+    // would then burn another pair and the lane would report the range exhausted
+    // rather than the real cause, a Tailscale outage.
+    const reservedAppPorts: number[] = [];
+    const broker: BrokerClient = {
+      async reserve(_runtimeId, requested) {
+        reservedAppPorts.push(requested[0]!.port);
+        return { handle: HANDLE, reservedPorts: requested.map((listener) => listener.port) };
+      },
+      async expose() {
+        return { handle: HANDLE, publicPorts: [] };
+      },
+      async remove() {
+        return { removedPorts: [] };
+      },
+      async list() {
+        return [];
+      },
+    };
+    installDeps({
+      broker,
+      resolveHostname: async () => {
+        throw new Error("MagicDNS unavailable");
+      },
+    });
+
+    for (let attempt = 0; attempt < 3; attempt += 1) {
+      await expect(
+        startRuntimeServicesForWorkspaceControl(startInput({ serviceName: "paperclip-dev" })),
+      ).rejects.toThrow(/MagicDNS hostname unavailable/);
+    }
+
+    // Every attempt reserved the same pair. A leaked claim would push each retry
+    // onto a fresh pair instead.
+    expect(reservedAppPorts).toHaveLength(3);
+    expect(new Set(reservedAppPorts).size).toBe(1);
+  }, 25_000);
 });
 
 /**
diff --git a/server/src/services/workspace-runtime.ts b/server/src/services/workspace-runtime.ts
index fcc4310b86..1007c05185 100644
--- a/server/src/services/workspace-runtime.ts
+++ b/server/src/services/workspace-runtime.ts
@@ -13,6 +13,7 @@ import {
   DEFAULT_TAILSCALE_HTTPS_EXPOSURE,
   deriveViteHmrPort,
   forceLoopbackBindInCommand,
+  isRuntimeExposureAppPort,
   listWorkspaceServiceCommandDefinitions,
   RUNTIME_EXPOSURE_BIND_HOST,
   RUNTIME_EXPOSURE_BIND_MODE,
@@ -68,6 +69,21 @@ import {
 } from "./runtime-exposure/exposure-manager.js";
 import { diagnoseRuntimeListenerBinds } from "./runtime-exposure/loopback-listener.js";
 import { allocateExposurePortPair } from "./runtime-exposure/port-pair.js";
+import {
+  buildExposureReservationLedger,
+  collectRowExposurePorts,
+  describeExposureReservationDrift,
+  ExposurePortOwnershipConflictError,
+  ExposurePortPairClaims,
+  findExposurePairConflict,
+  findExposureReservationDrift,
+  isExposureAdoptionPermitted,
+  type BrokerMappingSnapshot,
+  type ExposureOwnerIdentity,
+  type ExposureReservationLedger,
+  type InMemoryExposureSnapshot,
+  type PersistedExposureRowSnapshot,
+} from "./runtime-exposure/port-reservation.js";
 import { resolveTailscaleDnsName } from "./runtime-exposure/tailscale-hostname.js";
 
 export function resolveShell(): string {
@@ -216,6 +232,19 @@ const runtimeServicesByReuseKey = new Map<string, string>();
 const runtimeServiceLeasesByRun = new Map<string, string[]>();
 const runtimeProvisionByWorkspace = new Map<string, Promise<void>>();
 const quarantinedRuntimeExposurePorts = new Set<number>();
+/**
+ * Pair-atomic in-process claims for exposure allocations that have not bound a
+ * listener yet. Separate from `inFlightAllocatedPorts` (single ports, non-exposed
+ * runtimes) because an exposure claim must cover the app port and its HMR
+ * companion together or not at all.
+ */
+const exposurePortPairClaims = new ExposurePortPairClaims();
+/**
+ * Execution-workspace statuses that still hold an exclusive lease. `archived`
+ * is the only terminal state; everything else — including `idle` — is a lane an
+ * operator or agent can still return to, so its port pair stays reserved.
+ */
+const OPEN_EXECUTION_WORKSPACE_LEASE_STATUSES = ["active", "idle", "in_review"] as const;
 const DEFAULT_EXECUTE_PROCESS_OUTPUT_BYTES = 256 * 1024;
 export const WORKSPACE_RUNTIME_PORT_ALLOCATION_ATTEMPTS = 32;
 const ACTIVE_RUNTIME_PORT_RESERVATION_STATUSES = ["provisioning", "starting", "running"] as const;
@@ -448,6 +477,7 @@ export async function resetRuntimeServicesForTests(
   runtimeServiceLeasesByRun.clear();
   runtimeProvisionByWorkspace.clear();
   quarantinedRuntimeExposurePorts.clear();
+  exposurePortPairClaims.clear();
   workspaceRuntimeExposureDeps = defaultWorkspaceRuntimeExposureDeps();
 }
 
@@ -3848,22 +3878,188 @@ async function probeEphemeralPort(): Promise<number> {
   });
 }
 
-async function collectReservedExposurePorts(db: Db | undefined, companyId: string): Promise<Set<number>> {
-  const reserved = new Set(quarantinedRuntimeExposurePorts);
-  for (const record of runtimeServicesById.values()) {
-    if (record.companyId !== companyId || !record.exposure) continue;
-    for (const listener of record.exposure.listeners) reserved.add(listener.targetPort);
-  }
-  if (!db) return reserved;
+/**
+ * Execution workspaces whose exclusive lease is still open.
+ *
+ * "Open" is deliberately generous — every non-archived, non-closed status
+ * counts — because the reservation must outlive the *process*, not track it.
+ * A lane that is stopped, torn down, and reported `removed` still owns its
+ * pair until the workspace itself is released (PAP-17419).
+ */
+async function readActiveExecutionWorkspaceLeases(db: Db | undefined, companyId: string): Promise<Set<string>> {
+  if (!db) return new Set<string>();
   const rows = await db
-    .select({ exposure: workspaceRuntimeServices.exposure })
-    .from(workspaceRuntimeServices)
-    .where(eq(workspaceRuntimeServices.companyId, companyId));
-  for (const row of rows) {
-    if (!row.exposure || row.exposure.state === "removed") continue;
-    for (const listener of row.exposure.listeners) reserved.add(listener.targetPort);
+    .select({ id: executionWorkspaces.id })
+    .from(executionWorkspaces)
+    .where(
+      and(
+        eq(executionWorkspaces.companyId, companyId),
+        inArray(executionWorkspaces.status, [...OPEN_EXECUTION_WORKSPACE_LEASE_STATUSES]),
+        isNull(executionWorkspaces.closedAt),
+      ),
+    );
+  return new Set(rows.map((row) => row.id));
+}
+
+/**
+ * Every reservation view the allocator must respect, merged into one ledger.
+ *
+ * This replaced a set-of-ports that only ever saw rows whose `exposure.state`
+ * was not `removed`. That view could not represent the case that actually
+ * broke: a leased workspace whose exposure had been torn down. See
+ * `port-reservation.ts` for why the pair is re-derived from the `port` column.
+ */
+async function buildCompanyExposureReservationLedger(input: {
+  db?: Db;
+  companyId: string;
+  brokerMappings?: BrokerMappingSnapshot[];
+}): Promise<ExposureReservationLedger> {
+  const inMemoryRuntimes: InMemoryExposureSnapshot[] = [];
+  for (const record of runtimeServicesById.values()) {
+    if (record.companyId !== input.companyId || !record.exposure) continue;
+    inMemoryRuntimes.push({
+      runtimeServiceId: record.id,
+      executionWorkspaceId: record.executionWorkspaceId,
+      projectWorkspaceId: record.projectWorkspaceId,
+      issueId: record.issueId,
+      ports: record.exposure.listeners.map((listener) => listener.targetPort),
+    });
   }
-  return reserved;
+
+  const persistedRows: PersistedExposureRowSnapshot[] = input.db
+    ? (
+      await input.db
+        .select({
+          id: workspaceRuntimeServices.id,
+          status: workspaceRuntimeServices.status,
+          port: workspaceRuntimeServices.port,
+          exposure: workspaceRuntimeServices.exposure,
+          executionWorkspaceId: workspaceRuntimeServices.executionWorkspaceId,
+          projectWorkspaceId: workspaceRuntimeServices.projectWorkspaceId,
+          issueId: workspaceRuntimeServices.issueId,
+        })
+        .from(workspaceRuntimeServices)
+        .where(eq(workspaceRuntimeServices.companyId, input.companyId))
+    ).map((row) => ({
+      id: row.id,
+      status: row.status,
+      port: row.port,
+      exposure: row.exposure,
+      executionWorkspaceId: row.executionWorkspaceId,
+      projectWorkspaceId: row.projectWorkspaceId,
+      issueId: row.issueId,
+    }))
+    : [];
+
+  return buildExposureReservationLedger({
+    persistedRows,
+    inMemoryRuntimes,
+    brokerMappings: input.brokerMappings ?? [],
+    quarantinedPorts: quarantinedRuntimeExposurePorts,
+    activeExecutionWorkspaceIds: await readActiveExecutionWorkspaceLeases(input.db, input.companyId),
+    inFlightClaimedPorts: exposurePortPairClaims.activePorts(),
+  });
+}
+
+/**
+ * Rows Paperclip reports stopped/removed whose reserved pair is still live on
+ * the host or still mapped to someone else (PAP-17419 regression #3).
+ *
+ * The point is visibility. A false `stopped`/`removed` row used to be
+ * indistinguishable from a genuinely released one, so the pair silently
+ * returned to the free list and the next managed start collided with — or
+ * adopted — an unrelated workspace's service. Surfacing it does not stop or
+ * mutate the occupying service; that stays the owning issue's call.
+ */
+async function detectPersistedExposureReservationDrift(input: {
+  rows: ReadonlyArray<{
+    id: string;
+    status: string;
+    port: number | null;
+    exposure: RuntimeExposureStatus | null;
+    executionWorkspaceId: string | null;
+    projectWorkspaceId: string | null;
+    issueId: string | null;
+  }>;
+  ownedListeners: Awaited<ReturnType<BrokerClient["list"]>> | null;
+}) {
+  const snapshots: PersistedExposureRowSnapshot[] = input.rows.map((row) => ({
+    id: row.id,
+    status: row.status,
+    port: row.port,
+    exposure: row.exposure,
+    executionWorkspaceId: row.executionWorkspaceId,
+    projectWorkspaceId: row.projectWorkspaceId,
+    issueId: row.issueId,
+  }));
+
+  // Probe only the ports dormant rows actually reserve; a startup sweep must not
+  // walk the whole dedicated range.
+  const candidatePorts = new Set<number>();
+  for (const row of snapshots) {
+    if (row.status !== "stopped" && row.status !== "failed" && row.exposure && row.exposure.state !== "removed") {
+      continue;
+    }
+    for (const port of collectRowExposurePorts(row)) candidatePorts.add(port);
+  }
+
+  const livePorts = new Set<number>();
+  for (const port of candidatePorts) {
+    // "Not bindable" is the liveness signal the rest of this module already uses.
+    const available = await workspaceRuntimeExposureDeps.isPortAvailable(port).catch(() => true);
+    if (!available) livePorts.add(port);
+  }
+
+  return findExposureReservationDrift({
+    persistedRows: snapshots,
+    livePorts,
+    listenerOwners: await readExposureListenerOwners([...livePorts]),
+    brokerMappings: (input.ownedListeners ?? []).map((listener) => ({
+      runtimeId: listener.runtimeId,
+      port: listener.port,
+    })),
+  });
+}
+
+/** Paperclip-owned Serve mappings, or null when the broker cannot be read. */
+async function readBrokerExposureMappings(): Promise<BrokerMappingSnapshot[] | null> {
+  try {
+    const owned = await workspaceRuntimeExposureDeps.broker.list();
+    return owned.map((listener) => ({ runtimeId: listener.runtimeId, port: listener.port }));
+  } catch {
+    return null;
+  }
+}
+
+/**
+ * Resolve who owns the process listening on each of a pair's ports.
+ *
+ * A port with no listener is absent from the map; a port with a listener we
+ * cannot attribute maps to `null`, which the mediator treats as a conflict.
+ * Attribution goes through this process's own runtime records: a pid we did not
+ * start is by definition not ours to adopt.
+ */
+async function readExposureListenerOwners(ports: number[]): Promise<Map<number, ExposureOwnerIdentity | null>> {
+  const owners = new Map<number, ExposureOwnerIdentity | null>();
+  for (const port of ports) {
+    const ownerPid = await readLocalServicePortOwner(port).catch(() => null);
+    if (!ownerPid) continue;
+    let identity: ExposureOwnerIdentity | null = null;
+    for (const record of runtimeServicesById.values()) {
+      const recordPid = record.child?.pid ?? null;
+      if (recordPid === null) continue;
+      if (recordPid !== ownerPid && record.processGroupId !== ownerPid) continue;
+      identity = {
+        runtimeServiceId: record.id,
+        executionWorkspaceId: record.executionWorkspaceId,
+        projectWorkspaceId: record.projectWorkspaceId,
+        issueId: record.issueId,
+      };
+      break;
+    }
+    owners.set(port, identity);
+  }
+  return owners;
 }
 
 async function allocateAndReserveExposure(input: {
@@ -3871,34 +4067,85 @@ async function allocateAndReserveExposure(input: {
   companyId: string;
   runtimeId: string;
   config: RuntimeExposureConfigInput;
+  /** Identity claiming the pair; governs every ownership decision below. */
+  claimant: ExposureOwnerIdentity;
   /** Port this runtime already used, preserved when it is still safe to use. */
   preferredAppPort?: number | null;
-}): Promise<{ appPort: number; status: RuntimeExposureStatus; handle: string }> {
-  const reserved = await collectReservedExposurePorts(input.db, input.companyId);
+}): Promise<{ appPort: number; hmrPort: number; status: RuntimeExposureStatus; handle: string }> {
+  const brokerMappings = await readBrokerExposureMappings();
+  const ledger = await buildCompanyExposureReservationLedger({
+    db: input.db,
+    companyId: input.companyId,
+    brokerMappings: brokerMappings ?? [],
+  });
+  // Serve mappings are checked as their own view, not folded into the ledger:
+  // an unreadable broker must not silently downgrade to "no mapping exists".
+  const serveMappingOwners = new Map<number, ExposureOwnerIdentity | null>();
+  for (const mapping of brokerMappings ?? []) {
+    serveMappingOwners.set(mapping.port, ledger.reservationByPort.get(mapping.port)?.owner ?? null);
+  }
+
+  // Reserve only what this claimant may NOT have. A leaseholder restarting its
+  // own lane has to be offered its own pair back, or every restart would walk
+  // the range and undo "keep existing runtime ports when safe" (PAP-17158).
+  // Quarantined ports are withheld from everyone, including the owner.
+  const reserved = new Set<number>();
+  for (const [port, reservation] of ledger.reservationByPort) {
+    if (reservation.source === "quarantine" || !isExposureAdoptionPermitted(reservation.owner, input.claimant)) {
+      reserved.add(port);
+    }
+  }
   const retryable = new Set(["reservation_conflict", "manual_mapping_present", "quarantined"]);
+  const claimed: Array<{ appPort: number; hmrPort: number }> = [];
   let preferredAppPort = input.preferredAppPort ?? null;
-  while (true) {
-    const pair = await allocateExposurePortPair({
-      isPortAvailable: workspaceRuntimeExposureDeps.isPortAvailable,
-      reserved,
-      preferredAppPort,
-    });
-    const result = await reserveExposure(workspaceRuntimeExposureDeps, {
-      runtimeId: input.runtimeId,
-      config: input.config,
-      appPort: pair.appPort,
-    });
-    if (result.handle) {
-      return { appPort: pair.appPort, status: result.status, handle: result.handle };
+  try {
+    while (true) {
+      const pair = await allocateExposurePortPair({
+        isPortAvailable: workspaceRuntimeExposureDeps.isPortAvailable,
+        reserved,
+        preferredAppPort,
+        claimPair: (candidate) => exposurePortPairClaims.claim(candidate),
+      });
+      claimed.push(pair);
+
+      // Complete mediation before the broker is asked for anything: persisted
+      // reservations were already folded into `reserved`, so what remains is the
+      // live host — the listener actually bound, and the Serve mapping actually
+      // published. Either one belonging to a different execution workspace is
+      // terminal, never an adoption.
+      const conflict = findExposurePairConflict({
+        pair,
+        claimant: input.claimant,
+        ledger,
+        listenerOwners: await readExposureListenerOwners([pair.appPort, pair.hmrPort]),
+        serveMappingOwners,
+      });
+      if (conflict) throw new ExposurePortOwnershipConflictError(conflict);
+
+      const result = await reserveExposure(workspaceRuntimeExposureDeps, {
+        runtimeId: input.runtimeId,
+        config: input.config,
+        appPort: pair.appPort,
+      });
+      if (result.handle) {
+        // Keep this pair's claim; the caller releases it on stop/teardown.
+        claimed.pop();
+        return { appPort: pair.appPort, hmrPort: pair.hmrPort, status: result.status, handle: result.handle };
+      }
+      if (!result.status.lastError || !retryable.has(result.status.lastError)) {
+        throw new Error(`HTTPS exposure reservation failed: ${result.status.lastError ?? "unknown broker error"}`);
+      }
+      reserved.add(pair.appPort);
+      reserved.add(pair.hmrPort);
+      // The preference lost its race with a conflicting/manual/quarantined
+      // mapping; drop it so the retry scans instead of re-offering the same port.
+      preferredAppPort = null;
     }
-    if (!result.status.lastError || !retryable.has(result.status.lastError)) {
-      throw new Error(`HTTPS exposure reservation failed: ${result.status.lastError ?? "unknown broker error"}`);
-    }
-    reserved.add(pair.appPort);
-    reserved.add(pair.hmrPort);
-    // The preference lost its race with a conflicting/manual/quarantined
-    // mapping; drop it so the retry scans instead of re-offering the same port.
-    preferredAppPort = null;
+  } finally {
+    // Every pair this call took but did not hand back — rejected candidates and
+    // the in-flight pair on a thrown failure — goes back immediately. Leaving
+    // them held would burn the range down over a retry storm.
+    for (const pair of claimed) exposurePortPairClaims.release(pair);
   }
 }
 
@@ -4964,6 +5211,12 @@ async function spawnLocalRuntimeService(input: StartLocalRuntimeServiceInput): P
         companyId: input.agent.companyId,
         runtimeId,
         config: exposureConfig,
+        claimant: {
+          runtimeServiceId: runtimeId,
+          executionWorkspaceId: input.executionWorkspaceId ?? null,
+          projectWorkspaceId: input.workspace.workspaceId,
+          issueId: input.issue?.id ?? null,
+        },
         preferredAppPort: stoppedReuseCandidate?.port ?? (explicitPort > 0 ? explicitPort : null),
       })
     : null;
@@ -5032,6 +5285,16 @@ async function spawnLocalRuntimeService(input: StartLocalRuntimeServiceInput): P
       await workspaceRuntimeExposureDeps.broker
         .remove(runtimeId, reservedExposure.handle)
         .catch(() => undefined);
+      // The reservation deliberately keeps the winning pair's in-process claim for
+      // the caller to release on stop/teardown. No runtime record exists yet, so
+      // that teardown path can never run for this pair — releasing the broker
+      // reservation alone would leave the claim held. A hostname outage would then
+      // burn one pair per attempt, and a retry storm inside the claim TTL would
+      // report the range exhausted rather than the real cause.
+      exposurePortPairClaims.release({
+        appPort: reservedExposure.appPort,
+        hmrPort: reservedExposure.hmrPort,
+      });
       throw new Error("HTTPS exposure failed: Tailscale MagicDNS hostname unavailable");
     }
   }
@@ -5578,6 +5841,13 @@ async function cleanupRecordExposure(
     ports,
   });
   for (const port of result.quarantinedPorts) quarantinedRuntimeExposurePorts.add(port);
+  // Drop the in-process pair claim on teardown. The *lease* reservation is what
+  // still protects the pair from another workspace (PAP-17419) — this only
+  // releases the short-lived hold that keeps concurrent allocators apart, and
+  // keeping it would block this very lane's own restart.
+  if (record.port !== null && isRuntimeExposureAppPort(record.port)) {
+    exposurePortPairClaims.release({ appPort: record.port, hmrPort: deriveViteHmrPort(record.port) });
+  }
   if (result.status.state === "removed") {
     record.exposureHandle = null;
     record.exposure = options?.preserveFailure && previous.state === "failed"
@@ -6719,7 +6989,15 @@ export async function reconcilePersistedRuntimeServicesOnStartup(db: Db) {
   const readDeclaredExposureIntent = await buildPersistedRuntimeExposureIntentLookup(db);
 
   let ownedExposureListeners: Awaited<ReturnType<BrokerClient["list"]>> | null = [];
-  if (rows.some((row) => row.exposure && row.exposure.state !== "removed")) {
+  // Also fetch when a row merely *reserves* a dedicated-range port. The row that
+  // matters most to PAP-17419 is exactly the one with `exposure.state ===
+  // "removed"`: it claims nothing, yet its leased pair can still be mapped by
+  // someone else. Skipping the broker read for those rows is what let a false
+  // `removed` go unnoticed.
+  if (rows.some((row) => (
+    (row.exposure && row.exposure.state !== "removed")
+    || (row.port !== null && isRuntimeExposureAppPort(row.port))
+  ))) {
     try {
       ownedExposureListeners = await workspaceRuntimeExposureDeps.broker.list();
     } catch {
@@ -6727,11 +7005,54 @@ export async function reconcilePersistedRuntimeServicesOnStartup(db: Db) {
     }
   }
 
+  const exposureReservationDrift = await detectPersistedExposureReservationDrift({
+    rows,
+    ownedListeners: ownedExposureListeners,
+  });
+  const companyIdByRowId = new Map(rows.map((row) => [row.id, row.companyId] as const));
+  for (const entry of exposureReservationDrift) {
+    const description = describeExposureReservationDrift(entry);
+    console.warn(`[workspace-runtime] exposure reservation drift: ${description}`);
+    const companyId = companyIdByRowId.get(entry.runtimeServiceId);
+    if (!companyId) continue;
+    await logActivity(db, {
+      companyId,
+      actorType: "system",
+      actorId: "workspace_runtime",
+      action: "workspace_runtime.exposure_reservation_drift",
+      entityType: entry.owner.executionWorkspaceId ? "execution_workspace" : "workspace_runtime_service",
+      entityId: entry.owner.executionWorkspaceId ?? entry.runtimeServiceId,
+      issueId: entry.owner.issueId,
+      details: {
+        description,
+        runtimeServiceId: entry.runtimeServiceId,
+        port: entry.port,
+        reason: entry.reason,
+        executionWorkspaceId: entry.owner.executionWorkspaceId,
+        conflictingExecutionWorkspaceId: entry.conflictingOwner?.executionWorkspaceId ?? null,
+        conflictingRuntimeServiceId: entry.conflictingOwner?.runtimeServiceId ?? null,
+      },
+    }).catch(() => undefined);
+  }
+
   let reconciled = 0;
   let adopted = 0;
   let stopped = 0;
   let backfilled = 0;
+  const driftedRuntimeServiceIds = new Set(exposureReservationDrift.map((entry) => entry.runtimeServiceId));
   for (const row of rows) {
+    // PAP-17419: this row's reserved pair is live, or Serve-mapped, under an
+    // identity that is not this row's. Every branch below is unsafe for such a
+    // row — cleanup would remove a mapping that is now someone else's, adoption
+    // would take over another execution workspace's service and re-attribute it
+    // here, and the health branch would terminate it outright. None of that is
+    // this sweep's call to make, so leave the row and the occupying service
+    // exactly as they are. The drift is already reported above, and the ledger
+    // keeps the pair reserved so no start can be handed it either.
+    if (driftedRuntimeServiceIds.has(row.id)) {
+      reconciled += 1;
+      continue;
+    }
     if (row.status === "stopped" && row.exposure && row.exposure.state !== "removed") {
       // This branch is a GLOBAL sweep: `rows` spans every execution workspace and
       // company on the host, at any age, and it runs on every server start. It
@@ -6957,6 +7278,8 @@ export async function reconcilePersistedRuntimeServicesOnStartup(db: Db) {
     backfilled,
     restarted: desiredState.restarted,
     restartFailed: desiredState.failed,
+    /** Stopped/removed rows whose reserved ports are live or mapped elsewhere. */
+    exposureReservationDrift,
   };
 }