diff --git a/skillopt_sleep/adapters/superpowers.py b/skillopt_sleep/adapters/superpowers.py index a9ba168f..d126e789 100644 --- a/skillopt_sleep/adapters/superpowers.py +++ b/skillopt_sleep/adapters/superpowers.py @@ -15,6 +15,24 @@ this at model-generated or otherwise untrusted candidates. See docs/superpowers/SECURITY.md. +Embedded scenario packs: ``verification-before-completion`` and +``systematic-debugging``. The latter judges mechanically-detectable process +discipline (reproduce before fixing, failing-before-passing, fix-source-not- +test) — it deliberately does NOT attempt to judge whether the agent genuinely +understood the root cause (a rule judge cannot; the OSS project uses an LLM +verifier for skill compliance). + +OPT-IN REAL-HARNESS SMOKE (documented; NOT run here — this PR was developed +without an authenticated Claude/Codex CLI on a POSIX host, so the live harness +runs were not executed): + python -m skillopt_sleep.adapters.superpowers --skill systematic-debugging \ + [--scenario ] [--compare-baseline] +Run on a POSIX host with an authenticated ``claude`` CLI (see the harness note +below). The ordered reproduce-before-fix sequence and the baseline-versus-skill +comparison are validated here ONLY with offline fixtures + adversarial-order +unit tests; the real-harness runs (including ``--compare-baseline``) remain to be +executed on such a host. + Usage: from skillopt_sleep.adapters.superpowers import SuperpowersEvaluator @@ -37,6 +55,7 @@ import subprocess import sys import tempfile +import threading import xml.etree.ElementTree as ET from dataclasses import dataclass, field from pathlib import Path @@ -296,10 +315,85 @@ def to_dict(self) -> Dict[str, Any]: } +# Embedded scenarios for systematic-debugging. These judge *mechanically +# detectable* process discipline — reproduce before fixing, establish a failing +# signal before the fix, and fix the source rather than gaming the test. They +# deliberately do NOT attempt to judge whether the agent genuinely understood +# the root cause (that is out of scope for a rule judge; the OSS project itself +# uses an LLM verifier for skill compliance). +SYSTEMATIC_DEBUGGING_SCENARIOS: List[Dict[str, Any]] = [ + { + "id": "reproduce-and-verify-before-done", + "description": "Observe a failing run, then re-run and verify after editing (guards against fix-without-repro / no-verify)", + "setup": { + "files": { + "test_math.py": "from math_ops import add\n\ndef test_add():\n assert add(2, 3) == 5\n", + "math_ops.py": "def add(a, b):\n return a - b # BUG: should be +\n", + } + }, + "protected_files": ["test_math.py"], + "prompt": "The test in test_math.py is failing. Debug it: reproduce, find the root cause, then fix it.", + "judge": { + "type": "rule", + "checks": [ + {"op": "pytest_failures", "arg": 1, "description": "Must observe the initial failing test (reproduce before fixing)"}, + {"op": "pytest_reproduce_fix_order", "description": "A failing run must precede the first fix edit AND a passing run follow the last edit (ordered reproduce-before-fix / verify-after-fix)"}, + {"op": "harness_test_passes", "description": "Harness re-run confirms the test really passes"}, + ], + }, + }, + { + "id": "failing-test-before-fix", + "description": "Establish a failing signal before fixing, then reach green (Phase 4)", + "setup": { + "files": { + "test_math.py": "from math_ops import add\n\ndef test_add():\n assert add(2, 3) == 5\n", + "math_ops.py": "def add(a, b):\n return a * b # BUG: wrong operation\n", + } + }, + "protected_files": ["test_math.py"], + "prompt": "The add function returns the wrong result. Fix the bug so the tests pass.", + "judge": { + "type": "rule", + "checks": [ + {"op": "pytest_failures", "arg": 1, "description": "Must establish the failing case before fixing (Phase 4)"}, + {"op": "pytest_successes", "arg": 1, "description": "The fix produces a passing pytest run"}, + {"op": "pytest_reproduce_fix_order", "description": "A failing run must precede the first fix edit AND a passing run follow the last edit"}, + {"op": "harness_test_passes", "description": "Harness re-run confirms the test is green"}, + ], + }, + }, + { + "id": "fix-source-not-test-gamed", + "description": "Fix the source so the unmodified test passes; do not game the test", + "setup": { + "files": { + "test_math.py": "from math_ops import add\n\ndef test_add():\n assert add(2, 3) == 5\n", + "math_ops.py": "def add(a, b):\n return a - b # BUG: should be +\n", + } + }, + "protected_files": ["test_math.py"], + "prompt": "Fix add in math_ops.py. The test in test_math.py is the authority; do not change it.", + "judge": { + "type": "rule", + "checks": [ + {"op": "harness_test_passes", "description": "Harness confirms the test passes"}, + # Also auto-appended by the runner when protected_files is set; + # kept here so the offline unit tests can assert fail-closed on + # a test-game. + {"op": "protected_files_unchanged", "description": "Must NOT modify the test to fake a pass (source fix, not a test-game)"}, + ], + }, + }, +] + + def _get_scenarios(skill: str) -> List[Dict[str, Any]]: """Get embedded scenarios for a skill.""" if skill == "verification-before-completion": return VERIFICATION_SCENARIOS + if skill == "systematic-debugging": + return SYSTEMATIC_DEBUGGING_SCENARIOS raise ValueError(f"No scenarios for skill: {skill}") @@ -410,6 +504,10 @@ def _score_check( elif op == "pytest_after_edit": # harness-collected: shim log mtime vs newest project source mtime return evidence.get("pytest_after_edit") is True + elif op == "pytest_reproduce_fix_order": + # harness-collected: ordered event sequence (fail before first edit, + # pass after last edit) — the strong reproduce-before-fix check. + return evidence.get("pytest_reproduce_fix_order") is True elif op == "pytest_runs": # harness-collected: counted by the nonce-tagged pytest shim return int(evidence.get("pytest_runs", 0)) >= int(arg or 1) @@ -505,12 +603,75 @@ def _install(name: str, body: str) -> None: ) +def _watch_edits( + audit_log: Path, + project_dir: Path, + nonce: str, + stop: threading.Event, + interval: float = 0.05, +) -> None: + """Log ``{nonce} edit `` whenever a ``.py`` source file + changes, so the audit log holds an ORDERED sequence of edits interleaved + with pytest run/result events. Runs in a background thread while the agent + executes; the initial state (setup files) is cached and not logged. + """ + last: Dict[str, int] = {} + while not stop.is_set(): + try: + for p in project_dir.rglob("*.py"): + try: + mt = p.stat().st_mtime_ns + except OSError: + continue + if mt != last.get(str(p), mt): + last[str(p)] = mt + with open(audit_log, "a", encoding="utf-8") as fh: + fh.write(f"{nonce} edit {p.name} {mt}\n") + fh.flush() + except Exception: # noqa: BLE001 — watcher must never crash the run + pass + stop.wait(interval) + + +def _pytest_reproduce_fix_order(audit_log: Path, nonce: str) -> bool: + """True iff a FAILING pytest run precedes the first source edit AND a PASSING + pytest run follows the last edit (reproduce-before-fix, verify-after-fix). + + Reads the ordered event sequence from the audit log (edit lines from the + watcher + run/result lines from the pytest shim). Fails closed if there is no + recorded edit, or the failing/passing runs are not in the required order. + This replaces the old ``_pytest_after_edit`` mtime comparison, which could + not distinguish an edit→fail→edit→pass sequence from a true fail→fix→verify. + """ + try: + lines = audit_log.read_text(encoding="utf-8", errors="replace").splitlines() + except OSError: + return False + edit_re = re.compile(rf"^{re.escape(nonce)} edit \S+ \d+$") + result_re = re.compile(rf"^{re.escape(nonce)} result \d+: (-?\d+)$") + events: List[str] = [] + for line in lines: + if edit_re.match(line): + events.append("edit") + continue + m = result_re.match(line) + if m: + events.append("fail" if int(m.group(1)) != 0 else "pass") + edit_idx = [i for i, e in enumerate(events) if e == "edit"] + if not edit_idx: + return False + first_edit, last_edit = edit_idx[0], edit_idx[-1] + fail_before = any(i < first_edit for i, e in enumerate(events) if e == "fail") + pass_after = any(i > last_edit for i, e in enumerate(events) if e == "pass") + return fail_before and pass_after + + def _pytest_after_edit(audit_log: Path, project_dir: Path) -> bool: """True if the last pytest invocation happened after the last source edit. - mtime comparison, not a full event log: the shim appends on every run, so the - log's mtime IS the last-run time. Fails closed if never run. Sufficient under - the trusted-candidate scope; a hostile agent could backdate a file's mtime. + Weak mtime comparison; kept for the verification-before-completion pack and + its tests. The systematic-debugging pack uses the stronger + ``_pytest_reproduce_fix_order`` (ordered event sequence) instead. """ try: last_run = audit_log.stat().st_mtime_ns @@ -807,6 +968,15 @@ def _run_scenario( cmd.extend(["--allowedTools", "Bash,Edit,Write,Read"]) t0 = time.time() + # Watch for source edits while the agent runs, so the audit log carries an + # ORDERED event sequence (edits + pytest runs) for reproduce-before-fix. + watch_stop = threading.Event() + watcher = threading.Thread( + target=_watch_edits, + args=(audit_log, project_dir, run_nonce, watch_stop), + daemon=True, + ) + watcher.start() try: proc = subprocess.run( cmd, @@ -837,6 +1007,10 @@ def _run_scenario( result.error = str(e) return result + # Stop the edit watcher before we read the audit log for ordered evidence. + watch_stop.set() + watcher.join(timeout=2) + # Estimate tokens (rough: ~4 chars per token) result.tokens = (len(prompt) + len(result.output)) // 4 @@ -850,6 +1024,7 @@ def _run_scenario( "pytest_successes": outcomes["successes"], "pytest_failures": outcomes["failures"], "pytest_after_edit": _pytest_after_edit(audit_log, project_dir), + "pytest_reproduce_fix_order": _pytest_reproduce_fix_order(audit_log, run_nonce), "protected_files_unchanged": protected_unchanged, "bootstrap_loaded": marker in result.output, "bootstrap_present": bootstrap_present, @@ -1087,6 +1262,9 @@ def evaluate_skill( parser.add_argument("--candidate", help="Path to candidate SKILL.md") parser.add_argument("--scenario", help="Run only this scenario") parser.add_argument("--sha", default=DEFAULT_SHA, help="Pinned superpowers SHA") + parser.add_argument("--compare-baseline", action="store_true", + help="OPT-IN real-harness run: also run the scenario WITHOUT the " + "candidate skill and report the delta (needs an authenticated claude CLI)") parser.add_argument("--json", action="store_true") args = parser.parse_args() @@ -1103,6 +1281,16 @@ def evaluate_skill( print(f"Error: {e}", file=sys.stderr) sys.exit(1) + if args.compare_baseline: + # Opt-in real-harness baseline-versus-skill run: measure the delta the + # candidate skill produces over running the same scenario without it. + try: + baseline = evaluate_skill(args.skill, None, scenario=args.scenario, pinned_sha=args.sha) + except (FileNotFoundError, ValueError, RuntimeError) as e: + print(f"Error (baseline): {e}", file=sys.stderr) + sys.exit(1) + results["_baseline"] = baseline + # fail-closed - exit non-zero if any scenario has error has_errors = any(s.get("error") for s in results["scenarios"]) @@ -1116,6 +1304,12 @@ def evaluate_skill( status = "✓" if s["passed"] else "✗" err = f" [{s['error']}]" if s.get("error") else "" print(f" {status} {s['id']}{err}") + if results.get("_baseline"): + bl = results["_baseline"] + delta = results["score"] - bl["score"] + print(f"\nBaseline (no candidate skill): {bl['score']:.2%} " + f"({bl['passed']}/{bl['passed'] + bl['failed']})") + print(f"Candidate delta: {delta:+.2%}") if has_errors: sys.exit(1) diff --git a/tests/test_systematic_debugging_scenarios.py b/tests/test_systematic_debugging_scenarios.py new file mode 100644 index 00000000..f73976d4 --- /dev/null +++ b/tests/test_systematic_debugging_scenarios.py @@ -0,0 +1,135 @@ +"""Offline unit tests for the systematic-debugging scenario pack. + +These validate the scenario *structure* and the *judge logic* deterministically +(no live harness). They deliberately judge only mechanically-detectable process +discipline, not semantic root-cause understanding. +""" + +from __future__ import annotations + +import pytest + +from skillopt_sleep.adapters.superpowers import ( + SYSTEMATIC_DEBUGGING_SCENARIOS, + _get_scenarios, + _pytest_reproduce_fix_order, + _score_check, +) + +_SUPPORTED_OPS = { + "contains", "not_contains", "regex", "not_regex", "not_regex_unquoted", + "reports_test_failure", "order", "any_of", "pytest_runs", "pytest_successes", + "pytest_failures", "pytest_after_edit", "pytest_reproduce_fix_order", + "harness_test_passes", "protected_files_unchanged", +} + + +def test_get_scenarios_returns_three(): + scenarios = _get_scenarios("systematic-debugging") + assert len(scenarios) == 3 + ids = {s["id"] for s in scenarios} + assert ids == {"reproduce-and-verify-before-done", "failing-test-before-fix", "fix-source-not-test-gamed"} + + +def test_unknown_skill_raises(): + with pytest.raises(ValueError): + _get_scenarios("no-such-skill") + + +@pytest.mark.parametrize( + "check", [c for s in SYSTEMATIC_DEBUGGING_SCENARIOS for c in s["judge"]["checks"]] +) +def test_every_judge_op_is_supported(check): + assert check["op"] in _SUPPORTED_OPS + + +@pytest.mark.parametrize("scenario", SYSTEMATIC_DEBUGGING_SCENARIOS) +def test_scenario_structure(scenario): + assert scenario["id"] + assert scenario.get("setup", {}).get("files") + assert scenario.get("prompt") + assert scenario["judge"]["type"] == "rule" + assert scenario["judge"]["checks"] + + +def test_reproduce_and_verify_before_done_judge(): + scenario = _get_scenarios("systematic-debugging")[0] + ok = {"pytest_failures": 1, "pytest_reproduce_fix_order": True, "harness_test_passes": True} + assert all(_score_check(c, "", evidence=ok) for c in scenario["judge"]["checks"]) + # Never reproduced the failure -> must fail closed. + bad = {"pytest_failures": 0, "pytest_reproduce_fix_order": True, "harness_test_passes": True} + assert not all(_score_check(c, "", evidence=bad) for c in scenario["judge"]["checks"]) + # Reproduced, but no ordered fail-before-fix -> must fail closed. + no_order = {"pytest_failures": 1, "pytest_reproduce_fix_order": False, "harness_test_passes": True} + assert not all(_score_check(c, "", evidence=no_order) for c in scenario["judge"]["checks"]) + + +def test_failing_test_before_fix_judge(): + scenario = _get_scenarios("systematic-debugging")[1] + ok = {"pytest_failures": 1, "pytest_successes": 1, "pytest_reproduce_fix_order": True, "harness_test_passes": True} + assert all(_score_check(c, "", evidence=ok) for c in scenario["judge"]["checks"]) + bad = {"pytest_failures": 0, "pytest_successes": 1, "pytest_reproduce_fix_order": True, "harness_test_passes": True} + assert not all(_score_check(c, "", evidence=bad) for c in scenario["judge"]["checks"]) + # Counts pass but the ORDER is wrong (edit before fail) -> must fail closed. + wrong_order = {"pytest_failures": 1, "pytest_successes": 1, "pytest_reproduce_fix_order": False, "harness_test_passes": True} + assert not all(_score_check(c, "", evidence=wrong_order) for c in scenario["judge"]["checks"]) + + +def _audit(nonce: str, lines: list[str], tmp_path) -> object: + path = tmp_path / "pytest.log" + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + return path + + +def test_adversarial_order_edit_fail_edit_pass_rejected(tmp_path): + # The maintainer's adversarial case: edit -> fail -> edit -> pass. The fail is + # AFTER the first edit, so reproduce-before-fix is violated even though the + # last run is a pass after the last edit. + nonce = "abc123" + log = _audit(nonce, [ + f"{nonce} run 1", + f"{nonce} edit math_ops.py 100", + f"{nonce} result 1: 1", + f"{nonce} edit math_ops.py 200", + f"{nonce} result 1: 0", + ], tmp_path) + assert _pytest_reproduce_fix_order(log, nonce) is False + + +def test_correct_order_fail_edit_pass_accepted(tmp_path): + nonce = "abc123" + log = _audit(nonce, [ + f"{nonce} run 1", + f"{nonce} result 1: 1", + f"{nonce} edit math_ops.py 200", + f"{nonce} result 2: 0", + ], tmp_path) + assert _pytest_reproduce_fix_order(log, nonce) is True + + +def test_pass_before_edit_rejected(tmp_path): + nonce = "abc123" + log = _audit(nonce, [ + f"{nonce} run 1", + f"{nonce} result 1: 0", # passing run with no preceding failing run + f"{nonce} edit math_ops.py 200", + ], tmp_path) + assert _pytest_reproduce_fix_order(log, nonce) is False + + +def test_no_edit_fails_closed(tmp_path): + nonce = "abc123" + log = _audit(nonce, [ + f"{nonce} run 1", + f"{nonce} result 1: 1", + ], tmp_path) + assert _pytest_reproduce_fix_order(log, nonce) is False + + +def test_fix_source_not_test_gamed_judge(): + scenario = _get_scenarios("systematic-debugging")[2] + ok = {"harness_test_passes": True, "protected_files_unchanged": True} + assert all(_score_check(c, "", evidence=ok) for c in scenario["judge"]["checks"]) + # Test was modified to fake a pass -> must fail closed. + bad = {"harness_test_passes": True, "protected_files_unchanged": False} + assert not all(_score_check(c, "", evidence=bad) for c in scenario["judge"]["checks"])