"""Structured update receipts + post-update fleet version verification.

Phase 1 of the fleet-update reliability plan (#91277): the updater must
*prove* its outcome instead of assuming it.

Two additive capabilities, both designed so a failure inside them can never
break an update (every public entry point is exception-swallowing):

1. **Update receipt** — a machine-readable JSON record of what one
   ``hermes update`` run discovered, did, skipped (and why), written to
   ``<HERMES_HOME>/logs/update_receipts/``. Silent-failure classes this
   makes visible: #88848 (helper died after "success" printed), #74973
   (restart silently skipped), #85753 (restart phase never ran), #81193
   (desktop shows failure for a successful update).

2. **Fleet version verification** — after the restart phase, read every
   profile's ``gateway_state.json``, compare each live gateway's stamped
   ``code_sha`` (written by ``gateway/status.py`` on every runtime-status
   write) against the freshly-updated checkout's HEAD, and print a fleet
   version matrix. Mixed-version fleets (#88654, #69754, #77553, #56717)
   become a loud, actionable report instead of a latent state.

Deployment-kind awareness (docker/image-managed installs) rides on
``hermes_cli.version_info.get_code_identity()``: a packaged build reports
its install-stamp provenance (``source="docker"``/``"nix"``/…) and the
receipt records that the install is not in-place updatable.
"""

from __future__ import annotations

import contextvars
import copy
import json
import logging
import os
import sys
import time
import uuid
from contextlib import contextmanager, suppress
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional

logger = logging.getLogger(__name__)

_RECEIPT_KEEP = 20  # keep the last N receipts per profile home
COMMAND_BOUNDARY_STOP_REASON = "completed at command boundary"

# Receipt state is per-CONTEXT, not a module global: a nested
# ``hermes update`` receipt (or one in another thread) must never clobber
# the outer one, and the boundary finalize must see exactly its own
# process's receipt. Same pattern as pm.receipt's ContextVars — no
# manager object.
_current: contextvars.ContextVar[Optional["UpdateReceipt"]] = contextvars.ContextVar(
    "update_receipt_current", default=None
)


@contextmanager
def update_receipt_scope():
    """Keep the command's finalization guard away from an enclosing update."""
    token = _current.set(None)
    try:
        yield
    finally:
        _current.reset(token)


def current_correlation_id() -> Optional[str]:
    """The update correlation id in force in this context, or None.

    Derived from the OPEN update receipt itself — one source of truth, no
    duplicate id variable: a nested update's begin replaces the current
    receipt (so syncs begun under it capture the nested id), and its
    finalize RESTORES the outer receipt via the ContextVar token (so the
    outer id comes back for the rest of the outer update)."""
    current = _current.get()
    return current.correlation_id if current is not None else None


def _utc_now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def _code_identity(refresh: bool = False) -> dict[str, Any]:
    """Running-code identity, or ``{}`` when the probe fails."""
    with suppress(Exception):
        from hermes_cli.version_info import get_code_identity

        return get_code_identity(refresh=refresh) or {}
    return {}


def _str_records(entries: Any, keys: tuple[str, ...], *, pid: bool = False) -> list[dict[str, Any]]:
    """Dict entries reduced to stringified ``keys`` (plus an int ``pid`` first when requested)."""
    records = []
    for entry in entries:
        if not isinstance(entry, dict):
            continue
        record: dict[str, Any] = {"pid": int(entry.get("pid", 0) or 0)} if pid else {}
        record.update({key: str(entry.get(key, "")) for key in keys})
        records.append(record)
    return records


def _launcher_correlation_id() -> Optional[str]:
    """Correlation id handed in by an external launcher, or None.

    Desktop's managed SSH update exports ``HERMES_UPDATE_CORRELATION_ID`` and
    only accepts a receipt whose ``correlation_id`` equals it. This is separate
    from ``update_id``, which stays the CLI's own id for pm sync receipts."""
    return os.environ.get("HERMES_UPDATE_CORRELATION_ID", "").strip() or None


class UpdateReceipt:
    """Collects the observable facts of one ``hermes update`` run."""

    def __init__(self) -> None:
        self.data: dict[str, Any] = {
            "schema": 1, "started_at": _utc_now_iso(), "finished_at": None,
            "argv": list(sys.argv), "pid": os.getpid(),
            "outcome": "running",  # running | success | partial | failed
            "pre_update": _code_identity(), "post_update": {},
            "steps": [], "skips": [], "gateway_restart": {}, "fleet": [],
        }
        # The id binds this update to the pm sync receipts begun under it
        # (pm.receipt captures it via the _correlation ContextVar).
        self.correlation_id = uuid.uuid4().hex
        self.data["update_id"] = self.correlation_id
        self.data["correlation_id"] = _launcher_correlation_id()

    def step(self, name: str, ok: bool, detail: str = "") -> None:
        self.data["steps"].append({"name": name, "ok": bool(ok), "detail": detail, "at": _utc_now_iso()})

    def skip(self, name: str, reason: str) -> None:
        self.data["skips"].append({"name": name, "reason": reason, "at": _utc_now_iso()})

    def stage(self, name: str, outcome: str, **facts: str) -> None:
        # Stage END marks: a stage's duration is the gap since the previous mark (or started_at).
        self.data.setdefault("stages", []).append({"name": name, "outcome": outcome, "at": _utc_now_iso(), **facts})

    def fact(self, key: str, value: Any) -> None:
        self.data[key] = value

    def gateway_restart_result(
        self, *, restarted_services: list | None = None, relaunched_profiles: list | None = None,
        externally_supervised_profiles: list | None = None, killed_pids: list | None = None,
        failed_units: list | None = None, incomplete: bool = False, phase_error: str = "",
        fresh_recovery: dict[str, Any] | None = None,
    ) -> None:
        result: dict[str, Any] = {
            "restarted_services": list(restarted_services or []),
            "relaunched_profiles": list(relaunched_profiles or []),
            "externally_supervised_profiles": list(externally_supervised_profiles or []),
            "killed_pids": [int(p) for p in (killed_pids or [])],
            "failed_units": [str(u) for u in (failed_units or [])],
            "incomplete": bool(incomplete),
            "phase_error": phase_error,
        }
        if fresh_recovery is not None:
            # Conservative outcome vocabulary: "verified" is the only bucket allowed to claim
            # supervisor coverage; "relaunch_attempted" means the relaunch exited 0 without
            # independent supervisor observation. "skipped" preserves runtimes (manual gateways,
            # serve/dashboard entries) the pass deliberately did not touch.
            persisted: dict[str, Any] = {
                key: [str(profile) for profile in fresh_recovery.get(key, [])]
                for key in ("requested", "verified", "relaunch_attempted", "failed")
            }
            persisted["skipped"] = _str_records(
                fresh_recovery.get("skipped", []), ("profile", "kind", "supervisor", "reason")
            )
            # ``hermes serve`` hosts tui_gateway and is not a gateway profile, so neither the
            # per-profile buckets above nor the fleet-version matrix can describe it. Persist its
            # unit outcomes and any process that survived on the pre-update generation, or the
            # receipt keeps claiming a clean recovery the operator's box contradicts.
            serve_units = fresh_recovery.get("serve_units") or {}
            persisted["serve_units"] = {
                key: [str(unit) for unit in (serve_units.get(key) or [])] for key in ("verified", "failed")
            }
            persisted["stale_runtimes"] = _str_records(
                fresh_recovery.get("stale_runtimes", []), ("kind", "profile", "supervisor"), pid=True
            )
            result["fresh_recovery"] = persisted
        self.data["gateway_restart"] = result

    def finalize(self, outcome: str) -> None:
        self.data["outcome"] = outcome
        self.data["finished_at"] = _utc_now_iso()
        self.data["post_update"] = _code_identity(refresh=True)


def _receipt_dir() -> Path:
    # ``hermes_constants`` (stdlib-only), never ``hermes_cli.config``: the receipt must be
    # writable from the refused/failed paths where config loading itself may be what broke
    # (#112465, #112558).
    from hermes_constants import get_hermes_home

    return get_hermes_home() / "logs" / "update_receipts"


def begin_update_receipt(*, previous: dict | None = None, correlation_id: str | None = None) -> None:
    """Start recording a new update receipt.

    Nested updates are safe: the previous receipt (if any) is preserved
    behind the ContextVar token and comes back when this one finalizes —
    a nested begin/finalize never drops the outer update's receipt or
    correlation. Never raises."""
    try:
        receipt = UpdateReceipt()
        if previous:
            receipt.data.update(copy.deepcopy(previous))
        receipt.correlation_id = correlation_id or receipt.correlation_id
        receipt.data.update(update_id=receipt.correlation_id, outcome="running", finished_at=None)
        # A handoff receipt from an older interpreter may predate the field.
        receipt.data["correlation_id"] = receipt.data.get("correlation_id") or _launcher_correlation_id()
    except Exception as exc:  # pragma: no cover - defensive
        logger.debug("Could not start update receipt: %s", exc)
        return
    receipt.current_token = _current.set(receipt)


def _record(method: str, what: str, *args: Any, **kwargs: Any) -> None:
    """Invoke ``method`` on the active receipt; no-op when none, never raises.

    Copy-on-write, same rule as pm.receipt: a copied context (copy_context,
    asyncio.to_thread) inherits the SAME receipt object — mutate a clone
    and re-set it in THIS context only, so a child's records never leak
    into (or corrupt) the parent's receipt.
    """
    try:
        import copy

        current = _current.get()
        if current is None:
            return
        clone = copy.copy(current)
        clone.data = copy.deepcopy(current.data)
        getattr(clone, method)(*args, **kwargs)
        _current.set(clone)
    except Exception as exc:  # pragma: no cover - defensive
        logger.debug("Could not record %s: %s", what, exc)


def record_step(name: str, ok: bool, detail: str = "") -> None:
    """Record one update step outcome. No-op when no receipt is active."""
    _record("step", f"update step {name}", name, ok, detail)


def record_skip(name: str, reason: str) -> None:
    """Record a skipped step WITH the reason it was skipped."""
    _record("skip", f"update skip {name}", name, reason)


def record_stage(name: str, outcome: str, **facts: str) -> None:
    """Mark the END of a pipeline stage (``success``/``failed``/``skipped``) with a timestamp."""
    _record("stage", f"update stage {name}", name, outcome, **facts)


def record_fact(key: str, value: Any) -> None:
    """Set one top-level receipt field (e.g. ``initiator``)."""
    _record("fact", f"update fact {key}", key, value)


def record_gateway_restart(**kwargs: Any) -> None:
    """Record the gateway restart phase outcome (see UpdateReceipt)."""
    _record("gateway_restart_result", "gateway restart result", **kwargs)


def finalize_update_receipt(outcome: str, fleet: list | None = None, stop_reason: str = "") -> Optional[Path]:
    """Finalize + persist the receipt (``success``/``partial``/``failed``/``refused``); path or None.

    Exactly-once by construction: the context's receipt is popped first, so a second call (e.g. the
    command-boundary safety net after an inner path already finalized) is a no-op returning None.
    The receipt is popped via its OWN begin token, so a nested update's finalize RESTORES the outer
    update's open receipt and correlation instead of discarding them.
    """
    current = _current.get()
    if current is None:
        return None
    receipt = copy.copy(current)
    receipt.data = copy.deepcopy(current.data)
    token = getattr(receipt, "current_token", None)
    try:
        if token is not None:
            _current.reset(token)
        else:  # pragma: no cover - receipts begun before token binding
            _current.set(None)
    except ValueError:
        # Token from another context (finalize ran in a copied context) —
        # pop THIS context only, so exactly-once still holds.
        _current.set(None)
    try:
        receipt.finalize(outcome)
        if stop_reason:
            receipt.data["stop_reason"] = stop_reason
        if fleet is not None:
            receipt.data["fleet"] = fleet
        # Manual serve restart obligations outlive one receipt rotation: carry the previous
        # receipt's still-pending rows forward so the startup warning survives (see
        # update_serve_obligations).
        from hermes_cli.update_serve_obligations import retain_receipt_manual_serves
        pending = retain_receipt_manual_serves(read_latest_receipt() or {})
        if pending:
            receipt.data["pending_manual_serves"] = pending
        # EMBED the pm sync sections (the settled receipts contract): the
        # update's rebuild/bisect ran through pm's own sync receipt, which
        # finalizes before this one. Fold in only the completion filed
        # under THIS update's correlation id (pm.receipt.last_for_update,
        # per-context, never latest.json) — a sync from before this update
        # began, a standalone sync, or one in a concurrent thread cannot
        # be misattributed, and a nested update's sync cannot displace
        # this update's own. ONE file still carries the whole story
        # (desktop reads a single latest.json).
        try:
            from pm import receipt as pm_receipt

            sync = pm_receipt.last_for_update(receipt.correlation_id, consume=True)
            if isinstance(sync, dict):
                for key in ("venv_rebuild", "plugin_bisect", "feature_list", "steps", "exit_code"):
                    if sync.get(key) is not None:
                        receipt.data[f"pm_{key}"] = sync[key]
                if sync.get("outcome") is not None:
                    receipt.data["pm_sync_outcome"] = sync.get("outcome")
                if sync.get("warnings"):
                    receipt.data["pm_warnings"] = sync["warnings"]
                if sync.get("refusal") is not None:
                    receipt.data["pm_refusal"] = sync["refusal"]
        except Exception as exc:  # pragma: no cover — embedding is additive
            logger.debug("pm sync-section embed skipped: %s", exc)
        directory = _receipt_dir()
        directory.mkdir(parents=True, exist_ok=True)
        # Unique name: stamp+pid collides for nested/concurrent receipts in
        # the same process+second — the correlation id makes the name unique
        # per update run. Atomic write for BOTH the stamped receipt and the
        # latest.json pointer (no torn readers).
        from hermes_cli.runtime_state import _atomic_bytes

        path = directory / (
            f"update_{time.strftime('%Y%m%d_%H%M%S')}_{os.getpid()}_"
            f"{receipt.correlation_id}.json"
        )
        payload = (json.dumps(receipt.data, indent=2, default=str) + "\n").encode("utf-8")
        _atomic_bytes(path, payload)
        with suppress(Exception):  # stable pointer for the dashboard/desktop
            _atomic_bytes(directory / "latest.json", payload)
        _prune_old_receipts(directory)
        _publish_shared_metrics(receipt.data)
        return path
    except Exception as exc:
        # Visible, not debug: a run that pulled code and left no receipt is exactly the run
        # operators need to post-mortem, and INFO-level logs discard debug (#112465, #112558).
        logger.warning("Could not write update receipt (%s): %s", outcome, exc)
        print(f"  ⚠ Update receipt not written: {exc}")
        return None


def _collection_enabled_now() -> Optional[bool]:
    """Shared-metrics consent via the ALREADY-LOADED config module (never an import: this interpreter
    predates the checkout swap). None when it cannot tell (not loaded, unreadable config)."""
    reader = getattr(sys.modules.get("hermes_cli.config"), "read_raw_config_readonly", None)
    if reader is None:
        return None
    try:
        config: Any = reader()
    except Exception:
        return None
    if type(config) is not dict:  # FailedConfigRead: a fallback, not what the user chose
        return None
    for key in ("telemetry", "shared_metrics"):
        config = config.get(key) if isinstance(config, dict) else None
    return isinstance(config, dict) and config.get("enabled") is True


def _metric_receipt(data: dict[str, Any]) -> dict[str, Any]:
    """Only what shared_metrics_update.update_receipt_fields reads; never argv or step text."""
    pre = data.get("pre_update") if isinstance(data.get("pre_update"), dict) else {}
    return {
        "update_id": data.get("update_id"), "started_at": data.get("started_at"),
        "finished_at": data.get("finished_at"), "outcome": data.get("outcome"),
        "initiator": "desktop" if data.get("initiator") == "desktop" else None,
        "pre_update": {"commit_date": pre.get("commit_date")},
        "stages": [
            {key: mark[key] for key in ("name", "outcome", "at", "mode") if key in mark}
            for mark in data.get("stages") or () if isinstance(mark, dict)
        ],
        "steps": [
            {"name": "admission", "ok": bool(step.get("ok"))}
            for step in data.get("steps") or () if isinstance(step, dict) and step.get("name") == "admission"
        ],
        "fleet": [{"state": row.get("state")} for row in data.get("fleet") or () if isinstance(row, dict)],
    }


def _publish_shared_metrics(data: dict[str, Any]) -> None:
    """hermes.update.run/stage from this FINAL receipt; must never fail or slow the update."""
    with suppress(Exception):
        pre, post = data.get("pre_update") or {}, data.get("post_update") or {}
        if data.get("pid") == os.getpid() and not (pre.get("sha") and pre.get("sha") == post.get("sha")):
            # This interpreter began the run before the checkout swap: importing now would load
            # pulled code into it. Park the bounded fields (stdlib + loaded modules only); the next
            # Hermes start records them.
            from hermes_constants import get_hermes_home
            from hermes_cli.runtime_state import _atomic_bytes

            pending = get_hermes_home() / "telemetry" / "shared_metrics" / "pending_updates"  # = PENDING_DIRNAME
            enabled = _collection_enabled_now()
            if enabled is False:
                import shutil

                shutil.rmtree(pending, ignore_errors=True)  # opted out: nothing parked may be counted later
            elif enabled:
                pending.mkdir(parents=True, exist_ok=True)
                _atomic_bytes(pending / f"{data.get('update_id')}.json", json.dumps(_metric_receipt(data), default=str).encode())
            return
        from hermes_cli.observability.shared_metrics_update import record_update_receipt

        record_update_receipt(data)


def finalize_pending_update_receipt(exit_code: Optional[int] = None, stop_reason: str = "") -> Optional[Path]:
    """Command-boundary safety net: persist a still-open receipt, if any. Never raises.

    ``hermes update`` has many early ``sys.exit`` paths (preflight refusals, venv-holder refusal,
    fetch failure) predating the inner finalize calls; finalizing here means refused/failed runs —
    where a receipt matters most — leave a record. Exit 0/None → ``success``, exit 2 → ``refused``
    (preflight convention), else → ``failed``.

    No-op when no receipt is open (the inner paths already finalized — exactly-once via the popped
    per-context receipt) or when recording was never started. See #91283.
    """
    current = _current.get()
    if current is None:
        return None
    outcome = "success" if exit_code in (0, None) else "refused" if exit_code == 2 else "failed"
    if exit_code is not None:
        with suppress(Exception):
            clone = copy.copy(current)
            clone.data = copy.deepcopy(current.data)
            clone.data["exit_code"] = int(exit_code)
            _current.set(clone)
    return finalize_update_receipt(outcome, stop_reason=stop_reason)


def _prune_old_receipts(directory: Path) -> None:
    with suppress(Exception):
        receipts = (p for p in directory.glob("update_*.json") if p.is_file())
        for stale in sorted(receipts, key=lambda p: p.stat().st_mtime, reverse=True)[_RECEIPT_KEEP:]:
            with suppress(OSError):
                stale.unlink()


def settle_latest_receipt_fleet(fleet: list[dict[str, Any]], *, discharges) -> bool:
    """Record on ``latest.json`` that the fleet it still reports as owed now serves the checkout.

    A failed receipt whose plan rows cannot be matched to a live gateway (unknown identity,
    pre-pull SHAs) keeps ``hermes update`` exiting 1 and every CLI start warning about mixed
    modules, long after the operator's ``hermes gateway restart`` fixed the fleet (#117051). The
    caller has just verified every live row is current at the checkout SHA; persisting that
    matrix as the receipt's post-restart ``fleet`` (and un-flagging ``gateway_restart``) is what
    lets the stale-runtime readers see the recovery. ``discharges(settled_receipt)`` decides on
    the in-memory copy; ``latest.json`` is rewritten only when it answers True, so a catch-up
    that still exits 1 leaves the receipt byte-identical. Only the ``latest.json`` pointer is
    rewritten; the archived per-run file keeps the original outcome. Never raises.
    """
    try:
        path = _receipt_dir() / "latest.json"
        receipt = json.loads(path.read_text(encoding="utf-8-sig"))
        if not isinstance(receipt, dict):
            return False
        receipt["fleet"] = list(fleet)
        gateway_restart = receipt.get("gateway_restart")
        if not isinstance(gateway_restart, dict):
            gateway_restart = {}
        gateway_restart.update({"incomplete": False, "phase_error": ""})
        gateway_restart["settled_from_live_fleet_at"] = _utc_now_iso()
        receipt["gateway_restart"] = gateway_restart
        if not discharges(receipt):
            return False
        path.write_text(json.dumps(receipt, indent=2, default=str), encoding="utf-8")
        return True
    except Exception as exc:
        logger.debug("Could not settle latest update receipt from the live fleet: %s", exc)
        return False


def read_latest_receipt() -> Optional[dict[str, Any]]:
    """Read the most recent update receipt, or None. Never raises."""
    with suppress(Exception):
        path = _receipt_dir() / "latest.json"
        if not path.is_file():
            return None
        payload = json.loads(path.read_text(encoding="utf-8-sig"))
        return payload if isinstance(payload, dict) else None
    return None


def _profile_homes() -> list[tuple[str, Path]]:
    """``(profile, home)`` for the default home plus every valid named profile dir, sorted."""
    from hermes_cli.profiles import _get_default_hermes_home, _get_profiles_root, _PROFILE_ID_RE

    homes: list[tuple[str, Path]] = []
    default_home = _get_default_hermes_home()
    if default_home.is_dir():
        homes.append(("default", default_home))
    root = _get_profiles_root()
    if root.is_dir():
        homes.extend(
            (entry.name, entry)
            for entry in sorted(root.iterdir())
            if entry.is_dir() and entry.name != "default" and _PROFILE_ID_RE.match(entry.name)
        )
    return homes


def _socket_identity(home: Path) -> Optional[tuple[int, dict]]:
    """``(pid, identity)`` declared by the gateway owning ``home``'s control socket, else None.

    A live ``identify`` answer is authoritative — no PID-reuse or stale-file heuristics. Callers
    fall back to ``gateway_state.json`` for gateways that predate the socket or whose socket
    didn't bind.
    """
    try:
        # Prefer the gateway-owned control socket (#92091): identity declared by the process itself,
        # including its own supervisor provenance — no argv/PID inference. Scan fallback below.
        from gateway.control_socket import identify_gateway

        identity = identify_gateway(home)
        return (int(identity.get("pid")), identity) if identity else None
    except Exception:  # probe failure, no gateway, or an unparseable pid
        return None


_CODE_ROOT_MAX_DEPTH = 8


def _code_root_for_path(raw: Any) -> Optional[Path]:
    """Return the Hermes checkout containing an absolute process path."""
    if not isinstance(raw, str) or not raw:
        return None
    with suppress(Exception):
        candidate = Path(raw)
        if not candidate.is_absolute():
            return None
        for parent in [candidate, *candidate.parents][:_CODE_ROOT_MAX_DEPTH]:
            if (parent / "hermes_cli" / "main.py").is_file():
                return parent.resolve()
    return None


def _updater_code_root() -> Optional[Path]:
    return _code_root_for_path(str(Path(__file__).resolve()))


def _gateway_code_root(pid: int, home: Path) -> Optional[Path]:
    """Resolve the checkout served by a verified live gateway when possible."""
    # Older gateways cannot publish a new identity field, but their pid-guarded
    # status record already carries sys.argv (whose first item is the resolved
    # module path for ``python -m`` launches).
    with suppress(Exception):
        from gateway.status import read_runtime_status

        record = read_runtime_status(home / "gateway_state.json") or {}
        if int(record.get("pid")) == pid:
            for probe in record.get("argv") or []:
                root = _code_root_for_path(probe)
                if root:
                    return root

    with suppress(Exception):
        import psutil  # type: ignore

        process = psutil.Process(pid)
        with suppress(Exception):
            root = _code_root_for_path((process.environ() or {}).get("VIRTUAL_ENV"))
            if root:
                return root
        probes: list[Any] = []
        with suppress(Exception):
            probes.append(process.exe())
        with suppress(Exception):
            probes.extend(process.cmdline() or [])
        for probe in probes:
            root = _code_root_for_path(probe)
            if root:
                return root
    return None


EXTERNAL_STATE = "external"
# A gateway this updater runs INSIDE that accepted a self-restart request: it is still on the
# pre-update code by construction and restarts once the updater exits (#100179 / #119597).
RESTART_PENDING_STATE = "restart_pending"


def row_is_external(row: Any) -> bool:
    """Whether a fleet row serves a checkout this update did not touch."""
    return isinstance(row, dict) and row.get("state") == EXTERNAL_STATE


def _fleet_row(
    profile: str, pid: int, code_sha: Any, code_version: Any, expected_sha: Any,
    state: str = "unknown", code_root: Optional[Path] = None,
    expected_root: Optional[Path] = None, served_profiles: Any = None,
    self_restart_pending: Optional[set] = None,
) -> dict[str, Any]:
    if state == "unknown" and code_root and expected_root and code_root != expected_root:
        state = EXTERNAL_STATE
    if state == "unknown" and code_sha and expected_sha:
        state = "current" if str(code_sha) == str(expected_sha) else "stale"
    if state == "stale" and self_restart_pending and pid in self_restart_pending:
        state = RESTART_PENDING_STATE
    row = {
        "profile": profile, "pid": pid, "code_sha": str(code_sha) if code_sha else None,
        "code_version": code_version, "state": state,
        "code_root": str(code_root) if code_root else None,
    }
    # A live, identity-verified multiplexer represents every profile in this
    # list. Keep the field only when its shape is usable: callers use it to
    # discharge per-profile restart obligations, so corrupt status must not
    # widen coverage.
    if isinstance(served_profiles, list) and served_profiles and all(
        isinstance(name, str) and name for name in served_profiles
    ):
        row["served_profiles"] = list(dict.fromkeys(served_profiles))
    return row


# Runtime-status states that do not describe a gateway that should be running now — no down row.
_NOT_EXPECTED_STATES = {"stopped", "startup_failed"}


def collect_fleet_versions(
    *, pre_restart_pids: Optional[list[int]] = None, self_restart_pending: Optional[set] = None,
) -> list[dict[str, Any]]:
    """Snapshot every profile's gateway code identity vs. the current tree.

    ``self_restart_pending`` — pids of gateways that are ancestors of this updater and accepted a
    self-restart request (cron update inside the gateway tree, #100179). They can only restart after
    this process exits, so their pre-update ``code_sha`` is expected: such a row is
    ``restart_pending`` instead of ``stale`` and does not fail the matrix (#119597). Every other
    live gateway on the old sha keeps its ``stale`` verdict.

    Rollout safety: ``down`` requires membership in ``pre_restart_pids`` — a stale state file from a
    long-dead gateway (machine reboot, manual kill weeks ago) must NOT fail every future update.
    Without a pre-restart snapshot (``None``/empty) dead PIDs are skipped (historical behavior).

    ``stale``   — gateway stamped a code_sha that differs from the updated checkout's HEAD (it is still
    serving pre-update modules). ``unknown`` — gateway predates the code-identity stamp (started before this
    feature landed), identity could not be resolved, or the state file's live PID is not the
    verified gateway for that home (``live_gateway_pid_for_home``): ``write_runtime_status`` re-stamps
    ``pid``/``code_sha`` for whatever process writes it, so a foreign writer must never read as
    ``current`` (#110420, sibling of #109680). ``down``    — the gateway was ALIVE when this update
    started (``pre_restart_pids``), its runtime status still says running, but the PID is dead and no
    successor rewrote the record: the restart phase stopped it and nothing came back. Without this row a
    killed-and-never-replaced gateway produced NO entry at all and the matrix passed silently (Phase-1
    verification gap, #88848/#74973 class).
    """
    _pre_restart = {int(p) for p in (pre_restart_pids or []) if isinstance(p, int)}
    _pending = {int(p) for p in (self_restart_pending or ()) if isinstance(p, int)}
    results: list[dict[str, Any]] = []
    expected_sha = _code_identity(refresh=True).get("sha")
    expected_root = _updater_code_root()
    try:
        from gateway.status import (
            live_gateway_pid_for_home,
            read_runtime_status,
            runtime_status_pid_is_live,
        )

        for profile, home in _profile_homes():
            sock = _socket_identity(home)
            if sock is not None:
                pid, identity = sock
                row = _fleet_row(
                    profile, pid, identity.get("code_sha"), identity.get("code_version"), expected_sha,
                    served_profiles=identity.get("served_profiles"),
                    code_root=_gateway_code_root(pid, home), expected_root=expected_root,
                    self_restart_pending=_pending,
                )
                results.append({**row, "source": "socket"})
                continue
            record = read_runtime_status(home / "gateway_state.json")
            if not record:
                continue
            try:
                pid = int(record.get("pid"))
            except (TypeError, ValueError):
                continue
            # A state file is only a fallback claim. Its SHA is evidence about
            # its own PID only when the profile's canonical identity resolver
            # verifies that same live gateway.
            if live_gateway_pid_for_home(home) == pid:
                results.append(
                    _fleet_row(
                        profile, pid, record.get("code_sha"), record.get("code_version"), expected_sha,
                        served_profiles=record.get("served_profiles"),
                        code_root=_gateway_code_root(pid, home), expected_root=expected_root,
                        self_restart_pending=_pending,
                    )
                )
                continue
            # A live non-gateway (or a gateway for another profile) can write a
            # plausible state file. Keep the fail-open visibility row, but never
            # let that file's self-reported SHA or version classify or label
            # the process — both claims have the same trust problem.
            if runtime_status_pid_is_live(record):
                results.append(_fleet_row(profile, pid, None, None, None))
                continue
            # Dead PID (or a live PID recycled by an unrelated process during the update's own
            # churn): a DOWN row only when this exact pid was alive at update start AND the record
            # still claims a running state — "the restart phase stopped it and nothing came back."
            # Everything else (clean stop, startup failure, long-dead stale record) keeps the no-row
            # behavior so the rollout can't false-positive. ``_pre_restart`` is a bare PID set, not
            # (pid, start_time) pairs, so a recycled PID from gateway A landing in B's stale record
            # could still mislabel B as down — inherent to the snapshot's data model.
            # See #93258.
            gw_state = record.get("gateway_state")
            if pid in _pre_restart and isinstance(gw_state, str) and gw_state and gw_state not in _NOT_EXPECTED_STATES:
                results.append(_fleet_row(profile, pid, None, record.get("code_version"), None, state="down"))
    except Exception as exc:
        logger.debug("Fleet version probe failed: %s", exc)
    return results


_FLEET_ROW_LINES = {
    "current": "  ✓ {profile} (pid {pid}) @ {short} — up to date",
    "stale": "  ✗ {profile} (pid {pid}) @ {short} — STALE (pre-update code)",
    "down": "  ✗ {profile} — DOWN (gateway was running before the update; pid {pid} is gone and nothing replaced it)",
    "external": "  ◆ {profile} (pid {pid}) @ {short} — separate checkout, not updated by this run",
    RESTART_PENDING_STATE: (
        "  ↻ {profile} (pid {pid}) @ {short} — restart pending (deferred until this process exits;"
        " the update runs inside this gateway)"
    ),
}
_FLEET_ROW_UNKNOWN = "  ? {profile} (pid {pid}) — version unknown (gateway predates version stamping; restart to enable)"
# A gateway pid the pre-update snapshot did not know that had not published its code identity when
# the settle window closed (#112634): most likely the successor this update relaunched, still
# booting, so "restart to enable" would be wrong — but the poll never observed the restart itself,
# so the copy does not claim one.
_FLEET_ROW_IDENTITY_PENDING = (
    "  ? {profile} (pid {pid}) — new pid since the update, code identity not published yet"
    " — re-check with `hermes gateway status`"
)


def print_fleet_version_matrix(fleet: list[dict[str, Any]]) -> bool:
    """Print the post-update fleet version matrix.

    Returns True when at least one gateway is provably stale (still serving pre-update code) OR
    provably down (killed by the restart phase, nothing came back), so the caller can escalate.
    ``unknown`` entries are reported but do NOT fail the update: gateways started before the
    code-identity stamp existed have no sha to compare, and failing them would be a false-positive
    storm. ``restart_pending`` entries (the gateway this updater runs inside, self-restart
    accepted) are on the old code by construction and do not fail it either (#119597).
    """
    if not fleet:
        return False
    print()
    print("Fleet version check:")
    states = set()
    external_roots: list[str] = []
    for entry in fleet:
        sha = entry.get("code_sha")
        states.add(entry.get("state"))
        if row_is_external(entry) and entry.get("code_root"):
            external_roots.append(f"{entry.get('profile')}: {entry.get('code_root')}")
        fallback = _FLEET_ROW_IDENTITY_PENDING if entry.get("identity_pending") else _FLEET_ROW_UNKNOWN
        print(_FLEET_ROW_LINES.get(entry.get("state"), fallback).format(
            profile=entry.get("profile"), pid=entry.get("pid"), short=sha[:8] if isinstance(sha, str) and sha else "?",
        ))
    if external_roots:
        print()
        print("  ℹ These profiles run their own checkout and are updated separately:")
        for line in external_roots:
            print(f"      {line}")
    if RESTART_PENDING_STATE in states:
        print()
        print("  ℹ A restart-pending gateway has not yet been verified on the new code;")
        print("    check after this update exits with `hermes gateway status`.")
    stale_or_down = sum(1 for entry in fleet if entry.get("state") in ("stale", "down"))
    if stale_or_down:
        print()
        if "stale" in states:
            print("  ⚠ Stale gateways keep serving pre-update code until restarted.")
        if "down" in states:
            print("  ⚠ Down gateways stopped serving messaging entirely.")
        # ``✓ Update complete!`` was already printed before the restart phase (the exit code
        # must land before a systemd restart can kill this process), so this verdict has to
        # supersede it explicitly — otherwise the output says success while the exit code is 1.
        print()
        print(
            f"✗ Update not complete: {stale_or_down} gateway(s) still running the old code (or stopped).")
        print("  Run `hermes gateway restart` (or `hermes -p <profile> gateway restart` for a named")
        print("  profile), then `hermes gateway status` to confirm.")
    return stale_or_down > 0