openlatch-client 0.6.21

#!/usr/bin/env python3
"""Fail the build when something can start a daemon that owns the machine.

`AGENTS.md` says it plainly — "Anything that starts a daemon, installs hooks or
writes agent config runs inside an isolated instance" — and until this scanner
existed, nothing checked. The rule was kept by five different conventions in six
files, each correct, none enforced, and the one place that adopted none of them
(`tools/policy-seed`) took over a developer's machine for an hour.

What is actually at stake is one seam, and it is not the obvious one. A
throwaway `OPENLATCH_DIR` isolates *state*. It says nothing about the model
model_relay, which ships enabled and, left alone, binds a machine-global default
port. `ModelRelayConfig::owns_agent_wiring` (src/core/config/mod.rs) treats
holding that port as owning the machine's agent wiring, so
`wire_model_relay_config` (src/daemon/mod.rs) writes `ANTHROPIC_BASE_URL` into the
developer's real `~/.claude/settings.json`. Every live agent session on the host
is then proxied through a test daemon — and when that daemon exits, those
sessions get `Connection refused` from a URL still on disk. A temp directory
does not prevent one byte of it.

So: every file that can spawn the client binary must declare, in
`ci/baselines/daemon-spawners.txt`, which mechanism keeps it off that port.
Adding a new one becomes a deliberate act rather than an omission, exactly as
`ci/baselines/hook-deps.txt` made a new hot-path dependency deliberate.

Recognised mechanisms, all of which make `owns_agent_wiring()` false:

  model-relay-default-port-pin   `OPENLATCH_MODEL_RELAY_DEFAULT_PORT` moved to a
                              synthetic port. Must be paired with removing an
                              inherited `OPENLATCH_MODEL_RELAY_PORT`, or a shell
                              standing in a sandbox overrides the pin.
  model-relay-free-port          `OPENLATCH_MODEL_RELAY_PORT` set to an OS-assigned
                              free port on the spawned daemon.
  model-relay-disabled           `OPENLATCH_MODEL_RELAY_ENABLED=false` — no listener
                              at all.
  no-model-relay-flag            `init --no-model-relay` — writes the disable.
  no-daemon                   Runs the binary but never starts a daemon
                              (`--help`, `--version`, argv tables, file copies).
                              Requires a reason in the baseline, because this is
                              the entry a future edit is most likely to falsify.

Exit codes: 0 clean, 1 violations found, 2 bad invocation.
"""

from __future__ import annotations

import re
import sys
from dataclasses import dataclass
from pathlib import Path, PurePath

#: Spelled with `/` and kept a `str`: it is compared against and printed far
#: more often than it is opened, and `Path("a/b")` stringifies as `a\b` on
#: Windows. `root / BASELINE` still opens it.
BASELINE = "ci/baselines/daemon-spawners.txt"

#: Where a daemon can be started from. Product code is deliberately out of
#: scope: the daemon binding its own default port is the daemon working.
SCAN_DIRS = ("tests", "tools")
SCAN_SUFFIXES = (".rs", ".py")

#: Evidence that a file can run the client binary as a subprocess. Broad on
#: purpose — a false positive costs one baseline line stating why it is safe,
#: a false negative costs a developer their agent sessions.
#:
#: `daemon\s+start\b` carries the one piece of precision here, and it is load-
#: bearing: without the `\b`, the subcommand also matches the English word
#: "started". `tools/sandbox/src/commands.rs` prints "its daemon started after
#: the preview" in a user-facing message, and the scanner read that sentence as
#: a spawn and failed the build on a file that runs no binary at all. Comment
#: lines are already skipped; this is the same prose one line further in, inside
#: a string literal, where skipping by prefix cannot reach it.
SPAWN_MARKERS = (
    re.compile(r'cargo_bin\(\s*"openlatch"'),
    re.compile(r'"daemon"\s*,\s*"start"'),
    re.compile(r"daemon\s+start\b"),
    re.compile(r"ensure_daemon|spawn_daemon|HermeticDaemon"),
)

#: The marker that proves a claimed mechanism is still in the file. `no-daemon`
#: has none by definition and is checked against the baseline's reason instead.
MECHANISM_MARKERS = {
    "model-relay-default-port-pin": re.compile(r"OPENLATCH_MODEL_RELAY_DEFAULT_PORT"),
    "model-relay-free-port": re.compile(
        r'OPENLATCH_MODEL_RELAY_PORT"?\s*[,\]]?\s*[:=,]\s*.*free_port|'
        r'setdefault\(\s*"OPENLATCH_MODEL_RELAY_PORT"'
    ),
    "model-relay-disabled": re.compile(r'OPENLATCH_MODEL_RELAY_ENABLED"?\s*[,:=]\s*"?false'),
    "no-model-relay-flag": re.compile(r"--no-model-relay"),
    "no-daemon": None,
}

#: A `model-relay-default-port-pin` that leaves an inherited `OPENLATCH_MODEL_RELAY_PORT`
#: in place is not a pin: the explicit var overrides the default, and the shell an
#: OpenLatch developer runs the suite from is a sandbox that exports it.
PIN_NEEDS_REMOVAL = re.compile(r'env_remove\(\s*"OPENLATCH_MODEL_RELAY_PORT"')


def rel_to(path: PurePath, root: PurePath) -> str:
    """A scanned path in the baseline's own spelling: `/`-separated, always.

    `str(p.relative_to(root))` yields `tests\\cli_contract.rs` on Windows, which
    matches nothing in a baseline written with `/`. Every scanned file then
    reads as an undeclared spawner, so the check flips from "clean" to
    "everything violates" on OS alone — and it flips on the one OS this repo's
    CI never runs, which is how it stayed invisible.
    """
    return path.relative_to(root).as_posix()


@dataclass(frozen=True)
class Violation:
    path: str
    problem: str
    hint: str


def parse_baseline(root: Path) -> tuple[dict[str, tuple[str, str]], list[Violation]]:
    """`{path: (mechanism, reason)}` plus any complaints about the file itself."""
    path = root / BASELINE
    if not path.exists():
        return {}, [
            Violation(BASELINE, "baseline file is missing", "restore it from version control")
        ]
    entries: dict[str, tuple[str, str]] = {}
    problems: list[Violation] = []
    for lineno, raw in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
        line = raw.split("#", 1)[0].strip()
        if not line:
            continue
        parts = [p.strip() for p in line.split("|")]
        if len(parts) < 2:
            problems.append(
                Violation(
                    f"{BASELINE}:{lineno}",
                    f"malformed entry: {raw!r}",
                    "format is `path | mechanism | optional reason`",
                )
            )
            continue
        target, mechanism = parts[0], parts[1]
        reason = parts[2] if len(parts) > 2 else ""
        if mechanism not in MECHANISM_MARKERS:
            problems.append(
                Violation(
                    f"{BASELINE}:{lineno}",
                    f"unknown mechanism {mechanism!r}",
                    "one of: " + ", ".join(sorted(MECHANISM_MARKERS)),
                )
            )
            continue
        if mechanism == "no-daemon" and not reason:
            problems.append(
                Violation(
                    f"{BASELINE}:{lineno}",
                    f"`no-daemon` claimed for {target} with no reason",
                    "state what the file does with the binary instead, so the next "
                    "reader can tell when it stops being true",
                )
            )
            continue
        entries[target] = (mechanism, reason)
    return entries, problems


#: Line prefixes that make a match prose rather than code. `tests/
#: proxy_auth_resolve.rs` says "on every daemon start" in a module doc comment
#: and starts nothing; a scanner that cannot tell the difference teaches people
#: to add baseline entries for sentences, and a baseline nobody believes is a
#: baseline nobody reads.
COMMENT_PREFIXES = ("//", "#", "*", "/*")


def _spawns(text: str) -> bool:
    """True when a non-comment line carries a spawn marker."""
    for raw in text.splitlines():
        line = raw.strip()
        if not line or line.startswith(COMMENT_PREFIXES):
            continue
        if any(m.search(line) for m in SPAWN_MARKERS):
            return True
    return False


def spawners(root: Path) -> list[Path]:
    """Every scanned file carrying evidence it can run the client binary."""
    found: list[Path] = []
    for d in SCAN_DIRS:
        base = root / d
        if not base.is_dir():
            continue
        for p in sorted(base.rglob("*")):
            if p.suffix not in SCAN_SUFFIXES or not p.is_file():
                continue
            if "__pycache__" in p.parts or ".venv" in p.parts:
                continue
            try:
                text = p.read_text(encoding="utf-8", errors="replace")
            except OSError:
                continue
            if _spawns(text):
                found.append(p)
    return found


def check(root: Path) -> list[Violation]:
    entries, violations = parse_baseline(root)
    found = spawners(root)
    found_rel = {rel_to(p, root) for p in found}

    # A directory entry covers every scanned file beneath it: `tools/e2e` gates
    # once in `cli.py` for a package whose every module can reach a daemon, and
    # listing sixteen files would be noise that goes stale on the next rename.
    def covering_entry(rel: str) -> tuple[str, tuple[str, str]] | None:
        if rel in entries:
            return rel, entries[rel]
        for key, val in entries.items():
            if rel.startswith(key.rstrip("/") + "/"):
                return key, val
        return None

    for p in found:
        rel = rel_to(p, root)
        entry = covering_entry(rel)
        if entry is None:
            violations.append(
                Violation(
                    rel,
                    "can spawn the client binary but is not in the baseline",
                    "add `path | mechanism | reason` to "
                    f"{BASELINE}, or stop it from starting a daemon",
                )
            )
            continue
        key, (mechanism, _) = entry
        marker = MECHANISM_MARKERS[mechanism]
        if marker is None:
            continue
        # A directory entry is proven by any file under it carrying the marker;
        # a file entry must carry it itself.
        if key == rel:
            hay = p.read_text(encoding="utf-8", errors="replace")
            proven = bool(marker.search(hay))
        else:
            proven = any(
                marker.search(q.read_text(encoding="utf-8", errors="replace"))
                for q in (root / key).rglob("*")
                if q.suffix in SCAN_SUFFIXES
                and q.is_file()
                and "__pycache__" not in q.parts
            )
        if not proven:
            violations.append(
                Violation(
                    rel,
                    f"baseline claims `{mechanism}` but its marker is gone",
                    "restore the mechanism, or update the baseline to the one "
                    "actually in force",
                )
            )
        elif mechanism == "model-relay-default-port-pin" and key == rel:
            hay = p.read_text(encoding="utf-8", errors="replace")
            if not PIN_NEEDS_REMOVAL.search(hay):
                violations.append(
                    Violation(
                        rel,
                        "pins OPENLATCH_MODEL_RELAY_DEFAULT_PORT but never removes "
                        "an inherited OPENLATCH_MODEL_RELAY_PORT",
                        'add `.env_remove("OPENLATCH_MODEL_RELAY_PORT")` — an '
                        "explicit value overrides the default, and a sandboxed "
                        "shell exports one",
                    )
                )

    for target in sorted(entries):
        covered = target in found_rel or any(
            r.startswith(target.rstrip("/") + "/") for r in found_rel
        )
        if not covered and not (root / target).exists():
            violations.append(
                Violation(
                    target,
                    "baseline names a path that no longer exists",
                    f"remove the entry from {BASELINE}",
                )
            )
    return violations


def main(argv: list[str]) -> int:
    if len(argv) > 2:
        print(f"usage: {argv[0]} [repo-root]", file=sys.stderr)
        return 2
    # Windows hands stdout the ANSI code page (cp1252) whenever it is a pipe,
    # which is every way this actually runs — CI, pre-commit, `| head`. The
    # diagnostics below carry an em dash and an arrow, so without this the
    # script dies with UnicodeEncodeError *part-way through* listing the
    # violations it found, replacing a real finding with a stack trace.
    for stream in (sys.stdout, sys.stderr):
        reconfigure = getattr(stream, "reconfigure", None)
        if reconfigure is not None:
            reconfigure(encoding="utf-8", errors="replace")

    root = Path(argv[1] if len(argv) == 2 else ".").resolve()
    if not (root / "Cargo.toml").exists():
        print(f"{root} is not the repo root (no Cargo.toml)", file=sys.stderr)
        return 2

    violations = check(root)
    if not violations:
        n = len(spawners(root))
        print(f"daemon isolation: clean ({n} daemon-spawning file(s) accounted for)")
        return 0

    print("::error::a daemon could be started against the developer's own install:\n")
    for v in violations:
        print(f"  {v.path}")
        print(f"      {v.problem}")
        print(f"      → {v.hint}\n")
    print(f"{len(violations)} violation(s).")
    print(
        "A daemon on the DEFAULT model-relay port owns the machine's agent "
        "wiring and rewrites ~/.claude/settings.json — see this script's "
        "docstring for the mechanisms that prevent it."
    )
    return 1


if __name__ == "__main__":
    sys.exit(main(sys.argv))