hermes-agent/hermes_cli/gateway_runtime.py

"""Validated, credential-free gateway endpoint discovery for local clients."""

from __future__ import annotations

from dataclasses import dataclass
import ipaddress
import math
import os
from pathlib import Path
from typing import TYPE_CHECKING, Literal
from urllib.parse import urlsplit

if TYPE_CHECKING:
    from gateway.runtime_contract import RuntimeState

# Total startup deadline a client waits for a cold daemon. A cold boot imports the whole
# runtime, seeds the skill library and warms the turn machinery: ~10-20 s on a loaded
# 32-core CI runner with dozens of sibling boots, where 30 s tipped every attaching client
# of a parallel e2e shard into `starting: deadline`. A minute is still a verdict, not a hang.
DEFAULT_ENSURE_TIMEOUT = 60.0


@dataclass(frozen=True)
class GatewayEndpoint:
    profile_id: str
    instance_id: str
    authority_epoch: int
    runtime_protocol: int
    api_origin: str
    supervisor: Literal["none", "systemd", "launchd", "windows", "external"]
    capabilities: frozenset[str]
    # Home whose control socket answers for this endpoint: the profile's own home, or the
    # default multiplexer's root when the profile is served by it.
    control_home: str | None = None
    # Commit the owner BOOTED from (its ``identify`` ``code_sha``); None when it cannot name one.
    # A gateway that outlived ``hermes update`` keeps serving old code, so an attaching client
    # compares this with its own checkout and restarts the owner instead of re-attaching it.
    code_sha: str | None = None


@dataclass(frozen=True)
class GatewayDiscovery:
    state: RuntimeState
    endpoint: GatewayEndpoint | None = None
    reason_code: str | None = None
    # Human-readable cause behind a bounded ``reason_code`` when the owner published one (the
    # multiplexer's park reason), or the redacted output tail of a start this call made that died
    # (``runtime_exited``), or the unsafe local control node a discovery refused (path, owner uid or
    # mode, and the fix); never raw peer data.
    detail: str | None = None


def attachable(result: GatewayDiscovery) -> bool:
    """Whether a client may attach to this discovery/ensure verdict.

    ``ready`` owners, plus a restart's after-turn wait (``draining``/``restart_wait``): that owner
    still owns its turns and must stay reachable to answer a parked approval/clarify (#139041).
    Shared by every client bootstrap (``connect_gateway`` and Ink's ``gateway_bootstrap.py``)."""
    if result.endpoint is None:
        return False
    from gateway.drain_control import RESTART_WAIT_REASON
    return result.state == "ready" or (result.state == "draining" and result.reason_code == RESTART_WAIT_REASON)


def _canonical_home(home: str | Path) -> str:
    return os.path.normcase(str(Path(home).expanduser().resolve()))


def gateway_profile_home(home: str | Path) -> Path:
    """The profile home whose gateway serves *home*: the home itself, or for a private overlay home
    (a task dir mirroring a profile through symlinks, its ``gateway.lock`` among them) the profile its
    lock names. A client compares the owner's ``profile_id`` with this, never the overlay path.
    Not normcased: ``profile_id`` keeps the owner's own spelling (Windows)."""
    return _overlay_owner_home(Path(home).expanduser().resolve())


def _overlay_owner_home(home: Path) -> Path:
    """Return the owner named by a private overlay's gateway lock, if it has one."""
    if os.name == "nt":
        return home
    from hermes_cli.gateway_runtime_discovery import DiscoveryError, _private_lock, _private_node
    lock = home / "gateway.lock"
    try:
        # An overlay home intentionally shares only the owner's lock.  Do not turn an arbitrary
        # link into a profile alias: both homes and the target lock must meet ordinary discovery's
        # private-control checks.
        _private_node(home, kind="directory", home=True)
        target = lock.resolve(strict=True)
        if not lock.is_symlink() or target.name != "gateway.lock":
            return home
        _private_node(target.parent, kind="directory", home=True)
        _private_lock(target)
    except (FileNotFoundError, OSError, DiscoveryError):
        return home
    return target.parent


def _endpoint(payload: dict, home: Path, control_home: Path | None = None) -> GatewayDiscovery:
    if type(payload.get("runtime_protocol")) is not int or payload["runtime_protocol"] != 1:
        return GatewayDiscovery("incompatible", reason_code="runtime_protocol")
    profiles = payload.get("served_profiles")
    if not isinstance(profiles, list):
        return GatewayDiscovery("inaccessible", reason_code="profile_mismatch")
    # Canonicalize BOTH sides: on Windows normcase lower-cases the served spelling, so a
    # caller's mixed-case Path (WorkerRPC) never matched and every worker saw owner_unavailable.
    wanted = _canonical_home(home)
    matches = [p for p in profiles if isinstance(p, dict)
               and isinstance(p.get("home"), str) and _canonical_home(p["home"]) == wanted]
    if len(matches) != 1 or not isinstance(matches[0].get("profile_id"), str) or not matches[0]["profile_id"]:
        return GatewayDiscovery("inaccessible", reason_code="profile_mismatch")
    state = payload.get("state")
    # A graceful restart's after-turn wait still owns its in-flight turns (#139041): it stays
    # ``draining`` (exit 6, never a fresh-start target) but carries its endpoint, so an interactive
    # client can attach to answer a turn parked on approval/clarify. New input queues for the next owner.
    from gateway.drain_control import RESTART_WAIT_REASON
    restart_wait = state == "draining" and payload.get("drain_reason") == RESTART_WAIT_REASON
    if state in {"starting", "draining", "conflict"} and not restart_wait:
        # An auto-started gateway ending through its idle exit names that, so its clients wait
        # for it to go and start a fresh one instead of reporting a refusal.
        idle = state == "draining" and payload.get("drain_reason") == "idle_exit"
        return GatewayDiscovery(state, reason_code="idle_exit" if idle else None)
    if state != "ready" and not restart_wait:
        return GatewayDiscovery("inaccessible", reason_code="invalid_runtime_state")
    capabilities = payload.get("capabilities")
    if (not isinstance(capabilities, list) or not all(isinstance(c, str) for c in capabilities)
            or "session-authority-v1" not in capabilities):
        return GatewayDiscovery("incompatible", reason_code="session_authority_unavailable")
    epoch, instance = payload.get("authority_epoch"), payload.get("instance_id")
    if type(epoch) is not int or epoch <= 0 or not isinstance(instance, str) or not instance:
        return GatewayDiscovery("inaccessible", reason_code="invalid_runtime_identity")
    origin = payload.get("api_origin")
    if not isinstance(origin, str):
        return GatewayDiscovery("inaccessible", reason_code="invalid_api_origin")
    try:
        parsed = urlsplit(origin)
        if (parsed.scheme not in {"http", "https"} or not parsed.hostname
                or parsed.username is not None or parsed.password is not None
                or parsed.path or parsed.query or parsed.fragment or not parsed.port
                or not ipaddress.ip_address(parsed.hostname).is_loopback):
            return GatewayDiscovery("inaccessible", reason_code="invalid_api_origin")
    except ValueError:
        return GatewayDiscovery("inaccessible", reason_code="invalid_api_origin")
    supervisor = payload.get("supervisor")
    if supervisor not in {"none", "systemd", "launchd", "windows", "external"}:
        return GatewayDiscovery("inaccessible", reason_code="invalid_supervisor")
    return GatewayDiscovery("draining" if restart_wait else "ready", GatewayEndpoint(
        profile_id=matches[0]["profile_id"], instance_id=instance,
        authority_epoch=epoch, runtime_protocol=1, api_origin=origin,
        supervisor=supervisor, capabilities=frozenset(capabilities),
        control_home=str(control_home) if control_home is not None else None,
        code_sha=code_sha if isinstance(code_sha := payload.get("code_sha"), str) and code_sha else None,
    ), reason_code=RESTART_WAIT_REASON if restart_wait else None)


def control_home_for(home: Path, endpoint: GatewayEndpoint | None) -> Path:
    """Home whose control socket mints tickets for *endpoint* (the multiplexer's for a served profile)."""
    if endpoint is not None and endpoint.control_home:
        return Path(endpoint.control_home)
    return home


def _starting_multiplexer_root(home: Path) -> Path | None:
    """The root whose reservation is pending AND whose boot policy would serve *home*, else None.
    A ``gateway.standalone: true`` secondary is never served by it, so a reserved (or ready) root
    must not make its client wait out the deadline instead of starting the profile's own owner.
    One uncached config parse plus a lock probe: call it once per ensure iteration."""
    from hermes_cli.gateway_runtime_discovery import missing_owner_state
    from hermes_cli.gateway_runtime_multiplex import implied_host_root
    root = implied_host_root(home)
    return root if root is not None and missing_owner_state(root) == "starting" else None


def _standalone_multiplexer_reason(root: Path) -> str | None:
    """A live host's reason for not serving its named profiles, if it published one.

    A held root lock ordinarily means that a named profile should wait for its
    multiplexer.  A host which published a guard refusal is different: it is a
    standalone gateway and can never add the named profile during that boot. Read
    with the same liveness rules as ``hermes gateway status``'s standalone box.
    """
    from hermes_cli.gateway_multiplex_mode import recorded_standalone_reason
    reason = recorded_standalone_reason(root / "gateway_state.json")
    return reason[:500] if reason else None


def _parked_by_multiplexer(payload: dict, home: Path) -> GatewayDiscovery | None:
    """A secondary the multiplexer could not serve (unusable store, home owned elsewhere) is
    published under ``parked_profiles`` (name -> reason). That is a terminal verdict for its
    clients: waiting for a ``served_profiles`` entry that will never appear is the wrong answer."""
    parked = payload.get("parked_profiles")
    if not isinstance(parked, dict):
        return None
    from hermes_cli.profiles import normalize_profile_name
    name = normalize_profile_name(home.name)
    reason = next((r for n, r in parked.items() if normalize_profile_name(str(n)) == name), None)
    if reason is None:
        return None
    return GatewayDiscovery("inaccessible", reason_code="profile_parked", detail=str(reason)[:500] or None)


def _refused_owner_verdict(exc: PermissionError) -> GatewayDiscovery | None:
    """The named refusal when the client proved the boundary from the control pipe, else None."""
    from gateway.runtime_bootstrap_windows import RefusedOwner
    if isinstance(exc, RefusedOwner):
        return GatewayDiscovery("inaccessible", reason_code=exc.reason_code, detail=str(exc))
    return None


def _served_by_multiplexer(home: Path, *, timeout: float) -> GatewayDiscovery | None:
    """A served secondary has no socket of its own: the default multiplexer's control socket
    answers for it, and its ``identify`` lists the home under ``served_profiles``. Only the
    multiplexer's live descriptor proves service; a held ``gateway.lock`` alone does not."""
    from hermes_cli.gateway_runtime_discovery import DiscoveryError, query_identify
    from hermes_cli.gateway_runtime_multiplex import multiplexer_root_for
    root = multiplexer_root_for(home)
    if root is None:
        return None
    try:
        payload = query_identify(root, timeout=timeout)
    except PermissionError as exc:
        # The multiplexer's pipe proved the boundary: name it for the secondary too (#140706), unless
        # the boot policy never serves this home (a standalone secondary starts its own gateway). Any
        # other refusal is "not served" before a config is read, as on POSIX, where none is proven.
        named = _refused_owner_verdict(exc)
        if named is None:
            return None
        from hermes_cli.gateway_multiplex_mode import host_serves_named_profile
        if host_serves_named_profile(root, home) is False:
            return None
        return named
    except (FileNotFoundError, ConnectionRefusedError, TimeoutError, DiscoveryError,
            OSError, ValueError, TypeError):
        return None
    result = _endpoint(payload, home, control_home=root)
    if result.state == "inaccessible" and result.reason_code == "profile_mismatch":
        return _parked_by_multiplexer(payload, home)  # else: the multiplexer does not serve this profile
    return result


_NATIVE_WINDOWS = os.name == "nt"


def _owner_pid(home: Path) -> int | None:
    """The PID a refusal may name: one of *home*'s gateway records whose process is provably that
    gateway, else None (the caller keeps the generic ``authorization``).

    Proof is the shared identity chain: alive, its recorded start time known and equal to the live
    one (a recycled PID differs), its recorded home not another profile's, and its live command line
    (or, when that is unreadable across the elevation boundary, its record) a gateway's. A stale
    ``gateway.pid`` falls through to ``gateway.lock`` instead of masking it."""
    from gateway.status import (
        _get_process_start_time,
        _live_pid_from_record,
        _read_json_file,
        _record_matches_live_gateway_pid,
        recorded_gateway_home_conflicts,
    )
    for name in ("gateway.pid", "gateway.lock"):
        record = _read_json_file(home / name, bare_pid_ok=True)
        pid = _live_pid_from_record(record)
        if pid is None or pid == os.getpid():
            continue
        recorded = record.get("start_time")
        if recorded is None or recorded != _get_process_start_time(pid):
            continue  # unverifiable or recycled: never name it
        if recorded_gateway_home_conflicts(record, expected_home=home):
            continue
        if not _record_matches_live_gateway_pid(record, pid):
            continue
        return pid
    return None


def _refused(home: Path, *, windows: bool = _NATIVE_WINDOWS) -> GatewayDiscovery:
    """The owner's control channel refused this same-user client. On Windows (*windows*, the host
    fact) name the boundary (an elevated owner, or one in another logon session) with its remedy
    instead of the generic ``authorization`` that reads as a credential failure (#139845, #140252)."""
    if windows and (pid := _owner_pid(home)) is not None:
        from gateway.runtime_bootstrap_windows import probe_refused_owner
        try:
            named = probe_refused_owner(pid)
        except OSError:
            named = None
        if named is not None:
            return GatewayDiscovery("inaccessible", reason_code=named[0], detail=named[1])
    return GatewayDiscovery("inaccessible", reason_code="authorization")


def discover_gateway_endpoint(profile_home: str | Path, *, timeout: float = 2.0) -> GatewayDiscovery:
    """Read live control identity, preserving uncertainty instead of spawning.

    A named profile served by the default multiplexer resolves to the multiplexer's endpoint
    with ``profile_id`` = the profile's own home, so clients attach to it with that identity.
    """
    from hermes_cli.gateway_runtime_discovery import DiscoveryError, missing_owner_state, query_identify

    if not math.isfinite(timeout) or timeout <= 0:
        raise ValueError("timeout must be finite and positive")
    home = _overlay_owner_home(Path(_canonical_home(profile_home)))
    try:
        return _endpoint(query_identify(home, timeout=timeout), home)
    except (FileNotFoundError, ConnectionRefusedError):
        served = _served_by_multiplexer(home, timeout=timeout)
        if served is not None:
            return served
        state = missing_owner_state(home)
        if state == "inaccessible" and os.name != "nt":
            from gateway.runtime_ownership import unreadable_lock_recovery
            if detail := unreadable_lock_recovery(home / "gateway.lock"):
                return GatewayDiscovery(state, reason_code="foreign_stale_lock", detail=detail)
        return GatewayDiscovery(state)
    except TimeoutError:
        return GatewayDiscovery("inaccessible", reason_code="control_timeout")
    except DiscoveryError as exc:
        # The code stays bounded; the detail names the local node, its owner or mode, and the fix.
        return GatewayDiscovery("inaccessible", reason_code=exc.reason, detail=exc.bounded_detail(500))
    except PermissionError as exc:
        named = _refused_owner_verdict(exc)  # the client proved the boundary from the control pipe (#140706)
        return named if named is not None else _refused(home)
    except (OSError, ValueError, TypeError):
        # Raw peer data, paths and URLs may contain secrets; report bounded codes.
        return GatewayDiscovery("inaccessible", reason_code="invalid_control_peer")


def _update_fenced(homes) -> bool:
    """True while an update owns the install, judged the way main's launchers and the Desktop
    judge ``.hermes-update-in-progress``: a live (or still being written) marker, or a dead or
    malformed one while the checkout lock is held (a killed update's tree still runs). A dead
    marker over a free lock is a killed update: it never blocks a launch.

    Read-only: a client never clears the updater's fence; the launch-time reclaim
    (``update_lock.read_live_update``) and the next update remove a dead marker. A marker that
    exists but cannot be read is unjudgeable and fences.
    """
    import time
    from hermes_cli import update_lock

    for fence_home in homes:
        path = fence_home / update_lock.MARKER_NAME
        try:
            raw = path.read_bytes()
            if not raw:
                # An exclusive first publish in flight (contract A3), not yet a dead claim.
                if time.time() - path.stat().st_mtime < update_lock.EMPTY_MARKER_GRACE_SECONDS:
                    return True
        except FileNotFoundError:
            continue
        except OSError:
            return True
        if update_lock.judge_marker(raw)[0] in {"live", "ours"} or update_lock.checkout_lock_held():
            return True
    return False


def _fenced_verdict(home: Path, timeout: float) -> GatewayDiscovery:
    """What ensure answers while an update owns the install. The update's own drain
    (``pause-for-update``) is a restart's after-turn wait whose owner may hold a turn parked on
    approval/clarify, so it is handed out for clients to answer it (#139041). Discovery only:
    nothing is started or replaced while fenced, and any other owner state stays ``update_paused``."""
    observed = discover_gateway_endpoint(home, timeout=timeout)
    if observed.state == "draining" and attachable(observed):
        return observed
    return GatewayDiscovery("draining", reason_code="update_paused")


_REPLACE_FAILED_DETAIL = "an older Hermes gateway still owns this profile after a stop; run: hermes gateway stop"
# ``draining`` reasons after which the owner is going away by itself: its own idle exit, or the
# older-Hermes owner this call just stopped. Ensure waits it out and starts the current owner.
_RETIRING = frozenset({"idle_exit", "replaced"})


def _observe_replacing_stale(home: Path, deadline: float, replaced: set[Path]) -> GatewayDiscovery:
    """Discovery, except that a gateway an older Hermes left on this home is stopped once (like
    ``hermes gateway stop``) and reported as retiring; a refusal keeps ``incompatible`` + its remedy."""
    from hermes_cli.gateway_runtime_service import remaining
    observed = discover_gateway_endpoint(home, timeout=remaining(deadline))
    if observed.state != "incompatible" or observed.reason_code != "runtime_protocol":
        return observed
    if home in replaced:
        return GatewayDiscovery("incompatible", reason_code="runtime_protocol", detail=_REPLACE_FAILED_DETAIL)
    from hermes_cli import gateway_runtime_replace as replace
    replaced.add(home)
    refusal = replace.replace_incompatible_owner(home, deadline=deadline)
    if refusal is not None:
        return GatewayDiscovery("incompatible", reason_code="runtime_protocol",
                                detail=replace.REFUSAL_DETAIL.get(refusal, _REPLACE_FAILED_DETAIL))
    return GatewayDiscovery("draining", reason_code="replaced")


def ensure_gateway_runtime(profile_home: str | Path, *, timeout: float = DEFAULT_ENSURE_TIMEOUT,
                           idle_exit: bool = False) -> GatewayDiscovery:
    """Ensure once; pending remains pending at deadline.

    Never installs anything. The one owner it replaces is a gateway an older Hermes left running on
    this install's own home (``incompatible``/``runtime_protocol``, the upgrade case): it is stopped
    the way ``hermes gateway stop`` stops it, once, and the current one started like any absent
    owner. Another install's or a newer Hermes's gateway stays refused, with the reason in ``detail``
    (``hermes_cli.gateway_runtime_replace``).

    A successful service command or Popen is not session readiness. After an
    owner/start request is observed this invocation never launches another.
    ``idle_exit``: an unmanaged daemon this call starts ends itself once idle (client entrypoints:
    chat, TUI, cron/kanban/ACP clients). A gateway that is ending that way reports ``draining``
    (``idle_exit``); this call waits for it to go and starts the replacement transparently.
    """
    import time
    from hermes_cli.gateway_runtime_service import (
        RuntimeStartError, discover_existing_gateway_service, remaining,
        start_existing_gateway_service,
    )
    from hermes_constants import get_default_hermes_root, get_process_hermes_home

    if not math.isfinite(timeout) or timeout <= 0:
        raise ValueError("timeout must be finite and positive")
    deadline = time.monotonic() + timeout
    home = _overlay_owner_home(Path(_canonical_home(profile_home)))
    requested = False
    child = None
    child_home, child_offset = home, 0
    # A gateway ending through its own idle exit is not a refusal: wait for it to go (its lock is
    # still held for a moment after its control socket closes, which reads as "starting"), then
    # start the replacement this call owes the client.
    retiring = False
    replaced: set[Path] = set()
    delay = 0.025
    try:
        while True:
            # The install-root marker also covers profiles.
            if _update_fenced({home, get_process_hermes_home(), get_default_hermes_root()}):
                return _fenced_verdict(home, remaining(deadline))
            # Poll BEFORE discovering: an exit observed first, then "absent", cannot be a child that
            # published its owner a moment later. A winning racer shows up as starting/ready instead.
            status = child.poll() if child is not None else None
            observed = _observe_replacing_stale(home, deadline, replaced)
            if observed.reason_code == "control_timeout":
                return GatewayDiscovery("starting", reason_code="deadline")
            if observed.state == "draining" and observed.reason_code in _RETIRING:
                retiring, requested, child = True, False, None
                time.sleep(min(0.1, remaining(deadline)))
                continue
            if observed.state not in {"absent", "starting"}:
                return observed
            if status is not None and observed.state == "absent":
                # The one start this invocation requested died without leaving an owner: waiting
                # out the deadline cannot change that, and this call never launches a second one.
                # Its own (redacted) output is the answer the user needs, not a pointer to a file.
                from hermes_cli.gateway_runtime_start import startup_failure_report
                return GatewayDiscovery("inaccessible", reason_code="runtime_exited",
                                        detail=startup_failure_report(child_home, child_offset, status, child.pid))
            if observed.state == "starting" and not retiring:
                requested = True
            if observed.state == "absent":
                retiring = False
            reserved = _starting_multiplexer_root(home) if observed.state == "absent" and not requested else None
            if reserved is not None and (reason := _standalone_multiplexer_reason(reserved)):
                return GatewayDiscovery("inaccessible", reason_code="profile_parked", detail=reason)
            # A default multiplexer that is still starting will serve this named profile; never
            # spawn a competing per-profile daemon while its reservation is pending.
            if reserved is not None:
                requested = True
            if observed.state == "absent" and not requested:
                # A named profile the default multiplexer serves has no daemon of its own: start
                # (or await) the MULTIPLEXER. A per-profile spawn here would become a second owner
                # that blocks the multiplexer's next all-or-nothing reserve.
                from hermes_cli.gateway_runtime_multiplex import implied_host_root, multiplexer_serves_home
                target = multiplexer_serves_home(home) or home
                service = discover_existing_gateway_service(target, deadline=deadline)
                if service is None and target == home and (host := implied_host_root(home)) is not None:
                    # Main's guard refuses a named profile a gateway of its own unless it is standalone
                    # or already installed; with no multiplexer evidence yet, the host is still its owner.
                    target = host
                    service = discover_existing_gateway_service(target, deadline=deadline)
                # Runtime locks settle races remaining after this second probe.
                observed = discover_gateway_endpoint(home, timeout=remaining(deadline))
                if observed.state != "absent":
                    continue
                if service is not None:
                    start_existing_gateway_service(service, deadline=deadline)
                else:
                    from hermes_cli.gateway_runtime_start import spawn_unmanaged_gateway, stdio_log_size
                    child_home, child_offset = target, stdio_log_size(target)
                    child = spawn_unmanaged_gateway(target, deadline=deadline, idle_exit=idle_exit)
                requested = True
            time.sleep(min(delay, remaining(deadline)))
            # Capped low: each probe is a lock-file read or one local identify, and READY is noticed
            # within ~50 ms instead of up to 250 ms later on every cold start.
            delay = min(delay * 1.5, 0.05)
    except TimeoutError:
        return GatewayDiscovery("starting", reason_code="deadline")
    except RuntimeStartError as exc:
        return GatewayDiscovery(exc.state, reason_code=exc.reason)
    except PermissionError:
        return _refused(home)
    except OSError:
        return GatewayDiscovery("inaccessible", reason_code="runtime_start_failed")