#!/usr/bin/env python3
"""bullpen-grinder — the test-driven code grinder (harness v2), fixing what the lurker couldn't.

Runs ON boltzmann (local repo + local pytest + the hossenfelder proxy next door). Given a
failing pytest suite, it drives a model to a GREEN implementation, in an isolated git worktree,
iterating locally (no remote-edit-over-sic), with TIERED ESCALATION: start on a cheap local
model, escalate to a stronger one when the small model stalls. The tests are the contract.

  bullpen-grinder --once "<task>" tests/<file> [target/file ...]   # standalone, prints the result
  (service/room mode wires this into bullpen_worker later.)

Model tiers come from the proxy (OpenAI API) so ANY served model works — floor local, ceiling
deepseek-local. Never touches the real working copy: all edits happen in `git worktree`.
"""
import json, os, re, shutil, subprocess, sys, time, urllib.parse, urllib.request

BWRAP = shutil.which("bwrap")   # #76: sandbox the (model-generated) test run

REPO   = os.environ.get("GRIND_REPO", os.path.expanduser("~/src/bullpen"))
import os, sys
sys.path[:0] = [p for p in (os.path.join(os.path.dirname(os.path.realpath(__file__)), "..", "lib"), "/usr/local/lib/bullpen") if os.path.isdir(p)]
import bullpen_config as cfg
import bullpen_mem as mem
from bullpen_modeltag import model_tag                                    # #118
PROXY  = cfg.PROXY
# Name of the gateway, for the per-grind cost line. Derived from the configured proxy so
# it follows the fleet instead of being a literal.
GATEWAY = urllib.parse.urlparse(PROXY).hostname or "gateway"
TIERS  = os.environ.get("GRIND_TIERS", "[local] deepseek-v4-flash-dspark").split(",")  # floor -> ceiling
# Local deepseek-v4-flash-dspark is the floor (markus request 2026-07-22, 4b96b70: no auto-spend).
# [free] qwen/qwen3-coder is the ceiling: confirmed genuinely $0/call (2026-07-27 probe, unlike
# [free] openai/gpt-oss-120b which bills a fraction of a cent despite the [free] label) and
# previously verified to route + emit FILE: blocks correctly. A stuck floor model now actually
# has somewhere to escalate to instead of ESCALATE_AT being dead code.
# The coordinator is asked to APPLY a grind's diff, so test scratch must never appear in it.
# (The sic EPIPE grind returned 12 pytest tmp files alongside the 2 real source hunks.)
#
# `glob` magic is REQUIRED, not decoration: without it git treats `**` as an ordinary
# wildcard that will not span "zero directories", so `**/__pycache__/**` excluded a NESTED
# __pycache__ but not one at the repo root — which is exactly where pytest puts it for a
# flat repo. Every grind diff carried a .pyc because of it (@reviewer flagged the symptom
# 2026-08-02; measured with git diff --stat against both spellings).
# Module-level so tests/test_grind_noise.py can check it without running a grind.
NOISE = [":(exclude,glob)**/pytest-of-*/**",
         ":(exclude,glob)**/__pycache__/**",
         ":(exclude,glob)**/.pytest_cache/**"]

PY_TIERS = os.environ.get("PY_GRIND_TIERS", "[local] deepseek-v4-flash-dspark,[free] qwen/qwen3-coder")
MAX_ITERS   = 10          # total model attempts across all tiers
ESCALATE_AT = 4           # attempts on a tier without progress before escalating
LARGE_DIFF_LINES = 40     # insertions+deletions above which a GREEN still gets auto-flagged for review
COLLECTION_PENALTY = 500  # a module that won't import scores worse than any assertion-fail count
PYTEST_TIMEOUT = 120
GO_TEST_TIMEOUT = 600     # compile + test is slower than pytest, especially cold
JS_TEST_TIMEOUT = 180     # node --test: fast, but cold module resolution has overhead

def sh(cmd, cwd=None, timeout=PYTEST_TIMEOUT):
    return subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout)

OUT_FLOOR = 4000          # never ask for less than this
OUT_CEIL  = 32000         # never ask for more (proxy/model limits)

def out_budget(tgt_text):
    """Output-token budget for ONE reply that must reproduce every target file IN FULL.

    A fixed cap silently breaks two ways (both observed 2026-07-22):
      * the file simply does not fit, so the reply truncates mid-fence; or
      * a REASONING model spends the whole budget thinking and returns content=None.
        kimi-k3 burned 3997 of 4000 tokens on reasoning_tokens and emitted ZERO characters,
        six iterations running, indistinguishable from an incapable model.
    So scale with what we are asking it to reproduce, and leave explicit reasoning headroom."""
    need = len(tgt_text) // 3                       # chars -> tokens, deliberately conservative
    return max(OUT_FLOOR, min(OUT_CEIL, need * 2 + OUT_FLOOR))

def ask_model(model, system, user, max_tokens=OUT_FLOOR):
    """Returns (content, meta). meta carries finish_reason / reasoning_tokens so the caller can
    tell 'model had nothing to say' apart from 'model never got to speak' — previously both
    collapsed to "" and were burned as an ordinary no-progress iteration."""
    payload = json.dumps({"model": model, "temperature": 0.2, "max_tokens": max_tokens,
        "messages": [{"role": "system", "content": system}, {"role": "user", "content": user}]}).encode()
    req = urllib.request.Request(PROXY, payload, {"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=300) as r:
        d = json.load(r)
    meta = {"finish_reason": None, "reasoning_tokens": None, "budget": max_tokens}
    # hossenfelder is a failover/auto-routing proxy — the model NAME in the request is what was
    # ASKED for, not necessarily what SERVED it. OpenAI-compatible responses carry the actual
    # serving model in the body; capture it so the room reply can report reality, not the ask.
    meta["served_model"] = d.get("model")
    try:
        ch = d["choices"][0]
        meta["finish_reason"] = ch.get("finish_reason")
        u = d.get("usage") or {}
        meta["reasoning_tokens"] = (u.get("completion_tokens_details") or {}).get("reasoning_tokens")
        return (ch["message"].get("content") or ""), meta
    except Exception:
        return "", meta

GOMODCACHE = os.environ.get("GOMODCACHE", os.path.expanduser("~/go/pkg/mod"))

def test_kind(test_rel):
    """Which test runner grades this spec. The grinder was pytest-only, so Go work (sic is Go)
    could not be delegated at all — @godev exists to close that, sharing this one binary rather
    than forking it (a fork would have needed tonight's max_tokens bug fixed twice)."""
    if test_rel.endswith("_test.go"):
        return "go"
    if test_rel.endswith((".test.js", ".test.mjs", "_test.js", "_test.mjs")):
        return "js"
    return "py"


def _go_scores(out, rc):
    """Parse `go test` output into the same (failed, fails, collect_err) shape pytest gives.

    Go reports a compile failure as a build error with NO test results at all — the exact
    analogue of a pytest collection error, and it must score worse than any assertion failure
    so that 'it compiles now, N tests fail' reads as progress rather than regression."""
    fails = re.findall(r"(?m)^\s*--- FAIL: (\S+)", out)
    build_err = bool(re.search(r"(?m)^(# |.*\[build failed\]|.*cannot find package|"
                               r".*no required module provides)", out)) and not fails
    return fails, build_err


def _js_scores(out, rc):
    """`node --test` TAP output -> (fails, load_err), the same shape _go_scores gives. A
    syntax / import error exits non-zero with no test plan — the collection-error analogue,
    scored worse than any assertion failure."""
    fails = re.findall(r"(?m)^not ok \d+ - (.+?)\s*$", out)
    ran = bool(re.search(r"(?m)^# tests \d+", out) or re.search(r"(?m)^(?:ok|not ok) \d+", out))
    load_err = (not ran) and rc != 0
    return fails, load_err


def run_tests(work, test_rel):
    # SANDBOX (#76): model-generated code executes here. Run pytest in a bubblewrap jail —
    # --unshare-all (NO network), --tmpfs /home + /root (NO ssh/claude creds), read-only
    # system, only the worktree writable. Closes the fleet-reach blast radius.
    kind = test_kind(test_rel)
    extra, env_args = [], []
    if kind == "go":
        inner = ["go", "test", "./...", "-count=1"]
        # The jail has NO network (--unshare-all), so the module cache must be present or every
        # build fails on a download. Bind it READ-ONLY: a grind must never be able to poison the
        # real cache. GOFLAGS=-mod=mod + GOPROXY=off makes the failure honest ("missing module")
        # instead of a hang on an unreachable proxy. GOCACHE lands in the jail's tmpfs /tmp.
        if os.path.isdir(GOMODCACHE):
            extra = ["--ro-bind", GOMODCACHE, GOMODCACHE]
        env_args = ["--setenv", "GOMODCACHE", GOMODCACHE, "--setenv", "GOPROXY", "off",
                    "--setenv", "GOFLAGS", "-mod=mod", "--setenv", "GOCACHE", "/tmp/go-build",
                    "--setenv", "HOME", "/tmp"]
    elif kind == "js":
        inner = ["node", "--test", "--test-reporter=tap", test_rel]
        # NODE_PATH: resolve jsdom + the webgame-smoke harness (global modules) inside the
        # jail, so a .test.js can DOM-smoke a single-file HTML game headless.
        env_args = ["--setenv", "HOME", "/tmp", "--setenv", "NODE_PATH", "/usr/lib/node_modules"]
    else:
        inner = ["python3", "-m", "pytest", test_rel, "-q", "--no-header",
                 "-p", "no:cacheprovider", "-rf"]
    if BWRAP:
        # --tmpfs /tmp: the jail binds / READ-ONLY, so without a writable /tmp pytest cannot
        # create its tmp_path base and silently falls back to `pytest-of-<user>/` inside the
        # WORKTREE — which `git add -A` then swept into the returned diff (observed on the sic
        # EPIPE grind: 12 junk files, 150 insertions). Also more hermetic: a fresh empty /tmp.
        cmd = [BWRAP, "--unshare-all", "--ro-bind", "/", "/", "--dev", "/dev", "--proc", "/proc",
               "--tmpfs", "/home", "--tmpfs", "/root", "--tmpfs", "/tmp", *extra, *env_args,
               "--bind", work, work, "--chdir", work,
               "--die-with-parent", *inner]
        p = sh(cmd, timeout={"go": GO_TEST_TIMEOUT, "js": JS_TEST_TIMEOUT}.get(kind, PYTEST_TIMEOUT))
    else:
        sys.stderr.write(f"WARN: bwrap missing — {kind} tests running UNSANDBOXED\n")
        p = sh(inner, cwd=work, timeout={"go": GO_TEST_TIMEOUT, "js": JS_TEST_TIMEOUT}.get(kind, PYTEST_TIMEOUT))
    out = (p.stdout + p.stderr)
    if kind == "go":
        fails, build_err = _go_scores(out, p.returncode)
        passed = p.returncode == 0 and not fails and not build_err
        failed = 0 if passed else (COLLECTION_PENALTY + 1 if build_err else (len(fails) or 999))
        tail = "\n".join(out.splitlines()[-40:])
        if build_err:
            tail = ("BUILD FAILED — the package does not COMPILE. No test can run until it does; "
                    "fix the compile errors first.\n" + tail)
        return passed, failed, tail, fails
    if kind == "js":
        fails, load_err = _js_scores(out, p.returncode)
        passed = p.returncode == 0 and not fails and not load_err
        failed = 0 if passed else (COLLECTION_PENALTY + 1 if load_err else (len(fails) or 999))
        tail = "\n".join(out.splitlines()[-40:])
        if load_err:
            tail = ("LOAD FAILED — the suite could not run (syntax or import error). Fix that "
                    "first; no test runs until it loads.\n" + tail)
        return passed, failed, tail, fails
    n_failed = int((re.search(r"(\d+) failed", out) or [0, 0])[1])
    n_error  = int((re.search(r"(\d+) error",  out) or [0, 0])[1])
    n_passed = int((re.search(r"(\d+) passed", out) or [0, 0])[1])
    collect_err = ("during collection" in out.lower() or "interrupted" in out.lower()
                   or (n_error > 0 and n_passed == 0 and n_failed == 0))
    passed = n_passed > 0 and n_failed == 0 and n_error == 0
    # 'failed' drives escalation and MUST be able to DECREASE as the model makes progress.
    # A module that won't IMPORT (collection error) is strictly worse than any assertion
    # failure — score it above COLLECTION_PENALTY so "now imports, N fail" reads as progress
    # (N < penalty), not a regression, and a plain assertion-fail count decreases normally.
    if passed:
        failed = 0
    elif collect_err:
        failed = COLLECTION_PENALTY + n_error
    elif n_failed or n_error:
        failed = n_failed + n_error
    else:
        failed = 999                       # couldn't parse any result (pytest itself crashed?)
    # trim the failure report the model needs to see; foreground an import wall so it's actionable
    tail = "\n".join(out.splitlines()[-40:])
    if collect_err:
        tail = ("COLLECTION ERROR — your target module does NOT import (syntax / NameError / bad "
                "import). Nothing can be tested until it imports cleanly; fix THAT first.\n" + tail)
    # per-test node-ids of the failures (from -rf), so the loop can target ONE at a time and detect
    # a regression (a previously-passing test that a later change broke). Empty on collection error.
    fails = re.findall(r"(?m)^FAILED (\S+)", out)
    return passed, failed, tail, fails

# model emits `FILE: <path>` then a fenced block; apply each to the worktree
_FILE_RE = re.compile(r"FILE:\s*(?P<path>[^\n`]+)\n+```[a-zA-Z0-9]*\n(?P<body>.*?)\n```", re.S)

def _write_jailed(work, rel, body):
    dest = os.path.join(work, rel.strip().lstrip("/"))
    if not os.path.realpath(dest).startswith(os.path.realpath(work) + os.sep):
        return False                                    # jail: never escape the worktree
    os.makedirs(os.path.dirname(dest), exist_ok=True)
    open(dest, "w").write(body)
    return True

def apply_files(work, text, targets, test_rel=""):
    n = 0
    protected = test_rel.strip().lstrip("/")
    # split on FILE: markers; each section = a path line then the body (fenced OR bare code —
    # the 4B fences its output, deepseek emits bare code, so accept both).
    for part in re.split(r"(?m)^\s*FILE:\s*", text)[1:]:
        head, _, body = part.partition("\n")
        rel = head.strip().strip("`").strip()
        # NEVER let the model overwrite the spec: a model that "returns the test unchanged"
        # would clobber it -> 0 tests collected -> unparseable 999 -> false RED (this exact
        # bug killed slice 3.6). Protect the test file and anything under tests/.
        relnorm = rel.lstrip("/")
        if relnorm == protected or relnorm.startswith("tests/"):
            continue
        fm = re.match(r"\s*```[a-zA-Z0-9_]*\n(.*?)\n```", body, re.S)
        body = fm.group(1) if fm else re.sub(r"\n```\s*$", "", body).rstrip("\n")
        if rel and body.strip() and _write_jailed(work, rel, body.rstrip("\n") + "\n"):
            n += 1
    # fallback: no FILE: markers at all, single target, single fenced block
    if n == 0 and len(targets) == 1:
        blocks = re.findall(r"```[a-zA-Z0-9_]*\n(.*?)\n```", text, re.S)
        if len(blocks) == 1 and _write_jailed(work, targets[0], blocks[0].rstrip("\n") + "\n"):
            n = 1
    return n

def _infer_targets(work, test_rel, test_src):
    """Which files is the grinder allowed to edit, when the ticket doesn't say?

    The old version matched a hardcoded allowlist of top-level dirs, so every repo layout
    it hadn't been taught about (bullseye/, src/, lmcp-tools/, now gateway/) inferred
    NOTHING and the grind ran against a file the model could not see — three separate
    "the model is incapable" false alarms. Ask git for the file list instead: a repo file
    is a target if the test names it by path, or — since tests build paths piecewise, e.g.
    Path(__file__).parent.parent / "gateway" / "sicd" — by a basename unique in the repo."""
    files = [f for f in sh(["git", "-C", work, "ls-files"], timeout=30).stdout.split()
             if f != test_rel and not f.startswith("tests/")]
    if test_kind(test_rel) == "go":
        # Go tests live in the SAME package as the code and never name a source file — there is
        # no import path to match on, so the text heuristics below find nothing (they returned
        # zero targets on @godev's first ticket). The package IS the unit: every non-test .go
        # file in the spec's own directory.
        d = os.path.dirname(test_rel)
        return sorted(f for f in files
                      if os.path.dirname(f) == d
                      and f.endswith(".go") and not f.endswith("_test.go"))[:6]
    hits = sorted(f for f in files if f in test_src)
    if not hits:
        byname = {}
        for f in files:
            byname.setdefault(os.path.basename(f), []).append(f)
        hits = sorted(v[0] for k, v in byname.items()
                      if len(v) == 1 and re.search(r"\b%s\b" % re.escape(k), test_src))
    if not hits:
        # Flat single-module repo: both heuristics above need the filename WITH its
        # extension in the test source, but a pytest writes `import catalog_entry`, not
        # `catalog_entry.py`. Spec authors forget the extension reliably — it cost a
        # bounced ticket twice (modeltag and catalog_entry, both 2026-08-02) — and naming
        # the target explicitly does not help either, since serve()'s target regex requires
        # a slash and a root-level file has none.
        # Deliberately narrow: only when exactly ONE candidate remains after dropping docs
        # and dotfiles. With two, keep bouncing — guessing which one to grind costs more
        # than a rejected ticket does.
        DOCS = (".md", ".txt", ".rst", ".cfg", ".ini", ".toml", ".json", ".yaml", ".yml")
        src_files = [f for f in files
                     if not f.lower().endswith(DOCS) and not os.path.basename(f).startswith(".")]
        if len(src_files) == 1:
            hits = src_files
    return hits[:6]

def _eval_escalation_discipline(rid, escalations, passed, final_model):
    """Trajectory eval (mechanical, no LLM): was each tier-escalation actually earned
    (>=ESCALATE_AT stalled attempts on the tier), or a regression override (a genuine
    signal, allowed to jump immediately without a stall count) — vs a bare escalation
    with neither behind it. Per ~/claude/trajectory-evals-design.md §4b/§5. A run with
    zero escalations still gets a PASS record: a clean trajectory is itself informative,
    not a wasted check (same posture as the CLAUDE.md phase-5 review rule)."""
    if not escalations:
        verdict, note = "PASS", "no escalation needed (floor tier reached the result)"
    else:
        bad = [e for e in escalations
               if e["reason"] != "regression" and e["on_tier_before"] < ESCALATE_AT]
        if bad:
            verdict = "FAIL"
            note = f"{len(bad)}/{len(escalations)} escalation(s) fired before {ESCALATE_AT} stalled attempts: {bad}"
        else:
            verdict = "PASS"
            steps = "; ".join(f"{e['from']}->{e['to']} on {e['reason']} (on_tier={e['on_tier_before']})"
                              for e in escalations)
            note = f"{len(escalations)} escalation(s), all earned: {steps}"
    # mem.save() fails SOFT — a worker must not die because shared memory is unreachable.
    # The cost is that a discarded record is indistinguishable from a written one, and that
    # is exactly what happened: /etc/bullpen/mneme-token exists only on the room host, the
    # grind engine runs on the grind host, so EVERY grinder eval since this shipped was
    # thrown away while the room reply cheerfully said ESCALATION-EVAL: PASS. Say so.
    # final_model is QUOTED: since #118 a tier id carries its cost prefix ("[local] deepseek-…"),
    # so the value contains a space and an unquoted field truncates to "[local]". Caught by
    # bullpen-evals on its very first real read — which is the argument for having a reader.
    if mem.save(f"EVAL agent=grinder msg_id={rid} rubric=escalation-discipline verdict={verdict} "
                f"final_model=\"{final_model}\" passed={passed} note=\"{note}\" ts={int(time.time())}",
                ns="/trajectory/grinder") is None:
        sys.stderr.write(f"WARN: trajectory eval for msg {rid} was NOT persisted "
                         f"(mneme unreachable or no token at {cfg.MNEME_TOKEN_FILE}) — "
                         f"bullpen-evals will not see this run\n")
    return verdict, note

def _diffstat_lines(stat):
    """Sum insertions+deletions out of a `git diff --stat` summary line
    ('N files changed, A insertions(+), D deletions(-)'). 0 if unparsable."""
    ins = re.search(r"(\d+) insertion", stat)
    dels = re.search(r"(\d+) deletion", stat)
    return (int(ins.group(1)) if ins else 0) + (int(dels.group(1)) if dels else 0)

def grind(task, test_rel, targets):
    rid = str(int(time.time()))
    work = f"/tmp/grind-{rid}"
    result_file = os.environ.get("GRIND_RESULT")   # durable best-so-far sink (survives an outer SIGKILL)
    # AUTO-SCAFFOLD (build-from-scratch): a NEW artifact (e.g. a game from a spec) has loose
    # files, not a git repo. If REPO is not a git repo, seed it — create empty stub targets,
    # init + commit — so the grinder can BUILD a new file to satisfy a test, not only fix an
    # existing repo. Existing repos are untouched (the check only fires on a non-repo dir).
    if sh(["git", "-C", REPO, "rev-parse", "--git-dir"], timeout=10).returncode != 0:
        # SAFETY: only scaffold a FRESH dedicated build dir — never git-init a home/system
        # dir or a dir already full of files (@testdesigner once pointed a spec at ~ itself).
        real = os.path.realpath(REPO)
        n = len(os.listdir(REPO)) if os.path.isdir(REPO) else 0
        if (real in ("/", os.path.realpath(os.path.expanduser("~")))
                or real in ("/home", "/usr", "/etc", "/var", "/opt", "/root", "/tmp") or n > 50):
            return (f"INVALID TICKET \u274c — {REPO} is not a git repo and is unsafe to scaffold "
                    f"(a home/system dir or too many files). Point the build at a FRESH dedicated "
                    f"dir (e.g. /tmp/build-<name>) holding only the test + stub target.")
        for t in (targets or []):
            tp = os.path.join(REPO, t)
            os.makedirs(os.path.dirname(tp) or REPO, exist_ok=True)
            if not os.path.exists(tp):
                open(tp, "a").close()
        sh(["git", "-C", REPO, "init", "-q"], timeout=15)
        sh(["git", "-C", REPO, "add", "-A"], timeout=15)
        sh(["git", "-C", REPO, "-c", "user.email=grind@bullpen", "-c", "user.name=grinder",
            "commit", "-q", "--allow-empty", "-m", "scaffold: seed repo for build-from-scratch grind"], timeout=15)
    sh(["git", "-C", REPO, "worktree", "add", "-q", "--detach", work, "HEAD"], timeout=60)
    try:
        # Overlay the main repo's uncommitted + untracked (non-ignored) files onto the clean
        # worktree — so a fresh @testdesigner spec (not yet committed) IS present. Then commit
        # HEAD+working-tree as the grind baseline, so the final diff shows ONLY the grinder's
        # own changes, not the pre-existing WIP.
        for rel in sh(["git", "-C", REPO, "ls-files", "--modified", "--others",
                       "--exclude-standard"], timeout=30).stdout.split():
            src, dst = os.path.join(REPO, rel), os.path.join(work, rel)
            if os.path.isfile(src) and os.path.realpath(dst).startswith(os.path.realpath(work) + os.sep):
                os.makedirs(os.path.dirname(dst), exist_ok=True)
                shutil.copy2(src, dst)
        sh(["git", "-C", work, "add", "-A"], timeout=30)
        sh(["git", "-C", work, "-c", "user.email=grind@bullpen", "-c", "user.name=grinder",
            "commit", "-q", "--no-verify", "--allow-empty", "-m", "grind baseline (HEAD + working tree)"], timeout=30)
        baseline = sh(["git", "-C", work, "rev-parse", "HEAD"], timeout=30).stdout.strip()
        # If test_rel is absolute (e.g. from SPEC: /abs/path), convert to repo-relative
        # so os.path.join(work, test_rel) works inside the bwrap sandbox where /tmp is fresh tmpfs.
        if os.path.isabs(test_rel) and test_rel.startswith(REPO):
            test_rel = os.path.relpath(test_rel, REPO)
        try:
            test_src = open(os.path.join(work, test_rel)).read()
        except FileNotFoundError:
            return (f"INVALID TICKET ❌ — {test_rel} does not exist in {REPO}. "
                    f"If the tests live in another repo, say `repo=/abs/path` (or "
                    f"`REPO: host:/abs/path`) in the ticket body.")
        if not targets:
            targets = _infer_targets(work, test_rel, test_src)
        if not targets:
            return (f"INVALID TICKET ❌ — could not infer any target file from {test_rel}. "
                    f"Name the file(s) to edit explicitly in the ticket.")
        suite = {"go": "`go test ./...` suite", "js": "`node --test` suite"}.get(test_kind(test_rel), "pytest suite")
        system = (f"You are a code grinder. Make the given {suite} pass by editing ONLY the "
                  "target source file(s) — never the tests. Output each changed file EXACTLY as:\n"
                  "FILE: <repo/relative/path>\n```\n<the complete new file contents>\n```\n"
                  "Output nothing else. Change the minimum needed to go green.")

        # --- durability: track the BEST-scoring state, not the last one. Each new best is a git
        # checkpoint AND is flushed to GRIND_RESULT immediately, so an outer SIGKILL (coordinator
        # timeout) still leaves the best artifact behind for the coordinator to recover. On a
        # regression the next attempt resumes from the best checkpoint (hill-climb, don't wander).
        best = {"failed": 10**9, "sha": baseline, "passed": False, "tail": "",
                "model": TIERS[0], "served_model": None}
        # trajectory eval (§4b escalation-discipline): every tier bump already carries a reason
        # (stall/no-content/regression) and an on-tier attempt count — today that evidence goes
        # only to stderr and is lost. Capture it here so a completed grind can be scored on
        # whether escalation was EARNED, not just on whether it reached green.
        escalations = []

        def _model_tag(b):
            # #118: name the COST TIER, not just the model. The gateway may fail over, and
            # a same-model free->paid escalation used to be invisible here. bullpen_modeltag
            # only says "switch" when the name or the tier actually changed, so the common
            # case (asked == served, gateway echoing the id bare) stays quiet.
            return model_tag(b["model"], b.get("served_model"), gateway=GATEWAY)

        def report(b, eval_verdict=None):
            stat = sh(["git", "-C", work, "diff", "--stat", baseline, b["sha"], "--", ".", *NOISE],
                      timeout=60).stdout.strip()
            diff = sh(["git", "-C", work, "diff", baseline, b["sha"], "--", ".", *NOISE],
                      timeout=60).stdout
            if b["passed"]:
                # #2 of the W31 self-improvement findings: 125 GREENs shipped, only 41 ever
                # reviewed. Flag the risky subset for an automatic @reviewer dispatch (serve()
                # below parses this line) instead of leaving every GREEN's review to chance —
                # escalated past the floor tier, a regression happened along the way, or the
                # diff grew past a "minimal fix" size are exactly the cases most likely to hide
                # a real problem in code nobody looks at again.
                escalated = b["model"] != TIERS[0]
                n_regressions = sum(1 for e in escalations if e.get("reason") == "regression")
                difflines = _diffstat_lines(stat)
                reasons = []
                if escalated: reasons.append(f"escalated to {b['model']}")
                if n_regressions: reasons.append(f"{n_regressions} regression(s)")
                if difflines > LARGE_DIFF_LINES: reasons.append(f"large diff ({difflines} lines)")
                auto_review = ("AUTO-REVIEW: yes (" + ", ".join(reasons) + ")") if reasons else "AUTO-REVIEW: no"
                # #4 of the W31 findings: the escalation-discipline trajectory-eval (below) was
                # computed and saved to mneme on every grind but had no consumer — surface its
                # verdict right here, in the exact reply @reviewer reads on an auto-flagged GREEN,
                # instead of leaving it stranded in /trajectory/grinder for nobody to query.
                eval_line = f"\nESCALATION-EVAL: {eval_verdict}" if eval_verdict else ""
                return (f"GREEN ✅ — {test_rel} passes ({_model_tag(b)}).\n{stat}\n"
                        f"worktree: {work} (diff below, apply after review)\n{auto_review}{eval_line}\n\n{diff[:3500]}")
            return (f"RED ❌ — best-so-far {b['failed']} failing ({_model_tag(b)}); could not reach "
                    f"green in {MAX_ITERS} attempts. worktree left at {work} for a human.\n{stat}\n\n"
                    f"{diff[:3000]}\nlast pytest:\n{b['tail']}")

        def save_best(passed, failed, tail, model, served_model=None):
            sh(["git", "-C", work, "add", "-A"], timeout=30)
            sh(["git", "-C", work, "-c", "user.email=grind@bullpen", "-c", "user.name=grinder",
                "commit", "-q", "--no-verify", "--allow-empty",
                "-m", f"grind checkpoint ({failed} failing)"], timeout=30)
            best.update(failed=failed, passed=passed, tail=tail, model=model, served_model=served_model,
                        sha=sh(["git", "-C", work, "rev-parse", "HEAD"], timeout=30).stdout.strip())
            if result_file:                              # atomic flush — survives a SIGKILL mid-write
                try:
                    open(result_file + ".tmp", "w").write(report(best))
                    os.replace(result_file + ".tmp", result_file)
                except Exception:
                    pass

        passed, failed, tail, fails = run_tests(work, test_rel)
        if passed:
            # R1: a spec ALREADY GREEN at baseline is vacuous — nothing to implement, and reporting
            # "GREEN ✅" with an empty diff is a false pass. A grind spec must FAIL first.
            return (f"INVALID SPEC ❌ — {test_rel} is already GREEN at baseline (passes before any "
                    f"change). A grind spec must FAIL first; nothing to implement.")
        tier, on_tier = 0, 0
        if failed < best["failed"]:
            save_best(passed, failed, tail, TIERS[tier])
        history = f"Task: {task}\n\nTest suite ({test_rel}):\n{test_src}\n"

        def rewind_to_best():
            # undo the last change and re-derive state from the best checkpoint
            sh(["git", "-C", work, "reset", "-q", "--hard", best["sha"]], timeout=30)
            sh(["git", "-C", work, "clean", "-fdq"], timeout=30)
            return run_tests(work, test_rel)

        for it in range(MAX_ITERS):
            if best["passed"]:
                break
            model = TIERS[tier]
            prev_fails = set(fails)
            # ONE FAILING TEST AT A TIME: when >1 test fails, hand the (weak) model a single
            # digestible target instead of "make the whole suite green". With <=1 failing — or a
            # collection error, where there are no node-ids — this naturally IS the whole spec.
            target = fails[0] if len(fails) > 1 else None
            tgt = "\n\n".join(f"FILE: {t}\n{open(os.path.join(work,t)).read()}"
                              for t in targets if os.path.isfile(os.path.join(work, t)))
            focus = (f"\n{len(fails)} tests fail. Make THIS ONE pass first and DO NOT break any test "
                     f"that already passes:\n    {target}\n" if target
                     else "\nMake the failing test(s) pass without breaking any that already pass.\n")
            user = (history + f"\nCurrent target file(s):\n{tgt}\n\n"
                    f"Latest pytest output ({failed} failing):\n{tail}\n{focus}\n"
                    "Return the corrected file(s) in the FILE: format.")
            budget = out_budget(tgt)
            try:
                reply, meta = ask_model(model, system, user, max_tokens=budget)
            except Exception as e:
                tail += f"\n(model {model} error: {e})"; on_tier += 1
                if on_tier >= ESCALATE_AT and tier < len(TIERS) - 1:
                    escalations.append({"iter": it, "reason": "model-error", "on_tier_before": on_tier,
                                        "from": TIERS[tier], "to": TIERS[tier + 1]})
                    tier += 1; on_tier = 0
                continue
            # EMPTY REPLY IS NOT A NO-OP. A reasoning model can burn the entire output budget on
            # reasoning_tokens and return content=None (kimi-k3: 3997/4000 reasoning, 0 chars, six
            # iterations). Previously that was swallowed as an ordinary no-progress iteration and
            # was indistinguishable from an incapable model — it silently corrupted every
            # escalation decision. Say so loudly, raise the budget once, then escalate.
            if not reply.strip():
                fr, rt = meta.get("finish_reason"), meta.get("reasoning_tokens")
                sys.stderr.write(f"[iter {it} tier={model} EMPTY REPLY finish_reason={fr} "
                                 f"reasoning_tokens={rt} budget={budget}]\n")
                tail += (f"\n(model {model} returned NO CONTENT: finish_reason={fr}, "
                         f"reasoning_tokens={rt}, budget={budget})")
                # A reasoning model that hit the length cap with NO content spent the whole
                # budget thinking (kimi-k2.6: 6498 reasoning_tokens of a 6518 budget -> 0 chars).
                # Doubling creeps up one starved iteration at a time; go straight to the ceiling,
                # because we now KNOW it needs reasoning headroom + the file on top. (out_budget
                # sizes to the OUTPUT file and can't see reasoning coming; this reacts once it has.)
                if fr == "length" and budget < OUT_CEIL:
                    bumped = OUT_CEIL
                    sys.stderr.write(f"[reasoning starved: budget {budget} -> {bumped} (ceiling), retrying]\n")
                    try:
                        reply, meta = ask_model(model, system, user, max_tokens=bumped)
                    except Exception as e:
                        tail += f"\n(retry error: {e})"
                if not reply.strip():
                    on_tier += 1
                    if on_tier >= ESCALATE_AT and tier < len(TIERS) - 1:
                        escalations.append({"iter": it, "reason": "no-content", "on_tier_before": on_tier,
                                            "from": TIERS[tier], "to": TIERS[tier + 1]})
                        tier += 1; on_tier = 0
                        sys.stderr.write(f"[no-content -> escalate to {TIERS[tier]}]\n")
                    continue
            wrote = apply_files(work, reply, targets, test_rel)
            passed, failed, tail, fails = run_tests(work, test_rel)
            # REGRESSION = a test that was passing now fails (only meaningful when the previous state
            # had a real per-test failing set, not a collection error). This is the escalation signal
            # the test-manager wants: the cheap model fixed one thing and broke another, i.e. it can't
            # hold the whole picture — rewind the damage and hand UP a tier rather than let it thrash.
            regressed = bool(prev_fails) and bool(set(fails) - prev_fails)
            sys.stderr.write(f"[iter {it} tier={model} wrote={wrote} failed={failed} "
                             f"target={target or 'whole-spec'}{' REGRESSED' if regressed else ''}]\n")
            if regressed:
                broke = sorted(set(fails) - prev_fails)          # capture before rewind reassigns fails
                passed, failed, tail, fails = rewind_to_best()
                if tier < len(TIERS) - 1:
                    escalations.append({"iter": it, "reason": "regression", "on_tier_before": on_tier,
                                        "from": TIERS[tier], "to": TIERS[tier + 1], "broke": broke})
                    tier += 1; on_tier = 0
                    sys.stderr.write(f"[regression {broke} -> escalate to {TIERS[tier]}]\n")
                else:
                    on_tier += 1
                continue
            if passed or failed < best["failed"]:
                save_best(passed, failed, tail, model, meta.get("served_model"))   # new best -> checkpoint + durable flush
                on_tier = 0                               # progress — stay on this tier
            else:
                if failed > best["failed"]:               # worse by count without a named regression
                    passed, failed, tail, fails = rewind_to_best()
                on_tier += 1
            if not best["passed"] and on_tier >= ESCALATE_AT and tier < len(TIERS) - 1:
                escalations.append({"iter": it, "reason": "stall", "on_tier_before": on_tier,
                                    "from": TIERS[tier], "to": TIERS[tier + 1]})
                tier += 1; on_tier = 0
                sys.stderr.write(f"[stall -> escalate to {TIERS[tier]}]\n")
        # leave the worktree AT the best checkpoint so a human sees the best, not a regression
        sh(["git", "-C", work, "reset", "-q", "--hard", best["sha"]], timeout=30)
        verdict, note = _eval_escalation_discipline(rid, escalations, best["passed"], best["model"])
        return report(best, eval_verdict=f"{verdict} — {note}")
    finally:
        pass  # leave the worktree for inspection; `git worktree prune` + rm to clean up

# ---- room-worker mode: @<nick> reacts to a ticket, grinds, posts the diff ----
SIC   = "/usr/local/bin/sic"
TRUST = cfg.TRUST


# R4: privileged nicks must present the shared secret or room_say refuses the post.
# Read once at start; /etc/bullpen first (hertz), then the per-user copy (agent hosts).
def _post_secret():
    for p in ("/etc/bullpen/post-secret", os.path.expanduser("~/.config/bullpen/post-secret")):
        try:
            s = open(p).read().strip()
            if s:
                return s
        except OSError:
            pass
    return ""
POST_SECRET = _post_secret()

def _hz(*args, timeout=45):
    return subprocess.run([SIC, cfg.ROOM_HOST, "lmcp-tool", *args], capture_output=True, text=True, timeout=timeout)


GRIND_HOST = cfg.GRIND_HOST          # where the repos + pytest live
# Same path on noether and boltzmann (same username), so REPO doubles as the remote default.
DEFAULT_REPO = REPO

def _repo_test_files(repo):
    """The repo's pytest files, asked of git on the grind host. [] if anything goes wrong."""
    try:
        p = subprocess.run([SIC, GRIND_HOST, "git", "-C", repo, "ls-files",
                            "tests/test_*.py", "test_*.py", "*_test.go"],
                           capture_output=True, text=True, timeout=60)
        return sorted(f for f in p.stdout.split()
                      if f.endswith(".py") or f.endswith("_test.go"))
    except Exception:
        return []


def _resolve_test_rel(body, repo):
    """Find the executable spec a ticket refers to. Returns (path, error).

    Coordinators describe the FAILING TESTS ("test_epipe_… — BrokenPipeError at …") and name
    the repo, but often never write the FILE path — ticket 739 was a precise, well-formed
    ticket that @py still bounced because it lacked the literal string `tests/x.py`. The
    coordinator then has to re-send, burning a round on a regex rather than on the work.

    Resolution order: an explicit path; a bare test_*.py filename matched against the repo;
    or, when the repo has exactly ONE spec file, that one. Ambiguity is never guessed — it
    comes back as an error LISTING the candidates, so the reply is actionable.

    JS/TS specs (`.test.js` / `_test.js`, incl. single-file HTML-canvas game tests) and any
    ABSOLUTE spec path are matched too: the JS runner (test_kind/run_tests) already existed on
    the grind host, but this parser only knew tests/*.py and *_test.go, so every @jsdev ticket
    was silently bounced. An absolute match fully specifies the grind — serve() derives the repo
    from its directory (2026-07-24)."""
    m = (re.search(r"(?<!\w)/[\w./-]+\.(?:test|spec)\.(?:js|ts)\b", body or "")   # absolute JS/TS spec (guarded)
         or re.search(r"(?<!\w)/[\w./-]+_test\.(?:js|ts|go|py)\b", body or "")    # absolute *_test.* (guarded)
         or re.search(r"tests/[\w./-]+\.py", body or "")                          # tests/ relative py
         or re.search(r"(?<!\w)/[\w./-]+\.py\b", body or "")                     # absolute py (guarded)
         or re.search(r"[\w./-]*(?:\.test|\.spec|_test)\.(?:js|ts)\b", body or "")  # relative JS/TS
         or re.search(r"[\w./-]+_test\.go", body or ""))
    if m:
        return m.group(0), None

    files = _repo_test_files(repo)
    named = (set(re.findall(r"\b(test_[\w-]+\.py)\b", body or ""))
             | set(re.findall(r"\b([\w-]+_test\.go)\b", body or "")))
    if named:
        hits = sorted({f for f in files if os.path.basename(f) in named})
        if len(hits) == 1:
            return hits[0], None
        if len(hits) > 1:
            return None, ("ambiguous spec — the ticket names " + ", ".join(sorted(named)) +
                          " which match " + ", ".join(hits) + ". Give one path.")
        return None, ("the ticket names " + ", ".join(sorted(named)) +
                      f" but {repo} has no such file" +
                      (" (it has: " + ", ".join(files) + ")" if files else " and no tests at all"))
    if len(files) == 1:
        return files[0], None
    if not files:
        return None, (f"no executable spec in the ticket and {repo} has no test files — "
                      "write-tests -> @testdesigner first, then hand me the path.")
    return None, ("no spec path in the ticket, and " + repo + " has several: " +
                  ", ".join(files) + ". Add a line like `SPEC: " + files[0] + "`.")


def _repo_from_body(body):
    """A ticket may target a repo other than the default by saying `repo=<path>`.
    Without this @py could only ever grind ~/src/bullpen, so work on any other repo
    (e.g. ~/src/sic) was impossible to delegate.

    Coordinators also write the human form `REPO: boltzmann:/home/mfritsche/src/sic` —
    accept it, and drop any `host:` prefix, since the grind already runs on that host."""
    m = re.search(r"\brepo=(\S+)", body or "") or \
        re.search(r"^\s*REPO:\s*(\S+)", body or "", re.M | re.I)
    if not m:
        return None
    path = m.group(1).strip().rstrip(",;")
    return path.split(":", 1)[1] if re.match(r"^[\w.-]+:/", path) else path

def _tiers_from_body(body):
    """A ticket may escalate the model for ONE grind: `tiers=<a,b>` or `model=<m>` in the body
    overrides the default PY_TIERS. This makes markus' escalation policy expressible in-room —
    "if local DeepSeek can't crack a single stubborn test, hand that one task to a stronger
    model" — instead of only being settable via the service env. Comma-separated = a floor->
    ceiling ladder the grinder climbs on stall; a single name pins one model."""
    m = re.search(r"\b(?:tiers|models?)=([\w./,-]+)", body or "", re.I)
    return m.group(1).strip() if m else None

def extract_trace(stderr):
    """Pull the grinder's bracketed progress/reasoning trace out of raw stderr for the
    THINKING REVEAL: the [iter â¦], [EMPTY REPLY â¦], [reasoning starved â¦], [regression â¦]
    and [stall â¦] lines that _remote_grind otherwise discards. Pure: str -> str, order
    preserved, newline-joined, no trailing newline; pytest noise / warnings dropped."""
    out = [ln.strip() for ln in (stderr or "").splitlines() if ln.strip().startswith("[")]
    return "\n".join(out)


def _remote_grind(task, test_rel, targets):
    """serve() runs on noether (sic reaches both hertz-room and boltzmann); the grind engine
    (repo + pytest + proxy) runs on boltzmann. Shell the --once grind over there."""
    rf = f"/tmp/grind-result-{int(time.time())}.txt"
    try:
        tiers = _tiers_from_body(task) or PY_TIERS
        env_args = [f"GRIND_TIERS={tiers}", f"GRIND_RESULT={rf}"]
        repo = _repo_from_body(task)
        if repo:
            env_args.append(f"GRIND_REPO={repo}")
        p = subprocess.run([SIC, GRIND_HOST, "env", *env_args,
                            "bullpen-grinder", "--once", task, test_rel, *targets],
                           capture_output=True, text=True, timeout=3600)  # large units on slow boltzmann-local
        result = p.stdout.strip() or p.stderr.strip()[:2000] or f"(grinder: empty output rc{p.returncode})"
        trace = extract_trace(p.stderr)          # THINKING REVEAL: surface the discarded progress trace
        return (result + "\n\n" + trace) if trace else result
    except subprocess.TimeoutExpired:
        # the grinder flushes its best-so-far to GRIND_RESULT on every improvement, so even a hard
        # timeout-kill leaves the best artifact behind — recover it instead of throwing the grind away.
        try:
            best = subprocess.run([SIC, GRIND_HOST, "cat", rf],
                                  capture_output=True, text=True, timeout=45).stdout.strip()
            if best:
                return "(grinder hit the 3600s cap — best-so-far recovered:)\n\n" + best
        except Exception:
            pass
        return f"(grinder timed out on {cfg.GRIND_HOST}; no best-so-far artifact found)"
    except Exception as e:
        return f"(grinder invoke error: {e})"

def _room_read(since):
    try:
        r = _hz("room_read", f"since={since}")
    except Exception:
        return None
    if r.returncode != 0:
        return None
    out = []
    for line in r.stdout.splitlines():
        line = line.strip()
        if line:
            try: out.append(json.loads(line))
            except Exception: pass
    return out

def _auto_review_reason(reply):
    """Pulls the reason string out of grind()'s 'AUTO-REVIEW: yes (...)' marker line, or None."""
    m = re.search(r"^AUTO-REVIEW: yes \((.*)\)$", reply, re.M)
    return m.group(1) if m else None

def serve(nick):
    state = os.path.expanduser(f"~/.grinder-{nick}.since")
    def persist(v):
        try: open(state, "w").write(str(v))
        except Exception: pass
    def say(body, to="", typ="chat", rid=None):
        a = ["room_say", f"from={nick}", f"type={typ}", f"body={body}"]
        if POST_SECRET: a.append(f"secret={POST_SECRET}")
        if to: a.append(f"to={to}")
        if rid is not None: a.append(f"in_reply_to={rid}")
        try: return _hz(*a).returncode == 0
        except Exception: return False
    try:
        since = int(open(state).read().strip())
    except Exception:
        msgs = _room_read(0); since = max([m.get("id", 0) for m in msgs], default=0) if msgs else 0
        persist(since)
    sys.stderr.write(f"bullpen-grinder serving @{nick} from id {since}\n")
    while True:
        try:
            batch = _room_read(since)
            if batch is None:
                time.sleep(3); continue
            for m in batch:
                mid = m.get("id", since)
                if mid <= since: continue
                asker = (m.get("from") or "").lower().lstrip("@")
                if (m.get("from") == nick or m.get("type") not in ("ask", "chat")
                        or (m.get("to") or "").lstrip("@").lower() != nick or asker not in TRUST):
                    since = mid; persist(since); continue
                # R4: note if verified is missing (room API omits it when from has @ prefix),
                # but don't reject — TRUST above already authenticates.
                if not m.get("verified"):
                    sys.stderr.write(f"@{nick}: warning — ask without verified:true from '{asker}' (due to @ prefix?); TRUST auth suffices\n")
                    pass
                body = m.get("body", "")
                say("…@{} grinding".format(nick), to=f"@{asker}", typ="ack", rid=mid)
                test_rel, err = _resolve_test_rel(body, _repo_from_body(body) or DEFAULT_REPO)
                if err:
                    reply = "INVALID TICKET ❌ — " + err
                elif os.path.isabs(test_rel) and not _repo_from_body(body):
                    # An absolute spec fully specifies the grind: repo = its dir, spec = its
                    # basename. Inject repo=<dir> so _remote_grind sets GRIND_REPO, and let
                    # grind()'s git-backed inference pick the target (the scaffold stubs it) —
                    # the relative-path target regex below can't see /tmp/... paths anyway.
                    reply = _remote_grind(body + f"\nrepo={os.path.dirname(test_rel)}",
                                          os.path.basename(test_rel), [])
                else:
                    # Explicit targets only when they look like real files; anything else is
                    # left to grind()'s git-backed inference, which sees the actual repo.
                    # Never offer a test file as a target — the grinder must not edit the spec.
                    # `bin/[\w.-]+` alternative: bullpen's own scripts (bin/bullpen-*) have NO
                    # dot-extension, so the base dir/dir/file.ext pattern alone can never match
                    # them — meaning a ticket fixing bullpen's own code couldn't auto-infer its
                    # own target. tmp/ and absolute paths are excluded (never real repo targets);
                    # inclusionai/ling excluded because research tickets quoting model names like
                    # "inclusionai/Ling-flash-2.0" look exactly like a dir/file target to this
                    # regex otherwise. (Recovered 2026-07-28: this fix existed only as an
                    # undeployed hand-patch on boltzmann's /usr/local/bin copy, never committed —
                    # merged back in when that staleness was found.)
                    targets = [t for t in re.findall(r"\b(?:bin/[\w.-]+|[\w-]+(?:/[\w.-]+)*/[\w.-]+\.\w+)\b", body)
                               if not t.startswith("tests/") and t != test_rel
                               and not t.startswith("tmp/") and not t.startswith("/")
                               and not t.startswith("inclusionai") and not t.startswith("ling")]
                    reply = _remote_grind(body, test_rel, targets)
                review_reason = _auto_review_reason(reply)
                posted = False
                for _ in range(3):
                    if say(reply, to=f"@{asker}", typ="reply", rid=mid): posted = True; break
                    time.sleep(2)
                if posted: since = mid; persist(since)
                else: sys.stderr.write(f"@{nick}: reply failed id {mid}; retry\n"); break
                if review_reason:
                    say(f"🔍 auto-review — @{nick}'s grind for @{asker} flagged itself ({review_reason}); "
                        f"please review before it ships.\n\n{reply}", to="@reviewer", typ="ask")
        except Exception as e:
            sys.stderr.write(f"serve loop error: {e}\n")
        time.sleep(3)

if __name__ == "__main__":
    if len(sys.argv) >= 3 and sys.argv[1] == "--once":
        print(grind(sys.argv[2], sys.argv[3], sys.argv[4:]))
    elif len(sys.argv) >= 3 and sys.argv[1] == "--serve":
        serve(sys.argv[2].strip())
    else:
        print("usage: bullpen-grinder --once '<task>' tests/<f> [target ...] | --serve <nick>", file=sys.stderr)
        sys.exit(2)
