"""Client/session-aware runtime ownership and provenance (#948). Before this module the control plane had two disagreeing provenance models. ``gitea_get_runtime_context`` read the *live* process environment, while ``mcp_namespace_health.classify_namespace_probe`` read a filtered env summary whose allowlist never carried the provenance keys — so one surface reported ``client_managed`` while the other reported ``manual_launch`` for the very same process. Neither model could name *which* client or *which* session owned the runtime, so a healthy daemon serving a second client looked like a duplicate. This module is the single authority both surfaces now consult. The model it implements: * A **worker identity** is unique and is *not* a role or a profile. Roles and profiles are reusable capability definitions that many live workers may share concurrently; exclusivity belongs to an active lease target, never to a role. * A **generation** is one daemon launch. Exactly one live session may claim a generation at a time; a takeover mints a higher fencing epoch so the prior session is fenced if it resumes. * **Trusted evidence** is a live attachment record binding ``(client_instance_id, session_id, generation_id)`` to the running process. A process environment flag is corroborating evidence at best: it is set by whoever launched the process and says nothing about which session owns it *now*, so it can never establish ownership on its own. * Missing or self-contradictory evidence fails closed, and the refusal is scoped to the affected worker identity rather than to every worker sharing its profile. """ from __future__ import annotations import hashlib import os import re import secrets import sqlite3 import threading import time from contextlib import contextmanager from datetime import datetime, timezone from typing import Any, Iterator # --- Provenance verdicts ------------------------------------------------- #: A live attachment record binds this runtime to a client session. PROVENANCE_CLIENT_SESSION = "client_session_attached" #: Positive evidence that the process was launched outside any client session. PROVENANCE_MANUAL = "manual_launch" #: No trusted evidence either way — fails closed, distinct from "manual". PROVENANCE_UNPROVEN = "unproven" VALID_PROVENANCE = frozenset( {PROVENANCE_CLIENT_SESSION, PROVENANCE_MANUAL, PROVENANCE_UNPROVEN} ) # --- Evidence kinds ------------------------------------------------------ EVIDENCE_ATTACHMENT_RECORD = "live_attachment_record" EVIDENCE_NATIVE_TRANSPORT = "bound_native_transport" EVIDENCE_PROCESS_LIVENESS = "process_liveness" #: Never sufficient alone. Named so a report can say *why* it was not enough. EVIDENCE_ENV_FLAG = "process_env_flag" #: Evidence kinds that, on their own, establish current session ownership. SUFFICIENT_EVIDENCE = frozenset({EVIDENCE_ATTACHMENT_RECORD}) # --- Blocker kinds ------------------------------------------------------- BLOCKER_NONE = "none" BLOCKER_NO_ATTACHMENT = "session_attachment_missing" BLOCKER_CONFLICTING_SESSIONS = "conflicting_live_sessions" BLOCKER_FENCED = "superseded_generation_fenced" BLOCKER_IDENTITY_COLLISION = "worker_identity_collision" BLOCKER_CONTRADICTORY = "contradictory_provenance_evidence" # --- Client naming ------------------------------------------------------- #: Returned when no client name is supplied or the name is unusable. There is #: deliberately no default vendor here: defaulting an unknown client to a #: specific product emits reconnect steps for a UI the operator is not looking #: at, which is the #948 hardcoded-``codex`` defect. UNKNOWN_CLIENT = "unknown_client" #: Canonical spellings for clients whose aliases we recognise. Absence from #: this map is not an error — an unrecognised name is slugified and kept, so a #: new client is named accurately rather than relabelled as a known one. CLIENT_ALIASES: dict[str, str] = { "codex": "codex", "openai": "codex", "openai_codex": "codex", "claude": "claude_code", "claude_code": "claude_code", "claude_desktop": "claude_code", "anthropic": "claude_code", "gemini": "gemini", "google": "gemini", "antigravity": "antigravity", "gemini_antigravity": "antigravity", "grok": "grok", "xai": "grok", } _SLUG_RE = re.compile(r"[^a-z0-9]+") #: ``--`` per the #948 scope clarification. WORKER_IDENTITY_RE = re.compile( r"^(?P[a-z0-9][a-z0-9_]*)-(?P\d{8}T\d{6}Z)-(?P[0-9a-f]{12})$" ) IDENTITY_TIMESTAMP_FORMAT = "%Y%m%dT%H%M%SZ" IDENTITY_DIGEST_LENGTH = 12 # --- Registry defaults --------------------------------------------------- REGISTRY_PATH_ENV = "GITEA_WORKER_REGISTRY_DB" DEFAULT_REGISTRY_PATH = os.path.expanduser( "~/.cache/gitea-tools/control-plane/worker_registry.sqlite3" ) #: How long a registration stays live without a heartbeat. Chosen to outlast a #: full-suite run inside one tool call while still releasing an abandoned #: worker within a single operator coffee break. DEFAULT_HEARTBEAT_TTL_SECONDS = 900.0 STATUS_ACTIVE = "active" STATUS_SUPERSEDED = "superseded" STATUS_RELEASED = "released" _TRUE_VALUES = frozenset({"1", "true", "yes", "client_managed"}) _FALSE_VALUES = frozenset({"0", "false", "no", "manual", "manual_launch"}) #: Environment keys that historically stood in for provenance. They are read #: only as corroboration; see :func:`env_provenance_signal`. PROVENANCE_ENV_KEYS = ( "GITEA_CLIENT_MANAGED", "GITEA_MCP_CLIENT_MANAGED", "GITEA_SERVER_PROVENANCE", "GITEA_MCP_SANCTIONED_DAEMON", ) #: Explicit launch-provenance declarations, highest precedence first. Mirrors #: the historical ``_is_client_managed_process`` precedence exactly, so #: unifying the two models changes *who decides*, not *what is decided*. LAUNCH_DECLARATION_ENV_KEYS = ( "GITEA_CLIENT_MANAGED", "GITEA_MCP_CLIENT_MANAGED", "GITEA_SERVER_PROVENANCE", "GITEA_FORCE_CLIENT_MANAGED", ) #: Env keys a client launcher sets that imply a non-terminal launch. LAUNCH_CONFIG_ENV_KEYS = ( "GITEA_MCP_CONFIG", "GITEA_MCP_PROFILE", "GITEA_PROFILE_NAME", ) # --- Launch provenance verdicts ------------------------------------------ #: Launch-dimension values. These keep the literals the pre-#948 surfaces #: already published, so a consumer reading ``server_provenance`` sees the same #: strings it always did. LAUNCH_CLIENT_MANAGED = "client_managed" LAUNCH_MANUAL = PROVENANCE_MANUAL LAUNCH_UNPROVEN = PROVENANCE_UNPROVEN # --- Session ownership verdicts ------------------------------------------ #: A live attachment record binds this runtime to exactly one client session. OWNERSHIP_OWNED = "owned" #: No live attachment record. Fails closed for ownership-scoped decisions. OWNERSHIP_UNOWNED = "unowned" #: More than one live session claims the generation. Fails closed. OWNERSHIP_CONTESTED = "contested" _SCHEMA_SQL = """ CREATE TABLE IF NOT EXISTS worker_registrations ( worker_identity TEXT PRIMARY KEY, client_name TEXT NOT NULL, client_instance_id TEXT NOT NULL, session_id TEXT NOT NULL, generation_id TEXT NOT NULL, role TEXT, profile TEXT, namespace TEXT, remote TEXT, repository_binding TEXT, pid INTEGER, transport TEXT, token_fingerprint TEXT, started_at TEXT NOT NULL, last_heartbeat_at TEXT NOT NULL, heartbeat_ttl_seconds REAL NOT NULL, fencing_epoch INTEGER NOT NULL, status TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_worker_generation ON worker_registrations(generation_id, status); CREATE INDEX IF NOT EXISTS idx_worker_session ON worker_registrations(session_id, status); CREATE INDEX IF NOT EXISTS idx_worker_profile ON worker_registrations(profile, status); """ class WorkerRegistryError(RuntimeError): """Raised for registry misuse that is a programming error, not a refusal.""" def _utc_now() -> datetime: return datetime.now(timezone.utc) def _ts(value: datetime) -> str: return value.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _parse_ts(value: str | None) -> datetime | None: if not value: return None try: return datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace( tzinfo=timezone.utc ) except (TypeError, ValueError): return None def default_registry_path() -> str: raw = (os.environ.get(REGISTRY_PATH_ENV) or DEFAULT_REGISTRY_PATH).strip() return raw or DEFAULT_REGISTRY_PATH # --- Identity ------------------------------------------------------------ def normalize_client_name(raw: str | None) -> str: """Canonicalise a client name without inventing one. An empty or unusable name yields :data:`UNKNOWN_CLIENT` rather than a specific product, so downstream reconnect guidance describes a UI the operator is actually looking at (or says it does not know) instead of always naming one vendor. """ text = _SLUG_RE.sub("_", (raw or "").strip().lower()).strip("_") if not text: return UNKNOWN_CLIENT return CLIENT_ALIASES.get(text, text) def worker_identity_digest( client_name: str, session_id: str, timestamp_ns: int, nonce: str, ) -> str: """Collision-resistant digest over client, session, high-resolution time, nonce.""" payload = f"{client_name}\x1f{session_id}\x1f{timestamp_ns}\x1f{nonce}" return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:IDENTITY_DIGEST_LENGTH] def generate_worker_identity( client_name: str | None, session_id: str | None, *, timestamp_ns: int | None = None, nonce: str | None = None, now: datetime | None = None, ) -> str: """Mint ``--`` (#948 AC27-29). ``timestamp_ns`` and ``nonce`` are injectable so a test can force a collision deliberately; production callers omit both and get high-resolution time plus a fresh random nonce. """ client = normalize_client_name(client_name) session = (session_id or "").strip() or "no_session" ts_ns = int(timestamp_ns if timestamp_ns is not None else time.time_ns()) seed = nonce if nonce is not None else secrets.token_hex(16) stamp = (now or _utc_now()).astimezone(timezone.utc).strftime( IDENTITY_TIMESTAMP_FORMAT ) digest = worker_identity_digest(client, session, ts_ns, seed) return f"{client}-{stamp}-{digest}" def parse_worker_identity(identity: str | None) -> dict[str, Any]: """Structurally validate a worker identity.""" text = (identity or "").strip() match = WORKER_IDENTITY_RE.match(text) if not match: return { "valid": False, "worker_identity": text or None, "client_name": None, "minted_at": None, "digest": None, "reasons": [ "worker identity must match --, " "e.g. gemini-20260729T061500Z-1a2b3c4d5e6f" ], } return { "valid": True, "worker_identity": text, "client_name": match.group("client"), "minted_at": match.group("ts"), "digest": match.group("digest"), "reasons": [], } def new_generation_id(pid: int | None = None) -> str: """Mint a boot epoch for one daemon launch.""" return f"gen-{int(pid if pid is not None else os.getpid())}-{secrets.token_hex(8)}" # --- Environment signal -------------------------------------------------- def env_provenance_signal(env: dict[str, str] | None = None) -> dict[str, Any]: """Read the legacy env provenance flags as *corroborating* signal only. Returns the raw reading plus an explicit statement that it does not establish ownership. Callers must not branch on ``value`` alone; #948 requires a live attachment record for that. """ source = os.environ if env is None else env present: dict[str, str] = {} for key in PROVENANCE_ENV_KEYS: raw = source.get(key) if raw not in (None, ""): present[key] = str(raw) value: bool | None = None for key in PROVENANCE_ENV_KEYS: raw = (present.get(key) or "").strip().lower() if not raw: continue if raw in _FALSE_VALUES: value = False break if raw in _TRUE_VALUES: value = True break return { "evidence_kind": EVIDENCE_ENV_FLAG, "keys_present": sorted(present), "value": value, "proves_session_ownership": False, "reason": ( "process environment flags are set by whoever launched the process and " "carry no session binding; they corroborate a live attachment record but " "never establish current ownership on their own (#948)" ), } def assess_declared_launch_provenance(env: dict[str, str] | None = None) -> dict[str, Any]: """Launch provenance for a process we are observing from the outside. Deliberately stricter than :func:`assess_launch_provenance`. For a *peer* process there is no stdin to inspect, and launcher-configuration env is not evidence of anything: a shell that exports ``GITEA_MCP_PROFILE`` passes it to a hand-launched server too. So only an explicit declaration counts, and its absence is reported as ``unproven`` rather than asserted as a manual launch — naming a proof we do not have was the misdiagnosis #948 reports. """ source = os.environ if env is None else env for key in LAUNCH_DECLARATION_ENV_KEYS: raw = (source.get(key) or "").strip().lower() if not raw: continue if raw in _FALSE_VALUES: return { "launch_provenance": LAUNCH_MANUAL, "client_managed": False, "decided_by": f"env:{key}", "reasons": [f"{key}={raw!r} explicitly declares a manual launch"], } if raw in _TRUE_VALUES: return { "launch_provenance": LAUNCH_CLIENT_MANAGED, "client_managed": True, "decided_by": f"env:{key}", "reasons": [ f"{key}={raw!r} explicitly declares a client-managed launch" ], } return { "launch_provenance": LAUNCH_UNPROVEN, "client_managed": False, "decided_by": "no_declaration", "reasons": [ "no explicit client-managed declaration on the observed process; " "launcher-configuration env is not proof for a peer process because a " "hand-launched server inherits it from the shell" ], } def assess_launch_provenance( env: dict[str, str] | None = None, *, stdin_is_tty: bool | None = None, declared_only: bool = False, ) -> dict[str, Any]: """Decide *how the process was launched* — a different question from ownership. #948 separates two claims the old code ran together: * **Launch provenance** — was this started by a client, or hand-launched from a terminal? This is the #686 concern and it is what the environment can legitimately answer, because the launcher is exactly what sets it. * **Session ownership** — which live client session owns this runtime *now*? The environment cannot answer that; only a live attachment record can. Conflating them is what let one surface call a process ``client_managed`` while another called the same process ``manual_launch``. This function is now the only implementation of the first claim, so every surface gets one answer. ``declared_only`` selects the strategy for an *observed peer* process, where stdin cannot be inspected and launcher-config env proves nothing. Callers assessing their own process leave it false. """ if declared_only: return assess_declared_launch_provenance(env) source = os.environ if env is None else env reasons: list[str] = [] for key in LAUNCH_DECLARATION_ENV_KEYS: raw = (source.get(key) or "").strip().lower() if not raw: continue if raw in _FALSE_VALUES: return { "launch_provenance": LAUNCH_MANUAL, "client_managed": False, "decided_by": f"env:{key}", "reasons": [f"{key}={raw!r} explicitly declares a manual launch"], } if raw in _TRUE_VALUES: return { "launch_provenance": LAUNCH_CLIENT_MANAGED, "client_managed": True, "decided_by": f"env:{key}", "reasons": [f"{key}={raw!r} explicitly declares a client-managed launch"], } if stdin_is_tty: return { "launch_provenance": LAUNCH_MANUAL, "client_managed": False, "decided_by": "stdin_tty", "reasons": [ "stdin is an interactive terminal, so this process cannot be " "receiving client stdio" ], } present = [k for k in LAUNCH_CONFIG_ENV_KEYS if (source.get(k) or "").strip()] if present: return { "launch_provenance": LAUNCH_CLIENT_MANAGED, "client_managed": True, "decided_by": "env:launcher_config", "reasons": [ f"launcher-supplied configuration env present ({', '.join(present)}) " "with no terminal on stdin" ], } reasons.append( "no launch declaration, no launcher configuration env, and no terminal " "evidence; launch provenance is unproven rather than assumed manual" ) return { "launch_provenance": LAUNCH_UNPROVEN, "client_managed": False, "decided_by": "no_evidence", "reasons": reasons, } # --- Registry ------------------------------------------------------------ class WorkerRegistry: """Atomic, fail-closed registry of live client/session runtime ownership. Backed by SQLite with ``BEGIN IMMEDIATE`` so two clients racing to register the same identity, or to claim the same generation, resolve to exactly one winner across processes. """ def __init__(self, db_path: str | None = None) -> None: self.db_path = (db_path or default_registry_path()).strip() parent = os.path.dirname(self.db_path) if parent: os.makedirs(parent, mode=0o700, exist_ok=True) self._lock = threading.RLock() self._init_schema() # -- plumbing -- def _connect(self) -> sqlite3.Connection: conn = sqlite3.connect(self.db_path, timeout=30) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode = WAL") return conn @contextmanager def _tx(self) -> Iterator[sqlite3.Connection]: with self._lock: conn = self._connect() try: conn.execute("BEGIN IMMEDIATE") yield conn conn.commit() except Exception: try: conn.rollback() except sqlite3.Error: pass raise finally: conn.close() def _init_schema(self) -> None: with self._lock: conn = self._connect() try: conn.executescript(_SCHEMA_SQL) conn.commit() finally: conn.close() @staticmethod def _row_to_record(row: sqlite3.Row) -> dict[str, Any]: return {key: row[key] for key in row.keys()} # -- liveness -- @staticmethod def is_live( record: dict[str, Any] | None, *, now: datetime | None = None, pid_alive: bool | None = None, ) -> dict[str, Any]: """Liveness from heartbeat freshness, corroborated by process evidence. #948 AC7: abandoned generations are detected through authoritative liveness evidence, not PID comparison alone. A fresh heartbeat is the authority; ``pid_alive`` can only *withdraw* liveness, never grant it, because a long-lived daemon PID outlives every session it serves. """ if not record: return {"live": False, "reasons": ["no registration record"]} if record.get("status") != STATUS_ACTIVE: return { "live": False, "reasons": [f"registration status is {record.get('status')!r}"], } beat = _parse_ts(record.get("last_heartbeat_at")) if beat is None: return {"live": False, "reasons": ["registration has no parsable heartbeat"]} ttl = float(record.get("heartbeat_ttl_seconds") or DEFAULT_HEARTBEAT_TTL_SECONDS) age = ((now or _utc_now()) - beat).total_seconds() fresh = age <= ttl reasons: list[str] = [] if not fresh: reasons.append( f"heartbeat is {age:.0f}s old, past the {ttl:.0f}s liveness window" ) if pid_alive is False: reasons.append(f"recorded pid {record.get('pid')} is no longer running") return { "live": bool(fresh and pid_alive is not False), "heartbeat_age_seconds": age, "heartbeat_ttl_seconds": ttl, "heartbeat_fresh": fresh, "pid_alive": pid_alive, "reasons": reasons, } # -- reads -- def get(self, worker_identity: str) -> dict[str, Any] | None: conn = self._connect() try: row = conn.execute( "SELECT * FROM worker_registrations WHERE worker_identity = ?", (worker_identity,), ).fetchone() return self._row_to_record(row) if row else None finally: conn.close() def list_workers( self, *, status: str | None = STATUS_ACTIVE, generation_id: str | None = None, profile: str | None = None, role: str | None = None, ) -> list[dict[str, Any]]: clauses: list[str] = [] params: list[Any] = [] if status: clauses.append("status = ?") params.append(status) if generation_id: clauses.append("generation_id = ?") params.append(generation_id) if profile: clauses.append("profile = ?") params.append(profile) if role: clauses.append("role = ?") params.append(role) where = f" WHERE {' AND '.join(clauses)}" if clauses else "" conn = self._connect() try: rows = conn.execute( f"SELECT * FROM worker_registrations{where} ORDER BY started_at", params, ).fetchall() return [self._row_to_record(r) for r in rows] finally: conn.close() # -- writes -- def register( self, *, worker_identity: str, client_name: str | None, client_instance_id: str, session_id: str, generation_id: str, role: str | None = None, profile: str | None = None, namespace: str | None = None, remote: str | None = None, repository_binding: str | None = None, pid: int | None = None, transport: str | None = None, token_fingerprint: str | None = None, heartbeat_ttl_seconds: float = DEFAULT_HEARTBEAT_TTL_SECONDS, now: datetime | None = None, pid_alive_probe=None, ) -> dict[str, Any]: """Atomically register one worker identity. Fails closed on collision (#948 AC31): an identity already present is never replaced, adopted, merged with, or corrupted — the caller is told to mint a different identity and register that instead (AC32). The existing registration is returned untouched so the caller can see what it collided with. """ parsed = parse_worker_identity(worker_identity) if not parsed["valid"]: return { "success": False, "registered": False, "mutation_performed": False, "blocker_kind": BLOCKER_IDENTITY_COLLISION, "collision": False, "reasons": parsed["reasons"], "exact_next_action": ( "Mint a worker identity with generate_worker_identity() and " "register that." ), } stamp = _ts(now or _utc_now()) with self._tx() as conn: existing = conn.execute( "SELECT * FROM worker_registrations WHERE worker_identity = ?", (worker_identity,), ).fetchone() if existing is not None: record = self._row_to_record(existing) pid_alive = ( pid_alive_probe(record.get("pid")) if pid_alive_probe is not None else None ) liveness = self.is_live(record, now=now, pid_alive=pid_alive) return { "success": False, "registered": False, "mutation_performed": False, "blocker_kind": BLOCKER_IDENTITY_COLLISION, "collision": True, "collision_kind": ( "active_worker" if liveness["live"] else "stale_registration" ), "existing_registration": _public_record(record), "existing_liveness": liveness, "reasons": [ f"worker identity {worker_identity!r} is already registered " f"(status={record.get('status')}, live={liveness['live']}); " "registration is refused rather than replacing, adopting, or " "merging with the existing worker (#948 AC31)" ], "exact_next_action": ( "Generate a different worker identity and atomically register " "that replacement; do not reuse or take over this one." ), } epoch = self._next_epoch(conn, generation_id) conn.execute( """ INSERT INTO worker_registrations ( worker_identity, client_name, client_instance_id, session_id, generation_id, role, profile, namespace, remote, repository_binding, pid, transport, token_fingerprint, started_at, last_heartbeat_at, heartbeat_ttl_seconds, fencing_epoch, status ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) """, ( worker_identity, normalize_client_name(client_name or parsed["client_name"]), client_instance_id, session_id, generation_id, role, profile, namespace, remote, repository_binding, int(pid) if pid is not None else None, transport, token_fingerprint, stamp, stamp, float(heartbeat_ttl_seconds), epoch, STATUS_ACTIVE, ), ) row = conn.execute( "SELECT * FROM worker_registrations WHERE worker_identity = ?", (worker_identity,), ).fetchone() record = self._row_to_record(row) return { "success": True, "registered": True, "mutation_performed": True, "blocker_kind": BLOCKER_NONE, "collision": False, "registration": _public_record(record), "fencing_epoch": record["fencing_epoch"], "reasons": [], } @staticmethod def _next_epoch(conn: sqlite3.Connection, generation_id: str) -> int: row = conn.execute( "SELECT MAX(fencing_epoch) AS hi FROM worker_registrations " "WHERE generation_id = ?", (generation_id,), ).fetchone() current = row["hi"] if row and row["hi"] is not None else 0 return int(current) + 1 def heartbeat( self, *, worker_identity: str, fencing_epoch: int, now: datetime | None = None, ) -> dict[str, Any]: """Renew only the owning registration (#948 AC11). A stale epoch is refused rather than silently renewed, so a superseded session that resumes cannot heartbeat its way back into ownership. """ stamp = _ts(now or _utc_now()) with self._tx() as conn: row = conn.execute( "SELECT * FROM worker_registrations WHERE worker_identity = ?", (worker_identity,), ).fetchone() if row is None: return { "success": False, "renewed": False, "mutation_performed": False, "blocker_kind": BLOCKER_NO_ATTACHMENT, "reasons": [f"no registration for {worker_identity!r}"], } record = self._row_to_record(row) if record["status"] != STATUS_ACTIVE: return { "success": False, "renewed": False, "mutation_performed": False, "blocker_kind": BLOCKER_FENCED, "reasons": [ f"registration status is {record['status']!r}; a superseded " "registration cannot be renewed (#948 AC15)" ], } if int(record["fencing_epoch"]) != int(fencing_epoch): return { "success": False, "renewed": False, "mutation_performed": False, "blocker_kind": BLOCKER_FENCED, "recorded_epoch": int(record["fencing_epoch"]), "presented_epoch": int(fencing_epoch), "reasons": [ f"fencing epoch {fencing_epoch} does not match the recorded " f"epoch {record['fencing_epoch']}; this session has been " "superseded (#948 AC15/AC16)" ], } conn.execute( "UPDATE worker_registrations SET last_heartbeat_at = ? " "WHERE worker_identity = ?", (stamp, worker_identity), ) return { "success": True, "renewed": True, "mutation_performed": True, "blocker_kind": BLOCKER_NONE, "last_heartbeat_at": stamp, "fencing_epoch": int(fencing_epoch), "reasons": [], } def claim_generation( self, *, worker_identity: str, generation_id: str, now: datetime | None = None, pid_alive_probe=None, ) -> dict[str, Any]: """Bind a generation to this worker, superseding only a dead claimant. A generation held by another *live* session is refused: one daemon must never be simultaneously claimed by conflicting live sessions. A generation whose claimant is no longer live is taken over with a higher fencing epoch, so stale ownership cannot permanently strand a healthy daemon (#948 AC8/AC14). """ with self._tx() as conn: mine = conn.execute( "SELECT * FROM worker_registrations WHERE worker_identity = ?", (worker_identity,), ).fetchone() if mine is None: return { "success": False, "claimed": False, "mutation_performed": False, "blocker_kind": BLOCKER_NO_ATTACHMENT, "reasons": [ f"{worker_identity!r} is not registered; register before " "claiming a generation" ], } holders = [ self._row_to_record(r) for r in conn.execute( "SELECT * FROM worker_registrations " "WHERE generation_id = ? AND status = ? AND worker_identity != ?", (generation_id, STATUS_ACTIVE, worker_identity), ).fetchall() ] live_holders = [] for holder in holders: pid_alive = ( pid_alive_probe(holder.get("pid")) if pid_alive_probe is not None else None ) if self.is_live(holder, now=now, pid_alive=pid_alive)["live"]: live_holders.append(holder) if live_holders: return { "success": False, "claimed": False, "mutation_performed": False, "blocker_kind": BLOCKER_CONFLICTING_SESSIONS, "conflicting_owners": [_public_record(h) for h in live_holders], "reasons": [ f"generation {generation_id!r} is already claimed by " f"{len(live_holders)} live session(s): " + ", ".join( f"{h['worker_identity']} (session {h['session_id']})" for h in live_holders ) + "; a daemon must not be simultaneously claimed by " "conflicting live sessions (#948)" ], "exact_next_action": ( "Wait for the live owner's lease to lapse, or attach this " "worker to its own generation. Do not kill the other process." ), } superseded = [h["worker_identity"] for h in holders] for identity in superseded: conn.execute( "UPDATE worker_registrations SET status = ? " "WHERE worker_identity = ?", (STATUS_SUPERSEDED, identity), ) epoch = self._next_epoch(conn, generation_id) conn.execute( "UPDATE worker_registrations SET generation_id = ?, fencing_epoch = ?, " "status = ?, last_heartbeat_at = ? WHERE worker_identity = ?", ( generation_id, epoch, STATUS_ACTIVE, _ts(now or _utc_now()), worker_identity, ), ) row = conn.execute( "SELECT * FROM worker_registrations WHERE worker_identity = ?", (worker_identity,), ).fetchone() record = self._row_to_record(row) return { "success": True, "claimed": True, "mutation_performed": True, "blocker_kind": BLOCKER_NONE, "registration": _public_record(record), "fencing_epoch": record["fencing_epoch"], "superseded_workers": superseded, "reasons": ( [ f"took over generation {generation_id!r} from " f"{len(superseded)} non-live claimant(s) with fencing epoch " f"{record['fencing_epoch']}" ] if superseded else [] ), } def release( self, *, worker_identity: str, now: datetime | None = None ) -> dict[str, Any]: """Mark a registration released. Idempotent; never deletes history.""" with self._tx() as conn: row = conn.execute( "SELECT * FROM worker_registrations WHERE worker_identity = ?", (worker_identity,), ).fetchone() if row is None: return { "success": True, "released": False, "mutation_performed": False, "reasons": [f"no registration for {worker_identity!r}"], } conn.execute( "UPDATE worker_registrations SET status = ?, last_heartbeat_at = ? " "WHERE worker_identity = ?", (STATUS_RELEASED, _ts(now or _utc_now()), worker_identity), ) return { "success": True, "released": True, "mutation_performed": True, "reasons": [], } def _public_record(record: dict[str, Any]) -> dict[str, Any]: """Registry row minus anything that should not travel to an LLM surface.""" return { "worker_identity": record.get("worker_identity"), "client_name": record.get("client_name"), "client_instance_id": record.get("client_instance_id"), "session_id": record.get("session_id"), "generation_id": record.get("generation_id"), "role": record.get("role"), "profile": record.get("profile"), "namespace": record.get("namespace"), "remote": record.get("remote"), "repository_binding": record.get("repository_binding"), "pid": record.get("pid"), "transport": record.get("transport"), "started_at": record.get("started_at"), "last_heartbeat_at": record.get("last_heartbeat_at"), "fencing_epoch": record.get("fencing_epoch"), "status": record.get("status"), } # --- The single authoritative verdict ------------------------------------ def assess_provenance( *, registry: WorkerRegistry | None = None, worker_identity: str | None = None, generation_id: str | None = None, env: dict[str, str] | None = None, native_transport_bound: bool | None = None, namespace: str | None = None, profile: str | None = None, role: str | None = None, now: datetime | None = None, pid_alive_probe=None, stdin_is_tty: bool | None = None, declared_only: bool = False, ) -> dict[str, Any]: """Resolve client/session provenance once, for every surface to reuse. Runtime context, namespace health, namespace attachment, fleet inventory, capability resolution, and typed reconnect assessment all call this, so the contradiction that motivated #948 — one surface reporting ``client_managed`` while another reported ``manual_launch`` for the same process — cannot recur by construction. Two independent dimensions are reported: ``launch_provenance`` How the process was started. Answered from the environment, which is legitimate because the launcher is what sets it. Preserves the #686 manual-launch wall unchanged. ``session_ownership`` Which live client session owns this runtime now. Answered *only* from a live attachment record; no environment flag can establish it. Missing evidence yields :data:`PROVENANCE_UNPROVEN`, which is deliberately distinct from :data:`PROVENANCE_MANUAL`: "we cannot prove who owns this" is not the same claim as "a human launched this from a terminal", and only the latter justifies telling an operator to stop hand-launching servers. """ env_signal = env_provenance_signal(env) launch = assess_launch_provenance( env, stdin_is_tty=stdin_is_tty, declared_only=declared_only ) evidence: list[str] = [EVIDENCE_ENV_FLAG] if env_signal["keys_present"] else [] reasons: list[str] = [] record: dict[str, Any] | None = None liveness: dict[str, Any] = {"live": False, "reasons": ["no attachment record"]} if registry is not None and worker_identity: record = registry.get(worker_identity) if record is not None: pid_alive = ( pid_alive_probe(record.get("pid")) if pid_alive_probe is not None else None ) liveness = registry.is_live(record, now=now, pid_alive=pid_alive) if liveness["live"]: evidence.append(EVIDENCE_ATTACHMENT_RECORD) if pid_alive is not None: evidence.append(EVIDENCE_PROCESS_LIVENESS) if native_transport_bound: evidence.append(EVIDENCE_NATIVE_TRANSPORT) # Conflicting live claims on one generation are contradictory evidence: two # sessions cannot both currently own the same daemon, so neither claim is # trustworthy and the assessment fails closed rather than picking a winner. conflicting: list[dict[str, Any]] = [] effective_generation = generation_id or (record or {}).get("generation_id") if registry is not None and effective_generation: for holder in registry.list_workers(generation_id=effective_generation): if ( record is not None and holder["worker_identity"] == record["worker_identity"] ): continue pid_alive = ( pid_alive_probe(holder.get("pid")) if pid_alive_probe is not None else None ) if registry.is_live(holder, now=now, pid_alive=pid_alive)["live"]: conflicting.append(_public_record(holder)) trusted = bool(set(evidence) & SUFFICIENT_EVIDENCE) blocker = BLOCKER_NONE if conflicting and trusted: trusted = False blocker = BLOCKER_CONTRADICTORY reasons.append( f"generation {effective_generation!r} is claimed by " f"{len(conflicting) + 1} live sessions at once; contradictory ownership " "evidence fails closed (#948)" ) elif not trusted: blocker = BLOCKER_NO_ATTACHMENT if record is None: reasons.append( "no live client/session attachment record binds this runtime to a " "session; a process environment flag alone does not prove current " "session ownership (#948)" ) else: reasons.extend( f"attachment record is not live: {r}" for r in liveness["reasons"] ) if conflicting: session_ownership = OWNERSHIP_CONTESTED elif trusted: session_ownership = OWNERSHIP_OWNED else: session_ownership = OWNERSHIP_UNOWNED # The headline verdict. Session ownership decides it when proven; otherwise # the launch dimension supplies the more specific of "hand-launched" versus # "cannot tell", so remediation names the proof that is actually missing. if session_ownership == OWNERSHIP_OWNED: provenance = PROVENANCE_CLIENT_SESSION elif launch["launch_provenance"] == PROVENANCE_MANUAL: provenance = PROVENANCE_MANUAL reasons.extend(launch["reasons"]) else: provenance = PROVENANCE_UNPROVEN client_name = normalize_client_name( (record or {}).get("client_name") or parse_worker_identity(worker_identity)["client_name"] ) return { "provenance": provenance, # #948: the *launch* dimension, which is what this field has always # meant to the #686 manual-launch wall. Keeping it bound to launch # provenance means unifying the two models changes which code decides, # not what gets decided, so no mutation that used to be permitted is # newly refused. "is_client_managed": bool(launch["client_managed"]), "launch_provenance": launch["launch_provenance"], "launch_decided_by": launch["decided_by"], "launch_reasons": launch["reasons"], # The *ownership* dimension, which no environment flag can establish. "session_ownership": session_ownership, "session_owned": session_ownership == OWNERSHIP_OWNED, "trusted_evidence": trusted, "evidence": sorted(set(evidence)), "sufficient_evidence_kinds": sorted(SUFFICIENT_EVIDENCE), "env_signal": env_signal, "env_flag_only": bool(evidence) and not trusted and not conflicting, "attachment": _public_record(record) if record else None, "attachment_liveness": liveness, "client_name": client_name, "client_known": client_name != UNKNOWN_CLIENT, "session_id": (record or {}).get("session_id"), "client_instance_id": (record or {}).get("client_instance_id"), "generation_id": effective_generation, "worker_identity": (record or {}).get("worker_identity") or worker_identity, "fencing_epoch": (record or {}).get("fencing_epoch"), "conflicting_live_sessions": conflicting, "blocker_kind": blocker, "fail_closed": provenance != PROVENANCE_CLIENT_SESSION, # #948 AC17/AC41: a refusal names the worker it applies to, never the # profile, so unrelated healthy workers sharing that profile keep working. "scope": { "scope_kind": "worker_identity" if worker_identity else "generation", "worker_identity": worker_identity, "generation_id": effective_generation, "namespace": namespace, "profile": profile, "role": role, "profile_wide": False, "note": ( "role and profile are reusable capability definitions; this verdict " "binds only the named worker identity/generation" ), }, "reasons": reasons, } # --- Cohort scoping ------------------------------------------------------ def classify_cohort( records: list[dict[str, Any]], *, now: datetime | None = None, pid_alive_probe=None, ) -> dict[str, Any]: """Separate genuine duplicates from valid concurrent workers (#948 AC43). The pre-#948 scan flagged every additional process sharing a profile as a duplicate, which is exactly the fleet-wide wall #948 exists to remove. Sharing a role or profile is legitimate; reusing a *worker identity*, or two live sessions claiming one generation, is not. """ live: list[dict[str, Any]] = [] for record in records: pid_alive = ( pid_alive_probe(record.get("pid")) if pid_alive_probe is not None else None ) if WorkerRegistry.is_live(record, now=now, pid_alive=pid_alive)["live"]: live.append(record) by_identity: dict[str, list[dict[str, Any]]] = {} by_generation: dict[str, list[dict[str, Any]]] = {} for record in live: by_identity.setdefault(str(record.get("worker_identity")), []).append(record) by_generation.setdefault(str(record.get("generation_id")), []).append(record) duplicate_identities = sorted(k for k, v in by_identity.items() if len(v) > 1) contested_generations = sorted(k for k, v in by_generation.items() if len(v) > 1) shared_profiles = sorted( { str(r.get("profile")) for r in live if r.get("profile") and sum(1 for o in live if o.get("profile") == r.get("profile")) > 1 } ) blocked = bool(duplicate_identities or contested_generations) reasons: list[str] = [] if duplicate_identities: reasons.append( f"worker identity reused by more than one live worker: " f"{duplicate_identities}" ) if contested_generations: reasons.append( f"generation claimed by more than one live session: {contested_generations}" ) if shared_profiles and not blocked: reasons.append( f"profile(s) {shared_profiles} are shared by multiple live workers with " "distinct identities, which is permitted and does not block mutations (#948)" ) return { "live_worker_count": len(live), "live_workers": [_public_record(r) for r in live], "duplicate_identities": duplicate_identities, "contested_generations": contested_generations, "shared_profiles": shared_profiles, "profile_sharing_permitted": True, "blocked": blocked, "blocker_kind": ( BLOCKER_IDENTITY_COLLISION if duplicate_identities else BLOCKER_CONFLICTING_SESSIONS if contested_generations else BLOCKER_NONE ), # Never profile-wide: the block names the offending identities only. "blocked_worker_identities": sorted( set(duplicate_identities) | { str(r.get("worker_identity")) for gen in contested_generations for r in by_generation[gen] } ), "reasons": reasons, } def scope_runtime_failure( *, failure_kind: str, worker_identity: str | None = None, generation_id: str | None = None, profile: str | None = None, role: str | None = None, all_live_workers: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: """Scope a stale-runtime/reconnect failure to one worker (#948 AC41). Returns the set of workers actually affected and, separately, the healthy workers that must keep operating. A stale generation affects only its own ownership scope. """ workers = list(all_live_workers or []) affected = [ _public_record(w) for w in workers if (worker_identity and w.get("worker_identity") == worker_identity) or ( not worker_identity and generation_id and w.get("generation_id") == generation_id ) ] affected_ids = {a["worker_identity"] for a in affected} unaffected = [ _public_record(w) for w in workers if w.get("worker_identity") not in affected_ids ] return { "failure_kind": failure_kind, "scope_kind": "worker_identity" if worker_identity else "generation", "worker_identity": worker_identity, "generation_id": generation_id, "profile": profile, "role": role, "profile_wide": False, "fleet_wide": False, "affected_workers": affected, "unaffected_workers": unaffected, "unaffected_worker_count": len(unaffected), "reasons": [ f"{failure_kind} is scoped to " + ( f"worker identity {worker_identity!r}" if worker_identity else f"generation {generation_id!r}" ) + f"; {len(unaffected)} other live worker(s) sharing this role/profile " "are unaffected and remain able to mutate (#948 AC41)" ], } # --- Client-specific remediation ---------------------------------------- def reconnect_client_for( provenance: dict[str, Any] | None = None, *, fallback: str | None = None, ) -> str: """Name the client to address reconnect guidance to. Derived from the live attachment record, never a hardcoded vendor. When nothing identifies the client, :data:`UNKNOWN_CLIENT` is returned so the caller emits generic steps instead of steps for the wrong product. """ name = normalize_client_name((provenance or {}).get("client_name")) if name != UNKNOWN_CLIENT: return name return normalize_client_name(fallback) # --- Remote binding ------------------------------------------------------ def resolve_bound_remote( *, requested_remote: str | None, bound_remote: str | None, default_remote: str | None = None, ) -> dict[str, Any]: """Keep a namespace on the remote it is bound to (#948). A tool default is not a decision. When a namespace is bound to a remote, an omitted argument resolves to that binding rather than silently falling through to the library default and hitting the wrong host; an argument that contradicts the binding is refused rather than honoured. """ requested = (requested_remote or "").strip() or None bound = (bound_remote or "").strip() or None default = (default_remote or "").strip() or None if bound and requested and requested != bound: return { "remote": bound, "resolved_from": "session_binding", "drifted": True, "honoured_request": False, "reasons": [ f"requested remote {requested!r} contradicts the session binding " f"{bound!r}; the binding wins and the request is refused so a " "PRGS-bound namespace never mutates another host (#948)" ], } if bound and not requested: return { "remote": bound, "resolved_from": "session_binding", "drifted": False, "honoured_request": False, "reasons": [ f"remote omitted; resolved to the session binding {bound!r} rather " f"than the library default {default!r} (#948)" ], } if requested: return { "remote": requested, "resolved_from": "explicit_argument", "drifted": False, "honoured_request": True, "reasons": [], } return { "remote": default, "resolved_from": "library_default", "drifted": False, "honoured_request": False, "reasons": [ "no remote requested and no session binding recorded; falling back to " f"the library default {default!r}. Bind the session or pass remote " "explicitly to avoid host drift (#948)." ], }