| """Detect when the gateway is running stale code after a hot ``git pull``. |
| |
| The gateway's ``sys.modules`` is frozen at boot. If the checkout is updated |
| underneath it, a first-time lazy import can resolve a freshly-pulled module |
| against a stale cached dependency -> ImportError. We snapshot the revision at |
| startup so risky callers (e.g. ``/model`` switching) can refuse with a clear |
| "restart the gateway" message. If the revision can't be read (non-git install, |
| IO error) the boot snapshot stays ``None`` and detection no-ops — never a false positive. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from pathlib import Path |
|
|
| _PROJECT_ROOT = Path(__file__).resolve().parent.parent |
| _boot_fingerprint: str | None = None |
|
|
|
|
| def _fingerprint() -> str | None: |
| """Current checkout fingerprint via the CLI's worktree-aware git-rev reader |
| (``hermes_cli.main`` is always already imported in a gateway process).""" |
| try: |
| from hermes_cli.main import _read_git_revision_fingerprint |
|
|
| return _read_git_revision_fingerprint(_PROJECT_ROOT) |
| except Exception: |
| return None |
|
|
|
|
| def record_boot_fingerprint() -> None: |
| """Snapshot the checkout revision at gateway startup (idempotent).""" |
| global _boot_fingerprint |
| if _boot_fingerprint is None: |
| _boot_fingerprint = _fingerprint() |
|
|
|
|
| def _short(fingerprint: str) -> str: |
| """Render a ``git:<ref>:<sha>`` fingerprint as a compact label.""" |
| sha = fingerprint.rsplit(":", 1)[-1] |
| return sha[:10] if sha and sha != "unresolved" and len(sha) > 10 else (sha or fingerprint) |
|
|
|
|
| def detect_code_skew() -> tuple[str, str] | None: |
| """``(boot_rev, disk_rev)`` short labels if the checkout drifted since boot, else ``None``.""" |
| current = _fingerprint() if _boot_fingerprint is not None else None |
| if current is None or current == _boot_fingerprint: |
| return None |
| return _short(_boot_fingerprint), _short(current) |
|
|
