troupe-standup/wizard-work/volt/q118_verify.py

#!/usr/bin/env python3
"""Q118 instrument: a folder read is retried before it may be called a failure.

Q118's done-means has three parts and only one of them was ever graded:

  (a) a folder listing is retried -- 3 attempts, 2 s apart -- before it may be
      declared unreadable;
  (b) the failure line carries `attempts=` and `last=`, so the next occurrence
      is a measurement rather than an inference;
  (c) a folder that fails every attempt is still printed as unreadable and
      still changes the output (failure is a wake, never a silence).

Nothing in this repository reached `get_folder()`.  `CHECK-q115` grades the
carry-forward path and its harness *replaces* `get_folder` with a stub (see the
docstring of /root/.hermes/scripts/troupe_pipe_monitor_harness.py: "replaces
only the three boundaries ... `get` (Gitea HTTP), `get_folder` (the retried
folder read) and `token`"), and the one forced-failure run Q115 recorded drives
its stub to 502 on *all three* attempts -- which measures the exhausted branch
and never the recovery it exists for.

So this instrument drives the deployed retry loop itself, at the boundary the
module really crosses: it shells out to `curl`, so a fake `curl` first on PATH
answers a scripted sequence of HTTP codes and timestamps every call.  Nothing
inside the module is stubbed or re-implemented; the copy differs from the
deployed file only in four path/URL constants (rewritten into a temp dir) and,
for the mutant legs, one deliberate mutation.

Legs, all against the deployed bytes:
  recover        502 then 200 -> rc 0, both files listed, no UNREADABLE token,
                 exactly 2 calls, the gap between them ~2 s (the retry landed)
  exhaust        502/503/504  -> token names `attempts=3 last=504`, 3 calls,
                 two gaps ~2 s each, total span ~4 s (two sleeps, not three)
  firstok        200 on attempt 1 -> exactly 1 call, no token, no sleep
  wake_no_delete tick 1 healthy, tick 2 exhausted -> the file lines are
                 byte-identical and the token line is what differs: a wake, and
                 no fabricated deletion (this is Q115's repair, measured from
                 this side because Q118 (c) as written predates it)

Mutant legs prove the cells above are not vacuous -- each mutation must red the
cell it targets:
  no_retry       `range(1, tries + 1)` -> `range(1, 2)`   reds `recover`
  no_pause       `time.sleep(pause)`   -> `time.sleep(0.01)` reds `exhaust` gap
  last_first     `last = code...` -> keeps the first code   reds `exhaust` last=

Exit codes follow the repository's tools: 0 clean, 1 graded fault, 2 could not
grade (module or interpreter absent, or a constant anchor that no longer
matches -- a rewrite that silently did nothing would grade the live host).

Run: python3 wizard-work/volt/q118_verify.py [module.py]
The optional argument points the instrument at a copy of the module -- that is
how a whole-artifact negative control is taken (a copy whose default retry count
is 1 must red `recover` and `exhaust`).
"""
import hashlib
import json
import os
import re
import subprocess
import sys
import tempfile
import time
from pathlib import Path

DEFAULT_MODULE = "/root/.hermes/scripts/troupe_pipe_monitor.py"
MODULE = Path(sys.argv[1]) if len(sys.argv) > 1 else Path(DEFAULT_MODULE)
HOST_CACHES = (
    Path("/root/.hermes/cache/troupe_last_touch.json"),
    Path("/root/.hermes/cache/troupe_folder_listing.json"),
)
ITEMS = [
    {"type": "file", "path": "replies/2026-10-28-to-zephyrus-1.md", "sha": "aaa111bbb222"},
    {"type": "file", "path": "replies/2026-10-28-to-zephyrus-2.md", "sha": "ccc333ddd444"},
]
EXPECTED_FILES = sorted(i["path"] for i in ITEMS)
PAUSE = 2.0
GAP_LO, GAP_HI = PAUSE * 0.75, PAUSE * 1.9  # scheduler slack, not a wide band

MUTATIONS = {
    "no_retry": ("    for attempt in range(1, tries + 1):",
                 "    for attempt in range(1, 2):"),
    "no_pause": ("            time.sleep(pause)",
                 "            time.sleep(0.01)"),
    "last_first": ('        last = code.strip() or f"curl{proc.returncode}"',
                   '        last = last if last != "none" else '
                   '(code.strip() or f"curl{proc.returncode}")'),
}

FAKE_CURL = r'''#!/usr/bin/env python3
"""Stand-in for curl, first on PATH: scripted codes, every call timestamped."""
import json
import os
import sys
import time

args = sys.argv[1:]
url = args[-1]
want_code = "-w" in args
plan = json.load(open(os.environ["Q118_PLAN"]))
state = os.environ["Q118_STATE"]
idx = int(open(state).read().strip() or 0) if os.path.exists(state) else 0


def reply(code, body):
    with open(os.environ["Q118_LOG"], "a") as fh:
        fh.write("%.6f\t%s\t%s\n" % (time.time(), code, url))
    sys.stdout.write(body + ("\n" + code if want_code else ""))


if "/contents/replies" in url:
    i = min(idx, len(plan) - 1)
    code, body = plan[i]
    with open(state, "w") as fh:
        fh.write(str(idx + 1))
    reply(code, json.dumps(body) if code == "200" and body else body)
elif "/contents/" in url:
    reply("200", "[]")
elif "/branches/" in url:
    reply("200", json.dumps({"commit": {"sha": "deadbeefcafe0001"}}))
elif "/commits" in url:
    reply("200", json.dumps([{
        "sha": "beef1234cafe",
        "commit": {"author": {"name": "zephyrus-sparkborn",
                              "email": "z@users.noreply.github.com"}},
        "files": [],
    }]))
else:
    reply("404", "")
'''

FAULTS = []


def fault(leg, detail):
    FAULTS.append(leg)
    print(f"FAIL: {leg} -- {detail}")


def ok(leg, detail):
    print(f"ok   {leg} -- {detail}")


def digest(path):
    try:
        return hashlib.sha256(Path(path).read_bytes()).hexdigest()[:16]
    except OSError:
        return "absent"


def build_module(work: Path, mutate=None) -> Path:
    """Copy the deployed module with its four outside-world constants repointed."""
    text = MODULE.read_text(encoding="utf-8")
    if mutate:
        old, new = MUTATIONS[mutate]
        if text.count(old) != 1:
            raise RuntimeError(
                f"mutation anchor {mutate!r} matched {text.count(old)} time(s), need 1")
        text = text.replace(old, new)
    subs = (
        (r"^API = .*$", 'API = "http://127.0.0.1:9/api/v1"'),
        (r"^ENV = Path\(.*\)$", f"ENV = Path({str(work / 'env')!r})"),
        (r"^CACHE = Path\(.*\)$", f"CACHE = Path({str(work / 'attrib.json')!r})"),
        (r"^LISTING_CACHE = Path\(.*\)$",
         f"LISTING_CACHE = Path({str(work / 'listing.json')!r})"),
    )
    for pat, rep in subs:
        text, n = re.subn(pat, rep, text, count=1, flags=re.M)
        if n != 1:
            raise RuntimeError(f"constant anchor {pat!r} matched {n} time(s), need 1")
    dest = work / "mon.py"
    dest.write_text(text, encoding="utf-8")
    return dest


def run_ticks(plans, mutate=None):
    """Run the module once per plan; return (rcs, outputs, calls, seconds)."""
    work = Path(tempfile.mkdtemp(prefix="q118-"))
    (work / "bin").mkdir()
    fake = work / "bin" / "curl"
    fake.write_text(FAKE_CURL, encoding="utf-8")
    fake.chmod(0o755)
    (work / "env").write_text("GITEA_TOKEN=q118-fake\n", encoding="utf-8")
    module = build_module(work, mutate=mutate)

    rcs, outs, calls = [], [], []
    started = time.time()
    for n, plan in enumerate(plans):
        (work / f"plan{n}.json").write_text(
            json.dumps([list(pair) for pair in plan]), encoding="utf-8")
        log = work / f"log{n}.txt"
        log.unlink(missing_ok=True)
        state = work / f"state{n}.txt"
        state.unlink(missing_ok=True)
        env = dict(os.environ)
        env["PATH"] = str(work / "bin") + ":" + env.get("PATH", "/usr/bin:/bin")
        env["Q118_PLAN"] = str(work / f"plan{n}.json")
        env["Q118_LOG"] = str(log)
        env["Q118_STATE"] = str(state)
        proc = subprocess.run([sys.executable, str(module)], capture_output=True,
                              text=True, env=env, timeout=120)
        rcs.append(proc.returncode)
        outs.append(proc.stdout + proc.stderr)
        rows = []
        if log.exists():
            for line in log.read_text(encoding="utf-8").splitlines():
                ts, code, url = line.split("\t", 2)
                rows.append((float(ts), code, url))
        calls.append(rows)
    return rcs, outs, calls, time.time() - started


def path_lines(out):
    keep = []
    for line in out.splitlines():
        if line.startswith(("branch-head ", "head ", "UNREADABLE ", "NO-BASELINE ")):
            continue
        if line.startswith("state unreadable"):
            continue
        if not line.strip():
            continue
        keep.append(line.split(" ", 1)[0])
    return sorted(keep)


def folder_calls(rows):
    """Only the calls the retry loop under test makes (the folder read itself)."""
    return [r for r in rows if "/contents/replies" in r[2]]


def gaps(rows):
    return [b[0] - a[0] for a, b in zip(rows, rows[1:])]


def case_recover(mutate=None):
    rcs, outs, calls, _ = run_ticks([(("502", ""), ("200", ITEMS))], mutate=mutate)
    out, rows = outs[0], folder_calls(calls[0])
    listed = path_lines(out)
    if rcs[0] != 0:
        return False, f"rc {rcs[0]} != 0"
    if "UNREADABLE" in out:
        return False, "a transient failure was still reported as UNREADABLE"
    if listed != EXPECTED_FILES:
        return False, f"listing {listed} != {EXPECTED_FILES}"
    if len(rows) != 2:
        return False, f"{len(rows)} folder call(s), expected 2 (one retry)"
    g = gaps(rows)[0]
    if not GAP_LO <= g <= GAP_HI:
        return False, f"retry gap {g:.2f}s outside [{GAP_LO}, {GAP_HI}]"
    return True, f"recovered on attempt 2, gap {g:.2f}s, {len(listed)} files listed"


def case_exhaust(mutate=None):
    plan = (("502", ""), ("503", ""), ("504", ""))
    rcs, outs, calls, _ = run_ticks([plan], mutate=mutate)
    out, rows = outs[0], folder_calls(calls[0])
    if rcs[0] != 0:
        return False, f"rc {rcs[0]} != 0"
    want = "UNREADABLE replies attempts=3 last=504 carried=0 from=none"
    if want not in out:
        got = [l for l in out.splitlines() if l.startswith("UNREADABLE")] or ["<none>"]
        return False, f"token {got[0]!r} != {want!r}"
    if "NO-BASELINE replies" not in out:
        return False, "exhausted read with no baseline did not say NO-BASELINE"
    if len(rows) != 3:
        return False, f"{len(rows)} folder call(s), expected 3"
    g = gaps(rows)
    if len(g) != 2 or not all(GAP_LO <= x <= GAP_HI for x in g):
        return False, f"gaps {[round(x, 2) for x in g]} outside [{GAP_LO}, {GAP_HI}]"
    span = rows[-1][0] - rows[0][0]
    if not 2 * GAP_LO <= span <= 2 * PAUSE * 1.9:
        return False, f"span {span:.2f}s is not two pauses (three sleeps would be ~6s)"
    return True, (f"attempts=3 last=504, gaps {[round(x, 2) for x in g]}, "
                  f"span {span:.2f}s, wake token present")


def case_firstok(mutate=None):
    rcs, outs, calls, secs = run_ticks([(("200", ITEMS),)], mutate=mutate)
    out, rows = outs[0], folder_calls(calls[0])
    if rcs[0] != 0:
        return False, f"rc {rcs[0]} != 0"
    if "UNREADABLE" in out or "NO-BASELINE" in out:
        return False, "a healthy read printed a failure token"
    if path_lines(out) != EXPECTED_FILES:
        return False, f"listing {path_lines(out)} != {EXPECTED_FILES}"
    if len(rows) != 1:
        return False, f"{len(rows)} folder call(s), expected 1 (no retry when it lands)"
    if secs > 1.5:
        return False, f"one-attempt tick took {secs:.2f}s -- a pause ran on a success"
    return True, f"1 call, {len(path_lines(out))} files, {secs:.2f}s, no token"


def case_wake_no_delete(mutate=None):
    rcs, outs, calls, _ = run_ticks(
        [(("200", ITEMS),), (("502", ""), ("502", ""), ("502", ""))], mutate=mutate)
    if rcs != [0, 0]:
        return False, f"rcs {rcs} != [0, 0]"
    healthy, failed = outs
    if path_lines(healthy) != EXPECTED_FILES:
        return False, f"healthy tick listed {path_lines(healthy)}"
    if path_lines(failed) != EXPECTED_FILES:
        return False, (f"failure tick listed {path_lines(failed)} -- the carry did "
                       "not hold, so the delta reads as deletions")
    if "UNREADABLE replies attempts=3 last=502 carried=2" not in failed:
        return False, "failure tick did not name the carry"
    if healthy == failed:
        return False, "the ticks are byte-identical -- the failure is a silence"
    return True, "file lines identical across the failure, token line is the difference"


DEPLOYED_CASES = (
    ("recover", case_recover),
    ("exhaust", case_exhaust),
    ("firstok", case_firstok),
    ("wake_no_delete", case_wake_no_delete),
)
MUTANT_CASES = (
    ("no_retry", "recover", case_recover),
    ("no_pause", "exhaust", case_exhaust),
    ("last_first", "exhaust", case_exhaust),
)


def main() -> int:
    if not MODULE.is_file():
        print(f"FAIL: could not grade -- module missing at {MODULE}", file=sys.stderr)
        return 2
    before = {str(p): digest(p) for p in HOST_CACHES}
    passed = 0
    try:
        for name, fn in DEPLOYED_CASES:
            good, detail = fn()
            (ok if good else fault)(name, detail)
            passed += 1 if good else 0

        for mutation, target, fn in MUTANT_CASES:
            good, detail = fn(mutate=mutation)
            if good:
                fault(f"mutant:{mutation}", f"the {target} cell stayed green -- it does "
                                           f"not grade the retry it claims to")
            else:
                ok(f"mutant:{mutation}", f"caught by {target}: {detail}")
                passed += 1
    except RuntimeError as exc:
        print(f"FAIL: could not grade -- {exc}", file=sys.stderr)
        return 2

    after = {str(p): digest(p) for p in HOST_CACHES}
    if before != after:
        fault("host-cache", f"a graded run wrote host state: {before} -> {after}")
    else:
        ok("host-cache", "deployed caches byte-identical after the run")
        passed += 1

    total = len(DEPLOYED_CASES) + len(MUTANT_CASES) + 1
    print(f"q118: {total} leg(s), {passed} passed, {len(FAULTS)} fault(s)")
    return 0 if not FAULTS else 1


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