Gitea-Tools/mcp_worker_identity.py

"""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]+")

#: ``<llm-name>-<UTC-timestamp>-<short-sha>`` per the #948 scope clarification.
WORKER_IDENTITY_RE = re.compile(
    r"^(?P<client>[a-z0-9][a-z0-9_]*)-(?P<ts>\d{8}T\d{6}Z)-(?P<digest>[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 ``<llm-name>-<UTC-timestamp>-<short-sha>`` (#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 <llm-name>-<UTC-timestamp>-<short-sha>, "
                "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)."
        ],
    }