#!/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)