diff --git a/pyproject.toml b/pyproject.toml
index e247166..9b21b97 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -30,6 +30,7 @@ classifiers = [
 
 dependencies = [
   "httpx",
+  "psutil",
 ]
 
 [build-system]
@@ -43,6 +44,7 @@ dev = [
   "pytest-timeout",
   "ruff",
   "mypy",
+  "types-psutil",
   "build",
   "twine",
   "hatchling",
diff --git a/src/opencode_runtime/__init__.py b/src/opencode_runtime/__init__.py
index 50558d0..d526673 100644
--- a/src/opencode_runtime/__init__.py
+++ b/src/opencode_runtime/__init__.py
@@ -10,8 +10,8 @@
 from .session import OpenCodeSession
 
 __all__ = [
-    "OpenCodeRuntime",
-    "OpenCodeSession",
     "OpenCodeEvent",
     "OpenCodeResponse",
+    "OpenCodeRuntime",
+    "OpenCodeSession",
 ]
diff --git a/src/opencode_runtime/cli.py b/src/opencode_runtime/cli.py
index c151f5f..55cc1e9 100644
--- a/src/opencode_runtime/cli.py
+++ b/src/opencode_runtime/cli.py
@@ -17,7 +17,7 @@
 from datetime import datetime, timezone
 from pathlib import Path
 
-from .server import DisplayStatus, ServerManager, _compute_runtime_key
+from .server import RuntimeStatus, ServerManager, _compute_runtime_key
 
 # ---------------------------------------------------------------------------
 # ANSI
@@ -59,45 +59,31 @@ def _home(path: str) -> str:
         return path
 
 
-def _uptime(started_at: str, alive: bool) -> str:
-    try:
-        mins = max(
-            0,
-            int(
-                (datetime.now(timezone.utc) - datetime.fromisoformat(started_at)).total_seconds()
-                // 60
-            ),
-        )
-    except Exception:
-        return "?"
-    return f"Up {mins}m" if alive else f"Dead {mins}m"
-
-
 def _row(label: str, value: str) -> None:
     print(f"  {_cyan(f'{label:<9}')}  {value}")
 
 
-_STATUS_ICONS: dict[DisplayStatus, str] = {
-    DisplayStatus.RUNNING: "●",
-    DisplayStatus.STARTING: "◐",
-    DisplayStatus.UNHEALTHY: "▲",
-    DisplayStatus.STALE: "○",
-    DisplayStatus.FAILED: "✗",
+# None means "claimed but not yet spawned" (ServerStatus.status while pid is None).
+_STATUS_ICONS: dict[RuntimeStatus | None, str] = {
+    None: "◐",
+    RuntimeStatus.RUNNING: "●",
+    RuntimeStatus.UNHEALTHY: "▲",
+    RuntimeStatus.STALE: "○",
 }
 _STATUS_COLORS = {
-    DisplayStatus.RUNNING: _green,
-    DisplayStatus.STARTING: _yellow,
-    DisplayStatus.UNHEALTHY: _red,
-    DisplayStatus.STALE: _dim,
-    DisplayStatus.FAILED: _red,
+    None: _yellow,
+    RuntimeStatus.RUNNING: _green,
+    RuntimeStatus.UNHEALTHY: _red,
+    RuntimeStatus.STALE: _dim,
 }
 
 
-def _status_display(status: DisplayStatus) -> str:
-    """Render a ServerStatus.display value as a coloured icon + label."""
+def _status_display(status: RuntimeStatus | None) -> str:
+    """Render a ServerStatus.status value as a coloured icon + label."""
     icon = _STATUS_ICONS.get(status, "?")
     color = _STATUS_COLORS.get(status, _dim)
-    return color(f"{icon} {status.value}")
+    label = status.value if status is not None else "starting"
+    return color(f"{icon} {label}")
 
 
 # ---------------------------------------------------------------------------
@@ -121,12 +107,11 @@ async def _serve(args: argparse.Namespace) -> None:
     manager = ServerManager()
 
     existing = manager.find(key)
-    if existing is not None:
-        if manager.is_alive(key):
-            sys.exit(
-                _yellow(f"● Server already running  id={existing.key}  pid={existing.pid}\n")
-                + _dim(f"  use: opencode-runtime stop {existing.key}")
-            )
+    if existing is not None and manager.is_alive(key):
+        sys.exit(
+            _yellow(f"● Server already running  id={existing.key}  pid={existing.pid}\n")
+            + _dim(f"  use: opencode-runtime stop {existing.key}")
+        )
 
     server_dir: Path | None = None
     if runtime_dir is not None:
@@ -197,10 +182,10 @@ def cmd_ps(_args: argparse.Namespace) -> None:
 
     for st in statuses:
         e = st.entry
-        status_plain = f"{_STATUS_ICONS.get(st.display, '?')} {st.display.value}"
-        status_coloured = _status_display(st.display)
+        label = st.status.value if st.status is not None else "starting"
+        status_plain = f"{_STATUS_ICONS.get(st.status, '?')} {label}"
+        status_coloured = _status_display(st.status)
 
-        # Compute uptime
         try:
             started = datetime.fromisoformat(e.started_at)
             uptime_secs = int((datetime.now(timezone.utc) - started).total_seconds())
@@ -210,7 +195,7 @@ def cmd_ps(_args: argparse.Namespace) -> None:
                 uptime_str = f"{uptime_secs // 60}m"
             else:
                 uptime_str = f"{uptime_secs // 3600}h"
-        except Exception:
+        except Exception:  # noqa: BLE001 — malformed timestamp, unknown format
             uptime_str = "?"
 
         vals = [e.key, str(e.pid), str(e.port), status_plain, uptime_str]
@@ -261,7 +246,7 @@ def cmd_stop_all(_args: argparse.Namespace) -> None:
     asyncio.run(manager.stop_all())
 
     print(f"{_green(f'✓ Stopped {len(entries)} server(s)')}\n")
-    for e, _ in entries:
+    for e in entries:
         print(f"  {_dim(e.key)}   {_dim(f'pid {e.pid}')}")
 
 
@@ -277,7 +262,14 @@ def cmd_health(args: argparse.Namespace) -> None:
         sys.exit(_red(f"✗ ID {args.key!r} not found in registry"))
     entry = st.entry
 
-    if st.display == DisplayStatus.RUNNING:
+    if st.status is None:
+        try:
+            claimed = datetime.fromisoformat(entry.started_at)
+            age_secs = int((datetime.now(timezone.utc) - claimed).total_seconds())
+            sys.exit(_yellow(f"◐ starting: claimed {age_secs}s ago, health check pending"))
+        except Exception:  # noqa: BLE001 — malformed timestamp, unknown format
+            sys.exit(_yellow("◐ starting: awaiting health check"))
+    elif st.status == RuntimeStatus.RUNNING:
         try:
             result = asyncio.run(manager.health(args.key))
             version = result.get("version")
@@ -286,27 +278,16 @@ def cmd_health(args: argparse.Namespace) -> None:
                 + f"   {_dim(f'version {version}')}"
                 + f"   {_dim(f'http://127.0.0.1:{entry.port}')}"
             )
-        except Exception as exc:
+        except Exception as exc:  # noqa: BLE001 — network error, timeout, etc.
             sys.exit(_red(f"✗ unhealthy: /global/health failed: {exc}"))
-    elif st.display == DisplayStatus.STARTING:
-        try:
-            claimed = datetime.fromisoformat(entry.claimed_at)
-            age_secs = int((datetime.now(timezone.utc) - claimed).total_seconds())
-            sys.exit(_yellow(f"◐ starting: claimed {age_secs}s ago, health check pending"))
-        except Exception:
-            sys.exit(_yellow("◐ starting: awaiting health check"))
-    elif st.display == DisplayStatus.UNHEALTHY:
+    elif st.status == RuntimeStatus.UNHEALTHY:
         sys.exit(
             _red(
                 f"✗ unhealthy: process running (pid {entry.pid}) but /global/health endpoint failed"
             )
         )
-    elif st.display == DisplayStatus.STALE:
-        sys.exit(_red(f"✗ stale: registry entry exists but pid {entry.pid} is not running"))
-    elif st.display == DisplayStatus.FAILED:
-        sys.exit(_red("✗ failed: startup failed or lease expired"))
     else:
-        sys.exit(_red(f"✗ unknown: {st.display}"))
+        sys.exit(_red(f"✗ stale: registry entry exists but pid {entry.pid} is not running"))
 
 
 # ---------------------------------------------------------------------------
@@ -321,7 +302,6 @@ def cmd_inspect(args: argparse.Namespace) -> None:
         sys.exit(_red(f"✗ ID {args.key!r} not found in registry"))
     entry = st.entry
 
-    # Compute uptime
     try:
         started = datetime.fromisoformat(entry.started_at)
         uptime_secs = int((datetime.now(timezone.utc) - started).total_seconds())
@@ -331,28 +311,12 @@ def cmd_inspect(args: argparse.Namespace) -> None:
             uptime = f"{uptime_secs // 60}m {uptime_secs % 60}s"
         else:
             uptime = f"{uptime_secs // 3600}h {(uptime_secs % 3600) // 60}m"
-    except Exception:
+    except Exception:  # noqa: BLE001 — malformed timestamp, unknown format
         uptime = "?"
 
-    # Compute idle time (time since last use)
-    if entry.last_used_at:
-        try:
-            last_used = datetime.fromisoformat(entry.last_used_at)
-            idle_secs = int((datetime.now(timezone.utc) - last_used).total_seconds())
-            if idle_secs < 60:
-                idle = f"{idle_secs}s ago"
-            elif idle_secs < 3600:
-                idle = f"{idle_secs // 60}m ago"
-            else:
-                idle = f"{idle_secs // 3600}h ago"
-        except Exception:
-            idle = "?"
-    else:
-        idle = "-"
-
     print()
     _row("ID", entry.key)
-    _row("Status", _status_display(st.display))
+    _row("Status", _status_display(st.status))
     _row("Project", _home(entry.project_dir))
     if entry.workspace:
         _row("Workspace", entry.workspace)
@@ -361,7 +325,6 @@ def cmd_inspect(args: argparse.Namespace) -> None:
     _row("PID", _dim(str(entry.pid)) if entry.pid else _dim("(none)"))
     _row("Port", _dim(str(entry.port)))
     _row("Uptime", uptime)
-    _row("Last used", idle)
     if entry.runtime_version:
         _row("Runtime", entry.runtime_version)
     if entry.server_dir:
diff --git a/src/opencode_runtime/client.py b/src/opencode_runtime/client.py
index c1d7e5c..2d124ef 100644
--- a/src/opencode_runtime/client.py
+++ b/src/opencode_runtime/client.py
@@ -162,64 +162,63 @@ async def events(self, session_id: str) -> t.AsyncIterator[OpenCodeEvent]:
 
         Only ``sync`` bus-noise events are suppressed.
         """
-        async with self._http() as http:
-            async with http.stream("GET", "/global/event") as r:
+        async with self._http() as http, http.stream("GET", "/global/event") as r:
+            try:
+                r.raise_for_status()
+            except httpx.HTTPStatusError as exc:
+                raise OpenCodeServerError(
+                    f"SSE /global/event returned {exc.response.status_code}"
+                ) from exc
+
+            async for line in r.aiter_lines():
+                line = line.strip()
+
+                if not line.startswith("data:"):
+                    continue
+
+                raw_data = line[len("data:") :].strip()
+                if not raw_data:
+                    continue
+
                 try:
-                    r.raise_for_status()
-                except httpx.HTTPStatusError as exc:
-                    raise OpenCodeServerError(
-                        f"SSE /global/event returned {exc.response.status_code}"
-                    ) from exc
-
-                async for line in r.aiter_lines():
-                    line = line.strip()
-
-                    if not line.startswith("data:"):
-                        continue
-
-                    raw_data = line[len("data:") :].strip()
-                    if not raw_data:
-                        continue
-
-                    try:
-                        envelope = json.loads(raw_data)
-                    except json.JSONDecodeError:
-                        continue
-
-                    # /global/event wraps each event: {"payload": {"type": ..., "properties": ...}}
-                    payload = envelope.get("payload", {})
-                    if not isinstance(payload, dict):
-                        continue
-
-                    event_type = payload.get("type", "")
-
-                    # Suppress internal bus noise — never useful to callers.
-                    if event_type == "sync":
-                        continue
-
-                    props = payload.get("properties", {})
-                    if not isinstance(props, dict):
-                        props = {}
-
-                    # Filter to this session (events without sessionID are global, pass through)
-                    sid = props.get("sessionID")
-                    if sid is not None and sid != session_id:
-                        continue
-
-                    # Derive a convenience text field where it naturally exists,
-                    # so callers don't have to dig into raw for the common case.
-                    text: str | None = None
-                    if event_type == "message.part.delta" and props.get("field") == "text":
-                        text = props.get("delta") or None
-                    elif event_type == "message.part.updated":
-                        part = props.get("part") or {}
-                        if isinstance(part, dict) and part.get("type") == "text":
-                            text = part.get("text") or None
-
-                    yield OpenCodeEvent(type=event_type, text=text, raw=payload)
-
-                    # Terminal events — stop iterating after yielding.
-                    if event_type == "session.idle":
-                        return
-                    if event_type == "session.error":
-                        return
+                    envelope = json.loads(raw_data)
+                except json.JSONDecodeError:
+                    continue
+
+                # /global/event wraps each event: {"payload": {"type": ..., "properties": ...}}
+                payload = envelope.get("payload", {})
+                if not isinstance(payload, dict):
+                    continue
+
+                event_type = payload.get("type", "")
+
+                # Suppress internal bus noise — never useful to callers.
+                if event_type == "sync":
+                    continue
+
+                props = payload.get("properties", {})
+                if not isinstance(props, dict):
+                    props = {}
+
+                # Filter to this session (events without sessionID are global, pass through)
+                sid = props.get("sessionID")
+                if sid is not None and sid != session_id:
+                    continue
+
+                # Derive a convenience text field where it naturally exists,
+                # so callers don't have to dig into raw for the common case.
+                text: str | None = None
+                if event_type == "message.part.delta" and props.get("field") == "text":
+                    text = props.get("delta") or None
+                elif event_type == "message.part.updated":
+                    part = props.get("part") or {}
+                    if isinstance(part, dict) and part.get("type") == "text":
+                        text = part.get("text") or None
+
+                yield OpenCodeEvent(type=event_type, text=text, raw=payload)
+
+                # Terminal events — stop iterating after yielding.
+                if event_type == "session.idle":
+                    return
+                if event_type == "session.error":
+                    return
diff --git a/src/opencode_runtime/exceptions.py b/src/opencode_runtime/exceptions.py
index ab51414..af5e2a7 100644
--- a/src/opencode_runtime/exceptions.py
+++ b/src/opencode_runtime/exceptions.py
@@ -14,9 +14,5 @@ class OpenCodeTimeoutError(OpenCodeRuntimeError):
     """Raised when a health check or request exceeds the allowed timeout."""
 
 
-class RegistrySchemaError(OpenCodeRuntimeError):
-    """Raised when the registry database's schema version can't be handled.
-
-    Either it's newer than this version of opencode-runtime understands, or
-    no migration is registered to bring it up to the current version.
-    """
+class RegistryBusyError(OpenCodeRuntimeError):
+    """Raised when a registry entry's lock can't be acquired."""
diff --git a/src/opencode_runtime/process.py b/src/opencode_runtime/process.py
new file mode 100644
index 0000000..420e6b4
--- /dev/null
+++ b/src/opencode_runtime/process.py
@@ -0,0 +1,115 @@
+"""
+OS process primitives: spawning, identity, signaling, and liveness.
+
+Kept separate from registry.py (which only persists JSON state) and
+server.py (which orchestrates OpenCode-specific startup and health) so
+that everything OS-process-specific lives in one place — and so a future
+non-local backend (Docker, remote) could replace just this module.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import signal
+from typing import Any
+
+import psutil
+
+
+async def spawn(
+    *args: str, cwd: str, env: dict[str, str], output: Any
+) -> asyncio.subprocess.Process:
+    """Start a subprocess as its own process group leader.
+
+    start_new_session=True means kill_group() reaches children the process
+    spawns too, not just the pid we hold.
+    """
+    return await asyncio.create_subprocess_exec(
+        *args,
+        cwd=cwd,
+        env=env,
+        stdout=output,
+        stderr=output,
+        start_new_session=True,
+    )
+
+
+def is_alive(pid: int | None) -> bool:
+    """Return True if pid is set and a non-zombie process with it is running.
+
+    psutil.Process.is_running() returns False for zombie processes, unlike
+    os.kill(pid, 0) which returns True for zombies on Linux.
+    """
+    if pid is None:
+        return False
+    try:
+        proc = psutil.Process(pid)
+        return proc.is_running() and proc.status() != psutil.STATUS_ZOMBIE
+    except psutil.NoSuchProcess:
+        return False
+
+
+def start_time(pid: int | None) -> float | None:
+    """Return pid's process creation time as a Unix timestamp, or None.
+
+    Uses psutil for a portable, zombie-aware result on all platforms.
+    """
+    if pid is None:
+        return None
+    try:
+        return psutil.Process(pid).create_time()
+    except psutil.NoSuchProcess:
+        return None
+
+
+def is_same(pid: int | None, started_at: float | None) -> bool:
+    """Return True if pid is alive and its creation time matches started_at.
+
+    started_at is None for entries written before this field existed —
+    those fall back to a plain liveness check.
+    """
+    if not is_alive(pid):
+        return False
+    if started_at is None:
+        return True
+    return start_time(pid) == started_at
+
+
+def kill_group(pid: int, sig: int) -> bool:
+    """Send sig to pid's process group. Returns False if it's already gone."""
+    try:
+        os.killpg(os.getpgid(pid), sig)
+        return True
+    except (ProcessLookupError, PermissionError):
+        return False
+
+
+async def terminate(process: asyncio.subprocess.Process) -> None:
+    """Terminate a process group gracefully, kill if it doesn't exit within 5s."""
+    if process.returncode is not None:
+        return  # already exited
+    if not kill_group(process.pid, signal.SIGTERM):
+        return  # already dead
+    try:
+        await asyncio.wait_for(process.wait(), timeout=5.0)
+    except asyncio.TimeoutError:
+        kill_group(process.pid, signal.SIGKILL)
+
+
+async def wait_until_dead(pid: int, started_at: float | None, timeout: float) -> bool:
+    """Wait until pid is confirmed gone, or timeout elapses.
+
+    Uses psutil.wait_procs() for efficient OS-level waiting instead of a
+    polling loop. Returns True if the process is gone within timeout.
+    """
+    try:
+        p = psutil.Process(pid)
+        # Verify we have the right generation before waiting.
+        if started_at is not None and p.create_time() != started_at:
+            return True  # already a different process — original is gone
+    except psutil.NoSuchProcess:
+        return True  # already gone
+
+    gone, _ = await asyncio.to_thread(psutil.wait_procs, [p], timeout=timeout)
+    return len(gone) > 0
diff --git a/src/opencode_runtime/registry.py b/src/opencode_runtime/registry.py
index e82489b..53fc9ac 100644
--- a/src/opencode_runtime/registry.py
+++ b/src/opencode_runtime/registry.py
@@ -1,181 +1,191 @@
 """
 Registry for tracking OpenCode instance processes.
 
-Every running instance is a row in a SQLite database at:
-    ~/.opencode-runtime/servers/registry.db
+Every running instance is one file at:
+    ~/.opencode-runtime/servers/<key>.json
 
 Used by both the CLI (opencode-runtime serve/ps/stop) and the library
 (OpenCodeRuntime) — the registry is the shared source of truth for all
 running instances regardless of how they were started.
+
+This module only persists and locks JSON state. It has no opinion on
+whether a pid is actually alive (see process.py) or whether OpenCode is
+healthy (see server.py).
 """
 
 from __future__ import annotations
 
+import json
 import os
-import sqlite3
+import time
+import uuid
+from collections.abc import Generator
 from contextlib import contextmanager
-from dataclasses import asdict, dataclass
+from dataclasses import asdict, dataclass, fields
 from datetime import datetime, timedelta, timezone
-from enum import Enum
 from pathlib import Path
-from typing import Generator
 
-from .schema import SCHEMA, SCHEMA_VERSION, migrate
+from .exceptions import RegistryBusyError
 
 REGISTRY_DIR = Path(
     os.environ.get("OPENCODE_RUNTIME_REGISTRY_DIR")
     or (Path.home() / ".opencode-runtime" / "servers")
 )
 
-# A 'starting' row older than this is treated as abandoned (its starter
-# crashed — SIGKILL, host reboot — before reaching write()/delete()) and is
-# reclaimed by the next claim_starting() call. Comfortably above
-# _wait_healthy's default 60s startup timeout, so a claim only expires once
-# a start attempt has definitely either finished or died without cleaning
-# up after itself.
+# A claim (pid still None) older than this is treated as abandoned — its
+# starter crashed (SIGKILL, host reboot) before finishing _start(). Comfortably
+# above _wait_healthy's default 60s startup timeout.
 _START_LEASE_SECONDS = 90
 
-
-class ServerState(str, Enum):
-    """Server lifecycle state.
-
-    STARTING: claimed startup slot, awaiting health check.
-    RUNNING: process alive, health check passing.
-    STOPPING: shutdown initiated (reserved for future use).
-    FAILED: startup failed or lease expired (reserved for future use).
-    """
-
-    STARTING = "starting"
-    RUNNING = "running"
-    STOPPING = "stopping"
-    FAILED = "failed"
+# A lock file older than this is treated as abandoned — its holder crashed
+# mid read-check-write. Locks here are only ever held across a handful of
+# filesystem syscalls, so a few seconds of staleness tolerance is generous.
+_LOCK_STALE_SECONDS = 5
 
 
 @dataclass
 class RegistryEntry:
     """A server entry in the registry.
 
-    state: ServerState enum value. Display status is derived from state + observed health.
+    pid is None while the key is claimed but the process hasn't been
+    spawned yet. instance_id identifies this process generation, so a
+    delete can be scoped to "this generation only" via delete_if_instance().
+    pid_start_time guards against pid having been reused by an unrelated
+    process — see process.is_same().
     """
 
     key: str
-    state: ServerState
     pid: int | None
     port: int
     password: str
     project_dir: str
     server_dir: str | None
-    started_at: str  # ISO-8601
-    claimed_at: str  # ISO-8601; only meaningful while state == "starting"
+    started_at: str  # ISO-8601; when this generation was claimed
     workspace: str | None = None
     user_id: str | None = None
-    last_used_at: str | None = None
     runtime_version: str | None = None
+    instance_id: str | None = None
+    pid_start_time: float | None = None
 
 
-_INSERT = """
-INSERT INTO servers (key, state, pid, port, password, project_dir, server_dir,
-                     started_at, claimed_at, workspace, user_id, last_used_at,
-                     runtime_version)
-VALUES (:key, :state, :pid, :port, :password, :project_dir, :server_dir,
-        :started_at, :claimed_at, :workspace, :user_id, :last_used_at,
-        :runtime_version)
-"""
+_FIELD_NAMES = {f.name for f in fields(RegistryEntry)}
+
+
+def _path(key: str) -> Path:
+    return REGISTRY_DIR / f"{key}.json"
 
-_UPSERT = _INSERT.replace("INSERT", "INSERT OR REPLACE", 1)
+
+def _deserialize(text: str) -> RegistryEntry:
+    data = json.loads(text)
+    return RegistryEntry(**{k: v for k, v in data.items() if k in _FIELD_NAMES})
 
 
 @contextmanager
-def _connect() -> Generator[sqlite3.Connection, None, None]:
-    """Open the registry database, creating it on first use.
+def _locked(key: str, wait_seconds: float = 1.0) -> Generator[None, None, None]:
+    """Hold a short-lived exclusive lock on key for a read-check-write step.
 
-    Commits on clean exit, rolls back if the body raises. The 5s timeout is
-    SQLite's busy timeout — concurrent writers wait for each other instead
-    of failing immediately.
+    Legitimate holders never keep this past a few syscalls, so genuine
+    contention clears in milliseconds — wait_seconds is generous headroom,
+    not a real operation budget.
     """
     REGISTRY_DIR.mkdir(parents=True, exist_ok=True)
-    db_path = REGISTRY_DIR / "registry.db"
-    is_new = not db_path.exists()
-    conn = sqlite3.connect(db_path, timeout=5.0)
+    lock_path = REGISTRY_DIR / f"{key}.lock"
+    deadline = time.monotonic() + wait_seconds
+    while True:
+        try:
+            os.close(os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600))
+            break
+        except FileExistsError:
+            pass
+        try:
+            stale = time.time() - lock_path.stat().st_mtime > _LOCK_STALE_SECONDS
+        except FileNotFoundError:
+            stale = False
+        if stale:
+            lock_path.unlink(missing_ok=True)  # holder crashed; reclaim
+            continue
+        if time.monotonic() >= deadline:
+            raise RegistryBusyError(f"registry entry {key!r} is locked by another operation")
+        time.sleep(0.01)
     try:
-        conn.row_factory = sqlite3.Row
-        conn.execute(SCHEMA)
-        if is_new:
-            db_path.chmod(0o600)  # entries hold server passwords
-            conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
-        else:
-            current_version = conn.execute("PRAGMA user_version").fetchone()[0]
-            if current_version != SCHEMA_VERSION:
-                migrate(conn, current_version)
-        yield conn
-        conn.commit()
+        yield
     finally:
-        conn.close()
-
-
-def _entry(row: sqlite3.Row) -> RegistryEntry:
-    # Column names match RegistryEntry field names one-to-one.
-    return RegistryEntry(**{name: row[name] for name in row.keys()})
+        lock_path.unlink(missing_ok=True)
 
 
 def write(entry: RegistryEntry) -> None:
     """Write (insert or replace) a registry entry."""
-    with _connect() as conn:
-        conn.execute(_UPSERT, asdict(entry))
+    REGISTRY_DIR.mkdir(parents=True, exist_ok=True)
+    tmp = REGISTRY_DIR / f".{entry.key}.{uuid.uuid4().hex}.tmp"
+    tmp.write_text(json.dumps(asdict(entry)), encoding="utf-8")
+    tmp.chmod(0o600)
+    os.replace(tmp, _path(entry.key))  # atomic — readers never see a partial write
 
 
 def read(key: str) -> RegistryEntry | None:
     """Read a registry entry by key. Returns None if not found."""
-    with _connect() as conn:
-        row = conn.execute("SELECT * FROM servers WHERE key = ?", (key,)).fetchone()
-    return _entry(row) if row is not None else None
+    try:
+        return _deserialize(_path(key).read_text(encoding="utf-8"))
+    except FileNotFoundError:
+        return None
 
 
 def delete(key: str) -> None:
     """Remove a registry entry. No-op if not found."""
-    with _connect() as conn:
-        conn.execute("DELETE FROM servers WHERE key = ?", (key,))
+    _path(key).unlink(missing_ok=True)
 
 
 def list_all() -> list[RegistryEntry]:
     """Return all registry entries."""
-    with _connect() as conn:
-        rows = conn.execute("SELECT * FROM servers").fetchall()
-    return [_entry(row) for row in rows]
+    if not REGISTRY_DIR.exists():
+        return []
+    entries = []
+    for path in REGISTRY_DIR.glob("*.json"):
+        try:
+            entries.append(_deserialize(path.read_text(encoding="utf-8")))
+        except (FileNotFoundError, json.JSONDecodeError):
+            continue  # deleted mid-scan, or a crash left a partial write
+    return entries
 
 
 def claim_starting(entry: RegistryEntry) -> bool:
-    """Atomically insert a 'starting' row for entry.key.
+    """Claim entry.key for a new start attempt.
 
-    Returns True if this call claimed the key, False if a row for the key
-    already exists (a live 'starting' claim or a 'ready' server). A
-    'starting' row older than _START_LEASE_SECONDS is treated as abandoned
-    and reclaimed within the same transaction.
+    Returns True if this call claimed the key, False if a live claim or a
+    running server already occupies it. A claim (pid still None) older
+    than _START_LEASE_SECONDS is treated as abandoned and reclaimed.
     """
-    cutoff = (datetime.now(timezone.utc) - timedelta(seconds=_START_LEASE_SECONDS)).isoformat(
-        timespec="microseconds"
-    )
-    try:
-        with _connect() as conn:
-            conn.execute(
-                "DELETE FROM servers WHERE key = ? AND state = ? AND claimed_at < ?",
-                (entry.key, ServerState.STARTING.value, cutoff),
-            )
-            conn.execute(_INSERT, asdict(entry))
+    with _locked(entry.key):
+        existing = read(entry.key)
+        if existing is not None:
+            if existing.pid is not None:
+                return False
+            claimed_at = datetime.fromisoformat(existing.started_at)
+            if datetime.now(timezone.utc) - claimed_at < timedelta(seconds=_START_LEASE_SECONDS):
+                return False
+        write(entry)
         return True
-    except sqlite3.IntegrityError:  # PRIMARY KEY conflict — someone else holds the key
-        return False
 
 
-def is_alive(pid: int | None) -> bool:
-    """Return True if pid is set and a process with it is running."""
-    if pid is None:
-        return False
-    try:
-        os.kill(pid, 0)
+def write_if_instance(entry: RegistryEntry) -> bool:
+    """Write entry only if its generation still owns the registry key."""
+    with _locked(entry.key):
+        current = read(entry.key)
+        if current is None or current.instance_id != entry.instance_id:
+            return False
+
+        write(entry)
+        return True
+
+
+def delete_if_instance(key: str, instance_id: str | None) -> bool:
+    """Delete the entry for key iff its instance_id matches. Returns whether deleted."""
+    with _locked(key):
+        entry = read(key)
+        if entry is None or entry.instance_id != instance_id:
+            return False
+        delete(key)
         return True
-    except (ProcessLookupError, PermissionError):
-        return False
 
 
 def now_iso() -> str:
diff --git a/src/opencode_runtime/runtime.py b/src/opencode_runtime/runtime.py
index 032ccf4..1e42b37 100644
--- a/src/opencode_runtime/runtime.py
+++ b/src/opencode_runtime/runtime.py
@@ -1,8 +1,11 @@
 from __future__ import annotations
 
+import types
 import typing as t
 from pathlib import Path
 
+from typing_extensions import Self
+
 from .server import ServerManager, _compute_runtime_key
 
 if t.TYPE_CHECKING:
@@ -54,10 +57,15 @@ def __init__(
     # Lifecycle
     # ------------------------------------------------------------------
 
-    async def __aenter__(self) -> OpenCodeRuntime:
+    async def __aenter__(self) -> Self:
         return self
 
-    async def __aexit__(self, exc_type: t.Any, exc: t.Any, tb: t.Any) -> None:
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: types.TracebackType | None,
+    ) -> None:
         await self.close()
 
     async def close(self) -> None:
@@ -128,8 +136,6 @@ async def session(
             user_id=user_id,
         )
 
-        self._server_manager.touch(key)
-
         return OpenCodeSession(
             client=server.client,
             workspace=workspace,
diff --git a/src/opencode_runtime/schema.py b/src/opencode_runtime/schema.py
deleted file mode 100644
index 5717675..0000000
--- a/src/opencode_runtime/schema.py
+++ /dev/null
@@ -1,79 +0,0 @@
-"""
-SQLite schema and migration mechanism for the registry database.
-
-The schema version is stored in the database itself via SQLite's built-in
-`PRAGMA user_version` — no separate metadata table needed. A fresh database
-is stamped with SCHEMA_VERSION directly; an existing one is brought up to it
-by migrate().
-
-Bump SCHEMA_VERSION and add a step to _MIGRATIONS whenever the `servers`
-table shape changes. Never edit SCHEMA in place once a version has shipped —
-existing databases on disk still have the old shape and rely on a migration
-step to reach the new one.
-"""
-
-from __future__ import annotations
-
-import sqlite3
-from typing import Callable
-
-from .exceptions import RegistrySchemaError
-
-SCHEMA = """
-CREATE TABLE IF NOT EXISTS servers (
-    key             TEXT PRIMARY KEY,
-    state           TEXT NOT NULL,
-    pid             INTEGER,
-    port            INTEGER NOT NULL,
-    password        TEXT NOT NULL,
-    project_dir     TEXT NOT NULL,
-    server_dir      TEXT,
-    started_at      TEXT NOT NULL,
-    claimed_at      TEXT NOT NULL,
-    workspace       TEXT,
-    user_id         TEXT,
-    last_used_at    TEXT,
-    runtime_version TEXT
-)
-"""
-
-SCHEMA_VERSION = 1
-
-# Migration steps, keyed by the version they upgrade *from*. Each callable
-# receives the open connection (mid-transaction, table already created) and
-# must leave the database in the shape of `from_version + 1` — e.g. an
-# `ALTER TABLE servers ADD COLUMN ...`.
-#
-# Version 0 covers every database that predates this versioning scheme. Its
-# table shape is identical to version 1 (versioning was introduced with no
-# accompanying column change), so that migration is a no-op — it exists so
-# unversioned databases get stamped with a version the first time they're
-# opened under this scheme.
-_MIGRATIONS: dict[int, Callable[[sqlite3.Connection], None]] = {
-    0: lambda conn: None,
-}
-
-
-def migrate(conn: sqlite3.Connection, current_version: int) -> None:
-    """Bring conn's schema from current_version up to SCHEMA_VERSION, in place.
-
-    Raises RegistrySchemaError if current_version is newer than this code
-    understands, or if a required migration step isn't registered.
-    """
-    if current_version > SCHEMA_VERSION:
-        raise RegistrySchemaError(
-            f"registry database is schema v{current_version}, but this version of "
-            f"opencode-runtime only understands up to v{SCHEMA_VERSION}. Upgrade "
-            "opencode-runtime to use it."
-        )
-    version = current_version
-    while version < SCHEMA_VERSION:
-        step = _MIGRATIONS.get(version)
-        if step is None:
-            raise RegistrySchemaError(
-                f"no migration registered to bring the registry database from schema "
-                f"v{version} to v{version + 1}"
-            )
-        step(conn)
-        version += 1
-    conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
diff --git a/src/opencode_runtime/server.py b/src/opencode_runtime/server.py
index e1edaca..082ab00 100644
--- a/src/opencode_runtime/server.py
+++ b/src/opencode_runtime/server.py
@@ -15,6 +15,7 @@
 import shutil
 import signal
 import socket
+import uuid
 from dataclasses import dataclass
 from enum import Enum
 from pathlib import Path
@@ -27,7 +28,7 @@
 
 import httpx
 
-from . import registry
+from . import process, registry
 from .client import OpenCodeClient
 from .exceptions import (
     OpenCodeNotFoundError,
@@ -35,7 +36,7 @@
     OpenCodeServerError,
     OpenCodeTimeoutError,
 )
-from .registry import RegistryEntry, ServerState
+from .registry import RegistryEntry
 
 # One or more paths to overlay into the server dir before startup.
 # Module-level alias: inside ServerManager's class body, `list[...]` in an
@@ -53,30 +54,27 @@ class _ManagedServer:
     server_dir: Path | None  # None when runtime_dir is not set (no isolation)
 
 
-class DisplayStatus(str, Enum):
-    """User-facing status derived from registry state, process liveness, and health."""
+class RuntimeStatus(str, Enum):
+    """Health of a server that has a pid, derived from OS liveness + OpenCode health.
+
+    A registry entry whose pid is still None (claimed, not yet spawned)
+    has no RuntimeStatus — see ServerStatus.status.
+    """
 
-    STARTING = "starting"
     RUNNING = "running"
     UNHEALTHY = "unhealthy"
     STALE = "stale"
-    FAILED = "failed"
-    STOPPING = "stopping"
 
 
 @dataclass
 class ServerStatus:
-    """Computed liveness/health status for a registry entry."""
-
-    entry: RegistryEntry
-    process_alive: bool
-    health_ok: bool
-    display: DisplayStatus
+    """Computed status for a registry entry.
 
+    status is None while entry.pid hasn't been assigned yet (still starting).
+    """
 
-def _is_process_alive(pid: int | None) -> bool:
-    """Return True if pid is set and a process with it is running."""
-    return registry.is_alive(pid)
+    entry: RegistryEntry
+    status: RuntimeStatus | None
 
 
 async def _is_health_ok(client: OpenCodeClient, timeout: float = 3.0) -> bool:
@@ -84,27 +82,10 @@ async def _is_health_ok(client: OpenCodeClient, timeout: float = 3.0) -> bool:
     try:
         await asyncio.wait_for(client.health(), timeout=timeout)
         return True
-    except Exception:
+    except Exception:  # noqa: BLE001 — any network/timeout/parse error means unhealthy
         return False
 
 
-def _compute_display_status(
-    state: ServerState, process_alive: bool, health_ok: bool, lease_expired: bool = False
-) -> DisplayStatus:
-    """Derive user-facing display status from state, process liveness, and health."""
-    if state == ServerState.STARTING:
-        return DisplayStatus.FAILED if lease_expired else DisplayStatus.STARTING
-    if state == ServerState.STOPPING:
-        return DisplayStatus.STOPPING
-    if state == ServerState.FAILED:
-        return DisplayStatus.FAILED
-    if not process_alive:
-        return DisplayStatus.STALE
-    if not health_ok:
-        return DisplayStatus.UNHEALTHY
-    return DisplayStatus.RUNNING
-
-
 def _find_free_port(host: str = "127.0.0.1") -> int:
     """Bind to port 0 and let the OS pick a free ephemeral port."""
     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
@@ -127,7 +108,7 @@ async def _wait_healthy(
         try:
             await client.health()
             return
-        except Exception as exc:
+        except Exception as exc:  # noqa: BLE001 — any network/timeout/parse error while polling
             last_exc = exc
             await asyncio.sleep(1.0)
 
@@ -165,23 +146,6 @@ def _prepare_dir(
                 shutil.copy2(src, server_dir / src.name)
 
 
-async def _terminate_process(process: asyncio.subprocess.Process) -> None:
-    """Terminate a process gracefully, kill if it doesn't exit within 5s."""
-    if process.returncode is not None:
-        return  # already exited
-    try:
-        process.terminate()
-    except ProcessLookupError:
-        return  # already dead
-    try:
-        await asyncio.wait_for(process.wait(), timeout=5.0)
-    except asyncio.TimeoutError:
-        try:
-            process.kill()
-        except ProcessLookupError:
-            pass
-
-
 def _compute_runtime_key(
     workspace: str | None,
     user_id: str | None,
@@ -235,38 +199,28 @@ def __init__(self) -> None:
     def find(self, key: str) -> RegistryEntry | None:
         """Return the registry entry for key, or None if not found or still starting."""
         entry = registry.read(key)
-        return entry if entry is not None and entry.state == ServerState.RUNNING else None
+        return entry if entry is not None and entry.pid is not None else None
 
     def is_alive(self, key: str) -> bool:
         """Return True if the server for key is running."""
         entry = self.find(key)
-        return entry is not None and registry.is_alive(entry.pid)
-
-    def touch(self, key: str) -> None:
-        """Update last_used_at timestamp for a server. Call after session creation."""
-        entry = registry.read(key)
-        if entry is not None and entry.state == ServerState.RUNNING:
-            entry.last_used_at = registry.now_iso()
-            registry.write(entry)
+        return entry is not None and process.is_same(entry.pid, entry.pid_start_time)
 
     async def _status_for_entry(
         self, entry: RegistryEntry, *, health_timeout: float = 3.0
     ) -> ServerStatus:
         """Derive a ServerStatus for an already-fetched registry entry."""
-        process_alive = (
-            _is_process_alive(entry.pid) if entry.state == ServerState.RUNNING else False
-        )
-        health_ok = False
-        if process_alive:
-            client = OpenCodeClient(
-                base_url=f"http://127.0.0.1:{entry.port}",
-                password=entry.password,
-            )
-            health_ok = await _is_health_ok(client, timeout=health_timeout)
-        display = _compute_display_status(entry.state, process_alive, health_ok)
-        return ServerStatus(
-            entry=entry, process_alive=process_alive, health_ok=health_ok, display=display
+        if entry.pid is None:
+            return ServerStatus(entry=entry, status=None)
+        if not await asyncio.to_thread(process.is_same, entry.pid, entry.pid_start_time):
+            return ServerStatus(entry=entry, status=RuntimeStatus.STALE)
+        client = OpenCodeClient(
+            base_url=f"http://127.0.0.1:{entry.port}",
+            password=entry.password,
         )
+        if not await _is_health_ok(client, timeout=health_timeout):
+            return ServerStatus(entry=entry, status=RuntimeStatus.UNHEALTHY)
+        return ServerStatus(entry=entry, status=RuntimeStatus.RUNNING)
 
     async def status(self, key: str, *, health_timeout: float = 3.0) -> ServerStatus | None:
         """Return the computed status for key, or None if not in the registry."""
@@ -282,19 +236,15 @@ async def status(self, key: str, *, health_timeout: float = 3.0) -> ServerStatus
     # up top for the same issue). list_statuses()'s `-> list[ServerStatus]`
     # return annotation is defined above list() to avoid it.
     async def list_statuses(self, *, health_timeout: float = 1.0) -> list[ServerStatus]:
-        """Return computed status for every ready registry entry."""
+        """Return computed status for every entry that has a pid."""
         return [
             await self._status_for_entry(entry, health_timeout=health_timeout)
-            for entry, _ in self.list()
+            for entry in self.list()
         ]
 
-    def list(self) -> list[tuple[RegistryEntry, bool]]:
-        """Return all ready registry entries with their liveness status."""
-        return [
-            (entry, registry.is_alive(entry.pid))
-            for entry in registry.list_all()
-            if entry.state == ServerState.RUNNING
-        ]
+    def list(self) -> list[RegistryEntry]:
+        """Return every registry entry that has progressed past a bare claim."""
+        return [entry for entry in registry.list_all() if entry.pid is not None]
 
     async def health(self, key: str) -> dict[str, Any]:
         """Return health info for the server at key.
@@ -318,33 +268,37 @@ async def health(self, key: str) -> dict[str, Any]:
     async def stop(self, key: str) -> bool:
         """Kill the server for key and remove its registry entry.
 
-        Returns True if the process was alive and killed, False if it was
-        already dead or not found in the registry.
+        Returns True if the process was alive and was confirmed killed,
+        False if it was already dead, not found, or couldn't be confirmed
+        dead (in which case the entry is left in place, still discoverable,
+        rather than forgotten while the process may still be running).
         """
         entry = registry.read(key)
         if entry is None:
             return False
 
-        registry.delete(key)
+        async def forget() -> None:
+            await asyncio.to_thread(registry.delete_if_instance, key, entry.instance_id)
 
-        if entry.pid is None or not registry.is_alive(entry.pid):
+        if not process.is_same(entry.pid, entry.pid_start_time):
+            await forget()
             return False
 
-        try:
-            os.kill(entry.pid, signal.SIGTERM)
-        except ProcessLookupError:
+        assert entry.pid is not None
+        if not process.kill_group(entry.pid, signal.SIGTERM):
+            await forget()
             return False
-        # Wait for the process to exit (up to 5s, then SIGKILL).
-        deadline = asyncio.get_event_loop().time() + 5.0
-        while asyncio.get_event_loop().time() < deadline:
-            if not registry.is_alive(entry.pid):
-                return True
-            await asyncio.sleep(0.1)
-        try:
-            os.kill(entry.pid, signal.SIGKILL)
-        except ProcessLookupError:
-            pass
-        return True
+
+        if await process.wait_until_dead(entry.pid, entry.pid_start_time, timeout=5.0):
+            await forget()
+            return True
+
+        process.kill_group(entry.pid, signal.SIGKILL)
+        if await process.wait_until_dead(entry.pid, entry.pid_start_time, timeout=2.0):
+            await forget()
+            return True
+
+        return False
 
     async def stop_all(self) -> None:
         """Kill all servers tracked in the registry."""
@@ -375,37 +329,36 @@ async def get_or_start(
     ) -> _ManagedServer:
         """Return a client for the running server, starting one if needed."""
         entry = registry.read(key)
-        if entry is not None and entry.state == ServerState.RUNNING:
-            if registry.is_alive(entry.pid):
+        if entry is not None and entry.pid is not None:
+            if process.is_same(entry.pid, entry.pid_start_time):
                 return self._attach(entry)
-            # Stale — the process died after finishing startup.
-            registry.delete(key)
+            # Stale — the process died after finishing startup. Scoped to
+            # this generation so a concurrent fresh claim isn't clobbered.
+            registry.delete_if_instance(key, entry.instance_id)
 
-        # port/password are generated here (not inside _start) because a
-        # 'starting' claim row needs both up front — pid is the only field
+        # port/password/instance_id are generated here (not inside _start)
+        # because a claim needs them all up front — pid is the only field
         # that isn't known until the subprocess actually exists.
         port = _find_free_port()
         password = secrets.token_urlsafe(32)
+        instance_id = uuid.uuid4().hex
         timestamp = registry.now_iso()
 
         claimed = registry.claim_starting(
             RegistryEntry(
                 key=key,
-                state=ServerState.STARTING,
                 pid=None,
                 port=port,
                 password=password,
                 project_dir=str(project_dir),
                 server_dir=str(server_dir) if server_dir else None,
                 started_at=timestamp,
-                claimed_at=timestamp,
                 workspace=workspace,
                 user_id=user_id,
+                instance_id=instance_id,
             )
         )
         if not claimed:
-            # Another caller is already starting this key — wait for it
-            # instead of racing to spawn a second process for the same key.
             return await self._wait_for_ready(key)
 
         try:
@@ -418,11 +371,13 @@ async def get_or_start(
                 env=env,
                 port=port,
                 password=password,
+                instance_id=instance_id,
+                started_at=timestamp,
                 workspace=workspace,
                 user_id=user_id,
             )
         except Exception:
-            registry.delete(key)
+            registry.delete_if_instance(key, instance_id)
             raise
         self._owned.add(key)
         return server
@@ -448,7 +403,7 @@ async def _wait_for_ready(self, key: str, timeout: float = 60.0) -> _ManagedServ
                 raise OpenCodeServerError(
                     f"the caller starting server {key!r} failed before it became ready"
                 )
-            if entry.state == ServerState.RUNNING and registry.is_alive(entry.pid):
+            if entry.pid is not None and process.is_same(entry.pid, entry.pid_start_time):
                 return self._attach(entry)
             await asyncio.sleep(0.1)
 
@@ -467,6 +422,8 @@ async def _start(
         env: dict[str, str],
         port: int,
         password: str,
+        instance_id: str,
+        started_at: str,
         workspace: str | None = None,
         user_id: str | None = None,
     ) -> _ManagedServer:
@@ -486,11 +443,11 @@ async def _start(
             process_env["HOME"] = str(server_dir)
             process_env["TMPDIR"] = str(server_dir / "tmp")
             process_env["OPENCODE_CONFIG"] = str(server_dir / "opencode.json")
-            output: Any = open(server_dir / "opencode.log", "ab")
+            output: Any = open(server_dir / "opencode.log", "ab")  # noqa: ASYNC230, SIM115
         else:
             output = asyncio.subprocess.DEVNULL
 
-        process = await asyncio.create_subprocess_exec(
+        proc = await process.spawn(
             "opencode",
             "serve",
             "--hostname",
@@ -499,8 +456,7 @@ async def _start(
             str(port),
             cwd=str(project_dir),
             env=process_env,
-            stdout=output,
-            stderr=output,
+            output=output,
         )
 
         client = OpenCodeClient(
@@ -517,33 +473,35 @@ async def _start(
         )
 
         try:
-            await _wait_healthy(poll_client, process=process)
+            await _wait_healthy(poll_client, process=proc)
         except Exception:
-            await _terminate_process(process)
+            await process.terminate(proc)
             raise
 
-        timestamp = registry.now_iso()
-        registry.write(
-            RegistryEntry(
-                key=key,
-                state=ServerState.RUNNING,
-                pid=process.pid,
-                port=port,
-                password=password,
-                project_dir=str(project_dir),
-                server_dir=str(server_dir) if server_dir else None,
-                started_at=timestamp,
-                claimed_at=timestamp,
-                last_used_at=timestamp,
-                runtime_version=version("opencode-runtime"),
-                workspace=workspace,
-                user_id=user_id,
-            )
+        ready_entry = RegistryEntry(
+            key=key,
+            pid=proc.pid,
+            port=port,
+            password=password,
+            project_dir=str(project_dir),
+            server_dir=str(server_dir) if server_dir else None,
+            started_at=started_at,
+            runtime_version=version("opencode-runtime"),
+            workspace=workspace,
+            user_id=user_id,
+            instance_id=instance_id,
+            pid_start_time=process.start_time(proc.pid),
         )
 
+        if not registry.write_if_instance(ready_entry):
+            await process.terminate(proc)
+            raise OpenCodeServerError(
+                f"startup claim for server {key!r} was replaced before startup completed"
+            )
+
         return _ManagedServer(
             key=key,
-            process=process,
+            process=proc,
             client=client,
             server_dir=server_dir,
         )
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 648c1ea..ee900fb 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -14,10 +14,9 @@
 
 import pytest
 
-import opencode_runtime.registry as registry
+from opencode_runtime import process, registry
 from opencode_runtime.cli import cmd_health, cmd_ps, cmd_serve, cmd_stop, cmd_stop_all
-from opencode_runtime.registry import RegistryEntry, ServerState
-
+from opencode_runtime.registry import RegistryEntry
 
 # ---------------------------------------------------------------------------
 # fixtures
@@ -36,19 +35,17 @@ def ns(**kwargs: object) -> argparse.Namespace:
 
 
 def make_entry(**kwargs: object) -> RegistryEntry:
-    defaults: dict[str, object] = dict(
-        key="abc123def456abcd",
-        state=ServerState.RUNNING,
-        pid=os.getpid(),  # alive by default
-        port=54321,
-        password="secret",
-        project_dir="/tmp/project",
-        server_dir=None,
-        started_at="2026-07-05T00:00:00+00:00",
-        claimed_at="2026-07-05T00:00:00+00:00",
-        workspace=None,
-        user_id=None,
-    )
+    defaults: dict[str, object] = {
+        "key": "abc123def456abcd",
+        "pid": os.getpid(),  # alive by default
+        "port": 54321,
+        "password": "secret",
+        "project_dir": "/tmp/project",
+        "server_dir": None,
+        "started_at": "2026-07-05T00:00:00+00:00",
+        "workspace": None,
+        "user_id": None,
+    }
     defaults.update(kwargs)
     return RegistryEntry(**defaults)  # type: ignore[arg-type]
 
@@ -118,10 +115,11 @@ def test_stop_dead_process_warns_and_deletes(capsys):
 
 
 def test_stop_starting_entry_is_removed(capsys):
-    """A STARTING entry (e.g. an orphaned claim) must be stoppable by key,
-    not just RUNNING ones — cmd_stop used to look it up via find(), which
-    filters to RUNNING and reported STARTING entries as "not found"."""
-    registry.write(make_entry(state=ServerState.STARTING, pid=None))
+    """A claim entry (pid still None, e.g. an orphaned claim) must be
+    stoppable by key, not just entries with a pid — cmd_stop used to look
+    it up via find(), which filters to entries with a pid and reported
+    claims as "not found"."""
+    registry.write(make_entry(pid=None))
     cmd_stop(ns(key="abc123def456abcd"))
     out = capsys.readouterr().out
     assert "Server stopped" in out
@@ -135,7 +133,7 @@ def _is_truly_dead(pid: int) -> bool:
     exists in the process table but the process has already exited. We treat
     zombies as dead since we are not the parent and cannot reap them.
     """
-    if not registry.is_alive(pid):
+    if not process.is_alive(pid):
         return True
     try:
         with open(f"/proc/{pid}/status") as f:
@@ -161,7 +159,7 @@ def test_stop_live_server(tmp_path):
     entries = registry.list_all()
     assert len(entries) == 1
     entry = entries[0]
-    assert registry.is_alive(entry.pid)
+    assert process.is_alive(entry.pid)
 
     cmd_stop(ns(key=entry.key))
 
@@ -251,7 +249,7 @@ def test_serve_starts_server(tmp_path):
 
     entries = registry.list_all()
     assert len(entries) == 1
-    assert registry.is_alive(entries[0].pid)
+    assert process.is_alive(entries[0].pid)
 
     # cleanup
     cmd_stop(ns(key=entries[0].key))
@@ -278,9 +276,10 @@ def test_serve_duplicate_key_exits(tmp_path):
 
 def test_serve_stale_entry_cleaned_and_restarted(tmp_path):
     """Dead PID in registry — serve should clean it up and start fresh."""
-    from opencode_runtime.server import _compute_runtime_key
     from pathlib import Path
 
+    from opencode_runtime.server import _compute_runtime_key
+
     # Compute the same key serve will use
     key = _compute_runtime_key(
         workspace=None, user_id=None, project_dir=Path(tmp_path), materials=None, config={}
@@ -298,7 +297,7 @@ def test_serve_stale_entry_cleaned_and_restarted(tmp_path):
 
     entries = registry.list_all()
     assert len(entries) == 1
-    assert registry.is_alive(entries[0].pid)
+    assert process.is_alive(entries[0].pid)
 
     # cleanup
     cmd_stop(ns(key=entries[0].key))
diff --git a/tests/test_multi_tenant.py b/tests/test_multi_tenant.py
index 14d1928..409ecde 100644
--- a/tests/test_multi_tenant.py
+++ b/tests/test_multi_tenant.py
@@ -10,8 +10,7 @@
 
 import pytest
 
-import opencode_runtime.registry as registry
-from opencode_runtime import OpenCodeRuntime
+from opencode_runtime import OpenCodeRuntime, process, registry
 
 pytestmark = pytest.mark.asyncio
 
@@ -104,6 +103,6 @@ async def test_stop_single_tenant(self, tmp_path, monkeypatch):
             await r._server_manager.stop(acme_key)
 
             assert registry.read(acme_key) is None
-            assert not registry.is_alive(acme_entry.pid)
+            assert not process.is_alive(acme_entry.pid)
             assert registry.read(beta_key) is not None
-            assert registry.is_alive(beta_entry.pid)
+            assert process.is_alive(beta_entry.pid)
diff --git a/tests/test_process.py b/tests/test_process.py
new file mode 100644
index 0000000..027faad
--- /dev/null
+++ b/tests/test_process.py
@@ -0,0 +1,53 @@
+"""Tests for the process module (OS process identity and liveness)."""
+
+import os
+
+import pytest
+
+from opencode_runtime import process
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_is_alive_current_process():
+    assert process.is_alive(os.getpid()) is True
+
+
+async def test_is_alive_dead_pid():
+    assert process.is_alive(99999999) is False
+
+
+async def test_is_alive_none_pid():
+    assert process.is_alive(None) is False
+
+
+async def test_start_time_for_current_process():
+    assert process.start_time(os.getpid())
+
+
+async def test_start_time_for_dead_pid():
+    assert process.start_time(99999999) is None
+
+
+async def test_start_time_for_none_pid():
+    assert process.start_time(None) is None
+
+
+async def test_is_same_true_for_matching_pid_and_start_time():
+    started_at = process.start_time(os.getpid())
+    assert process.is_same(os.getpid(), started_at) is True
+
+
+async def test_is_same_false_for_mismatched_start_time():
+    """A pid that's alive but whose start time doesn't match is a different
+    process generation — e.g. the original died and the pid was reused."""
+    assert process.is_same(os.getpid(), 0.0) is False
+
+
+async def test_is_same_false_for_dead_pid():
+    assert process.is_same(99999999, 0.0) is False
+
+
+async def test_is_same_falls_back_to_liveness_when_start_time_unknown():
+    """Entries written before pid_start_time existed have it as None."""
+    assert process.is_same(os.getpid(), None) is True
diff --git a/tests/test_registry.py b/tests/test_registry.py
index 6205520..bd409a9 100644
--- a/tests/test_registry.py
+++ b/tests/test_registry.py
@@ -1,14 +1,17 @@
 """Tests for the registry module."""
 
 import asyncio
+import json
 import os
 import stat
+import time
 from concurrent.futures import ThreadPoolExecutor
 
 import pytest
 
-import opencode_runtime.registry as registry
-from opencode_runtime.registry import RegistryEntry, ServerState
+from opencode_runtime import registry
+from opencode_runtime.exceptions import RegistryBusyError
+from opencode_runtime.registry import RegistryEntry
 
 pytestmark = pytest.mark.asyncio
 
@@ -20,30 +23,26 @@ def isolated_registry(tmp_path, monkeypatch):
 
 
 def make_entry(**kwargs: object) -> RegistryEntry:
-    defaults: dict[str, object] = dict(
-        key="abc123def456abcd",
-        state=ServerState.RUNNING,
-        pid=99999,
-        port=54321,
-        password="secret",
-        project_dir="/tmp/project",
-        server_dir=None,
-        started_at="2026-07-05T00:00:00+00:00",
-        claimed_at="2026-07-05T00:00:00+00:00",
-    )
+    defaults: dict[str, object] = {
+        "key": "abc123def456abcd",
+        "pid": 99999,
+        "port": 54321,
+        "password": "secret",
+        "project_dir": "/tmp/project",
+        "server_dir": None,
+        "started_at": "2026-07-05T00:00:00+00:00",
+    }
     defaults.update(kwargs)
     return RegistryEntry(**defaults)  # type: ignore[arg-type]
 
 
 def make_claim(**kwargs: object) -> RegistryEntry:
-    """A 'starting' entry, as claim_starting() expects — pid is unknown yet.
+    """A claim entry, as claim_starting() expects — pid is unknown yet.
 
-    claimed_at defaults to now (not make_entry()'s fixed placeholder date),
+    started_at defaults to now (not make_entry()'s fixed placeholder date),
     since claim_starting()'s lease check compares it against the real clock.
     """
-    defaults: dict[str, object] = dict(
-        pid=None, state=ServerState.STARTING, claimed_at=registry.now_iso()
-    )
+    defaults: dict[str, object] = {"pid": None, "started_at": registry.now_iso()}
     defaults.update(kwargs)
     return make_entry(**defaults)
 
@@ -84,13 +83,12 @@ async def test_write_twice_replaces_entry():
 
 
 async def test_write_with_null_pid():
-    """A 'starting' row has no pid yet."""
-    entry = make_entry(state=ServerState.STARTING, pid=None)
+    """A claim entry has no pid yet."""
+    entry = make_entry(pid=None)
     registry.write(entry)
     result = registry.read(entry.key)
     assert result is not None
     assert result.pid is None
-    assert result.state == ServerState.STARTING
 
 
 # ---------------------------------------------------------------------------
@@ -129,7 +127,6 @@ async def test_delete_is_noop_for_missing_key():
 
 
 async def test_list_all_empty_when_no_registry_dir():
-    # No db file created yet
     assert registry.list_all() == []
 
 
@@ -147,11 +144,12 @@ async def test_list_all_returns_all_entries():
 # ---------------------------------------------------------------------------
 
 
-async def test_db_file_created_with_permissions_600():
-    registry.write(make_entry())
-    db_path = registry.REGISTRY_DIR / "registry.db"
-    assert db_path.exists()
-    mode = stat.S_IMODE(db_path.stat().st_mode)
+async def test_entry_file_created_with_permissions_600():
+    entry = make_entry()
+    registry.write(entry)
+    path = registry.REGISTRY_DIR / f"{entry.key}.json"
+    assert path.exists()
+    mode = stat.S_IMODE(path.stat().st_mode)
     assert mode == 0o600
 
 
@@ -165,7 +163,6 @@ async def test_claim_starting_succeeds_for_new_key():
     assert registry.claim_starting(entry) is True
     result = registry.read(entry.key)
     assert result is not None
-    assert result.state == ServerState.STARTING
     assert result.pid is None
 
 
@@ -211,29 +208,96 @@ async def test_claim_starting_does_not_reclaim_before_lease_expires():
     assert registry.claim_starting(make_claim(port=55555)) is False
 
 
-async def test_claim_starting_does_not_reclaim_ready_row():
-    """A live 'running' row isn't touched by claim_starting's lease logic."""
-    entry = make_entry()  # state=ServerState.RUNNING
+async def test_claim_starting_does_not_reclaim_entry_with_pid():
+    """An entry that already has a pid isn't touched by the lease logic,
+    no matter how old its started_at is."""
+    entry = make_entry()  # has a pid
     registry.write(entry)
     assert registry.claim_starting(make_claim(port=55555)) is False
     result = registry.read(entry.key)
     assert result is not None
-    assert result.state == ServerState.RUNNING
+    assert result.pid == entry.pid
     assert result.port == entry.port
 
 
 # ---------------------------------------------------------------------------
-# is_alive
+# delete_if_instance
+# ---------------------------------------------------------------------------
+
+
+async def test_delete_if_instance_deletes_on_match():
+    entry = make_entry(instance_id="gen-1")
+    registry.write(entry)
+    assert registry.delete_if_instance(entry.key, "gen-1") is True
+    assert registry.read(entry.key) is None
+
+
+async def test_delete_if_instance_leaves_mismatched_generation():
+    entry = make_entry(instance_id="gen-2")
+    registry.write(entry)
+    assert registry.delete_if_instance(entry.key, "gen-1") is False
+    assert registry.read(entry.key) is not None
+
+
+async def test_delete_if_instance_false_for_missing_key():
+    assert registry.delete_if_instance("doesnotexist", "gen-1") is False
+
+
+async def test_delete_if_instance_reclaims_stale_lock():
+    """A lock file left behind by a crashed holder doesn't wedge the registry."""
+    entry = make_entry(instance_id="gen-1")
+    registry.write(entry)
+    lock_path = registry.REGISTRY_DIR / f"{entry.key}.lock"
+    lock_path.write_text("", encoding="utf-8")
+    stale = time.time() - registry._LOCK_STALE_SECONDS - 1
+    os.utime(lock_path, (stale, stale))
+
+    assert registry.delete_if_instance(entry.key, "gen-1") is True
+    assert registry.read(entry.key) is None
+
+
+async def test_delete_if_instance_raises_when_genuinely_locked():
+    entry = make_entry(instance_id="gen-1")
+    registry.write(entry)
+    lock_path = registry.REGISTRY_DIR / f"{entry.key}.lock"
+    lock_path.write_text("", encoding="utf-8")
+
+    with pytest.raises(RegistryBusyError):
+        registry.delete_if_instance(entry.key, "gen-1")
+
+
+# ---------------------------------------------------------------------------
+# forward/backward compatibility
 # ---------------------------------------------------------------------------
 
 
-async def test_is_alive_current_process():
-    assert registry.is_alive(os.getpid()) is True
+async def test_read_tolerates_unknown_fields():
+    """A file written by a newer version can carry fields this version
+    doesn't know about — they should be ignored, not raise."""
+    entry = make_entry()
+    registry.write(entry)
+    path = registry.REGISTRY_DIR / f"{entry.key}.json"
+    data = json.loads(path.read_text(encoding="utf-8"))
+    data["from_a_future_version"] = "whatever"
+    path.write_text(json.dumps(data), encoding="utf-8")
 
+    result = registry.read(entry.key)
+    assert result is not None
+    assert result.key == entry.key
 
-async def test_is_alive_dead_pid():
-    assert registry.is_alive(99999999) is False
 
+async def test_read_defaults_fields_missing_from_older_writer():
+    """A file written before instance_id/pid_start_time existed should load
+    fine, with those fields defaulting to None."""
+    entry = make_entry()
+    registry.write(entry)
+    path = registry.REGISTRY_DIR / f"{entry.key}.json"
+    data = json.loads(path.read_text(encoding="utf-8"))
+    del data["instance_id"]
+    del data["pid_start_time"]
+    path.write_text(json.dumps(data), encoding="utf-8")
 
-async def test_is_alive_none_pid():
-    assert registry.is_alive(None) is False
+    result = registry.read(entry.key)
+    assert result is not None
+    assert result.instance_id is None
+    assert result.pid_start_time is None
diff --git a/tests/test_runtime.py b/tests/test_runtime.py
index cce5852..e3c8cc4 100644
--- a/tests/test_runtime.py
+++ b/tests/test_runtime.py
@@ -8,8 +8,7 @@
 
 import pytest
 
-import opencode_runtime.registry as registry
-from opencode_runtime import OpenCodeRuntime
+from opencode_runtime import OpenCodeRuntime, process, registry
 
 pytestmark = pytest.mark.asyncio
 
@@ -45,7 +44,7 @@ async def test_server_running_after_start(self, tmp_path, monkeypatch):
         await r.session()
         entries = registry.list_all()
         assert len(entries) == 1
-        assert registry.is_alive(entries[0].pid)
+        assert process.is_alive(entries[0].pid)
         await r.close()
 
     async def test_client_reachable_after_start(self, runtime):
diff --git a/tests/test_schema.py b/tests/test_schema.py
deleted file mode 100644
index db9e63a..0000000
--- a/tests/test_schema.py
+++ /dev/null
@@ -1,106 +0,0 @@
-"""Tests for the registry's schema versioning and migration mechanism."""
-
-import sqlite3
-
-import pytest
-
-import opencode_runtime.registry as registry
-from opencode_runtime import schema
-from opencode_runtime.exceptions import RegistrySchemaError
-from opencode_runtime.registry import RegistryEntry, ServerState
-
-
-@pytest.fixture(autouse=True)
-def isolated_registry(tmp_path, monkeypatch):
-    """Redirect REGISTRY_DIR to a temp path for every test."""
-    monkeypatch.setattr(registry, "REGISTRY_DIR", tmp_path / "servers")
-
-
-def make_entry(**kwargs: object) -> RegistryEntry:
-    defaults: dict[str, object] = dict(
-        key="abc123def456abcd",
-        state=ServerState.RUNNING,
-        pid=99999,
-        port=54321,
-        password="secret",
-        project_dir="/tmp/project",
-        server_dir=None,
-        started_at="2026-07-05T00:00:00+00:00",
-        claimed_at="2026-07-05T00:00:00+00:00",
-    )
-    defaults.update(kwargs)
-    return RegistryEntry(**defaults)  # type: ignore[arg-type]
-
-
-def _db_path(tmp_path):
-    return tmp_path / "servers" / "registry.db"
-
-
-def _user_version(db_path) -> int:
-    conn = sqlite3.connect(db_path)
-    try:
-        return conn.execute("PRAGMA user_version").fetchone()[0]
-    finally:
-        conn.close()
-
-
-def test_fresh_database_is_stamped_with_current_version(tmp_path):
-    registry.write(make_entry())
-    assert _user_version(_db_path(tmp_path)) == schema.SCHEMA_VERSION
-
-
-def test_legacy_unversioned_database_is_migrated_on_open(tmp_path):
-    """A database created before versioning existed has user_version 0 by
-    default. Opening it through the registry should bring it up to the
-    current version without losing data or erroring."""
-    db_path = _db_path(tmp_path)
-    db_path.parent.mkdir(parents=True)
-    conn = sqlite3.connect(db_path)
-    try:
-        conn.execute(schema.SCHEMA)
-        conn.execute(
-            "INSERT INTO servers (key, state, pid, port, password, project_dir, "
-            "server_dir, started_at, claimed_at) VALUES "
-            "('legacykey00000000', 'running', 123, 4096, 'pw', '/tmp/p', NULL, "
-            "'2026-07-05T00:00:00+00:00', '2026-07-05T00:00:00+00:00')"
-        )
-        conn.commit()
-    finally:
-        conn.close()
-    assert _user_version(db_path) == 0
-
-    entry = registry.read("legacykey00000000")
-    assert entry is not None
-    assert entry.pid == 123
-    assert _user_version(db_path) == schema.SCHEMA_VERSION
-
-
-def test_future_schema_version_raises(tmp_path):
-    """A database written by a newer opencode-runtime must not be silently
-    treated as compatible — it should fail loudly instead of risking data
-    loss or a confusing runtime error deeper in the stack."""
-    db_path = _db_path(tmp_path)
-    db_path.parent.mkdir(parents=True)
-    conn = sqlite3.connect(db_path)
-    try:
-        conn.execute(schema.SCHEMA)
-        conn.execute(f"PRAGMA user_version = {schema.SCHEMA_VERSION + 1}")
-        conn.commit()
-    finally:
-        conn.close()
-
-    with pytest.raises(RegistrySchemaError):
-        registry.read("doesnotexist")
-
-
-def test_migrate_raises_on_missing_step(monkeypatch):
-    """If SCHEMA_VERSION is bumped without registering the matching migration
-    step, migrate() must fail loudly rather than silently skip a step."""
-    monkeypatch.setattr(schema, "SCHEMA_VERSION", 2)
-    with pytest.raises(RegistrySchemaError):
-        schema.migrate(sqlite3.connect(":memory:"), current_version=0)
-
-
-def test_migrate_raises_on_newer_version():
-    with pytest.raises(RegistrySchemaError):
-        schema.migrate(sqlite3.connect(":memory:"), current_version=schema.SCHEMA_VERSION + 1)
diff --git a/tests/test_server.py b/tests/test_server.py
index 07f77a0..d1f2c22 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -10,12 +10,11 @@
 
 import pytest
 
-import opencode_runtime.registry as registry
-from opencode_runtime.registry import ServerState
+from opencode_runtime import process, registry
 from opencode_runtime.server import (
     ServerManager,
-    _ManagedServer,
     _compute_runtime_key,
+    _ManagedServer,
 )
 
 
@@ -97,7 +96,7 @@ async def test_get_or_start_starts_server(self, tmp_path):
         )
         entry = registry.read(key)
         assert entry is not None
-        assert registry.is_alive(entry.pid)
+        assert process.is_alive(entry.pid)
         assert server.client is not None
         await manager.stop_all()
 
@@ -176,8 +175,8 @@ async def test_stop_all_terminates_all(self, tmp_path):
 
         await manager.stop_all()
 
-        assert not registry.is_alive(e1_before.pid)
-        assert not registry.is_alive(e2_before.pid)
+        assert not process.is_alive(e1_before.pid)
+        assert not process.is_alive(e2_before.pid)
         assert registry.read(k1) is None
         assert registry.read(k2) is None
 
@@ -226,9 +225,9 @@ async def test_stop_single_server(self, tmp_path):
 
         await manager.stop(k1)
         assert registry.read(k1) is None
-        assert not registry.is_alive(e1.pid)
+        assert not process.is_alive(e1.pid)
         assert registry.read(k2) is not None  # still running
-        assert registry.is_alive(e2.pid)
+        assert process.is_alive(e2.pid)
         await manager.stop_all()
 
     async def test_stop_nonexistent_key_is_noop(self, tmp_path):
@@ -252,7 +251,7 @@ async def test_start_writes_registry_entry(self, tmp_path, monkeypatch):
         )
         entry = registry.read(key)
         assert entry is not None
-        assert registry.is_alive(entry.pid)
+        assert process.is_alive(entry.pid)
         assert entry.port == int(server.client.base_url.split(":")[-1])
         await manager.stop_all()
 
@@ -301,7 +300,7 @@ async def test_attaches_to_alive_registry_entry(self, tmp_path, monkeypatch):
         # manager2 stop kills the process and deletes registry
         await manager2.stop(key)
         assert registry.read(key) is None
-        assert not registry.is_alive(e1.pid)
+        assert not process.is_alive(e1.pid)
 
         # manager1 stop is now a no-op (already gone)
         await manager1.stop_all()
@@ -316,14 +315,12 @@ async def test_stale_registry_entry_cleaned_on_attach(self, tmp_path, monkeypatc
         registry.write(
             RegistryEntry(
                 key=key,
-                state=ServerState.RUNNING,
                 pid=99999999,
                 port=54321,
                 password="stale",
                 project_dir=str(tmp_path),
                 server_dir=None,
                 started_at=timestamp,
-                claimed_at=timestamp,
             )
         )
 
@@ -339,7 +336,7 @@ async def test_stale_registry_entry_cleaned_on_attach(self, tmp_path, monkeypatc
         # Fresh server was spawned — registry has a new alive entry
         entry = registry.read(key)
         assert entry is not None
-        assert registry.is_alive(entry.pid)
+        assert process.is_alive(entry.pid)
         assert entry.pid != 99999999
         assert entry.port == int(server.client.base_url.split(":")[-1])
         await manager.stop_all()
@@ -398,7 +395,7 @@ async def test_server_restarted_after_external_kill(self, tmp_path, monkeypatch)
         new_entry = registry.read(key)
         assert new_entry is not None
         assert new_entry.pid != old_pid
-        assert registry.is_alive(new_entry.pid)
+        assert process.is_alive(new_entry.pid)
         await manager.stop_all()
 
 
@@ -468,8 +465,7 @@ async def test_list_returns_entries_with_liveness(self, tmp_path, monkeypatch):
         )
         entries = manager.list()
         assert len(entries) == 2
-        assert all(alive is True for _, alive in entries)
-        keys = {e.key for e, _ in entries}
+        keys = {e.key for e in entries}
         assert k1 in keys and k2 in keys
         await manager.stop_all()
 
@@ -529,14 +525,12 @@ async def test_stop_returns_false_for_dead_process(self, tmp_path, monkeypatch):
         registry.write(
             RegistryEntry(
                 key=key,
-                state=ServerState.RUNNING,
                 pid=99999999,
                 port=54321,
                 password="x",
                 project_dir=str(tmp_path),
                 server_dir=None,
                 started_at=timestamp,
-                claimed_at=timestamp,
             )
         )
         was_alive = await ServerManager().stop(key)
@@ -559,14 +553,12 @@ async def fake_start(_self, *, port, password, **kwargs):
             registry.write(
                 RegistryEntry(
                     key=key,
-                    state=ServerState.RUNNING,
                     pid=os.getpid(),
                     port=port,
                     password=password,
                     project_dir=str(tmp_path),
                     server_dir=None,
                     started_at=timestamp,
-                    claimed_at=timestamp,
                 )
             )
             return _ManagedServer(key=key, process=None, client=None, server_dir=None)  # type: ignore[arg-type]
@@ -647,7 +639,7 @@ async def always_fails(_self, **kwargs):
         monkeypatch.setattr(ServerManager, "_start", always_fails)
         manager = ServerManager()
 
-        with pytest.raises(Exception):
+        with pytest.raises(Exception):  # noqa: B017
             await asyncio.gather(
                 manager.get_or_start(
                     key=key,