test_gates.py

#!/usr/bin/env python3
"""Self-check for the goal-contract loop-fix + anti-stub hardening (2026-07-02).
Run: python3 test_gates.py  — exits non-zero on any failure. Stdlib only."""
import importlib.util, json, os, sys, tempfile

HERE = os.path.dirname(os.path.abspath(__file__))


def load(name, fname):
    spec = importlib.util.spec_from_file_location(name, os.path.join(HERE, fname))
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


w = load("witness", "witness.py")
g = load("dgate", "disposition-gate.py")
gg = load("ggate", "goal-gate.py")


def check(cond, msg):
    if not cond:
        print(f"  ✗ {msg}")
        sys.exit(1)
    print(f"  ✓ {msg}")


print("witness: FIX 5 — data/marker guards")
check(w.classify_line('{"total_count":3,"items":[{"title":"[bug]: agent cache"}]}') is None,
      "JSON API output (the 631-flood shape) is not witnessed")
check(w.classify_line("type/bug\tsomething isn't working\t1d2a3a") is None,
      "gh label listing (lowercase 'bug', no comment syntax) is not witnessed")
check(w.classify_line("##  [bug]: gateway telegram polling dies") is None,
      "markdown-header issue title is not witnessed")
check(w.classify_line("# TODO: migrate this before v2") == "marker",
      "real #-comment TODO still witnessed")
check(w.classify_line("// FIXME wrong offset under DST") == "marker",
      "real //-comment FIXME still witnessed")
check(w.classify_line("=== 3 failed, 12 passed in 1.21s ===") == "failing_test",
      "failing test line still witnessed")
check(w.classify_line("98 / 98 failed: 0") is None, "FIX 1 zero-failure shape still ignored")

print("witness: FIX 6 — digit-normalized dedup (done/undone goalposts)")
c1 = w.conditions_in("=== 3 failed, 12 passed in 1.21s ===\n=== 5 failed, 10 passed in 0.98s ===")
check(len(c1) == 1, "same failure shape with different counts/timings mints ONE condition")

print("witness: FIX 7 — stub markers in written content")
stubs = w.stub_conditions_in("src/auth/login.py",
                             "def login(user):\n    raise NotImplementedError\n")
check(len(stubs) == 1 and stubs[0]["kind"] == "stub_marker", "NotImplementedError write witnessed")
check(w.stub_conditions_in("src/api.ts", "return null; // TODO: implement refresh\n"),
      "comment-TODO stub in TS write witnessed")
check(w.stub_conditions_in("tests/test_login.py", "raise NotImplementedError\n") == [],
      "test files exempt")
check(w.stub_conditions_in("docs/notes.md", "TODO: write more\n") == [],
      "docs exempt")
check(w.stub_conditions_in("src/real.py", "def f():\n    return count + 1\n") == [],
      "clean code writes nothing")
path, text = w.written_text({"file_path": "a.py", "edits": [{"new_string": "todo!()"}]})
check("todo!()" in text and path == "a.py", "MultiEdit edits[].new_string extracted")

print("disposition-gate: FIX 8 — key-matched closure")
conds = [{"kind": "failing_test", "evidence": "1 failed", "key": "aaaaaaaaaaaa"},
         {"kind": "stub_marker", "evidence": "x.py: raise NotImplementedError", "key": "bbbbbbbbbbbb"}]
ok_text = ("DISPOSITION LOG\n- [FIXED] aaaaaaaaaaaa — reran suite, green\n"
           "- [FIXED_NOW] bbbbbbbbbbbb — implemented for real\n")
passes, *_ = g.evaluate(ok_text, conds)
check(passes, "both keys closed on tag lines -> passes")
# 2026-07-06: header trap removed — keyed closure alone passes, no "DISPOSITION LOG" needed
no_header = "- [FIXED] aaaaaaaaaaaa — reran suite, green\n- [FIXED_NOW] bbbbbbbbbbbb — done for real\n"
passes, *_ = g.evaluate(no_header, conds)
check(passes, "keyed closures WITHOUT the DISPOSITION LOG header still pass (header trap removed)")
fake_text = "DISPOSITION LOG\n- [FIXED] everything\n- [FIXED] all good\n- [FIXED] trust me\n"
passes, _, unclosed, *_ = g.evaluate(fake_text, conds)
check(not passes and len(unclosed) == 2, "tag-count paperwork without keys -> blocked (old gate passed this)")
half = "DISPOSITION LOG\n- [FIXED] aaaaaaaaaaaa — fixed\n- bbbbbbbbbbbb looked at it\n"
passes, _, unclosed, *_ = g.evaluate(half, conds)
check(not passes and unclosed[0]["key"] == "bbbbbbbbbbbb", "key without closing tag on its line -> that condition stays open")
owned = "DISPOSITION LOG\n- [FIXED] aaaaaaaaaaaa ok\n- [OWNED] bbbbbbbbbbbb later\n"
passes, *_ = g.evaluate(owned, conds)
check(not passes, "[OWNED] still banned at final")
legacy = [{"kind": "marker", "evidence": "# TODO: port 8080 hardcoded"}]  # no stored key
k = g._cond_key(legacy[0])
passes, *_ = g.evaluate(f"DISPOSITION LOG\n- [FIXED] {k} — made configurable\n", legacy)
check(passes, "legacy keyless witness entries get recomputed keys")

print("goal-gate: FIX 10 — pending hands off, failed is bounded")
with tempfile.TemporaryDirectory() as td:
    proof = os.path.join(td, "t.proof.json")
    task = {"task_id": "t", "manifest": os.path.join(td, "t.json"), "proof": proof}
    json.dump({"id": "t", "done_when": []}, open(task["manifest"], "w"))
    json.dump({"id": "t", "signed_off": "pending", "kickback_reason": "human_review awaits"}, open(proof, "w"))
    d, _ = gg.decide(task, {"pending": 0, "failed": 0})
    check(d == "block", "pending blocks once (forces the handoff report)")
    d, _ = gg.decide(task, {"pending": 1, "failed": 0})
    check(d == "yield_pending", "pending yields to operator after 1 block (was: forever)")
    json.dump({"id": "t", "signed_off": False, "kickback_reason": "failed checks: suite"}, open(proof, "w"))
    d, _ = gg.decide(task, {"pending": 0, "failed": 2})
    check(d == "block", "failed checks keep blocking while attempts remain")
    d, _ = gg.decide(task, {"pending": 0, "failed": 5})
    check(d == "yield_failed", "failed checks yield LOUDLY after 5 blocks (proof stays failed)")
    json.dump({"id": "t", "signed_off": True}, open(proof, "w"))
    d, _ = gg.decide(task, {"pending": 9, "failed": 9})
    check(d == "approve", "signed proof approves regardless of counters")

print("disarm: clears the whole gate set")
d = load("disarm", "disarm.py")
with tempfile.TemporaryDirectory() as td:
    d.ACTIVE_DIR = os.path.join(td, "active"); d.PTR_DIR = os.path.join(td, "active-goal")
    d.WITNESS_DIR = os.path.join(td, "witness")
    for p in (d.ACTIVE_DIR, d.PTR_DIR, d.WITNESS_DIR):
        os.makedirs(p)
    sid = "deadbeef-0000"
    for p in (os.path.join(d.ACTIVE_DIR, sid), os.path.join(d.ACTIVE_DIR, f"{sid}.gatestate"),
              os.path.join(d.PTR_DIR, f"{sid}.json"), os.path.join(d.PTR_DIR, f"{sid}.blocks")):
        open(p, "w").write("{}")
    d._remove({"sid": sid, "path": os.path.join(d.ACTIVE_DIR, sid), "armed_age": 60, "wcount": 0}, False)
    left = [f for sub in ("active", "active-goal") for f in os.listdir(os.path.join(td, sub))]
    check(left == [], f"sentinel + gatestate + warden pointer + blocks all cleared (left: {left})")

print("gate_notify: double-TTS throttle (2026-07-04)")
gn = load("gate_notify", "gate_notify.py")
with tempfile.TemporaryDirectory() as td:
    gn.LOCKDIR = os.path.join(td, "voicelocks")
    t = 3000.0  # mid-bucket, injected clock
    # the exact incident: two Stop gates fire for one session on one Stop
    votes = [gn._won_bucket("hermes", t), gn._won_bucket("hermes", t)]
    check(votes == [True, False], "two gates, one stop -> exactly one voice (double-TTS collapsed)")
    check(gn._won_bucket("other", t) is True, "a different session still gets its own voice")
    check(gn._won_bucket("hermes", t + gn.WINDOW + 1) is True, "same session, a later stop, pages again")

print("proctor_notify: Slack push rail — fingerprint, dedup, clear, fail-open (2026-07-06)")
pn = load("pnotify", "proctor_notify.py")
import socket as _socket
with tempfile.TemporaryDirectory() as td:
    pn.STATE = os.path.join(td, ".notify-state.json")
    pn.LOCK = os.path.join(td, ".notify.lock")
    pn.ERRLOG = os.path.join(td, "err.log")
    board = os.path.join(td, "board")
    calls = []
    def fake_post(url, payload, timeout):
        calls.append(payload); return True
    cfg = {"enabled": True, "slack_webhook": "https://hooks.slack.com/services/T/B/X",
           "vault_board_dir": board, "min_interval_s": 0}
    def sess(sid, project, human=None, failed=None):
        return {"sid": sid, "project": project, "goal": "g", "human": human or [], "failed": failed or []}
    # fingerprint: identity over checklist + failing NAMES + class, NOT volatile output
    check(pn.fingerprint(sess("a", "p", human=["flip it"])) == pn.fingerprint(sess("a", "p", human=["flip it"])),
          "fingerprint stable for identical content")
    check(pn.fingerprint(sess("a", "p", failed=[("test x", "exit=1 :: 3 failed")]))
          == pn.fingerprint(sess("a", "p", failed=[("test x", "exit=1 :: 5 failed @12:00")])),
          "fingerprint IGNORES volatile output (same failing check name -> same fp)")
    check(pn.fingerprint(sess("a", "p", human=["flip it"])) != pn.fingerprint(sess("a", "p", human=["flip it", "pull usb"])),
          "checklist change -> new fingerprint")
    s_safe = sess("a", "proj", human=["flip it"])
    pn.notify([s_safe], cfg=cfg, post=fake_post); check(len(calls) == 1, "first ask posts to Slack")
    pn.notify([s_safe], cfg=cfg, post=fake_post); check(len(calls) == 1, "identical ask on the next Stop does NOT re-post (dedup)")
    pn.notify([sess("a", "proj", failed=[("suite", "exit=1")])], cfg=cfg, post=fake_post)
    check(len(calls) == 2, "state transition safe->failed re-posts exactly once")
    pn.notify([], cfg=cfg, post=fake_post)
    check(len(calls) == 3 and "Cleared" in json.dumps(calls[-1]), "resolved session posts a Cleared line")
    pn.notify([], cfg=cfg, post=fake_post); check(len(calls) == 3, "nothing waiting + nothing changed -> ZERO further posts")
    # fail-open + self-heal
    calls.clear()
    def boom(u, p, t):
        raise RuntimeError("slack down")
    pn.notify([sess("b", "p2", human=["do x"])], cfg=cfg, post=boom)  # must not raise
    check(True, "a raising poster does not propagate (fail-open)")
    pn.notify([sess("b", "p2", human=["do x"])], cfg=cfg, post=fake_post)
    check(len(calls) == 1, "failed post self-heals — state uncommitted, retried on next Stop")
    # config gating + secret validation
    calls.clear()
    pn.notify([sess("c", "p3", human=["y"])], cfg={"enabled": False}, post=fake_post)
    check(len(calls) == 0, "disabled config -> no post")
    pn.notify([sess("c", "p3", human=["y"])], cfg={"enabled": True, "slack_webhook": "http://evil.example"}, post=fake_post)
    check(len(calls) == 0, "non-hooks.slack.com webhook rejected -> no post")
    # per-hostname vault board written
    host = _socket.gethostname().split(".")[0] or "host"
    bpath = os.path.join(board, f"Operator Inbox - {host}.md")
    check(os.path.exists(bpath), "per-hostname vault board file written (multi-machine safe)")
    check("Proctor Inbox" in open(bpath).read(), "board has content")

    def reset():
        for p in (pn.STATE, pn.LOCK):
            try: os.remove(p)
            except Exception: pass

    # overflow: >MAX_MSG sessions -> one post (8 shown + '+N more'); the un-shown overflow is
    # NOT committed, so it re-posts next round (review finding #1 — no silent drop off Slack)
    reset(); calls.clear()
    many = [sess(f"o{i}", "proj", human=["x"]) for i in range(10)]
    pn.notify(many, cfg=cfg, post=fake_post)
    check(len(calls) == 1 and "+2 more" in json.dumps(calls[-1]), "overflow: 10 sessions -> one post, 8 shown + '+2 more'")
    check(len(json.load(open(pn.STATE))["sessions"]) == 8, "overflow: only the 8 rendered are committed")
    calls.clear(); pn.notify(many[8:], cfg=cfg, post=fake_post)
    check(len(calls) == 1, "overflow: the 2 un-shown sessions DO re-post next round (no silent drop)")

    # not-silent bias: a corrupt state file re-announces current asks
    reset(); open(pn.STATE, "w").write("{bad json"); calls.clear()
    pn.notify([sess("z", "p", human=["y"])], cfg=cfg, post=fake_post)
    check(len(calls) == 1, "corrupt state file -> re-announce (not-silent bias)")

    # min_interval throttle: a new ask within the window is held (uncommitted), posts later
    reset(); calls.clear()
    tcfg = {**cfg, "min_interval_s": 9999}
    pn.notify([sess("t1", "p", human=["a"])], cfg=tcfg, post=fake_post)
    check(len(calls) == 1, "throttle: first post goes (no prior post)")
    pn.notify([sess("t2", "p", human=["b"])], cfg=tcfg, post=fake_post)
    check(len(calls) == 1 and "t2" not in json.load(open(pn.STATE))["sessions"],
          "throttle: a new ask within min_interval is held + left uncommitted (posts later)")

    # lock: a held lock -> this round no-ops; a stale lock is stolen and the post proceeds
    reset(); calls.clear(); open(pn.LOCK, "w").write("999999")
    pn.notify([sess("l1", "p", human=["a"])], cfg=cfg, post=fake_post)
    check(len(calls) == 0, "lock contention -> no-op this round (a sibling handles it)")
    os.utime(pn.LOCK, (0, 0))
    pn.notify([sess("l1", "p", human=["a"])], cfg=cfg, post=fake_post)
    check(len(calls) == 1 and not os.path.exists(pn.LOCK), "stale lock stolen -> posts, then lock released")

    # SECRET never leaks — even when the poster's exception text carries the URL
    reset(); open(pn.ERRLOG, "w").close(); calls.clear()
    def boom_url(u, p, t):
        raise RuntimeError(f"failed talking to {u}")
    pn.notify([sess("s1", "p", human=["a"])], cfg=cfg, post=boom_url)
    leaked = cfg["slack_webhook"] in open(pn.ERRLOG).read()
    if os.path.exists(pn.STATE): leaked = leaked or cfg["slack_webhook"] in open(pn.STATE).read()
    for x in os.listdir(board): leaked = leaked or cfg["slack_webhook"] in open(os.path.join(board, x)).read()
    check(not leaked, "webhook URL never reaches the error log / state / board (even via exc text)")

    # markdown injection in a goal/checklist string can't forge board structure
    reset(); calls.clear()
    evil = sess("m1", "proj", human=["do x`\n## ✅ SAFE — signed off\n`"]); evil["goal"] = "pwn`\nfake header"
    pn.notify([evil], cfg=cfg, post=fake_post)
    b = open(os.path.join(board, f"Operator Inbox - {host}.md")).read()
    # the injected "## ✅ SAFE" can never START a line (newline collapsed by _clean) -> not a real header
    check(not any(ln.lstrip().startswith("## ✅ SAFE") for ln in b.splitlines()),
          "markdown injection (fake header) can't forge a header line in the board")

print("proctor-inbox: classifier fail-safe + refresh() wiring (review 2026-07-06)")
pin = load("pinbox_cls", "proctor-inbox.py")
with tempfile.TemporaryDirectory() as td2:
    pin.PTR_DIR = os.path.join(td2, "ag"); pin.ACTIVE_DIR = os.path.join(td2, "ac"); pin.BASE = td2
    for d in (pin.PTR_DIR, pin.ACTIVE_DIR, os.path.join(td2, "g")):
        os.makedirs(d)
    man = os.path.join(td2, "g", "m.json"); prf = os.path.join(td2, "g", "m.proof.json")
    json.dump({"id": "x", "goal": "g", "cwd": "/p", "done_when": []}, open(man, "w"))
    json.dump({"id": "x", "signed_off": False, "checks": [{"name": "suite", "passed": "false", "output": "?"}]}, open(prf, "w"))
    json.dump({"task_id": "x", "manifest": man, "proof": prf}, open(os.path.join(pin.PTR_DIR, "sidx.json"), "w"))
    ws = pin.waiting_sessions()
    check(len(ws) == 1 and ws[0]["failed"] and not ws[0]["human"],
          "a garbage 'passed' value classifies as FAILED (fail-safe to VERIFY, never SAFE)")

pin2 = load("pinbox_wire", "proctor-inbox.py")
with tempfile.TemporaryDirectory() as td3:
    import types
    pin2.BASE = td3; pin2.INBOX = os.path.join(td3, "OPERATOR-INBOX.md")
    stub = types.ModuleType("proctor_notify"); rec = {}
    stub.notify = lambda sessions, **k: rec.__setitem__("sessions", sessions)
    sys.modules["proctor_notify"] = stub
    pin2.waiting_sessions = lambda: [{"sid": "w", "project": "p", "goal": "g", "human": ["x"], "failed": []}]
    pin2.refresh()
    check(rec.get("sessions") == [{"sid": "w", "project": "p", "goal": "g", "human": ["x"], "failed": []}],
          "refresh() calls proctor_notify.notify with the sessions (wiring covered)")
    del sys.modules["proctor_notify"]

print("disposition-gate: closure ledger persists across turns (field defect #4)")
conds2 = [{"kind": "error", "evidence": "x", "key": "aaaaaaaaaaaa"},
          {"kind": "error", "evidence": "y", "key": "bbbbbbbbbbbb"}]
m1 = "- [FIXED] aaaaaaaaaaaa ok\n- [FIXED] bbbbbbbbbbbb ok\n"
check(g.closed_keys_in(m1) == {"aaaaaaaaaaaa", "bbbbbbbbbbbb"}, "closed_keys_in extracts keys on closing-tag lines")
check(g.closed_keys_in("- [FIXED] no key here\naaaaaaaaaaaa (no tag on this line)") == set(),
      "a key without a tag on its line is NOT counted closed")
p, *_ = g.evaluate(m1, conds2); check(p, "both keys closed in the message -> passes")
led = g.closed_keys_in(m1)
p, *_ = g.evaluate("just a one-line status update, no closures", conds2, led)
check(p, "keys in the ledger stay closed on a later bare message (no re-paste needed)")
p, _, unc, *_ = g.evaluate("noise", conds2, {"aaaaaaaaaaaa"})
check(not p and len(unc) == 1 and unc[0]["key"] == "bbbbbbbbbbbb", "a key neither in ledger nor message stays open")
# forged/unwitnessed key: closed_keys_in harvests it, but main() banks only keys in the witnessed set
forged = g.closed_keys_in("- [FIXED] deadbeefcafe — a key that was never witnessed")
witnessed = {"aaaaaaaaaaaa", "bbbbbbbbbbbb"}
check((forged & witnessed) == set(), "a closing tag citing a non-witnessed key banks nothing (intersection guard, defect #4)")

print("arm-contract: reset only on a genuinely NEW contract (field defect #5)")
ac = load("armc", "arm-contract.py")
with tempfile.TemporaryDirectory() as td5:
    ac.CONTRACTS_DIR = os.path.join(td5, "contracts"); os.makedirs(ac.CONTRACTS_DIR)
    check(ac._is_new_contract("s1", "GOAL: x") is True, "no stored contract -> new (reset)")
    open(os.path.join(ac.CONTRACTS_DIR, "s1.txt"), "w").write("GOAL: x")
    check(ac._is_new_contract("s1", "GOAL: x  ") is False, "identical re-paste (whitespace-insensitive) -> NOT new (no wipe)")
    check(ac._is_new_contract("s1", "GOAL: y") is True, "changed contract -> new (reset)")
    ac._reset_session_state("../evil")  # path-traversal guard: must be a no-op, must not raise
    check(True, "reset with a traversal session id is a safe no-op")

print("disposition-gate: a yielded/flooded session voices ONCE, not every stop (2026-07-07)")
voiced = []
g._voice = lambda sid, msg: voiced.append(msg)
g._save_state = lambda sid, s: None            # keep it off disk
st = {}
try: g._yield_open("s", st, "witness log flooded (221 conditions)", "dumpster")
except SystemExit: pass
check(len(voiced) == 1 and st.get("yielded"), "first flood yield speaks once + records the yield")
try: g._yield_open("s", st, "witness log flooded (221 conditions)", "dumpster")
except SystemExit: pass
check(len(voiced) == 1, "a repeat yield on the SAME session stays SILENT (no audio every turn)")

print("bridge: REPO/workdir parse (field defect #1)")
B2 = load("bridge_repo", "manifest-bridge.py")
check(B2.workdir("REPO: ~/Projects/hb\n") == os.path.expanduser("~/Projects/hb"), "REPO: section parsed + ~ expanded")
check(B2.workdir("# run in: /Users/x/code/foo\nGOAL: x") == "/Users/x/code/foo", "'# run in:' hint parsed")
check(B2.workdir("GOAL: x\n") is None, "no repo declared -> None")
manr = B2.bridge_contract("=== LOCKED CONTRACT ===\nGOAL: g\nACCEPTANCE:\n  - t: `pytest`\nREPO: /tmp/x\n", "tid")
check(manr.get("repo") == "/tmp/x", "bridge_contract puts the declared repo in the manifest")

print("disarm: purges witness by DEFAULT, --keep-witness opts out (field defect #6)")
d2 = load("disarm2", "disarm.py")
with tempfile.TemporaryDirectory() as td4:
    d2.ACTIVE_DIR = os.path.join(td4, "active"); d2.WITNESS_DIR = os.path.join(td4, "witness")
    d2.PTR_DIR = os.path.join(td4, "active-goal")
    for p in (d2.ACTIVE_DIR, d2.WITNESS_DIR, d2.PTR_DIR):
        os.makedirs(p)
    def seed(sid):
        open(os.path.join(d2.ACTIVE_DIR, sid), "w").write("armed")
        open(os.path.join(d2.WITNESS_DIR, f"{sid}.jsonl"), "w").write('{"k":1}\n')
    seed("dead0001"); d2.main(["dead0001"])
    check(not os.path.exists(os.path.join(d2.WITNESS_DIR, "dead0001.jsonl")), "default disarm purges the witness log")
    seed("dead0002"); d2.main(["dead0002", "--keep-witness"])
    check(os.path.exists(os.path.join(d2.WITNESS_DIR, "dead0002.jsonl")), "--keep-witness retains the witness log")

print("\nall green")