coord/store.py

#!/usr/bin/env python3
"""Transactional storage for inter-session coordination.

This module is deliberately independent of the ``coord`` command so it can be
integrated incrementally by the CLI, lifecycle hook, and write-veto hook.

The important identity distinction is:

``conversation_id``
    The durable Claude/Codex conversation or thread identifier. A resumed
    conversation retains this identifier.

``runtime_id``
    A UUID for one live process attachment to that conversation. Every resume
    registers a fresh runtime, so a stale process cannot impersonate the
    resumed one or accidentally renew its leases.

Lease keys are generic resource names, not necessarily filesystem paths. For
example, callers may claim ``integration/master``, ``push/origin/master``,
``activate/home/tpp15s``, or ``deploy/system/nix-control`` with the same atomic
semantics as ``tests/checks.nix``.

All read/check/write decisions that must be atomic use ``BEGIN IMMEDIATE``.
SQLite serializes those transactions even when callers are separate processes.
WAL mode lets readers continue while a writer commits. Runtime state is local:
the containing directory is forced to 0700 and database files to 0600.
"""

from __future__ import annotations

import contextlib
import dataclasses
import hashlib
import hmac
import json
import os
import pathlib
import sqlite3
import time
import uuid
from collections.abc import Callable, Iterable, Iterator, Sequence
from typing import Any, TextIO


SCHEMA_VERSION = 1
DEFAULT_INSTANCE_TTL = 900.0
DEFAULT_LEASE_TTL = 3600.0
VALID_LEASE_MODES = frozenset(("exclusive", "append"))

# Human-facing handles deliberately have much less entropy than the UUIDs they
# represent: they are exact, collision-checked aliases, not security tokens.
# Five 7-bit words provide a 35-bit display namespace while normally costing
# far fewer model tokens than a 36-character UUID. Full IDs remain canonical
# in SQLite, tmux metadata, hook environment variables, and audit exports.
HANDLE_WORDS = (
    "acorn", "alpine", "amber", "apple", "arrow", "ash", "atlas", "aurora",
    "autumn", "badger", "bamboo", "bay", "beacon", "bear", "birch", "blue",
    "breeze", "brook", "cedar", "cloud", "coast", "coral", "crane", "creek",
    "dawn", "delta", "dune", "eagle", "earth", "ember", "fern", "field",
    "finch", "flame", "flora", "fog", "forest", "fox", "frost", "garden",
    "glade", "gold", "grove", "gull", "harbor", "hare", "hawk", "hazel",
    "hill", "ice", "iris", "island", "ivy", "jade", "jay", "lake",
    "lark", "leaf", "lemon", "light", "lily", "lion", "lotus", "maple",
    "marsh", "meadow", "mint", "moon", "moss", "north", "oak", "ocean",
    "olive", "orchid", "otter", "owl", "panda", "path", "peach", "pear",
    "pine", "plum", "pond", "poppy", "rain", "raven", "reed", "ridge",
    "river", "robin", "rock", "rose", "ruby", "sage", "sand", "seal",
    "shadow", "shore", "silver", "sky", "snow", "south", "sparrow", "spring",
    "star", "stone", "storm", "sun", "swift", "tide", "tiger", "trail",
    "tree", "tulip", "valley", "violet", "wave", "west", "whale", "willow",
    "wind", "wing", "winter", "wolf", "wood", "wren", "yarrow", "zenith",
)


def friendly_handle(value: str, namespace: str) -> str:
    """Return a stable five-word exact alias for an opaque internal ID."""
    if not value:
        raise ValueError("cannot derive a handle from an empty identity")
    if len(HANDLE_WORDS) != 128 or len(set(HANDLE_WORDS)) != 128:
        raise RuntimeError("friendly handle vocabulary must contain 128 unique words")
    digest = hashlib.sha256(f"{namespace}\0{value}".encode("utf-8")).digest()
    bits = int.from_bytes(digest[:5], "big") >> 5
    indices = [((bits >> shift) & 0x7F) for shift in (28, 21, 14, 7, 0)]
    return "-".join(HANDLE_WORDS[index] for index in indices)


@dataclasses.dataclass(frozen=True)
class LeaseConflict:
    path: str
    runtime_id: str
    mode: str
    purpose: str
    expires_at: float


class ClaimRefused(RuntimeError):
    """An atomic lease claim conflicted; no path in the request was claimed."""

    def __init__(self, conflicts: Sequence[LeaseConflict]):
        self.conflicts = tuple(conflicts)
        detail = ", ".join(
            f"{item.path} ({item.mode} by {item.runtime_id})" for item in conflicts
        )
        super().__init__(f"lease claim refused: {detail}")


@dataclasses.dataclass(frozen=True)
class Delivery:
    message_id: int
    sender_runtime_id: str
    kind: str
    body: str
    refs: tuple[str, ...]
    created_at: float
    state: str


class CoordStore:
    """Small process-safe coordination database.

    A store object holds no long-lived connection and is safe to share between
    threads. Each operation opens its own connection. Independent processes may
    construct their own objects pointed at the same database.
    """

    def __init__(
        self,
        db_path: str | os.PathLike[str],
        *,
        clock: Callable[[], float] = time.time,
        timeout: float = 30.0,
    ) -> None:
        self.path = pathlib.Path(db_path)
        self.clock = clock
        self.timeout = timeout
        self._secure_parent()
        self._initialize()

    def _secure_parent(self) -> None:
        self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
        os.chmod(self.path.parent, 0o700)

    def _secure_database_files(self) -> None:
        for suffix in ("", "-wal", "-shm"):
            candidate = pathlib.Path(f"{self.path}{suffix}")
            with contextlib.suppress(FileNotFoundError):
                os.chmod(candidate, 0o600)

    def _connect(self) -> sqlite3.Connection:
        conn = sqlite3.connect(
            self.path,
            timeout=self.timeout,
            isolation_level=None,
        )
        conn.row_factory = sqlite3.Row
        conn.execute("PRAGMA foreign_keys = ON")
        conn.execute("PRAGMA busy_timeout = %d" % int(self.timeout * 1000))
        self._secure_database_files()
        return conn

    @contextlib.contextmanager
    def _transaction(self) -> Iterator[sqlite3.Connection]:
        conn = self._connect()
        try:
            conn.execute("BEGIN IMMEDIATE")
            yield conn
            conn.commit()
        except BaseException:
            conn.rollback()
            raise
        finally:
            conn.close()
            self._secure_database_files()

    def _initialize(self) -> None:
        # journal_mode cannot be changed while a transaction is active. Set it
        # once before creating the schema; the mode persists in the database.
        conn = self._connect()
        try:
            mode = conn.execute("PRAGMA journal_mode = WAL").fetchone()[0]
            if str(mode).lower() != "wal":
                raise RuntimeError(f"could not enable SQLite WAL mode (got {mode!r})")
            conn.execute("PRAGMA synchronous = FULL")
        finally:
            conn.close()
            self._secure_database_files()
        with self._transaction() as conn:
            conn.executescript(
                """
                CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                );

                CREATE TABLE IF NOT EXISTS instances (
                    runtime_id TEXT PRIMARY KEY,
                    conversation_id TEXT NOT NULL,
                    harness TEXT NOT NULL,
                    host TEXT NOT NULL,
                    cwd TEXT NOT NULL,
                    label TEXT NOT NULL,
                    registered_at REAL NOT NULL,
                    heartbeat_at REAL NOT NULL,
                    heartbeat_expires_at REAL NOT NULL,
                    ended_at REAL
                );
                CREATE INDEX IF NOT EXISTS instances_conversation
                    ON instances(conversation_id);

                CREATE TABLE IF NOT EXISTS workspace_records (
                    instance_id TEXT PRIMARY KEY,
                    record_json TEXT NOT NULL,
                    updated_at REAL NOT NULL
                );

                CREATE TABLE IF NOT EXISTS leases (
                    path TEXT NOT NULL,
                    runtime_id TEXT NOT NULL REFERENCES instances(runtime_id),
                    mode TEXT NOT NULL CHECK(mode IN ('exclusive', 'append')),
                    purpose TEXT NOT NULL,
                    claimed_at REAL NOT NULL,
                    renewed_at REAL NOT NULL,
                    expires_at REAL NOT NULL,
                    PRIMARY KEY(path, runtime_id)
                );
                CREATE INDEX IF NOT EXISTS leases_expiry ON leases(expires_at);

                CREATE TABLE IF NOT EXISTS messages (
                    message_id INTEGER PRIMARY KEY AUTOINCREMENT,
                    sender_runtime_id TEXT NOT NULL REFERENCES instances(runtime_id),
                    requested_recipient TEXT NOT NULL,
                    kind TEXT NOT NULL,
                    body TEXT NOT NULL,
                    refs_json TEXT NOT NULL,
                    created_at REAL NOT NULL
                );

                CREATE TABLE IF NOT EXISTS message_deliveries (
                    message_id INTEGER NOT NULL REFERENCES messages(message_id),
                    recipient_runtime_id TEXT NOT NULL REFERENCES instances(runtime_id),
                    state TEXT NOT NULL CHECK(state IN ('queued', 'delivered', 'acked')),
                    delivered_at REAL,
                    acked_at REAL,
                    PRIMARY KEY(message_id, recipient_runtime_id)
                );
                CREATE INDEX IF NOT EXISTS deliveries_inbox
                    ON message_deliveries(recipient_runtime_id, state, message_id);

                CREATE TABLE IF NOT EXISTS audit_events (
                    event_id INTEGER PRIMARY KEY AUTOINCREMENT,
                    occurred_at REAL NOT NULL,
                    event TEXT NOT NULL,
                    runtime_id TEXT,
                    details_json TEXT NOT NULL
                );
                """
            )
            row = conn.execute(
                "SELECT value FROM metadata WHERE key = 'schema_version'"
            ).fetchone()
            if row is None:
                conn.execute(
                    "INSERT INTO metadata(key, value) VALUES('schema_version', ?)",
                    (str(SCHEMA_VERSION),),
                )
            elif int(row["value"]) != SCHEMA_VERSION:
                raise RuntimeError(
                    f"unsupported coordination schema {row['value']}; "
                    f"expected {SCHEMA_VERSION}"
                )

    @staticmethod
    def _normalize_paths(paths: Iterable[str]) -> tuple[str, ...]:
        normalized = tuple(
            dict.fromkeys(os.path.normpath(path.strip()) for path in paths if path.strip())
        )
        if not normalized:
            raise ValueError("at least one non-empty path is required")
        if any(path == "." or os.path.isabs(path) or path.startswith("../") for path in normalized):
            raise ValueError("lease paths must be repository-relative and may not escape it")
        return normalized

    @staticmethod
    def resources_overlap(left: str, right: str) -> bool:
        """Whether two exact resource names are equal or one contains the other."""
        return (
            left == right
            or left.startswith(right.rstrip("/") + "/")
            or right.startswith(left.rstrip("/") + "/")
        )

    @staticmethod
    def _audit(
        conn: sqlite3.Connection,
        now: float,
        event: str,
        runtime_id: str | None,
        details: dict[str, Any],
    ) -> None:
        conn.execute(
            """
            INSERT INTO audit_events(occurred_at, event, runtime_id, details_json)
            VALUES(?, ?, ?, ?)
            """,
            (now, event, runtime_id, json.dumps(details, sort_keys=True, separators=(",", ":"))),
        )

    @staticmethod
    def _require_live(
        conn: sqlite3.Connection, runtime_id: str, now: float
    ) -> sqlite3.Row:
        row = conn.execute(
            """
            SELECT * FROM instances
            WHERE runtime_id = ? AND ended_at IS NULL AND heartbeat_expires_at > ?
            """,
            (runtime_id, now),
        ).fetchone()
        if row is None:
            raise KeyError(f"runtime instance is not registered and live: {runtime_id}")
        return row

    @staticmethod
    def _reap(conn: sqlite3.Connection, now: float) -> int:
        cursor = conn.execute(
            """
            DELETE FROM leases
            WHERE expires_at <= ?
               OR runtime_id IN (
                   SELECT runtime_id FROM instances
                   WHERE ended_at IS NOT NULL OR heartbeat_expires_at <= ?
               )
            """,
            (now, now),
        )
        return cursor.rowcount

    def register_instance(
        self,
        conversation_id: str,
        *,
        runtime_id: str | None = None,
        harness: str = "",
        host: str = "",
        cwd: str = "",
        label: str = "",
        ttl_seconds: float = DEFAULT_INSTANCE_TTL,
    ) -> str:
        if not conversation_id:
            raise ValueError("conversation_id is required")
        if ttl_seconds <= 0:
            raise ValueError("ttl_seconds must be positive")
        runtime_id = runtime_id or str(uuid.uuid4())
        if runtime_id == conversation_id:
            raise ValueError("runtime_id must be distinct from conversation_id")
        now = self.clock()
        with self._transaction() as conn:
            conn.execute(
                """
                INSERT INTO instances(
                    runtime_id, conversation_id, harness, host, cwd, label,
                    registered_at, heartbeat_at, heartbeat_expires_at, ended_at
                ) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, NULL)
                """,
                (
                    runtime_id,
                    conversation_id,
                    harness,
                    host,
                    cwd,
                    label,
                    now,
                    now,
                    now + ttl_seconds,
                ),
            )
            # Carry undelivered work across a genuine resume. Only dead/expired
            # attachments of the same conversation are migrated; concurrently
            # live forks retain independent inboxes.
            stale_deliveries = conn.execute(
                """
                SELECT d.message_id, d.recipient_runtime_id
                FROM message_deliveries AS d
                JOIN instances AS old ON old.runtime_id = d.recipient_runtime_id
                WHERE old.conversation_id = ? AND old.runtime_id != ?
                  AND (old.ended_at IS NOT NULL OR old.heartbeat_expires_at <= ?)
                  AND d.state != 'acked'
                ORDER BY d.message_id, d.recipient_runtime_id
                """,
                (conversation_id, runtime_id, now),
            ).fetchall()
            for delivery in stale_deliveries:
                conn.execute(
                    """
                    INSERT INTO message_deliveries(
                        message_id, recipient_runtime_id, state
                    ) VALUES(?, ?, 'queued')
                    ON CONFLICT(message_id, recipient_runtime_id) DO NOTHING
                    """,
                    (delivery["message_id"], runtime_id),
                )
                conn.execute(
                    """
                    DELETE FROM message_deliveries
                    WHERE message_id = ? AND recipient_runtime_id = ?
                    """,
                    (delivery["message_id"], delivery["recipient_runtime_id"]),
                )
            self._audit(
                conn,
                now,
                "instance.registered",
                runtime_id,
                {
                    "conversation_id": conversation_id,
                    "ttl_seconds": ttl_seconds,
                    "deliveries_migrated": len(stale_deliveries),
                },
            )
        return runtime_id

    def heartbeat(
        self,
        runtime_id: str,
        *,
        ttl_seconds: float = DEFAULT_INSTANCE_TTL,
        renew_lease_ttl_seconds: float | None = None,
    ) -> None:
        if ttl_seconds <= 0:
            raise ValueError("ttl_seconds must be positive")
        if renew_lease_ttl_seconds is not None and renew_lease_ttl_seconds <= 0:
            raise ValueError("renew_lease_ttl_seconds must be positive")
        now = self.clock()
        with self._transaction() as conn:
            row = conn.execute(
                """
                SELECT ended_at, heartbeat_expires_at
                FROM instances WHERE runtime_id = ?
                """,
                (runtime_id,),
            ).fetchone()
            if (
                row is None
                or row["ended_at"] is not None
                or row["heartbeat_expires_at"] <= now
            ):
                raise KeyError(f"runtime instance is not active: {runtime_id}")
            conn.execute(
                """
                UPDATE instances
                SET heartbeat_at = ?, heartbeat_expires_at = ?
                WHERE runtime_id = ?
                """,
                (now, now + ttl_seconds, runtime_id),
            )
            renewed = 0
            if renew_lease_ttl_seconds is not None:
                renewed = conn.execute(
                    """
                    UPDATE leases SET renewed_at = ?, expires_at = ?
                    WHERE runtime_id = ? AND expires_at > ?
                    """,
                    (now, now + renew_lease_ttl_seconds, runtime_id, now),
                ).rowcount
            self._audit(
                conn,
                now,
                "instance.heartbeat",
                runtime_id,
                {"ttl_seconds": ttl_seconds, "leases_renewed": renewed},
            )

    def end_instance(self, runtime_id: str, *, reason: str = "clean") -> None:
        now = self.clock()
        with self._transaction() as conn:
            cursor = conn.execute(
                """
                UPDATE instances SET ended_at = ?
                WHERE runtime_id = ? AND ended_at IS NULL
                """,
                (now, runtime_id),
            )
            if cursor.rowcount != 1:
                raise KeyError(f"runtime instance is not active: {runtime_id}")
            released = conn.execute(
                "DELETE FROM leases WHERE runtime_id = ?", (runtime_id,)
            ).rowcount
            self._audit(
                conn,
                now,
                "instance.ended",
                runtime_id,
                {"reason": reason, "leases_released": released},
            )

    def claim(
        self,
        runtime_id: str,
        paths: Iterable[str],
        *,
        mode: str,
        purpose: str = "",
        ttl_seconds: float = DEFAULT_LEASE_TTL,
    ) -> None:
        paths = self._normalize_paths(paths)
        if mode not in VALID_LEASE_MODES:
            raise ValueError(f"invalid lease mode: {mode}")
        if ttl_seconds <= 0:
            raise ValueError("ttl_seconds must be positive")
        now = self.clock()
        with self._transaction() as conn:
            self._require_live(conn, runtime_id, now)
            self._reap(conn, now)
            rows = conn.execute(
                """
                SELECT path, runtime_id, mode, purpose, expires_at
                FROM leases
                WHERE runtime_id != ?
                  AND (? = 'exclusive' OR mode = 'exclusive')
                ORDER BY path, runtime_id
                """,
                (runtime_id, mode),
            ).fetchall()
            conflicts = [
                row
                for row in rows
                if any(self.resources_overlap(path, row["path"]) for path in paths)
            ]
            if conflicts:
                raise ClaimRefused(
                    [
                        LeaseConflict(
                            row["path"],
                            row["runtime_id"],
                            row["mode"],
                            row["purpose"],
                            row["expires_at"],
                        )
                        for row in conflicts
                    ]
                )
            for path in paths:
                conn.execute(
                    """
                    INSERT INTO leases(
                        path, runtime_id, mode, purpose,
                        claimed_at, renewed_at, expires_at
                    ) VALUES(?, ?, ?, ?, ?, ?, ?)
                    ON CONFLICT(path, runtime_id) DO UPDATE SET
                        mode = excluded.mode,
                        purpose = excluded.purpose,
                        renewed_at = excluded.renewed_at,
                        expires_at = excluded.expires_at
                    """,
                    (path, runtime_id, mode, purpose, now, now, now + ttl_seconds),
                )
            self._audit(
                conn,
                now,
                "lease.claimed",
                runtime_id,
                {
                    "paths": paths,
                    "mode": mode,
                    "purpose": purpose,
                    "ttl_seconds": ttl_seconds,
                },
            )

    def release(self, runtime_id: str, paths: Iterable[str]) -> tuple[str, ...]:
        paths = self._normalize_paths(paths)
        now = self.clock()
        with self._transaction() as conn:
            self._require_live(conn, runtime_id, now)
            placeholders = ",".join("?" for _ in paths)
            rows = conn.execute(
                f"""
                SELECT path FROM leases
                WHERE runtime_id = ? AND path IN ({placeholders})
                ORDER BY path
                """,
                (runtime_id, *paths),
            ).fetchall()
            released = tuple(row["path"] for row in rows)
            conn.execute(
                f"""
                DELETE FROM leases
                WHERE runtime_id = ? AND path IN ({placeholders})
                """,
                (runtime_id, *paths),
            )
            self._audit(
                conn, now, "lease.released", runtime_id, {"paths": released}
            )
        return released

    def renew_leases(
        self,
        runtime_id: str,
        *,
        paths: Iterable[str] | None = None,
        ttl_seconds: float = DEFAULT_LEASE_TTL,
    ) -> tuple[str, ...]:
        if ttl_seconds <= 0:
            raise ValueError("ttl_seconds must be positive")
        normalized = self._normalize_paths(paths) if paths is not None else None
        now = self.clock()
        with self._transaction() as conn:
            self._require_live(conn, runtime_id, now)
            self._reap(conn, now)
            parameters: list[Any] = [now, now + ttl_seconds, runtime_id]
            suffix = ""
            if normalized is not None:
                suffix = " AND path IN (%s)" % ",".join("?" for _ in normalized)
                parameters.extend(normalized)
            conn.execute(
                f"""
                UPDATE leases SET renewed_at = ?, expires_at = ?
                WHERE runtime_id = ?{suffix}
                """,
                parameters,
            )
            rows = conn.execute(
                f"""
                SELECT path FROM leases
                WHERE runtime_id = ?{suffix}
                ORDER BY path
                """,
                (runtime_id, *(normalized or ())),
            ).fetchall()
            renewed = tuple(row["path"] for row in rows)
            self._audit(
                conn,
                now,
                "lease.renewed",
                runtime_id,
                {"paths": renewed, "ttl_seconds": ttl_seconds},
            )
        return renewed

    # Resource-named aliases make it explicit that leases cover global operations
    # and external systems as well as repository files.
    def claim_resources(
        self,
        runtime_id: str,
        resources: Iterable[str],
        *,
        mode: str,
        purpose: str = "",
        ttl_seconds: float = DEFAULT_LEASE_TTL,
    ) -> None:
        self.claim(
            runtime_id,
            resources,
            mode=mode,
            purpose=purpose,
            ttl_seconds=ttl_seconds,
        )

    def release_resources(
        self, runtime_id: str, resources: Iterable[str]
    ) -> tuple[str, ...]:
        return self.release(runtime_id, resources)

    def renew_resources(
        self,
        runtime_id: str,
        *,
        resources: Iterable[str] | None = None,
        ttl_seconds: float = DEFAULT_LEASE_TTL,
    ) -> tuple[str, ...]:
        return self.renew_leases(
            runtime_id, paths=resources, ttl_seconds=ttl_seconds
        )

    def active_leases(self) -> list[dict[str, Any]]:
        now = self.clock()
        with self._transaction() as conn:
            self._reap(conn, now)
            return [
                dict(row)
                for row in conn.execute(
                    """
                    SELECT l.* FROM leases AS l
                    JOIN instances AS i USING(runtime_id)
                    WHERE l.expires_at > ? AND i.ended_at IS NULL
                      AND i.heartbeat_expires_at > ?
                    ORDER BY l.path, l.runtime_id
                    """,
                    (now, now),
                )
            ]

    def live_instances(self) -> list[dict[str, Any]]:
        """Return live runtime attachments with their full conversation IDs."""
        now = self.clock()
        conn = self._connect()
        try:
            return [
                dict(row)
                for row in conn.execute(
                    """
                    SELECT * FROM instances
                    WHERE ended_at IS NULL AND heartbeat_expires_at > ?
                    ORDER BY registered_at, runtime_id
                    """,
                    (now,),
                )
            ]
        finally:
            conn.close()

    def resolve_runtime(self, identity: str) -> str:
        """Resolve an exact runtime, friendly handle, or live conversation.

        Prefix matching is intentionally forbidden. Friendly handles are
        derived from the full runtime ID and must match exactly; the extremely
        unlikely ambiguous handle is refused rather than guessed.
        """
        now = self.clock()
        conn = self._connect()
        try:
            exact = conn.execute(
                """
                SELECT runtime_id FROM instances
                WHERE runtime_id = ? AND ended_at IS NULL
                  AND heartbeat_expires_at > ?
                """,
                (identity, now),
            ).fetchone()
            if exact is not None:
                return str(exact["runtime_id"])
            handle_matches = [
                str(row["runtime_id"])
                for row in conn.execute(
                    """
                    SELECT runtime_id FROM instances
                    WHERE ended_at IS NULL AND heartbeat_expires_at > ?
                    ORDER BY runtime_id
                    """,
                    (now,),
                )
                if hmac.compare_digest(
                    friendly_handle(str(row["runtime_id"]), "runtime"), identity
                )
            ]
            if len(handle_matches) == 1:
                return handle_matches[0]
            if len(handle_matches) > 1:
                raise KeyError(
                    f"ambiguous runtime handle: {identity}; use a full runtime ID"
                )
            latest = conn.execute(
                """
                SELECT runtime_id FROM instances
                WHERE conversation_id = ? AND ended_at IS NULL
                  AND heartbeat_expires_at > ?
                ORDER BY registered_at DESC, runtime_id DESC
                LIMIT 1
                """,
                (identity, now),
            ).fetchone()
            if latest is None:
                raise KeyError(f"no live runtime for exact identity: {identity}")
            return str(latest["runtime_id"])
        finally:
            conn.close()

    def save_workspace_record(
        self,
        event: str,
        record: dict[str, Any],
    ) -> None:
        """Transactionally retain the latest full lifecycle record."""
        instance_id = str(record.get("instance_id", ""))
        if not instance_id:
            raise ValueError("workspace record requires instance_id")
        now = self.clock()
        payload = json.dumps(record, sort_keys=True, separators=(",", ":"))
        with self._transaction() as conn:
            conn.execute(
                """
                INSERT INTO workspace_records(instance_id, record_json, updated_at)
                VALUES(?, ?, ?)
                ON CONFLICT(instance_id) DO UPDATE SET
                    record_json = excluded.record_json,
                    updated_at = excluded.updated_at
                """,
                (instance_id, payload, now),
            )
            self._audit(
                conn,
                now,
                event,
                str(record.get("runtime_id") or "") or None,
                {"instance_id": instance_id, "record": record},
            )

    def workspace_record(self, instance_id: str) -> dict[str, Any]:
        conn = self._connect()
        try:
            row = conn.execute(
                "SELECT record_json FROM workspace_records WHERE instance_id = ?",
                (instance_id,),
            ).fetchone()
            if row is not None:
                return dict(json.loads(row["record_json"]))
            matches = []
            for candidate in conn.execute(
                "SELECT instance_id, record_json FROM workspace_records ORDER BY instance_id"
            ):
                if hmac.compare_digest(
                    friendly_handle(str(candidate["instance_id"]), "work"),
                    instance_id,
                ):
                    matches.append(dict(json.loads(candidate["record_json"])))
            if len(matches) == 1:
                return matches[0]
            if len(matches) > 1:
                raise KeyError(
                    f"ambiguous workspace handle: {instance_id}; use a full instance ID"
                )
            raise KeyError(f"unknown workspace instance: {instance_id}")
        finally:
            conn.close()

    def workspace_records(self) -> list[dict[str, Any]]:
        conn = self._connect()
        try:
            return [
                dict(json.loads(row["record_json"]))
                for row in conn.execute(
                    "SELECT record_json FROM workspace_records ORDER BY updated_at, instance_id"
                )
            ]
        finally:
            conn.close()

    def is_managed_workspace(self, path: str | os.PathLike[str]) -> bool:
        candidate = pathlib.Path(path).resolve()
        for record in self.workspace_records():
            actual = record.get("actual_path")
            if (
                not actual
                or record.get("lifecycle_state") != "active"
                or pathlib.Path(str(actual)).resolve() != candidate
            ):
                continue
            try:
                marker = json.loads(
                    (candidate / ".jj" / "fleet-coord-owner.json").read_text(
                        encoding="utf-8"
                    )
                )
            except (FileNotFoundError, OSError, json.JSONDecodeError):
                return False
            return (
                marker.get("instance_id") == record.get("instance_id")
                and hmac.compare_digest(
                    str(marker.get("ownership_token", "")),
                    str(record.get("ownership_token", "")),
                )
            )
        return False

    def record_event(
        self,
        event: str,
        *,
        runtime_id: str | None = None,
        details: dict[str, Any] | None = None,
    ) -> int:
        """Append a generic structured event for integrations layered above this API."""
        if not event:
            raise ValueError("event is required")
        now = self.clock()
        with self._transaction() as conn:
            if runtime_id is not None:
                self._require_live(conn, runtime_id, now)
            self._audit(conn, now, event, runtime_id, details or {})
            return int(conn.execute("SELECT last_insert_rowid()").fetchone()[0])

    def send_message(
        self,
        sender_runtime_id: str,
        *,
        recipient: str | Iterable[str] = "*",
        kind: str = "fyi",
        body: str,
        refs: Iterable[str] = (),
    ) -> int:
        if not body:
            raise ValueError("message body is required")
        now = self.clock()
        with self._transaction() as conn:
            self._require_live(conn, sender_runtime_id, now)
            if recipient == "*":
                requested = "*"
                recipients = [
                    row["runtime_id"]
                    for row in conn.execute(
                        """
                        SELECT runtime_id FROM instances
                        WHERE runtime_id != ? AND ended_at IS NULL
                          AND heartbeat_expires_at > ?
                        ORDER BY runtime_id
                        """,
                        (sender_runtime_id, now),
                    )
                ]
            else:
                recipients = (
                    (recipient,)
                    if isinstance(recipient, str)
                    else tuple(dict.fromkeys(recipient))
                )
                requested = ",".join(recipients)
                if not recipients:
                    raise ValueError("at least one recipient is required")
                placeholders = ",".join("?" for _ in recipients)
                found = {
                    row["runtime_id"]
                    for row in conn.execute(
                        f"""
                        SELECT runtime_id FROM instances
                        WHERE runtime_id IN ({placeholders})
                          AND ended_at IS NULL AND heartbeat_expires_at > ?
                        """,
                        (*recipients, now),
                    )
                }
                missing = set(recipients) - found
                if missing:
                    raise KeyError(f"recipient runtime is not live: {sorted(missing)}")
            cursor = conn.execute(
                """
                INSERT INTO messages(
                    sender_runtime_id, requested_recipient, kind,
                    body, refs_json, created_at
                ) VALUES(?, ?, ?, ?, ?, ?)
                """,
                (
                    sender_runtime_id,
                    requested,
                    kind,
                    body,
                    json.dumps(tuple(refs), separators=(",", ":")),
                    now,
                ),
            )
            message_id = int(cursor.lastrowid)
            conn.executemany(
                """
                INSERT INTO message_deliveries(
                    message_id, recipient_runtime_id, state
                ) VALUES(?, ?, 'queued')
                """,
                ((message_id, target) for target in recipients),
            )
            self._audit(
                conn,
                now,
                "message.queued",
                sender_runtime_id,
                {
                    "message_id": message_id,
                    "requested_recipient": requested,
                    "recipient_snapshot": recipients,
                },
            )
        return message_id

    def inbox(
        self,
        runtime_id: str,
        *,
        limit: int = 100,
        mark_delivered: bool = True,
    ) -> list[Delivery]:
        if limit <= 0:
            return []
        now = self.clock()
        with self._transaction() as conn:
            self._require_live(conn, runtime_id, now)
            rows = conn.execute(
                """
                SELECT m.*, d.state
                FROM message_deliveries AS d
                JOIN messages AS m USING(message_id)
                WHERE d.recipient_runtime_id = ? AND d.state != 'acked'
                ORDER BY m.message_id
                LIMIT ?
                """,
                (runtime_id, limit),
            ).fetchall()
            ids = [row["message_id"] for row in rows]
            if mark_delivered and ids:
                placeholders = ",".join("?" for _ in ids)
                conn.execute(
                    f"""
                    UPDATE message_deliveries
                    SET state = 'delivered', delivered_at = COALESCE(delivered_at, ?)
                    WHERE recipient_runtime_id = ?
                      AND message_id IN ({placeholders}) AND state = 'queued'
                    """,
                    (now, runtime_id, *ids),
                )
                self._audit(
                    conn,
                    now,
                    "message.delivered",
                    runtime_id,
                    {"message_ids": ids},
                )
            return [
                Delivery(
                    message_id=row["message_id"],
                    sender_runtime_id=row["sender_runtime_id"],
                    kind=row["kind"],
                    body=row["body"],
                    refs=tuple(json.loads(row["refs_json"])),
                    created_at=row["created_at"],
                    # Report the state observed by this read. This transaction
                    # updates queued rows after reading them, while the hook
                    # needs to distinguish first delivery from a reminder.
                    state=row["state"],
                )
                for row in rows
            ]

    def ack(self, runtime_id: str, message_ids: Iterable[int]) -> tuple[int, ...]:
        ids = tuple(dict.fromkeys(int(item) for item in message_ids))
        if not ids:
            return ()
        now = self.clock()
        with self._transaction() as conn:
            self._require_live(conn, runtime_id, now)
            placeholders = ",".join("?" for _ in ids)
            rows = conn.execute(
                f"""
                SELECT message_id FROM message_deliveries
                WHERE recipient_runtime_id = ?
                  AND message_id IN ({placeholders}) AND state != 'acked'
                ORDER BY message_id
                """,
                (runtime_id, *ids),
            ).fetchall()
            acked = tuple(row["message_id"] for row in rows)
            conn.execute(
                f"""
                UPDATE message_deliveries
                SET state = 'acked', acked_at = ?
                WHERE recipient_runtime_id = ?
                  AND message_id IN ({placeholders}) AND state != 'acked'
                """,
                (now, runtime_id, *ids),
            )
            self._audit(
                conn, now, "message.acked", runtime_id, {"message_ids": acked}
            )
        return acked

    def delivery_state(self, message_id: int, runtime_id: str) -> str | None:
        conn = self._connect()
        try:
            row = conn.execute(
                """
                SELECT state FROM message_deliveries
                WHERE message_id = ? AND recipient_runtime_id = ?
                """,
                (message_id, runtime_id),
            ).fetchone()
            return None if row is None else str(row["state"])
        finally:
            conn.close()

    def audit_events(self, *, after_id: int = 0) -> list[dict[str, Any]]:
        conn = self._connect()
        try:
            rows = conn.execute(
                """
                SELECT * FROM audit_events
                WHERE event_id > ? ORDER BY event_id
                """,
                (after_id,),
            ).fetchall()
            return [
                {
                    "event_id": row["event_id"],
                    "occurred_at": row["occurred_at"],
                    "event": row["event"],
                    "runtime_id": row["runtime_id"],
                    "details": json.loads(row["details_json"]),
                }
                for row in rows
            ]
        finally:
            conn.close()

    def export_audit_jsonl(
        self,
        destination: str | os.PathLike[str] | TextIO,
        *,
        after_id: int = 0,
        include_secrets: bool = False,
    ) -> int:
        def without_secrets(value: Any) -> Any:
            if isinstance(value, dict):
                return {
                    key: without_secrets(item)
                    for key, item in value.items()
                    if key != "ownership_token"
                }
            if isinstance(value, list):
                return [without_secrets(item) for item in value]
            return value

        events = self.audit_events(after_id=after_id)
        close = False
        if hasattr(destination, "write"):
            stream = destination
        else:
            target = pathlib.Path(destination)
            target.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
            fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
            os.fchmod(fd, 0o600)
            stream = os.fdopen(fd, "w", encoding="utf-8")
            close = True
        try:
            for event in events:
                exported = event if include_secrets else without_secrets(event)
                stream.write(
                    json.dumps(exported, sort_keys=True, separators=(",", ":"))
                    + "\n"
                )
        finally:
            if close:
                stream.close()
        return len(events)