Add persistent Boat computer environments and runner ownership

Introduce company-scoped computer attachments, bounded runner and viewer ownership, persistent files and worktrees, Boat ingress, and native and legacy runtime integration behind the experimental Boat flag. Keep desktop UI and extended acceptance cases in the stacked feature branch.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/modules/computers/application/ports.ts

@@ -0,0 +1,67 @@
+import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
+import type {
+  ComputerRecord,
+  Ledger,
+  Owner,
+  ProcessClaim,
+} from "../domain/ledger.js";
+export interface ComputerRepository {
+  create(record: ComputerRecord): Promise<void>;
+  get(scope: {
+    companyId: string;
+    environmentId: string;
+  }): Promise<ComputerRecord>;
+  update<T>(
+    scope: { companyId: string; environmentId: string },
+    fn: (record: ComputerRecord) => T,
+  ): Promise<T>;
+  all(): Promise<ComputerRecord[]>;
+  runState(
+    scope: { companyId: string },
+    runId: string,
+    agentId: string,
+  ): Promise<"active" | "terminal" | "missing">;
+}
+export interface ComputerBackend {
+  inspect(record: ComputerRecord): Promise<{
+    state: string;
+    snapshots: boolean;
+    stop: null | { id: string; status: string };
+  }>;
+  ready(record: ComputerRecord): Promise<void>;
+  claim(record: ComputerRecord): Promise<void>;
+  advance(record: ComputerRecord, owner: Owner): Promise<void>;
+  runner(record: ComputerRecord): Promise<CommandManagedRuntimeRunner>;
+  launch(
+    record: ComputerRecord,
+    owner: Owner,
+    input: {
+      command: string;
+      args?: string[];
+      cwd?: string;
+      env?: Record<string, string>;
+    },
+  ): Promise<ProcessClaim>;
+  inspectProcess(
+    record: ComputerRecord,
+    owner: Owner,
+  ): Promise<{ running: boolean; claim: ProcessClaim | null }>;
+  retire(record: ComputerRecord, owner: Owner): Promise<void>;
+  stop(record: ComputerRecord): Promise<{ id: string; status: string }>;
+  stopStatus(
+    record: ComputerRecord,
+    id: string,
+  ): Promise<{ id: string; status: string }>;
+  renew(record: ComputerRecord): Promise<void>;
+  desktop(
+    record: ComputerRecord,
+  ): Promise<{ viewerUrl: string; expiresAt: string }>;
+  ingress(
+    record: ComputerRecord,
+    port: number,
+    path: string,
+  ): Promise<{ url: string; secretHeaders: Record<string, string> }>;
+  preview(record: ComputerRecord, port: number): Promise<{ url: string }>;
+  remote(record: ComputerRecord, input: Record<string, unknown>): Promise<any>;
+}
+export type { Ledger };

server/src/modules/computers/application/service.test.ts

@@ -0,0 +1,565 @@
+import { describe, it, expect, vi } from "vitest";
+import {
+  existsSync,
+  mkdtempSync,
+  readFileSync,
+  realpathSync,
+  rmSync,
+  mkdirSync,
+  writeFileSync,
+} from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { spawnSync } from "node:child_process";
+import { remoteProgram } from "../adapters/remote-program.js";
+import { createComputerService } from "./service.js";
+import type { ComputerBackend, ComputerRepository } from "./ports.js";
+import { ComputerError, type ComputerRecord } from "../domain/ledger.js";
+
+function fixture() {
+  let record: ComputerRecord | undefined;
+  let clock = new Date("2026-10-10T12:00:00Z");
+  let state = "ready";
+  let stopStatus = "pending";
+  const repository: ComputerRepository = {
+    create: async (value) => {
+      record = structuredClone(value);
+    },
+    get: async (scope) => {
+      if (
+        !record ||
+        record.companyId !== scope.companyId ||
+        record.environmentId !== scope.environmentId
+      )
+        throw new ComputerError("not_found", "missing");
+      return structuredClone(record);
+    },
+    update: async (scope, fn) => {
+      const next = await repository.get(scope);
+      const result = fn(next);
+      record = next;
+      return result;
+    },
+    all: async () => (record ? [structuredClone(record)] : []),
+    runState: vi.fn(async () => "active" as const),
+  };
+  const backend: ComputerBackend = {
+    inspect: vi.fn(async () => ({ state, snapshots: true, stop: null })),
+    ready: vi.fn(async () => {
+      state = "ready";
+    }),
+    claim: vi.fn(async () => {}),
+    advance: vi.fn(async () => {}),
+    runner: vi.fn(async () => ({ execute: vi.fn() })),
+    launch: vi.fn(async (_record, owner) => ({
+      ...owner.process!,
+      bootId: "boot",
+    })),
+    inspectProcess: vi.fn(async (_record, owner) => ({
+      running: true,
+      claim: owner.process,
+    })),
+    retire: vi.fn(async () => {}),
+    stop: vi.fn(async () => ({ id: "stop_1", status: stopStatus })),
+    stopStatus: vi.fn(async () => ({ id: "stop_1", status: stopStatus })),
+    renew: vi.fn(async () => {}),
+    desktop: vi.fn(async () => ({
+      viewerUrl: "https://test.on.boat.dev/#credential",
+      expiresAt: new Date(clock.getTime() + 540_000).toISOString(),
+    })),
+    ingress: vi.fn(async () => ({
+      url: "wss://test.on.boat.dev/",
+      secretHeaders: { Cookie: "private" },
+    })),
+    preview: vi.fn(async () => ({
+      url: "https://test.on.boat.dev/?_token=private",
+    })),
+    remote: vi.fn(async () => ({
+      remoteCwd: "/home/user/paperclip/company/projects/project/checkout",
+    })),
+  };
+  const service = createComputerService(repository, backend, () => clock, {
+    admissionWaitMs: 0,
+  });
+  const scope = { companyId: "company", environmentId: "environment" };
+  const attach = () =>
+    service.attach({
+      ...scope,
+      sandboxId: "bx_fixture",
+      apiKeySecretRef: { type: "secret_ref", secretId: "secret" },
+    });
+  const admit = (agentId = "agent", sessionKey = "session") =>
+    service.admit({
+      ...scope,
+      agentId,
+      sessionKey,
+      runId: "run",
+      idleTimeoutMs: 60_000,
+    });
+  return {
+    service,
+    backend,
+    scope,
+    attach,
+    admit,
+    repository,
+    advance: (ms: number) => {
+      clock = new Date(clock.getTime() + ms);
+    },
+    completeStop: () => {
+      stopStatus = "completed";
+      state = "stopped";
+    },
+  };
+}
+describe("computer ownership", () => {
+  it("explains an unavailable dev server without masking provider errors", async () => {
+    const f = fixture();
+    await f.attach();
+    const binding = await f.admit();
+    const input = { ...f.scope, owner: binding.owner, port: 5173 };
+    vi.mocked(f.backend.remote).mockRejectedValueOnce(new ComputerError("conflict", "Computer operation: conflict"));
+    await expect(f.service.preview(input)).rejects.toMatchObject({
+      code: "conflict", message: "No dev server is running on this port for this task. Start it in the task and try again.",
+    });
+    const offline = new ComputerError("provider_error", "Provider unavailable");
+    vi.mocked(f.backend.remote).mockRejectedValueOnce(offline);
+    await expect(f.service.preview(input)).rejects.toBe(offline);
+    expect(f.backend.preview).not.toHaveBeenCalled();
+  });
+  it("reuses warm process identity and port while fencing stale callbacks", async () => {
+    const f = fixture();
+    await f.attach();
+    const first = await f.admit();
+    const claim = await first.launch({ command: "runnerd" });
+    await f.service.retainWarm({
+      ...f.scope,
+      owner: first.owner,
+      idleTimeoutMs: 60_000,
+    });
+    const next = await f.admit();
+    expect(next.owner.generation).toBe(first.owner.generation + 1);
+    expect(next.listenerPort).toBe(first.listenerPort);
+    expect((await next.inspectProcess()).claim?.nonce).toBe(claim.nonce);
+    await f.service.retire({ ...f.scope, owner: first.owner });
+    expect(f.backend.retire).not.toHaveBeenCalled();
+    await f.service.retire({ ...f.scope, owner: next.owner });
+    expect(f.backend.retire).toHaveBeenCalledOnce();
+  });
+  it("does not suspend another agent and fences admission until provider stop completes", async () => {
+    const f = fixture();
+    await f.attach();
+    const a = await f.admit("agent-a", "a");
+    const b = await f.admit("agent-b", "b");
+    expect(a.listenerPort).not.toBe(b.listenerPort);
+    expect(a.agentHome).not.toBe(b.agentHome);
+    await f.service.retire({ ...f.scope, owner: a.owner });
+    await f.service.reconcile();
+    expect(f.backend.stop).not.toHaveBeenCalled();
+    await f.service.retire({ ...f.scope, owner: b.owner });
+    await f.service.reconcile();
+    expect(f.backend.stop).toHaveBeenCalledOnce();
+    await expect(f.admit()).rejects.toMatchObject({ code: "conflict" });
+    await f.service.reconcile();
+    expect(f.backend.stopStatus).toHaveBeenCalledOnce();
+    f.completeStop();
+    await f.service.reconcile();
+    await expect(f.admit()).resolves.toHaveProperty("owner");
+  });
+  it("retains a failed stop intent and retries the same idempotent operation", async () => {
+    const f = fixture();
+    await f.attach();
+    vi.mocked(f.backend.stop).mockRejectedValueOnce(
+      new Error("network response lost"),
+    );
+    await expect(f.service.reconcile()).rejects.toBeInstanceOf(AggregateError);
+    await expect(f.admit()).rejects.toMatchObject({ code: "conflict" });
+    await f.service.reconcile();
+    expect(f.backend.stop).toHaveBeenCalledTimes(2);
+    expect(
+      (await f.repository.get(f.scope)).ledger.action?.providerStopId,
+    ).toBe("stop_1");
+  });
+  it("viewer renewals cannot extend the absolute warm deadline or use another user", async () => {
+    const f = fixture();
+    await f.attach();
+    const viewer = await f.service.connect({
+      ...f.scope,
+      userId: "alice",
+      idleTimeoutMs: 10_000,
+    });
+    expect(viewer.expiresAt).toBe("2026-10-10T12:00:10.000Z");
+    await expect(
+      f.service.renewViewer({ ...f.scope, owner: viewer.owner, userId: "bob" }),
+    ).rejects.toMatchObject({ code: "forbidden" });
+    f.advance(9_000);
+    expect(
+      (
+        await f.service.renewViewer({
+          ...f.scope,
+          owner: viewer.owner,
+          userId: "alice",
+        })
+      ).expiresAt,
+    ).toBe(viewer.expiresAt);
+    f.advance(1_001);
+    await expect(
+      f.service.renewViewer({
+        ...f.scope,
+        owner: viewer.owner,
+        userId: "alice",
+      }),
+    ).rejects.toMatchObject({ code: "conflict" });
+    await f.service.reconcile();
+    expect(f.backend.stop).toHaveBeenCalledOnce();
+  });
+  it("a failed exact-process retirement keeps the machine held", async () => {
+    const f = fixture();
+    await f.attach();
+    const a = await f.admit();
+    vi.mocked(f.backend.retire).mockRejectedValue(new Error("not confirmed"));
+    await expect(
+      f.service.retire({ ...f.scope, owner: a.owner }),
+    ).rejects.toThrow("not confirmed");
+    await expect(f.service.reconcile()).rejects.toBeInstanceOf(AggregateError);
+    expect(f.backend.stop).not.toHaveBeenCalled();
+  });
+  it("keeps snapshot and attachment proof when detached", async () => {
+    const f = fixture();
+    await f.attach();
+    await f.admit();
+    await f.service.detach(f.scope);
+    expect((await f.repository.get(f.scope)).ledger.status).toBe("detaching");
+    f.completeStop();
+    await f.service.reconcile();
+    expect((await f.repository.get(f.scope)).ledger.status).toBe("detached");
+    await expect(f.admit()).rejects.toMatchObject({ code: "conflict" });
+  });
+  it("rejects cross company scopes before provider access", async () => {
+    const f = fixture();
+    await f.attach();
+    vi.mocked(f.backend.inspect).mockClear();
+    await expect(
+      f.service.inspect({ ...f.scope, companyId: "another" }),
+    ).rejects.toMatchObject({ code: "not_found" });
+    expect(f.backend.inspect).not.toHaveBeenCalled();
+  });
+  it("keeps process capabilities across attempt generations only for the exact same process", async () => {
+    const f = fixture();
+    await f.attach();
+    const first = await f.admit();
+    await first.launch({ command: "runnerd" });
+    await f.service.retainWarm({
+      ...f.scope,
+      owner: first.owner,
+      idleTimeoutMs: 60_000,
+    });
+    await f.admit();
+    await expect(first.ingress()).rejects.toMatchObject({ code: "conflict" });
+    await expect(first.process.ingress()).resolves.toHaveProperty("url");
+    await f.repository.update(f.scope, (record) => {
+      record.ledger.owners[0]!.process!.nonce = "replacement-process";
+    });
+    await expect(first.process.ingress()).rejects.toMatchObject({
+      code: "conflict",
+    });
+  });
+
+  it.each(["terminal", "missing"] as const)(
+    "retires an orphan active owner with a %s run after admission grace",
+    async (state) => {
+      const f = fixture();
+      await f.attach();
+      await f.admit();
+      vi.mocked(f.repository.runState).mockResolvedValue(state);
+      f.advance(119_000);
+      await f.service.reconcile();
+      expect(f.backend.retire).not.toHaveBeenCalled();
+      f.advance(1_001);
+      await f.service.reconcile();
+      expect(f.backend.retire).toHaveBeenCalledOnce();
+      expect(f.backend.stop).toHaveBeenCalledOnce();
+    },
+  );
+  it("preserves recoverable running owners and gives completed warm owners their idle interval", async () => {
+    const f = fixture();
+    await f.attach();
+    const binding = await f.admit();
+    f.advance(180_000);
+    await f.service.reconcile();
+    expect(f.backend.retire).not.toHaveBeenCalled();
+    await f.service.retainWarm({
+      ...f.scope,
+      owner: binding.owner,
+      idleTimeoutMs: 60_000,
+    });
+    vi.mocked(f.repository.runState).mockResolvedValue("terminal");
+    await f.service.reconcile();
+    expect(f.backend.retire).not.toHaveBeenCalled();
+    f.advance(60_001);
+    await f.service.reconcile();
+    expect(f.backend.retire).toHaveBeenCalledOnce();
+  });
+  it("coalesces concurrent reconciliation before provider stop dispatch", async () => {
+    const f = fixture();
+    await f.attach();
+    let complete!: (value: { id: string; status: string }) => void;
+    vi.mocked(f.backend.stop).mockImplementation(
+      () =>
+        new Promise((resolve) => {
+          complete = resolve;
+        }),
+    );
+    const first = f.service.reconcile();
+    await vi.waitFor(() => expect(f.backend.stop).toHaveBeenCalledOnce());
+    const second = f.service.reconcile();
+    await Promise.resolve();
+    expect(f.backend.stop).toHaveBeenCalledOnce();
+    complete({ id: "stop_1", status: "pending" });
+    await Promise.all([first, second]);
+    expect(f.backend.stop).toHaveBeenCalledOnce();
+  });
+  it.each(["retire", "warm"] as const)("waits for terminal predecessor %s before admitting its follow-up", async (transition) => {
+    const f = fixture();
+    await f.attach();
+    const first = await f.admit();
+    const claim = await first.launch({ command: "runnerd" });
+    vi.mocked(f.repository.runState).mockResolvedValue("terminal");
+    const wait = vi.fn(async () => {
+      const incumbent = (await f.repository.get(f.scope)).ledger.owners.find(owner => owner.id === first.owner.ownerId)!;
+      expect(incumbent.phase).toBe("active");
+      expect(incumbent.generation).toBe(first.owner.generation);
+      expect(f.backend.retire).not.toHaveBeenCalled();
+      if (transition === "retire") await f.service.retire({ ...f.scope, owner: first.owner });
+      else await f.service.retainWarm({ ...f.scope, owner: first.owner, idleTimeoutMs: 60_000 });
+    });
+    const waiting = createComputerService(f.repository, f.backend, () => new Date("2026-10-10T12:00:00Z"), { admissionWaitMs: 2000, wait });
+    const next = await waiting.admit({ ...f.scope, agentId: "agent", runId: "follow-up", sessionKey: "session", idleTimeoutMs: 60_000 });
+    expect(wait).toHaveBeenCalledOnce();
+    expect(f.repository.runState).toHaveBeenCalledWith(expect.objectContaining(f.scope), "run", "agent");
+    if (transition === "warm") {
+      expect(next.owner.ownerId).toBe(first.owner.ownerId);
+      expect(next.owner.generation).toBe(first.owner.generation + 1);
+      expect((await next.inspectProcess()).claim?.nonce).toBe(claim.nonce);
+      expect(f.backend.retire).not.toHaveBeenCalled();
+    } else {
+      expect(next.owner.ownerId).not.toBe(first.owner.ownerId);
+      expect(f.backend.retire).toHaveBeenCalledOnce();
+    }
+  });
+
+  it.each(["active", "missing"] as const)("does not wait on or supersede a %s predecessor", async (state) => {
+    const f = fixture();
+    await f.attach();
+    const first = await f.admit();
+    vi.mocked(f.repository.runState).mockResolvedValue(state);
+    const wait = vi.fn(async () => {});
+    const waiting = createComputerService(f.repository, f.backend, () => new Date("2026-10-10T12:00:00Z"), { admissionWaitMs: 2000, wait });
+    await expect(waiting.admit({ ...f.scope, agentId: "agent", runId: "follow-up", sessionKey: "session", idleTimeoutMs: 60_000 }))
+      .rejects.toMatchObject({ code: "conflict" });
+    expect(wait).not.toHaveBeenCalled();
+    expect(f.backend.retire).not.toHaveBeenCalled();
+    expect((await f.repository.get(f.scope)).ledger.owners[0]?.generation).toBe(first.owner.generation);
+  });
+
+  it("bounds the wait for terminal predecessor teardown without forcing retirement", async () => {
+    const f = fixture();
+    await f.attach();
+    await f.admit();
+    vi.mocked(f.repository.runState).mockResolvedValue("terminal");
+    const wait = vi.fn(async () => {});
+    const waiting = createComputerService(f.repository, f.backend, () => new Date("2026-10-10T12:00:00Z"), { admissionWaitMs: 2000, wait });
+    await expect(waiting.admit({ ...f.scope, agentId: "agent", runId: "follow-up", sessionKey: "session", idleTimeoutMs: 60_000 }))
+      .rejects.toMatchObject({ code: "conflict" });
+    expect(wait).toHaveBeenCalledTimes(2);
+    expect(f.backend.retire).not.toHaveBeenCalled();
+  });
+
+  it("waits for a normal provider snapshot stop before admitting the next turn", async () => {
+    const f = fixture();
+    await f.attach();
+    await f.service.reconcile();
+    const waiting = createComputerService(
+      f.repository,
+      f.backend,
+      () => new Date("2026-10-10T12:00:00Z"),
+      {
+        admissionWaitMs: 2000,
+        wait: async () => {
+          f.completeStop();
+        },
+      },
+    );
+    await expect(
+      waiting.admit({
+        ...f.scope,
+        agentId: "agent",
+        runId: "run",
+        sessionKey: "session",
+        idleTimeoutMs: 60_000,
+      }),
+    ).resolves.toHaveProperty("owner");
+    expect(f.backend.stop).toHaveBeenCalledOnce();
+  });
+  it("keeps a short bounded file batch hold between editor operations", async () => {
+    const f = fixture();
+    await f.attach();
+    const files = await f.service.files({ ...f.scope, agentId: "agent" });
+    await files.seed({});
+    await f.service.reconcile();
+    expect(f.backend.stop).not.toHaveBeenCalled();
+    f.advance(5001);
+    await f.service.reconcile();
+    expect(f.backend.stop).toHaveBeenCalledOnce();
+  });
+  it("returns foreground output without waiting for detached command descendants", async () => {
+    const f = fixture();
+    const temp = realpathSync(mkdtempSync(join(tmpdir(), "computer-command-")));
+    let backgroundPid: number | undefined;
+    const runPath = join(temp, "systemd-run");
+    // The shim starts the real supervisor. Live Boat coverage additionally
+    // verifies systemd keeps the descendant in the owner slice and retires it.
+    writeFileSync(
+      runPath,
+      `#!/usr/bin/env python3
+import sys,subprocess,json
+args=sys.argv[1:]
+assert '--property=ExitType=cgroup' in args
+assert '--property=KillMode=control-group' in args
+assert '--wait' not in args
+output=next(v.split('append:',1)[1] for v in args if v.startswith('--property=StandardOutput='))
+error=next(v.split('append:',1)[1] for v in args if v.startswith('--property=StandardError='))
+cwd=next(v.split('=',1)[1] for v in args if v.startswith('--working-directory='))
+with open(output,'wb') as out,open(error,'wb') as err:
+ subprocess.Popen(args[args.index('--')+1:],stdout=out,stderr=err,cwd=cwd,start_new_session=True)
+`,
+      { mode: 0o700 },
+    );
+    writeFileSync(join(temp, "systemctl"), "#!/bin/sh\necho active\n", {
+      mode: 0o700,
+    });
+    vi.mocked(f.backend.runner).mockResolvedValue({
+      execute: async (input) => {
+        const payload = JSON.parse(input.stdin!);
+        const ownerRoot = join(temp, payload.owner);
+        mkdirSync(ownerRoot, { recursive: true });
+        writeFileSync(
+          join(ownerRoot, "generation"),
+          String(payload.generation),
+        );
+        const result = spawnSync(
+          input.command,
+          input.args!.map((arg) =>
+            arg.replaceAll("/home/user/.paperclip-owners/", `${temp}/`),
+          ),
+          {
+            input: input.stdin,
+            encoding: "utf8",
+            timeout: 5000,
+            env: { ...process.env, PATH: `${temp}:${process.env.PATH}` },
+          },
+        );
+        if (result.error) throw result.error;
+        return {
+          stdout: result.stdout,
+          stderr: result.stderr,
+          exitCode: result.status,
+          timedOut: false,
+        } as any;
+      },
+    });
+    try {
+      await f.attach();
+      const binding = await f.admit();
+      const result = await binding.runner.execute({
+        command: "sh",
+        args: [
+          "-c",
+          "sleep 30 </dev/null >/dev/null 2>&1 & echo $!; read word; echo $word >&2; exit 7",
+        ],
+        cwd: temp,
+        stdin: "foreground-input\n",
+      });
+      backgroundPid = Number(result.stdout.trim());
+      expect(result.exitCode).toBe(7);
+      expect(result.stderr).toBe("foreground-input\n");
+      expect(backgroundPid).toBeGreaterThan(1);
+      expect(() => process.kill(backgroundPid!, 0)).not.toThrow();
+      expect(existsSync(join(temp, binding.owner.ownerId))).toBe(true);
+    } finally {
+      if (backgroundPid) process.kill(backgroundPid, "SIGKILL");
+      rmSync(temp, { recursive: true, force: true });
+    }
+  });
+  it("realizes probe cwd before readiness without consuming the initial personal-home seed", async () => {
+    const f = fixture();
+    const temp = realpathSync(mkdtempSync(join(tmpdir(), "computer-probe-")));
+    const local = (value: string) =>
+      value.replace("/home/user/paperclip/", `${temp}/`);
+    vi.mocked(f.backend.remote).mockImplementation(async (_record, input) => {
+      const result = spawnSync(
+        "python3",
+        ["-c", remoteProgram.replaceAll("/home/user/paperclip/", `${temp}/`)],
+        {
+          input: JSON.stringify({
+            ...input,
+            root: local(input.root as string),
+          }),
+          encoding: "utf8",
+        },
+      );
+      if (result.status !== 0) throw new Error(result.stderr);
+      return JSON.parse(result.stdout);
+    });
+    try {
+      await f.attach();
+      const probe = await f.service.admitProbe({
+        ...f.scope,
+        agentId: "agent",
+        probeId: "probe",
+        idleTimeoutMs: 60_000,
+      });
+      expect(probe.remoteCwd).not.toBe(probe.agentHome);
+      const ready = spawnSync("sh", ["-c", "pwd"], {
+        cwd: local(probe.remoteCwd),
+        encoding: "utf8",
+      });
+      expect(ready.status).toBe(0);
+      expect(ready.stdout.trim()).toBe(local(probe.remoteCwd));
+      expect(existsSync(local(probe.agentHome))).toBe(false);
+      const run = await f.admit();
+      expect(run.remoteCwd).toBe(probe.agentHome);
+      expect(existsSync(local(run.agentHome))).toBe(false);
+      const files = await f.service.files({ ...f.scope, agentId: "agent" });
+      await expect(
+        files.seed({ "AGENTS.md": "initial managed instructions" }),
+      ).resolves.toEqual({ seeded: true });
+      expect(
+        readFileSync(join(local(run.agentHome), "AGENTS.md"), "utf8"),
+      ).toBe("initial managed instructions");
+    } finally {
+      rmSync(temp, { recursive: true, force: true });
+    }
+  });
+  it("gives harness probes a finite owner without fabricating a heartbeat run", async () => {
+    const f = fixture();
+    await f.attach();
+    const probe = await f.service.admitProbe({
+      ...f.scope,
+      agentId: "agent",
+      probeId: "probe",
+      idleTimeoutMs: 60_000,
+    });
+    await expect(
+      f.service.retainWarm({
+        ...f.scope,
+        owner: probe.owner,
+        idleTimeoutMs: 60_000,
+      }),
+    ).rejects.toMatchObject({ code: "conflict" });
+    f.advance(60_001);
+    await f.service.reconcile();
+    expect(f.repository.runState).not.toHaveBeenCalled();
+    expect(f.backend.retire).toHaveBeenCalledOnce();
+  });
+});

server/src/modules/computers/application/service.ts

@@ -0,0 +1,964 @@
+import { randomUUID } from "node:crypto";
+import type { ComputerRepository, ComputerBackend } from "./ports.js";
+import {
+  ComputerError,
+  assertAdmission,
+  exactOwner,
+  expired,
+  liveOwners,
+  nextPort,
+  segment,
+  timeout,
+  type ComputerRecord,
+  type OwnerRef,
+  type Owner,
+} from "../domain/ledger.js";
+import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
+
+// A terminal run can enqueue its successor before remote teardown has finished.
+// Keep the incumbent owner fenced until its own cleanup or warm handoff settles.
+class ComputerSessionBusyError extends ComputerError {
+  constructor(readonly runId: string | undefined, readonly agentId: string | undefined) {
+    super("conflict", "Computer session already has an active turn");
+  }
+}
+
+type Scope = { companyId: string; environmentId: string };
+export function createComputerService(
+  repository: ComputerRepository,
+  backend: ComputerBackend,
+  now: () => Date = () => new Date(),
+  options: {
+    admissionWaitMs?: number;
+    wait?: (ms: number) => Promise<void>;
+  } = {},
+) {
+  const ref = (record: ComputerRecord, owner: Owner): OwnerRef => ({
+    computerId: record.id,
+    ownerId: owner.id,
+    generation: owner.generation,
+  });
+  const base = (record: ComputerRecord) =>
+    `/home/user/paperclip/${segment(record.companyId)}`;
+  async function admitRecord<T>(
+    scope: Scope,
+    change: (record: ComputerRecord) => T,
+  ): Promise<T> {
+    const attempts = Math.ceil((options.admissionWaitMs ?? 120_000) / 1000);
+    const wait =
+      options.wait ??
+      ((ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms)));
+    for (let attempt = 0; ; attempt++) {
+      try {
+        return await repository.update(scope, (record) => {
+          assertAdmission(record.ledger);
+          return change(record);
+        });
+      } catch (error) {
+        if (
+          !(error instanceof ComputerError) ||
+          error.code !== "conflict" ||
+          attempt >= attempts
+        )
+          throw error;
+        const current = await repository.get(scope);
+        if (current.ledger.status !== "attached") throw error;
+        if (error instanceof ComputerSessionBusyError && error.runId && error.agentId) {
+          // A terminal database status does not prove the process has stopped.
+          // Wait for the existing owner transition; never retire it here or
+          // advance its generation while final commands might still run.
+          if (await repository.runState(scope, error.runId, error.agentId) !== "terminal") throw error;
+          await wait(1000);
+          continue;
+        }
+        if (!current.ledger.action) throw error;
+        await reconcileRecord(current);
+        if ((await repository.get(scope)).ledger.action) await wait(1000);
+      }
+    }
+  }
+  async function scoped(input: Scope & { owner: OwnerRef }) {
+    const record = await repository.get(input);
+    return { record, owner: exactOwner(record, input.owner) };
+  }
+  async function inspect(input: Scope) {
+    const record = await repository.get(input);
+    const status = await backend.inspect(record);
+    return {
+      computerId: record.id,
+      sandboxId: record.providerId,
+      status: record.ledger.status,
+      state: status.state,
+    };
+  }
+  async function attach(
+    input: Scope & {
+      sandboxId: string;
+      apiKeySecretRef: {
+        type: "secret_ref";
+        secretId: string;
+        version?: number | "latest";
+      };
+    },
+  ) {
+    segment(input.companyId);
+    segment(input.environmentId);
+    segment(input.sandboxId);
+    segment(input.apiKeySecretRef.secretId);
+    let record: ComputerRecord;
+    try {
+      record = await repository.get(input);
+      if (record.providerId !== input.sandboxId)
+        throw new ComputerError(
+          "conflict",
+          "Environment already attaches another computer",
+        );
+    } catch (error) {
+      if (!(error instanceof ComputerError) || error.code !== "not_found")
+        throw error;
+      record = {
+        id: randomUUID(),
+        companyId: input.companyId,
+        environmentId: input.environmentId,
+        providerId: input.sandboxId,
+        ledger: {
+          controllerId: randomUUID(),
+          status: "attaching",
+          secretRef: input.apiKeySecretRef,
+          owners: [],
+          placements: {},
+          action: null,
+        },
+      };
+      await repository.create(record);
+    }
+    if (
+      JSON.stringify(record.ledger.secretRef) !==
+      JSON.stringify(input.apiKeySecretRef)
+    ) {
+      const candidate = structuredClone(record);
+      candidate.ledger.secretRef = input.apiKeySecretRef;
+      const status = await backend.inspect(candidate);
+      if (!status.snapshots)
+        throw new ComputerError(
+          "invalid",
+          "Boat snapshots must remain enabled",
+        );
+      await repository.update(input, (current) => {
+        current.ledger.secretRef = input.apiKeySecretRef;
+      });
+      record = candidate;
+    }
+    if (record.ledger.status === "detaching" || record.ledger.action)
+      throw new ComputerError("conflict", "Computer cleanup is pending");
+    if (record.ledger.status !== "attached") {
+      await backend.ready(record);
+      await backend.claim(record);
+      await repository.update(input, (current) => {
+        if (
+          current.ledger.status === "attaching" ||
+          current.ledger.status === "detached"
+        )
+          current.ledger.status = "attached";
+      });
+    }
+    return { computerId: record.id };
+  }
+  async function binding(record: ComputerRecord, owner: Owner) {
+    const ownerRef = ref(record, owner);
+    const placement = record.ledger.placements[owner.sessionKey!];
+    if (!placement)
+      throw new ComputerError("conflict", "Computer placement is missing");
+    const raw = await backend.runner(record);
+    let pinnedProcess = owner.process ? structuredClone(owner.process) : null;
+    async function processScope() {
+      if (!pinnedProcess) return scoped({ ...record, owner: ownerRef });
+      const latest = await repository.get(record);
+      const candidate = latest.ledger.owners.find(
+        (value) => value.id === owner.id,
+      );
+      if (
+        !candidate?.process ||
+        !["active", "starting", "warm"].includes(candidate.phase) ||
+        candidate.process.nonce !== pinnedProcess.nonce ||
+        candidate.process.unitName !== pinnedProcess.unitName ||
+        candidate.process.bootId !== pinnedProcess.bootId ||
+        candidate.process.launchGeneration !== pinnedProcess.launchGeneration
+      ) {
+        throw new ComputerError(
+          "conflict",
+          "Computer process capability has expired",
+        );
+      }
+      return { record: latest, owner: candidate };
+    }
+    const runnerFor = (
+      processScoped: boolean,
+    ): CommandManagedRuntimeRunner => ({
+      execute: async (input) => {
+        const current = await (processScoped
+          ? processScope()
+          : scoped({ ...record, owner: ownerRef }));
+        if (!["active", "starting", "warm"].includes(current.owner.phase))
+          throw new ComputerError("conflict", "Computer owner is retired");
+        // Start a command unit under the same owner slice while holding the remote tombstone lock.
+        // The lock is released only after systemd knows the unit, so retirement cannot miss a delayed launch.
+        return raw.execute({
+          command: "python3",
+          args: [
+            "-c",
+            String.raw`
+import os,sys,json,fcntl,subprocess,time,tempfile,shutil
+p=json.load(sys.stdin);root='/home/user/.paperclip-owners/'+p['owner'];os.makedirs(root,exist_ok=True)
+# A receipt reports foreground completion independently from cgroup lifetime.
+# ExitType=cgroup keeps nohup/setsid descendants owned until owner retirement.
+supervisor="""
+import os,sys,json,subprocess
+payload=sys.argv[1]
+with open(payload) as f:p=json.load(f)
+os.unlink(payload)
+i=p['input']
+try:
+ with open(p['stdin'],'rb') as source:
+  os.unlink(p['stdin'])
+  child=subprocess.Popen([i['command']]+i.get('args',[]),stdin=source)
+  code=child.wait()
+except Exception as error:
+ print(str(error),file=sys.stderr);code=127
+with open(p['receipt']+'.tmp','w') as f:json.dump({'exitCode':code},f)
+os.replace(p['receipt']+'.tmp',p['receipt'])
+"""
+work=None
+try:
+ with open(root+'/lock','a') as lock:
+  fcntl.flock(lock,fcntl.LOCK_EX)
+  if os.path.exists(root+'/retired'):raise RuntimeError('computer owner retired')
+  if p.get('processClaim'):
+   claim=json.load(open(root+'/claim.json'))
+   if any(claim[k]!=p['processClaim'][k] for k in ['nonce','unitName','bootId','launchGeneration']) or claim['bootId']!=open('/proc/sys/kernel/random/boot_id').read().strip():raise RuntimeError('computer process superseded')
+  elif not os.path.exists(root+'/generation') or int(open(root+'/generation').read())!=p['generation']:raise RuntimeError('computer owner superseded')
+  work=tempfile.mkdtemp(prefix='command-',dir=root)
+  output=work+'/stdout';error=work+'/stderr';receipt=work+'/receipt';payload=work+'/input.json';source=work+'/stdin'
+  for path in [output,error,source]:open(path,'wb').close()
+  i=p['input']
+  with open(source,'w') as f:f.write(i.get('stdin',''))
+  with open(payload,'w') as f:json.dump({'input':i,'stdin':source,'receipt':receipt},f)
+  args=['systemd-run','--user','--quiet','--collect','--unit='+p['unit'],'--slice=paperclip-'+p['owner']+'.slice','--property=Type=exec','--property=ExitType=cgroup','--property=KillMode=control-group','--property=StandardOutput=append:'+output,'--property=StandardError=append:'+error,'--working-directory='+i.get('cwd',p['cwd'])]
+  for k,v in i.get('env',{}).items():args.append('--setenv='+k+'='+v)
+  started=subprocess.run(args+['--','python3','-c',supervisor,payload],capture_output=True,text=True)
+  if started.returncode:sys.stderr.write(started.stderr);sys.exit(started.returncode)
+ with open(output,'rb') as out,open(error,'rb') as err:
+  while True:
+   done=os.path.exists(receipt)
+   for stream,dest in [(out,sys.stdout.buffer),(err,sys.stderr.buffer)]:
+    chunk=stream.read()
+    if chunk:dest.write(chunk);dest.flush()
+   if done:break
+   state=subprocess.run(['systemctl','--user','show',p['unit'],'--property=ActiveState','--value'],capture_output=True,text=True).stdout.strip()
+   if state not in ('active','activating','reloading'):
+    if os.path.exists(receipt):continue
+    raise RuntimeError('computer command stopped before completion')
+   time.sleep(.05)
+ with open(receipt) as f:code=json.load(f)['exitCode']
+ sys.exit(code if code>=0 else 128-code)
+finally:
+ if work:shutil.rmtree(work,ignore_errors=True)
+`,
+          ],
+          stdin: JSON.stringify({
+            owner: owner.id,
+            generation: current.owner.generation,
+            processClaim: processScoped ? pinnedProcess : null,
+            unit: `paperclip-command-${randomUUID()}.service`,
+            cwd: placement.cwd,
+            input: {
+              command: input.command,
+              args: input.args,
+              cwd: input.cwd,
+              env: input.env,
+              stdin: input.stdin,
+            },
+          }),
+          timeoutMs: input.timeoutMs,
+          onLog: input.onLog,
+          onSpawn: input.onSpawn,
+        });
+      },
+    });
+    return {
+      owner: ownerRef,
+      runner: runnerFor(false),
+      process: {
+        runner: runnerFor(true),
+        async ingress(input?: { port?: number; path?: string }) {
+          const current = await processScope();
+          return backend.ingress(
+            current.record,
+            input?.port ?? current.owner.port,
+            input?.path ?? "/",
+          );
+        },
+      },
+      listenerPort: owner.port,
+      remoteCwd: placement.cwd,
+      agentHome: placement.root,
+      placementId: placement.id,
+      async launch(input: {
+        command: string;
+        args?: string[];
+        cwd?: string;
+        env?: Record<string, string>;
+      }) {
+        const before = await scoped({ ...record, owner: ownerRef });
+        const existing = before.owner.process
+          ? await backend.inspectProcess(before.record, before.owner)
+          : null;
+        const claimed = await repository.update(record, (current) => {
+          const o = exactOwner(current, ownerRef);
+          if (o.phase !== "active" && o.phase !== "starting")
+            throw new ComputerError("conflict", "Computer owner cannot launch");
+          if (existing?.running && existing.claim) o.process = existing.claim;
+          else if (!o.process || !existing?.running)
+            o.process = {
+              bootId: "",
+              unitName: `paperclip-${o.id}.service`,
+              nonce: randomUUID(),
+              launchGeneration: (o.process?.launchGeneration ?? 0) + 1,
+            };
+          return structuredClone(current);
+        });
+        const claim = await backend.launch(
+          claimed,
+          exactOwner(claimed, ownerRef),
+          input,
+        );
+        await repository.update(record, (current) => {
+          exactOwner(current, ownerRef).process = claim;
+        });
+        pinnedProcess = structuredClone(claim);
+        return claim;
+      },
+      async inspectProcess() {
+        const current = await scoped({ ...record, owner: ownerRef });
+        return backend.inspectProcess(current.record, current.owner);
+      },
+      async ingress(input?: { port?: number; path?: string }) {
+        const current = await scoped({ ...record, owner: ownerRef });
+        if (current.owner.phase === "retired")
+          throw new ComputerError("conflict", "Computer owner retired");
+        return backend.ingress(
+          current.record,
+          input?.port ?? owner.port,
+          input?.path ?? "/",
+        );
+      },
+    };
+  }
+  async function admit(
+    input: Scope & {
+      agentId: string;
+      runId?: string;
+      probeId?: string;
+      sessionKey: string;
+      idleTimeoutMs: number;
+    },
+  ) {
+    segment(input.agentId);
+    if ((!input.runId && !input.probeId) || (input.runId && input.probeId))
+      throw new ComputerError(
+        "invalid",
+        "Computer admission requires one run or probe identity",
+      );
+    if (input.probeId) segment(input.probeId);
+    if (!/^[a-zA-Z0-9_-]{1,128}$/.test(input.sessionKey))
+      throw new ComputerError("invalid", "Invalid computer session key");
+    timeout(input.idleTimeoutMs);
+    const result = await admitRecord(input, (record) => {
+      let owner = record.ledger.owners.find(
+        (o) =>
+          o.kind === "runner" &&
+          o.agentId === input.agentId &&
+          o.sessionKey === input.sessionKey &&
+          o.phase !== "retired",
+      );
+      if (owner && owner.phase !== "warm")
+        throw new ComputerSessionBusyError(owner.runId, owner.agentId);
+      if (owner && expired(owner, now()))
+        throw new ComputerError(
+          "conflict",
+          "Computer warm session is retiring",
+        );
+      if (owner) {
+        owner.generation++;
+        owner.phase = "starting";
+        owner.runId = input.runId;
+        owner.admittedAt = now().toISOString();
+        owner.deadline = new Date(now().getTime() + 120_000).toISOString();
+      } else {
+        owner = {
+          id: randomUUID(),
+          generation: 1,
+          kind: "runner",
+          phase: "starting",
+          agentId: input.agentId,
+          runId: input.runId,
+          probeId: input.probeId,
+          admittedAt: now().toISOString(),
+          sessionKey: input.sessionKey,
+          port: nextPort(record.ledger.owners),
+          deadline: new Date(now().getTime() + 120_000).toISOString(),
+          absoluteDeadline: null,
+          process: null,
+        };
+        record.ledger.owners.push(owner);
+      }
+      record.ledger.placements[input.agentId] ??= {
+        id: randomUUID(),
+        root: `${base(record)}/agents/${input.agentId}`,
+        cwd: `${base(record)}/agents/${input.agentId}`,
+      };
+      record.ledger.placements[input.sessionKey] ??= {
+        id: randomUUID(),
+        root: `${base(record)}/agents/${input.agentId}`,
+        cwd: input.probeId
+          ? `${base(record)}/probes/${input.probeId}`
+          : `${base(record)}/agents/${input.agentId}`,
+      };
+      return { record: structuredClone(record), owner: structuredClone(owner) };
+    });
+    await backend.ready(result.record);
+    await backend.claim(result.record);
+    await backend.advance(result.record, result.owner);
+    await backend.renew(result.record);
+    if (input.probeId) {
+      // Readiness commands run before managed-home preparation. Give them an
+      // existing workspace without creating the personal home: an absent home
+      // must remain eligible for the first atomic managed-file seed.
+      await backend.remote(result.record, {
+        action: "seed",
+        root: result.record.ledger.placements[input.sessionKey]!.cwd,
+        files: {},
+      });
+    }
+    const active = await repository.update(input, (record) => {
+      const owner = exactOwner(record, ref(result.record, result.owner));
+      if (owner.phase !== "starting")
+        throw new ComputerError("conflict", "Computer admission was retired");
+      owner.phase = "active";
+      owner.deadline = owner.probeId
+        ? new Date(
+            now().getTime() + Math.min(input.idleTimeoutMs, 120_000),
+          ).toISOString()
+        : null;
+      return { record: structuredClone(record), owner: structuredClone(owner) };
+    });
+    return binding(active.record, active.owner);
+  }
+  async function recover(input: Scope & { owner: OwnerRef }) {
+    const { record, owner } = await scoped(input);
+    if (!["active", "warm"].includes(owner.phase) || expired(owner, now()))
+      throw new ComputerError(
+        "conflict",
+        "Computer owner is no longer recoverable",
+      );
+    await backend.ready(record);
+    await backend.claim(record);
+    return binding(record, owner);
+  }
+  async function retainWarm(
+    input: Scope & { owner: OwnerRef; idleTimeoutMs: number },
+  ) {
+    timeout(input.idleTimeoutMs);
+    await repository.update(input, (record) => {
+      const owner = exactOwner(record, input.owner);
+      if (owner.kind !== "runner" || owner.probeId || owner.phase !== "active")
+        throw new ComputerError(
+          "conflict",
+          "Computer owner cannot become warm",
+        );
+      owner.phase = "warm";
+      owner.deadline = new Date(
+        now().getTime() + input.idleTimeoutMs,
+      ).toISOString();
+    });
+  }
+  async function retire(input: Scope & { owner: OwnerRef }) {
+    const state = await repository.update(input, (record) => {
+      const owner = record.ledger.owners.find(
+        (value) => value.id === input.owner.ownerId,
+      );
+      if (record.id !== input.owner.computerId || !owner)
+        throw new ComputerError("not_found", "Computer owner not found");
+      if (owner.generation !== input.owner.generation) return null;
+      if (owner.phase !== "retired") owner.phase = "retiring";
+      return { record: structuredClone(record), owner: structuredClone(owner) };
+    });
+    if (!state) return { retired: false };
+    if (state.owner.phase === "retired") return { retired: true };
+    if (state.owner.kind === "runner") {
+      const status = await backend.inspect(state.record);
+      if (!["stopped", "archived"].includes(status.state))
+        await backend.retire(state.record, state.owner);
+    }
+    await repository.update(input, (record) => {
+      const owner = exactOwner(record, input.owner);
+      owner.phase = "retired";
+      owner.deadline = null;
+    });
+    return { retired: true };
+  }
+  async function connect(
+    input: Scope & { userId: string; idleTimeoutMs: number },
+  ) {
+    timeout(input.idleTimeoutMs);
+    const deadline = new Date(
+      now().getTime() + input.idleTimeoutMs,
+    ).toISOString();
+    const data = await admitRecord(input, (record) => {
+      const owner: Owner = {
+        id: randomUUID(),
+        generation: 1,
+        kind: "viewer",
+        phase: "active",
+        userId: input.userId,
+        port: 0,
+        deadline,
+        absoluteDeadline: deadline,
+        process: null,
+      };
+      record.ledger.owners.push(owner);
+      return { record: structuredClone(record), owner };
+    });
+    await backend.ready(data.record);
+    await backend.claim(data.record);
+    await backend.renew(data.record);
+    const viewer = await backend.desktop(data.record);
+    return {
+      ...viewer,
+      expiresAt: new Date(
+        Math.min(Date.parse(viewer.expiresAt), Date.parse(deadline)),
+      ).toISOString(),
+      owner: ref(data.record, data.owner),
+    };
+  }
+  async function renewViewer(
+    input: Scope & { owner: OwnerRef; userId: string },
+  ) {
+    const { record, owner } = await scoped(input);
+    if (owner.kind !== "viewer" || owner.userId !== input.userId)
+      throw new ComputerError(
+        "forbidden",
+        "Computer viewer belongs to another user",
+      );
+    if (owner.phase !== "active" || expired(owner, now()))
+      throw new ComputerError(
+        "conflict",
+        "Computer connection expired; connect again",
+      );
+    const viewer = await backend.desktop(record);
+    return {
+      ...viewer,
+      expiresAt: new Date(
+        Math.min(
+          Date.parse(viewer.expiresAt),
+          Date.parse(owner.absoluteDeadline!),
+        ),
+      ).toISOString(),
+      owner: input.owner,
+    };
+  }
+  async function disconnectViewer(
+    input: Scope & { owner: OwnerRef; userId: string },
+  ) {
+    const { owner } = await scoped(input);
+    if (owner.kind !== "viewer" || owner.userId !== input.userId)
+      throw new ComputerError(
+        "forbidden",
+        "Computer viewer belongs to another user",
+      );
+    await retire(input);
+  }
+  async function preview(input: Scope & { owner: OwnerRef; port: number }) {
+    const { record, owner } = await scoped(input);
+    if (
+      owner.kind !== "runner" ||
+      !["active", "warm"].includes(owner.phase) ||
+      expired(owner, now())
+    )
+      throw new ComputerError("conflict", "Preview owner is not live");
+    if (input.port >= 43127 && input.port < 44127)
+      throw new ComputerError(
+        "forbidden",
+        "Runner ports are not application previews",
+      );
+    try {
+      await backend.remote(record, {
+        root: base(record),
+        action: "owned-port",
+        port: input.port,
+        ownerId: owner.id,
+      });
+    } catch (error) {
+      if (error instanceof ComputerError && error.code === "conflict") {
+        throw new ComputerError("conflict", "No dev server is running on this port for this task. Start it in the task and try again.");
+      }
+      throw error;
+    }
+    return backend.preview(record, input.port);
+  }
+  async function withFiles<T>(
+    input: Scope,
+    root: string,
+    fn: (record: ComputerRecord) => Promise<T>,
+  ): Promise<T> {
+    const data = await admitRecord(input, (record) => {
+      const owner: Owner = {
+        id: randomUUID(),
+        generation: 1,
+        kind: "file-operation",
+        phase: "active",
+        port: 0,
+        deadline: new Date(now().getTime() + 120_000).toISOString(),
+        absoluteDeadline: null,
+        process: null,
+      };
+      record.ledger.owners.push(owner);
+      return { record: structuredClone(record), owner };
+    });
+    try {
+      await backend.ready(data.record);
+      await backend.claim(data.record);
+      return await fn(data.record);
+    } finally {
+      await repository.update(input, (record) => {
+        const owner = record.ledger.owners.find(
+          (value) => value.id === data.owner.id,
+        );
+        if (owner?.phase === "active") {
+          owner.phase = "warm";
+          owner.deadline = new Date(now().getTime() + 5000).toISOString();
+        }
+      });
+    }
+  }
+  function fileAccess(input: Scope, root: string) {
+    const remote = (payload: Record<string, unknown>) =>
+      withFiles(input, root, (record) =>
+        backend.remote(record, { ...payload, root }),
+      );
+    const readBytes = async (
+      path: string,
+      maxBytes = 16 * 1024 * 1024,
+    ): Promise<{ bytes: Buffer; sha256: string }> => {
+      if (
+        !Number.isSafeInteger(maxBytes) ||
+        maxBytes < 0 ||
+        maxBytes > 16 * 1024 * 1024
+      )
+        throw new ComputerError("invalid", "Invalid remote file size limit");
+      const value = await remote({ action: "read", path, maxBytes });
+      return {
+        bytes: Buffer.from(value.base64, "base64"),
+        sha256: value.sha256,
+      };
+    };
+    const writeBytes = async (
+      path: string,
+      bytes: Buffer,
+      expectedSha256: string | null,
+    ): Promise<{ sha256: string }> =>
+      remote({
+        action: "write",
+        path,
+        base64: bytes.toString("base64"),
+        expectedSha256,
+      });
+    return {
+      root,
+      readBytes,
+      writeBytes,
+      async read(path: string) {
+        const value = await readBytes(path);
+        return { content: value.bytes.toString("utf8"), sha256: value.sha256 };
+      },
+      async write(
+        path: string,
+        content: string,
+        expectedSha256: string | null,
+      ) {
+        return writeBytes(path, Buffer.from(content), expectedSha256);
+      },
+      async listPage(path = "", options: { limit?: number } = {}): Promise<{
+        entries: Array<{ name: string; kind: "file" | "directory"; size: number; mtimeMs: number }>;
+        truncated: boolean;
+      }> {
+        return remote({ action: "list", path, limit: options.limit ?? 1000 });
+      },
+      async list(path = "") {
+        const page = await this.listPage(path);
+        if (page.truncated) throw new ComputerError("invalid", "Computer directory exceeds the listing limit");
+        return page.entries;
+      },
+      async stat(path: string): Promise<{ name: string; kind: "file" | "directory"; size: number; mtimeMs: number }> {
+        return remote({ action: "stat", path });
+      },
+      async remove(path: string, expectedSha256: string) {
+        await remote({ action: "remove", path, expectedSha256 });
+      },
+      async move(
+        from: string,
+        to: string,
+        expectedSha256: string,
+      ): Promise<{ sha256: string }> {
+        return remote({ action: "move", path: from, to, expectedSha256 });
+      },
+      async seedBytes(
+        files: Record<string, Buffer>,
+      ): Promise<{ seeded: boolean }> {
+        return remote({
+          action: "seed",
+          files: Object.fromEntries(
+            Object.entries(files).map(([key, value]) => [
+              key,
+              value.toString("base64"),
+            ]),
+          ),
+        });
+      },
+      async seed(files: Record<string, string>): Promise<{ seeded: boolean }> {
+        return remote({
+          action: "seed",
+          files: Object.fromEntries(
+            Object.entries(files).map(([key, value]) => [
+              key,
+              Buffer.from(value).toString("base64"),
+            ]),
+          ),
+        });
+      },
+    };
+  }
+  async function files(input: Scope & { agentId: string }) {
+    segment(input.agentId);
+    const placement = await admitRecord(input, (record) => {
+      return (record.ledger.placements[input.agentId] ??= {
+        id: randomUUID(),
+        root: `${base(record)}/agents/${input.agentId}`,
+        cwd: `${base(record)}/agents/${input.agentId}`,
+      });
+    });
+    return fileAccess(input, placement.root);
+  }
+  async function workspaceFiles(input: Scope & { placementId: string }) {
+    const record = await repository.get(input);
+    const placement = Object.values(record.ledger.placements).find(
+      (p) => p.id === input.placementId,
+    );
+    if (!placement)
+      throw new ComputerError(
+        "not_found",
+        "Computer workspace placement not found",
+      );
+    return fileAccess(input, placement.cwd);
+  }
+  async function realizeWorkspace(
+    input: Scope & {
+      owner: OwnerRef;
+      projectId?: string;
+      taskId?: string;
+      repositoryUrl?: string;
+      branch?: string;
+      baseRef?: string;
+      mode: "shared" | "worktree";
+    },
+  ) {
+    const { record, owner } = await scoped(input);
+    if (!owner.agentId || owner.phase !== "active")
+      throw new ComputerError("conflict", "Workspace owner is not active");
+    const placement = record.ledger.placements[owner.sessionKey!]!;
+    if (!input.projectId)
+      return {
+        remoteCwd: placement.cwd,
+        agentHome: placement.root,
+        placementId: placement.id,
+      };
+    const root = `${base(record)}/projects/${segment(input.projectId)}`;
+    if (input.taskId) segment(input.taskId);
+    const result = await backend.remote(record, {
+      ...input,
+      root,
+      action: "workspace",
+    });
+    await repository.update(input, (current) => {
+      exactOwner(current, input.owner);
+      const placementKey = `workspace:${owner.agentId}:${input.projectId}:${input.mode}:${input.taskId ?? "shared"}:${result.remoteCwd}`;
+      current.ledger.placements[placementKey] ??= {
+        id: randomUUID(),
+        root: placement.root,
+        cwd: result.remoteCwd,
+      };
+      current.ledger.placements[owner.sessionKey!] =
+        current.ledger.placements[placementKey]!;
+    });
+    const updated = await repository.get(input);
+    return {
+      remoteCwd: result.remoteCwd as string,
+      agentHome: placement.root,
+      placementId: updated.ledger.placements[owner.sessionKey!]!.id,
+    };
+  }
+  async function detach(input: Scope) {
+    const record = await repository.update(input, (current) => {
+      current.ledger.status = "detaching";
+      return structuredClone(current);
+    });
+    for (const owner of liveOwners(record.ledger))
+      await retire({ ...input, owner: ref(record, owner) });
+    await reconcileRecord(await repository.get(input));
+  }
+  const reconciliation = new Map<string, Promise<void>>();
+  async function reconcileRecord(record: ComputerRecord): Promise<void> {
+    const existing = reconciliation.get(record.id);
+    if (existing) return existing;
+    const pending = (async () =>
+      reconcileOwnedRecord(await repository.get(record)))();
+    reconciliation.set(record.id, pending);
+    try {
+      await pending;
+    } finally {
+      if (reconciliation.get(record.id) === pending)
+        reconciliation.delete(record.id);
+    }
+  }
+  async function reconcileOwnedRecord(record: ComputerRecord) {
+    if (record.ledger.status === "attaching") return;
+    // A crash can happen after admission but before an environment lease exists.
+    // Run identity, rather than lease presence, decides whether that active owner
+    // still has work to recover. Warm owners keep their own idle deadline.
+    for (const owner of liveOwners(record.ledger)) {
+      const admittedAt = owner.admittedAt ? Date.parse(owner.admittedAt) : 0;
+      if (
+        owner.kind === "runner" &&
+        owner.phase === "active" &&
+        owner.runId &&
+        owner.agentId &&
+        now().getTime() - admittedAt >= 120_000
+      ) {
+        const runState = await repository.runState(
+          record,
+          owner.runId,
+          owner.agentId,
+        );
+        if (runState !== "active")
+          await retire({ ...record, owner: ref(record, owner) });
+      }
+    }
+    record = await repository.get(record);
+    for (const owner of liveOwners(record.ledger))
+      if (owner.phase === "retiring" || expired(owner, now()))
+        await retire({ ...record, owner: ref(record, owner) });
+    await repository.update(record, (current) => {
+      current.ledger.owners = current.ledger.owners.filter(
+        (owner) => owner.kind !== "file-operation" || owner.phase !== "retired",
+      );
+    });
+    record = await repository.get(record);
+    if (record.ledger.action) {
+      const action = record.ledger.action;
+      // A lost stop response is retried only while admission is fenced. Boat returns its same pending ID.
+      const stop = action.providerStopId
+        ? await backend.stopStatus(record, action.providerStopId)
+        : await backend.stop(record);
+      await repository.update(record, (current) => {
+        if (current.ledger.action?.id !== action.id) return;
+        current.ledger.action.providerStopId = stop.id;
+        if (stop.status === "completed") {
+          current.ledger.action = null;
+          if (current.ledger.status === "detaching")
+            current.ledger.status = "detached";
+        }
+      });
+      return;
+    }
+    if (liveOwners(record.ledger).length) {
+      await backend.renew(record);
+      return;
+    }
+    if (record.ledger.status === "detached") return;
+    const status = await backend.inspect(record);
+    if (["stopped", "archived"].includes(status.state)) {
+      if (record.ledger.status === "detaching")
+        await repository.update(record, (current) => {
+          current.ledger.status = "detached";
+        });
+      return;
+    }
+    const actionId = randomUUID();
+    const claimed = await repository.update(record, (current) => {
+      if (current.ledger.action || liveOwners(current.ledger).length)
+        return false;
+      current.ledger.action = {
+        kind: "stop",
+        id: actionId,
+        providerStopId: null,
+      };
+      return true;
+    });
+    if (!claimed) return;
+    const stop = await backend.stop(record);
+    await repository.update(record, (current) => {
+      if (current.ledger.action?.id === actionId)
+        current.ledger.action.providerStopId = stop.id;
+    });
+  }
+  async function reconcile() {
+    const failures: unknown[] = [];
+    for (const record of await repository.all())
+      try {
+        await reconcileRecord(record);
+      } catch (error) {
+        failures.push(error);
+      }
+    if (failures.length)
+      throw new AggregateError(
+        failures,
+        "Computer reconciliation requires retry",
+      );
+  }
+  return {
+    async list(companyId: string) {
+      return (await repository.all())
+        .filter((record) => record.companyId === companyId)
+        .map((record) => ({
+          computerId: record.id,
+          environmentId: record.environmentId,
+          sandboxId: record.providerId,
+          status: record.ledger.status,
+        }));
+    },
+    attach,
+    admit,
+    admitProbe(
+      input: Scope & {
+        agentId: string;
+        probeId: string;
+        idleTimeoutMs: number;
+      },
+    ) {
+      return admit({ ...input, sessionKey: `probe-${segment(input.probeId)}` });
+    },
+    recover,
+    retainWarm,
+    retire,
+    connect,
+    renewViewer,
+    disconnectViewer,
+    preview,
+    files,
+    workspaceFiles,
+    realizeWorkspace,
+    detach,
+    reconcile,
+    inspect,
+  };
+}