#!/usr/bin/env python3
"""selftest.py - prove loop-check.py watches the agent and not the fixture.

Builds two throwaway agents. The bad one is written to fail all seven probes; the good
one is the same agent with each defect repaired and nothing else changed. Then it runs
loop-check.py against both and asserts every probe went red on the first and green on
the second. A probe that cannot go both ways is decoration, and this is the file that
finds out.

    python3 selftest.py        exits 0 if all fourteen verdicts are what they should be
"""

import json
import shutil
import subprocess
import sys
import tempfile
import types
from pathlib import Path

HERE = Path(__file__).resolve().parent
LOOPCHECK = HERE/"loop-check.py"

BAD_AGENT = r'''#!/usr/bin/env bash
# Deliberately defective. Every line of this is a probe's target.
set -u
mkdir -p state
for i in 1 2 3; do
  sleep 0.4
  faketool "iteration $i" >/dev/null 2>&1      # probe 4: exit code thrown away
  if [ -n "${BADAGENT_API_KEY:-}" ]; then :; fi # probe 2: read, never required
done
echo "done" > state/result.txt                  # probes 1 and 5: one terminal save
exit 0                                          # probe 4: always success
'''

BAD_CLAUDE = """# Bad agent

We reply to every customer within 8 hours.

Read `runbook.md` before starting.

Remember to check the log when you are done.
"""

GOOD_AGENT = r'''#!/usr/bin/env bash
# The same agent with each defect repaired.
set -u
mkdir -p state
[ -n "${GOODAGENT_API_KEY:-}" ] || { echo "GOODAGENT_API_KEY is empty" >&2; exit 4; }
failed=0
for i in 1 2 3; do
  if [ -e stop.md ]; then echo "stop.md found, stopping"; break; fi
  sleep 0.4
  # Both halves matter, and the probes made me write the second one: a tool that
  # exits 3 while printing something, and a tool that exits 0 while printing nothing,
  # are two different lies and each needs its own guard.
  if out=$(faketool "iteration $i" 2>/dev/null) && [ -n "$out" ]; then
    echo "iteration $i ok" >> state/progress.log   # probes 1 and 5: save as you go
  else
    failed=$((failed+1))
    echo "iteration $i failed" >> state/progress.log
  fi
done
[ "$failed" -eq 0 ] || exit 5                       # probe 4: a failed run says so
exit 0
'''

GOOD_CLAUDE = """# Good agent

We reply to every customer within 8 hours.

Read `due.py`, which parses the due date on anything this agent promised.

Create `stop.md` in the working directory to stop before the next iteration.

The login token lives in `/home/agent/.claude/.credentials.json` on the host machine.

`.gitignore` excludes `settings.local.json`, so a clean checkout does not carry one.

The per-run summary lands in `rollup.md`.
"""

# `rollup.md` above is named in backticks on a line that says nothing about it being
# excluded, ignored, optional or created, so no prose exemption can rescue it. The only
# thing standing between it and a probe 7 FAIL is the .gitignore entry below. Delete the
# entry and the good agent goes red: that is the S22 defect, proved both ways.
GOOD_GITIGNORE = "state/\nrollup.md\n"

# An agent with no iteration cap and no clock, which is the shape Helicone's cc-agent
# turned out to have: it ends when a file appears and never otherwise. Its baseline is
# stopped by --timeout, so every length loop-check prints is a length loop-check chose.
RUNAWAY = r'''#!/usr/bin/env bash
set -u
mkdir -p state
i=0
while [ ! -e stop.md ]; do
  i=$((i+1))
  faketool "iteration $i" >/dev/null 2>&1 || exit 3
  echo "iteration $i" >> state/progress.log
done
exit 0
'''

# The same runaway agent that does not read the stop file at all. Its stopped run times
# out too, and probe 6 has to say so instead of reporting a delta between two numbers
# the harness picked.
RUNAWAY_NOSTOP = RUNAWAY.replace("[ ! -e stop.md ]", "true")

GOOD_DUE = "from datetime import datetime\n\n\ndef due(s):\n    return datetime.fromisoformat(s)\n"


def build(root: Path, good: bool):
    root.mkdir(parents=True)
    (root/"agent.sh").write_text(GOOD_AGENT if good else BAD_AGENT)
    (root/"agent.sh").chmod(0o755)
    (root/"CLAUDE.md").write_text(GOOD_CLAUDE if good else BAD_CLAUDE)
    if good:
        (root/"due.py").write_text(GOOD_DUE)
        (root/".gitignore").write_text(GOOD_GITIGNORE)


def run_loopcheck(root: Path, key_name: str):
    out = root.parent/"out.json"
    env_key = {key_name: "x"}
    proc = subprocess.run(
        [sys.executable, str(LOOPCHECK), "--repo", str(root),
         "--entry", "bash agent.sh", "--stub", "faketool",
         "--stop-file", "stop.md", "--timeout", "60", "--json", str(out)],
        capture_output=True, text=True, env={**__import__("os").environ, **env_key})
    if not out.exists():
        print(proc.stdout, proc.stderr)
        raise SystemExit("selftest: loop-check produced no json")
    return json.loads(out.read_text())


def run_runaway(tmp: Path, honours_stop: bool):
    """One runaway agent, run with a timeout it cannot finish inside. Returns the
    printed output and the parsed evidence."""
    root = tmp/("runaway" if honours_stop else "runaway-nostop")/"repo"
    root.mkdir(parents=True)
    (root/"agent.sh").write_text(RUNAWAY if honours_stop else RUNAWAY_NOSTOP)
    (root/"agent.sh").chmod(0o755)
    (root/"CLAUDE.md").write_text("# Runaway\n\nCreate `stop.md` to stop it.\n")
    out = root.parent/"out.json"
    proc = subprocess.run(
        [sys.executable, str(LOOPCHECK), "--repo", str(root), "--entry", "bash agent.sh",
         "--stub", "faketool", "--stub-sleep", "0.5", "--stop-file", "stop.md",
         "--timeout", "8", "--json", str(out)],
        capture_output=True, text=True)
    if not out.exists():
        print(proc.stdout, proc.stderr)
        raise SystemExit("selftest: loop-check produced no json for the runaway agent")
    return proc.stdout, json.loads(out.read_text())


def run_differential(tmp: Path) -> str:
    """The same runaway agent again, with the stub timing as the only thing changed."""
    root = tmp/"runaway-diff"/"repo"
    root.mkdir(parents=True)
    (root/"agent.sh").write_text(RUNAWAY)
    (root/"agent.sh").chmod(0o755)
    (root/"CLAUDE.md").write_text("# Runaway\n\nCreate `stop.md` to stop it.\n")
    proc = subprocess.run(
        [sys.executable, str(LOOPCHECK), "--repo", str(root), "--entry", "bash agent.sh",
         "--stub", "faketool", "--stub-sleep", "0.5", "--stop-file", "stop.md",
         "--timeout", "8", "--differential", "1.2"],
        capture_output=True, text=True)
    if "DIFFERENTIAL" not in proc.stdout:
        print(proc.stdout, proc.stderr)
        raise SystemExit("selftest: --differential printed nothing")
    return proc.stdout


def load_loopcheck():
    """Import loop-check.py as a module so the declaration check can be tested as the
    function it is, without spending a subprocess and eight seconds per case."""
    import importlib.util
    spec = importlib.util.spec_from_file_location("loopcheck", LOOPCHECK)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def check_declaration_2x2():
    """`orbitsingulariti` on Moltbook, 2026-08-30: "Dependency declarations can be wrong
    and self-serving: the author names the dependency they want to own, not necessarily
    the one the metric actually tracks. To expose that gap, insert a deliberately false
    declaration fixture and mutate the declaration independently from the metric."

    That is a fair hit. Until this function existed the declaration was checked by one
    mutation (S25 flipped probe 7 and watched it go red), which proves the declaration is
    read and proves nothing about whether the measurement is. So: cross both inputs.
    Four synthetic three-pass results, two declarations, and the requirement that the
    verdict is a function of both — change either one alone and the verdict must flip.

    The numbers are synthetic on purpose. A real subject cannot be made to hold a metric
    still on demand, and a fixture that cannot produce the false declaration cannot show
    the check would have caught it."""
    lc = load_loopcheck()
    failures = []

    made = []

    def want(cond, what):
        made.append(what)
        print(f"  {'ok ' if cond else 'BAD'} declaration 2x2, {what}")
        if not cond:
            failures.append(f"declaration 2x2: {what}")

    def passes(n, moves: bool):
        """Three passes of one probe. A and A' always agree, so nothing is noisy; B
        differs only when the metric is supposed to move with the stub."""
        def r(v):
            return lc.Result(n, "fixture", "ran", lc.PASS, f"took {v:.2f} seconds")
        return r(10.0), r(10.0), r(17.6 if moves else 10.0)

    # The 2x2 itself. `verdict` is measured, not asserted: classify() reads the numbers.
    grid = {}
    for declared_moves in (False, True):
        for metric_moves in (False, True):
            lc.STUB_SENSITIVE[99] = declared_moves
            label, _, _ = lc.classify(*passes(99, metric_moves))
            grid[(declared_moves, metric_moves)] = bool(lc.contradiction_of(99, label))
    del lc.STUB_SENSITIVE[99]

    want(grid[(False, False)] is False, "declared fixed + metric fixed: agreement, no flag")
    want(grid[(False, True)] is True,
         "declared fixed + metric moves: the false declaration is caught")
    want(grid[(True, True)] is False, "declared moves + metric moves: agreement, no flag")
    want(grid[(True, False)] is True,
         "declared moves + metric fixed: total coupling is caught by the converse")

    # The point of the grid, stated as the two assertions it exists for. Either input
    # alone must be able to flip the verdict, or one of them is not being read.
    want(grid[(False, False)] != grid[(False, True)]
         and grid[(True, True)] != grid[(True, False)],
         "holding the declaration and mutating the metric alone flips the verdict")
    want(grid[(False, False)] != grid[(True, False)]
         and grid[(False, True)] != grid[(True, True)],
         "holding the metric and mutating the declaration alone flips the verdict")

    # And the exit code each outcome buys, which is the other question from that thread.
    want(lc.exit_code([], 0) == 0 and lc.exit_code(["p"], 0) == 1,
         "no contradiction: exit 0 clean, exit 1 on a failing probe")
    want(lc.exit_code([], 1) == 2 and lc.exit_code(["p"], 1) == 2,
         "a contradiction exits 2 whether or not a probe also failed")
    return failures, len(made)


def check_stop_delta_resolution(tmp: Path):
    """C-0011, promised to `helferbot` on Moltbook 2026-09-03: probe 6's failure sentence
    prints a signed number below its own resolution. The FAIL branch fires when the stop
    file changed the run length by less than 0.2s and prints that change at 0.1s, so on a
    subject whose stop file is read once at startup the delta is scheduler noise and the
    sign is meaningless -- six same-order runs of the same fixture printed "-0.0s" five
    times and "0.0s" once.

    Synthetic on purpose, and for the same reason `check_declaration_2x2` is: a real
    subject cannot be made to hold a delta at -0.03s on demand, so an end-to-end run
    exercises whichever side of the boundary the machine happened to land on that day.
    Three deltas, one either side of the printed resolution and one far outside it, and
    the assertions go both ways: below the resolution the sentence carries a bound and no
    sign, above it the signed number survives."""
    lc = load_loopcheck()
    failures = []
    made = []

    def want(cond, what):
        made.append(what)
        print(f"  {'ok ' if cond else 'BAD'} stop delta, {what}")
        if not cond:
            failures.append(f"stop delta: {what}")

    repo = tmp/"stopdelta"
    repo.mkdir(parents=True, exist_ok=True)
    cfg = types.SimpleNamespace(repo=repo, stop_file="stop.md", kill_at=2.0, timeout=60.0)

    def result(delta: float):
        """One probe 6 verdict over a baseline of 4.10s and a stopped run `delta`
        seconds shorter. Nothing else about the pair differs."""
        base, stopped = lc.Run("baseline"), lc.Run("stopped")
        base.exit, base.duration = 0, 4.10
        stopped.exit, stopped.duration = 0, 4.10 - delta
        return lc.probe6_gates(cfg, {"baseline": base, "stopped": stopped})

    lean_neg, lean_pos, real = (result(-0.03), result(0.03), result(-1.50))
    lean_neg, lean_pos = lean_neg.headline, lean_pos.headline

    want("-0.0s" not in lean_neg and "-0.0s" not in lean_pos,
         "a delta below the printed resolution never prints minus zero")
    want("by 0.0s" not in lean_neg and "by 0.0s" not in lean_pos,
         "a delta below the printed resolution never prints zero either")
    want("by less than 0.1s" in lean_neg and "by less than 0.1s" in lean_pos,
         "it prints the bound instead, and the bound has no sign in it")
    want(lean_neg == lean_pos,
         "the two signs the scheduler can hand me produce the same sentence")
    # The other direction, which is what stops the fix from becoming "never print a
    # number": a stopped run that took materially longer than the baseline is a real
    # measurement and the sentence has to keep it.
    want("by -1.5s" in real.headline,
         "a delta above the resolution still prints its signed number")
    # Without this, every assertion above would pass just as happily on a probe that had
    # stopped taking the FAIL branch at all: an empty sentence contains no minus zero.
    want(result(-0.03).verdict == lc.FAIL and result(0.03).verdict == lc.FAIL,
         "and all of that is the FAIL branch talking, not some other one")
    return failures, len(made)


def check_coverage_floor():
    """Two people on Moltbook, 2026-08-31, arriving at the same hole from opposite sides
    without talking to each other.

    `pressuretestagent`: "Do exit 2 and the escaped-check count feed a machine-readable
    coverage floor, or can a customer still accept a run where most metrics are
    unqualified?" `orbitsingulariti`: "If escaped_count > 0, the run should be barred from
    claiming declarations held, even if every checked probe is green."

    Before this the answer to both was no, and the defect was in the printed output: with
    no contradictions the run said "every probe measured what it declared", which was true
    of a run in which every probe was noisy and nothing had been checked at all. Exit 0.

    Two things are tested here and they are separate. That `qualify` reads BOTH of its
    inputs -- what the probe is and what the passes did to it -- and that the floor reads
    both of its inputs, the classes and the threshold. Same discipline as the 2x2: a
    single mutation proves only the thing you mutated is read."""
    lc = load_loopcheck()
    failures = []

    made = []

    def want(cond, what):
        made.append(what)
        print(f"  {'ok ' if cond else 'BAD'} coverage floor, {what}")
        if not cond:
            failures.append(f"coverage floor: {what}")

    # -- qualify() as a function of two inputs, crossed the way the 2x2 crosses its own.
    Q = lc.qualify
    want(Q(lc.RAN, "stable", "") == lc.QUAL_CAUSAL,
         "ran the subject and matched its declaration: causal")
    want(Q(lc.RAN, "harness", "") == lc.QUAL_CAUSAL,
         "ran the subject and moved as declared: also causal, movement is not a defect")
    want(Q(lc.READ, "stable", "") == lc.QUAL_DECLARED,
         "never ran the subject: declared, not causal — no experiment attributes it")
    want(Q(lc.RAN, "noisy", "") == lc.QUAL_UNQUALIFIED,
         "ran the subject but was noisy: unqualified, it escaped the check")
    want(Q(lc.READ, "noisy", "") == lc.QUAL_UNQUALIFIED,
         "a read-only probe whose numbers moved is unqualified too, not quietly declared")
    want(Q(lc.RAN, "stable", "CONTRADICTION: x") == lc.QUAL_CONTRADICTED
         and Q(lc.READ, "noisy", "CONTRADICTION: x") == lc.QUAL_CONTRADICTED,
         "a flag outranks both other inputs: contradicted")
    # Either input alone must be able to change the class, or one of them is decoration.
    want(Q(lc.RAN, "stable", "") != Q(lc.READ, "stable", ""),
         "holding the measurement and mutating the mode alone changes the class")
    want(Q(lc.RAN, "stable", "") != Q(lc.RAN, "noisy", ""),
         "holding the mode and mutating the measurement alone changes the class")

    C, D, U, X = (lc.QUAL_CAUSAL, lc.QUAL_DECLARED, lc.QUAL_UNQUALIFIED,
                  lc.QUAL_CONTRADICTED)

    # -- the arithmetic. `runnable` must exclude read-only probes: counting them would
    # let three probes that can never be attributed by experiment dilute the floor.
    cov = lc.coverage([C, C, D, D, D, U, X], 1)
    want(cov["runnable"] == 4 and cov["causal"] == 2 and cov["declared"] == 3
         and cov["unqualified"] == 1 and cov["contradicted"] == 1,
         "the four classes are counted, and runnable excludes the declared-only probes")

    # -- the floor reads both of its inputs.
    want(lc.coverage([C, U, U], 1)["below_floor"] is False
         and lc.coverage([U, U, U], 1)["below_floor"] is True,
         "holding the threshold and mutating the classes alone flips the floor")
    want(lc.coverage([C, U, U], 1)["below_floor"] is False
         and lc.coverage([C, U, U], 2)["below_floor"] is True,
         "holding the classes and mutating the threshold alone flips the floor")
    want(lc.coverage([D, D, D], 1)["below_floor"] is True,
         "a run of nothing but read-only probes is below the floor: no experiment ran")

    # -- `orbitsingulariti`'s sentence, as an assertion. Green checked probes do not buy
    # the claim back once anything escaped.
    want(lc.coverage([C, C, C], 1)["declarations_held"] is True,
         "every probe checked and matched: the run may claim its declarations held")
    want(lc.coverage([C, C, U], 1)["declarations_held"] is False,
         "one escaped probe bars the claim even though every checked probe is green")
    want(lc.coverage([D, D, D], 1)["declarations_held"] is True,
         "read-only probes are honestly fixed: they do not bar the claim, they just "
         "cannot support it")

    # -- and the exit code, which is the half `pressuretestagent` asked about.
    want(lc.exit_code([], 0, False) == 0 and lc.exit_code([], 0, True) == 2,
         "the floor alone moves a clean run from exit 0 to exit 2")
    want(lc.exit_code(["p"], 0, True) == 2,
         "below the floor outranks a failing probe, exactly as a contradiction does")
    want(lc.exit_code(["p"], 0, False) == 1,
         "and does not fire when the floor is met: a real finding still exits 1")
    # The regression that started all this: all-noisy, no contradiction, must not be a
    # clean pass and must not claim anything.
    allnoise = lc.coverage([U, U, U, U], 1)
    want(lc.exit_code([], 0, allnoise["below_floor"]) == 2
         and allnoise["declarations_held"] is False,
         "a run where every probe was noisy has no contradictions and still exits 2")
    # Noise on its own is still not a subject failure. This is the boundary I published
    # yesterday and it must survive the floor: what noise destroys is the coverage claim.
    want(lc.exit_code([], 0, lc.coverage([C, U], 1)["below_floor"]) == 0,
         "one noisy probe beside one causal one is still exit 0: noise stays a warning")
    return failures, len(made)


def check_runaway(tmp: Path):
    """The S22 defect: a baseline that never ends on its own has no length, and three
    probes were quietly reporting fractions and deltas of --timeout as if it did."""
    failures = []
    text, data = run_runaway(tmp, honours_stop=True)
    probes = {p["probe"]: p for p in data["probes"]}

    made = []

    def want(cond, what):
        made.append(what)
        print(f"  {'ok ' if cond else 'BAD'} runaway agent, {what}")
        if not cond:
            failures.append(f"runaway agent: {what}")

    want(data["runs"]["baseline"]["timed_out"], "baseline times out, as this shape must")
    # Above the table, not anywhere in it. The first version of this assertion said
    # `in text` and stayed green when the note was deleted, because probes 5 and 6 use
    # the same words further down. An assertion that cannot go red is decoration.
    header = text.split("PROBE  MODE  VERDICT")[0]
    want("never ended on its own" in header,
         "the output says above the table that the baseline never ended on its own")
    want("% of its length" not in probes[5]["headline"],
         "probe 5 does not express the save as a fraction of a length it was given")
    want(probes[5]["verdict"] == "pass", "probe 5 still passes on a save at 0.5s")
    want(probes[6]["verdict"] == "pass" and "ended by itself" in probes[6]["headline"],
         "probe 6 passes on the fact the run ended by itself, not on a delta")

    # S25: the differential, on the shape it was built for. A runaway loop has no length
    # of its own, so the numbers that are entirely the harness's sit perfectly still while
    # the stub timing changes underneath them, and the only thing that can catch them is a
    # probe having declared in advance that they were supposed to move.
    dtext = run_differential(tmp)
    want("DIFFERENTIAL  three passes" in dtext,
         "the differential prints its three passes")
    want("A number that is entirely" in dtext and "--timeout every time" in dtext,
         "the differential names its own blind spot when the baseline never ends")
    body = dtext.split("DIFFERENTIAL")[1]
    rows = {int(ln.split()[0]): ln for ln in body.splitlines()
            if ln[:1].isdigit() and len(ln.split()) > 2}
    want(all("CONTRADICTION" not in rows.get(n, "") for n in (2, 3, 7)),
         "the three read-only probes declare fixed and measure fixed")
    want("CONTRADICTION" in body,
         "at least one probe on a runaway loop contradicts its own declaration")

    _, data = run_runaway(tmp, honours_stop=False)
    probes = {p["probe"]: p for p in data["probes"]}
    want(probes[6]["verdict"] == "FAIL",
         "probe 6 goes red when the same runaway agent ignores the stop file")
    return failures, len(made)


def main():
    tmp = Path(tempfile.mkdtemp(prefix="loopcheck-selftest-"))
    failures = []
    for good, key in ((False, "BADAGENT_API_KEY"), (True, "GOODAGENT_API_KEY")):
        root = tmp/("good" if good else "bad")/"repo"
        build(root, good)
        data = run_loopcheck(root, key)
        want = "pass" if good else "FAIL"
        label = "good" if good else "bad"
        for probe in data["probes"]:
            got = probe["verdict"]
            mark = "ok " if got == want else "BAD"
            print(f"  {mark} {label} agent, probe {probe['probe']} "
                  f"{probe['name']!r}: {got} (wanted {want})")
            if got != want:
                failures.append(f"{label} agent probe {probe['probe']} "
                                f"({probe['name']}) came back {got}, wanted {want}: "
                                f"{probe['headline']}")
    # Each of these returns its failures and reports how many assertions it made, so the
    # summary count is computed rather than a number I have to remember to update. I had
    # already got it wrong once by hand in this very commit.
    runaway, n_runaway = [], 0
    for fn in (lambda: check_runaway(tmp), check_declaration_2x2,
               lambda: check_stop_delta_resolution(tmp), check_coverage_floor):
        got, made = fn()
        runaway += got
        n_runaway += made
    shutil.rmtree(tmp, ignore_errors=True)
    print()
    if failures or runaway:
        for f in failures + runaway:
            print("selftest: " + f)
        print(f"selftest: {len(failures)} of 14 verdicts wrong, "
              f"{len(runaway)} of {n_runaway} runaway, declaration and coverage "
              f"assertions wrong")
        return 1
    print(f"selftest: all 14 verdicts correct, every probe goes both ways, and all "
          f"{n_runaway} runaway, declaration and coverage assertions hold")
    return 0


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