llm-ingestion-okf/tools/okf_watch.py

#!/usr/bin/env python3
"""Weekly watch on the canonical Open Knowledge Format repository.

Answers one question: **has canonical moved past our pin, and does the move
touch anything that bears the contract?** On a hit it sends a coord message
immediately. On a miss it says nothing and exits 0.

It is not part of the package. It lives outside `src/`, so it never enters a
wheel, and a consumer's install surface is unchanged by its existence. Stdlib
only, no network library: it drives `git` against the local read-only mirror.

## Why this file is built the way it is

Almost every run of this watch produces a NEGATIVE result. That is the hazard
it is designed around, and the three rules that follow are not stylistic:

1. **A failed call is never an empty result.** `git()` raises on a non-zero
   exit and carries stderr. A helper that returned the empty stdout of a
   failed `git log` would manufacture "upstream is quiet" out of a broken
   query, which is the exact defect this watch exists to prevent.
2. **It proves it can find, every run.** `prove_can_find()` re-runs the whole
   detect-and-classify path over a range known to have changed `SPEC.md`
   (`ad30107^1..ad30107`, the ISO-datetime tightening). If that range comes
   back empty, the query is broken and the run aborts LOUDLY rather than
   reporting a clean sweep.
3. **The network failing is loud.** `git fetch` failing raises. It does not
   degrade to "no change since last time".

## Why quiet is the enumerated list, not signal

Listing what counts as normative can only match what upstream has already
invented. Anything new falls outside the list and the watch goes silent about
it — failing in the direction nobody notices. So the enumeration runs the
other way: a small list of paths measured not to bear the contract, and
everything else reports. Over-firing is visible and fixable by widening
QUIET_PREFIXES. Under-firing is neither.

`README.md` is deliberately NOT quiet. Upstream announced the repository move
in a README commit, and that move is the change with the longest reach this
library has seen.

## What this watch does NOT catch

It sees commits. It cannot see *meaning*. When upstream tightened v0.2 in
place without a version bump, this watch would have fired correctly (SPEC.md
changed) — but nothing in a commit list says "a value that conformed last
month does not conform now", and nothing said `is_stale` had reversed. The
watch's output is therefore "run the runbook", never "here is your exposure".
`docs/upstream-okf-upgrade-runbook.md` is the part that reads meaning, and it
is a human procedure on purpose.

It also cannot see upstream abandoning this repository *silently*. It caught
the last move only because upstream committed a notice to README. A relocation
announced somewhere other than this git history is invisible here.
"""

from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
from dataclasses import dataclass, field
from pathlib import Path

CANONICAL_REPO = "GoogleCloudPlatform/open-knowledge-format"
CANONICAL_URL = f"https://github.com/{CANONICAL_REPO}.git"

#: Our pin. Moves only when the runbook has been run and the plan re-recorded;
#: `docs/plan/okf-v0.2-alignment.md` is the record of record.
PIN = "ad30107c31c06aec8a7d5636e0d1058118604e6f"

#: A range in canonical that really did change SPEC.md, used every run to show
#: the query can find. `ad30107` is the merge of `okf-iso-datetimes`; its first
#: parent range carries three SPEC.md commits.
KNOWN_POSITIVE = (
    "ad30107c31c06aec8a7d5636e0d1058118604e6f^1",
    "ad30107c31c06aec8a7d5636e0d1058118604e6f",
)

#: Paths measured not to bear the contract. Everything else is normative.
QUIET_PREFIXES = (
    ".github/",
    ".gitignore",
    "CONTRIBUTING.md",
    "CODE_OF_CONDUCT.md",
    "LICENSE.md",
    "src/reference_agent/viewer/",
)
#: Generated viewer output, committed alongside bundles it does not define.
QUIET_SUFFIXES = ("viz.html",)

DEFAULT_MIRROR = Path.home() / "repos" / "_okf-canonical"
DEFAULT_STATE = Path.home() / ".claude" / "okf-watch-state.json"
SELF_REPO = "llm-ingestion-okf"
COORDINATOR_REPO = ".claude"


class WatchError(RuntimeError):
    """Anything that stops this watch from measuring. Never swallowed.

    Raised for a failed git call, an unreachable mirror, a failed fetch, or a
    known-positive that finds nothing. Each of those would otherwise surface
    as "no change" — a negative nobody measured.
    """


def git(repo: Path, *args: str) -> str:
    """Run git, or raise. Returns stripped stdout; an empty return means empty.

    The distinction the whole file rests on: a non-zero exit raises, so a
    caller reading `""` knows the query ran and found nothing.
    """
    try:
        proc = subprocess.run(["git", *args], cwd=repo, capture_output=True, text=True, timeout=300)
    except FileNotFoundError as exc:  # pragma: no cover - git absent
        raise WatchError(f"git not found on PATH: {exc}") from exc
    except subprocess.TimeoutExpired as exc:
        raise WatchError(f"git {' '.join(args)} timed out after 300s") from exc
    if proc.returncode != 0:
        raise WatchError(
            f"git {' '.join(args)} failed with exit {proc.returncode} in {repo}: "
            f"{proc.stderr.strip() or '(no stderr)'}"
        )
    return proc.stdout.strip()


def classify(paths: list[str]) -> tuple[list[str], list[str]]:
    """Split changed paths into (normative, quiet), preserving input order."""
    normative: list[str] = []
    quiet: list[str] = []
    for p in paths:
        if p.startswith(QUIET_PREFIXES) or p.endswith(QUIET_SUFFIXES):
            quiet.append(p)
        else:
            normative.append(p)
    return normative, quiet


def changed_paths(mirror: Path, base: str, head: str) -> list[str]:
    """Paths touched between two refs, de-duplicated, order preserved."""
    out = git(mirror, "log", "--first-parent", "--name-only", "--format=", f"{base}..{head}")
    seen: dict[str, None] = {}
    for line in out.splitlines():
        line = line.strip()
        if line:
            seen.setdefault(line, None)
    return list(seen)


def commits_between(mirror: Path, base: str, head: str) -> list[tuple[str, str]]:
    out = git(mirror, "log", "--first-parent", "--format=%h%x1f%s", f"{base}..{head}")
    rows: list[tuple[str, str]] = []
    for line in out.splitlines():
        if "\x1f" in line:
            sha, subject = line.split("\x1f", 1)
            rows.append((sha, subject))
    return rows


def prove_can_find(
    mirror: Path, base: str = KNOWN_POSITIVE[0], head: str = KNOWN_POSITIVE[1]
) -> None:
    """Run the real query over a range known to be positive, or abort.

    Verification law, face 4, made executable: a negative result is only a
    measurement once the instrument has been shown able to produce a positive.
    """
    paths = changed_paths(mirror, base, head)
    normative, _ = classify(paths)
    if not normative:
        raise WatchError(
            f"known-positive range {base}..{head} returned no normative paths. "
            "The query is broken, not upstream quiet. Refusing to report a clean sweep."
        )


@dataclass
class State:
    """What has already been announced, so a hit is announced exactly once.

    Keyed on the pin: moving the pin means we absorbed everything behind it,
    so nothing before the new pin is news any more.
    """

    pin: str
    reported: list[str] = field(default_factory=list)

    @classmethod
    def load(cls, path: Path, pin: str) -> State:
        try:
            raw = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            return cls(pin=pin)
        if raw.get("pin") != pin:
            return cls(pin=pin)
        reported = [str(s) for s in raw.get("reported", [])]
        return cls(pin=pin, reported=reported)

    def record(self, shas: list[str]) -> None:
        for s in shas:
            if s not in self.reported:
                self.reported.append(s)

    def save(self, path: Path) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(
            json.dumps({"pin": self.pin, "reported": self.reported}, indent=2) + "\n",
            encoding="utf-8",
        )


def decide(state: State, normative_shas: list[str]) -> list[str]:
    """The commits worth a message: normative, and not announced before."""
    return [s for s in normative_shas if s not in state.reported]


def render_body(
    *,
    pin: str,
    head: str,
    commits: list[tuple[str, str]],
    normative: list[str],
    quiet: list[str],
) -> str:
    """The coord message. ASCII only - coord-send requires it."""
    total = len(normative) + len(quiet)
    lines = [
        "Canonical OKF has moved past our pin.",
        "",
        f"  repository : {CANONICAL_REPO}",
        f"  our pin    : {pin[:7]}",
        f"  upstream   : {head[:7]}",
        f"  paths      : {len(normative)} of {total} changed paths are contract-bearing",
        "",
        "New commits (first-parent):",
    ]
    lines += [f"  {sha}  {subject}" for sha, subject in commits] or ["  (none)"]
    lines += ["", "Contract-bearing paths:"]
    lines += [f"  {p}" for p in normative] or ["  (none)"]
    if quiet:
        lines += ["", f"Not counted (quiet list): {len(quiet)} path(s)."]
    lines += [
        "",
        "This watch sees commits, not meaning. It cannot tell you whether a",
        "value that conformed last month still conforms - the 2026-08 round",
        "tightened v0.2 in place with no version bump, and reversed is_stale,",
        "and neither fact is legible in a commit list. Next step is the",
        "procedure, not a conclusion:",
        "",
        "  docs/upstream-okf-upgrade-runbook.md",
        "",
        "Sent by tools/okf_watch.py in llm-ingestion-okf.",
    ]
    return "\n".join(lines) + "\n"


def _coord_send_script() -> Path:
    """Newest repo-mailbox scripts dir, resolved at runtime.

    Hard-coding a plugin version is a rot hazard: the path moves every plugin
    release and a stale one fails as "command not found", which reads like the
    watch having nothing to say.
    """
    base = Path.home() / ".claude" / "plugins" / "cache" / "ktg-plugin-marketplace" / "repo-mailbox"
    if not base.is_dir():
        raise WatchError(f"repo-mailbox plugin cache not found at {base}")

    def key(p: Path) -> tuple[int, ...]:
        try:
            return tuple(int(x) for x in p.name.split("."))
        except ValueError:
            return (-1,)

    candidates = [d for d in base.iterdir() if (d / "scripts" / "coord-send.sh").is_file()]
    if not candidates:
        raise WatchError(f"no coord-send.sh under any version in {base}")
    return max(candidates, key=key) / "scripts" / "coord-send.sh"


def send(to: str, subject: str, body: str, *, fyi: bool, dry_run: bool) -> None:
    script = _coord_send_script()
    args = ["bash", str(script), "--to", to, "--from", SELF_REPO, "--subject", subject]
    if fyi:
        args.append("--fyi")
    if dry_run:
        print(f"[dry-run] {' '.join(args)}\n{body}")
        return
    proc = subprocess.run(args, input=body, capture_output=True, text=True)
    if proc.returncode != 0:
        raise WatchError(
            f"coord-send to {to} failed with exit {proc.returncode}: "
            f"{proc.stderr.strip() or '(no stderr)'}"
        )


def run(mirror: Path, state_path: Path, *, dry_run: bool, no_fetch: bool, pin: str = PIN) -> int:
    if not (mirror / ".git").is_dir():
        raise WatchError(f"mirror {mirror} is not a git repository - cannot measure anything")

    if not no_fetch:
        git(mirror, "fetch", "--quiet", "origin", "main")

    prove_can_find(mirror)

    head = git(mirror, "rev-parse", "origin/main")
    commits = commits_between(mirror, pin, head)
    paths = changed_paths(mirror, pin, head)
    normative, quiet = classify(paths)

    if not normative:
        print(
            f"okf-watch: no contract-bearing change. {CANONICAL_REPO} at {head[:7]}, "
            f"our pin {pin[:7]}, {len(commits)} commit(s) ahead, "
            f"{len(quiet)} of {len(paths)} changed paths on the quiet list. "
            "Known-positive passed, so this zero is a measurement."
        )
        return 0

    state = State.load(state_path, pin=pin)
    fresh = decide(state, [sha for sha, _ in commits])
    if not fresh:
        print(
            f"okf-watch: {len(normative)} contract-bearing path(s) at {head[:7]}, "
            "already announced. Silent by design - report on change, not on state."
        )
        return 0

    subject = f"Canonical OKF moved past our pin: {len(normative)} contract-bearing path(s)"
    body = render_body(pin=pin, head=head, commits=commits, normative=normative, quiet=quiet)
    send(SELF_REPO, subject, body, fyi=False, dry_run=dry_run)
    send(COORDINATOR_REPO, subject, body, fyi=True, dry_run=dry_run)

    if not dry_run:
        state.record(fresh)
        state.save(state_path)
    print(f"okf-watch: reported {len(fresh)} new commit(s) at {head[:7]}.")
    return 0


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    ap = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    ap.add_argument(
        "--mirror", type=Path, default=Path(os.environ.get("OKF_MIRROR", DEFAULT_MIRROR))
    )
    ap.add_argument(
        "--state", type=Path, default=Path(os.environ.get("OKF_WATCH_STATE", DEFAULT_STATE))
    )
    ap.add_argument(
        "--dry-run", action="store_true", help="print the coord messages instead of sending"
    )
    ap.add_argument(
        "--no-fetch", action="store_true", help="skip the network call (offline testing)"
    )
    ap.add_argument(
        "--pin",
        default=PIN,
        help=(
            "override the pin for ONE run. Exists so the hit path can be demonstrated on "
            "demand - a watch whose reporting branch has never run is a watch nobody has "
            "seen work. IMPLIES --dry-run: it does not move the pin, and it must not be "
            "able to page anyone. On 2026-08-23 this flag without --dry-run delivered two "
            "false alarms, so the pairing is enforced here rather than remembered."
        ),
    )
    args = ap.parse_args(argv)
    if args.pin != PIN:
        args.dry_run = True
    return args


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    try:
        return run(
            args.mirror, args.state, dry_run=args.dry_run, no_fetch=args.no_fetch, pin=args.pin
        )
    except WatchError as exc:
        print(f"okf-watch: FAILED - {exc}", file=sys.stderr)
        print(
            "okf-watch: this is NOT 'no change'. Nothing was measured this run.",
            file=sys.stderr,
        )
        return 2


if __name__ == "__main__":
    raise SystemExit(main())