diff --git a/skillopt_sleep/adapters/superpowers.py b/skillopt_sleep/adapters/superpowers.py index 3ddf6868..729603aa 100644 --- a/skillopt_sleep/adapters/superpowers.py +++ b/skillopt_sleep/adapters/superpowers.py @@ -15,6 +15,23 @@ 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 ] +Run on a POSIX host with an authenticated ``claude`` CLI (see the harness note +below). The ordered reproduce-before-fix sequence is validated here ONLY with +offline fixtures + adversarial-order unit tests; the real-harness run remains to +be executed on such a host. + Usage: from skillopt_sleep.adapters.superpowers import SuperpowersEvaluator @@ -296,10 +313,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 +502,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) @@ -433,7 +529,10 @@ def _score_check( return False -def _write_pytest_shims(bin_dir: Path, audit_log: Path, nonce: str) -> None: +def _write_pytest_shims( + bin_dir: Path, audit_log: Path, nonce: str, project_dir: Path, + source_names: Optional[List[str]] = None, +) -> None: """Install `pytest`/`python` shims that log real invocations, tagged with a per-run nonce the parent generated. @@ -455,6 +554,7 @@ def _write_pytest_shims(bin_dir: Path, audit_log: Path, nonce: str) -> None: f"audit_log={shlex.quote(str(audit_log))}; " f"audit_dir={shlex.quote(str(audit_log.parent))}; " f"real_python={shlex.quote(real_python)}; " + f"project_dir={shlex.quote(str(project_dir))}; " ) def _install(name: str, body: str) -> None: @@ -463,12 +563,19 @@ def _install(name: str, body: str) -> None: path.chmod(path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH) # attempt number = (nonce-tagged start lines so far) + 1, stamped for flaky tests + source_args = " ".join(shlex.quote(n) for n in (source_names or [])) rec = ( f'count=$(grep -c "^{nonce} run " "$audit_log" 2>/dev/null || true); ' 'n=$(( ${count:-0} + 1 )); ' + # Snapshot the source state at THIS test boundary, before running, so the + # judge ties edit evidence to a real test boundary instead of the watcher's + # polling order. Uses the same snippet as _source_fingerprint (run-start/end), + # scoped to the same source files so the hashes are comparable. + f'snap_hash=$("$real_python" -c {shlex.quote(_FINGERPRINT_SNIPPET)} "$project_dir" {source_args}); ' + f'printf "{nonce} run %s: %s\\n" "$n" "$*" >> "$audit_log"; ' + f'printf "{nonce} snap %s\\n" "$snap_hash" >> "$audit_log"; ' f'report="$audit_dir/pytest-{nonce}-$n.xml"; ' f'report_local=".skillopt-pytest-{nonce}-$n-$$.xml"; ' - f'printf "{nonce} run %s: %s\\n" "$n" "$*" >> "$audit_log"; ' 'export SKILLOPT_ATTEMPT="$n"; ' f'export PYTHONPYCACHEPREFIX="$audit_dir/pycache-{nonce}-$n"; ' ) @@ -505,12 +612,88 @@ def _install(name: str, body: str) -> None: ) +def _pytest_reproduce_fix_order(audit_log: Path, nonce: str) -> bool: + """True iff a failing pytest run on the ORIGINAL source is followed by a + passing run on an edited source, and the final on-disk source equals that + last verified state (reproduce-before-fix, verify-after-fix). + + Edit evidence is tied to synchronous source snapshots at each test boundary + (``snap ``), plus a ``start``/``end`` baseline and final reconciliation. + These snapshots are authoritative (a polling mtime watcher was replaced, since + its append order can coalesce or omit edits across a scan interval and so + cannot establish the invariant). Fails closed when the required order, a + fail/pass pair, or the final reconciliation cannot be established. + """ + try: + lines = audit_log.read_text(encoding="utf-8", errors="replace").splitlines() + except OSError: + return False + start_re = re.compile(rf"^{re.escape(nonce)} start ([a-f0-9]+)$") + end_re = re.compile(rf"^{re.escape(nonce)} end ([a-f0-9]+)$") + snap_re = re.compile(rf"^{re.escape(nonce)} snap ([a-f0-9]+)$") + result_re = re.compile(rf"^{re.escape(nonce)} result \d+: (-?\d+)$") + + start_hash = end_hash = None + pending_snap = None + calls: List[tuple] = [] # (snapshot_hash, outcome) per pytest invocation + for line in lines: + m = start_re.match(line) + if m: + start_hash = m.group(1) + continue + m = end_re.match(line) + if m: + end_hash = m.group(1) + continue + m = snap_re.match(line) + if m: + pending_snap = m.group(1) + continue + m = result_re.match(line) + if m: + calls.append((pending_snap, "pass" if int(m.group(1)) == 0 else "fail")) + pending_snap = None + + if start_hash is None or end_hash is None or not calls: + return False + + first_fail = next( + (i for i, (snap, out) in enumerate(calls) if out == "fail"), None + ) + if first_fail is None: + return False + if calls[first_fail][0] != start_hash: + # The failing run was NOT on the original source: source was edited + # before reproduction, so reproduce-before-fix is violated. + return False + + # A passing run after the fail, on a source different from the reproduce one. + verify = [ + (snap, i) + for i, (snap, out) in enumerate(calls) + if i > first_fail and out == "pass" and snap != calls[first_fail][0] + ] + if not verify: + return False + last_verified_snap = verify[-1][0] + + # Final reconciliation: the last verified source must be the on-disk end + # state, i.e. no source edit was made after the final verification. + if last_verified_snap != end_hash: + return False + # Note: each `snap`/`result` is paired by its order in the shared log. If an + # agent ever runs two pytest shims concurrently, their lines can interleave + # and mis-pair; this fails closed (a safe default). Correlate by pid only if + # concurrent pytest runs become a supported path. + return True + + 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 @@ -646,6 +829,57 @@ def _protected_files_unchanged(project_dir: Path, snapshot: Dict[str, str]) -> b return True +# Content fingerprint of the project's *.py sources. The pytest shim and the +# harness both run this exact snippet so the ``snap``/``start``/``end`` lines +# record the same authoritative source state, instead of relying on a polling +# watcher whose append-order can coalesce or omit edits. ``argv[1]`` is the +# project dir to scan; ``argv[2:]`` are the source filenames to hash. When no +# names are given it falls back to every ``**/*.py`` under the dir, so the +# fingerprint can be scoped to the scenario's source files (excluding added +# aux files) while staying backward-compatible for callers that pass none. +_FINGERPRINT_SNIPPET = ( + "import glob, hashlib, os, sys\n" + "os.chdir(sys.argv[1])\n" + "names = sys.argv[2:]\n" + "if not names:\n" + " names = sorted(glob.glob('**/*.py', recursive=True))\n" + "h = hashlib.sha256()\n" + "for n in sorted(names):\n" + " try:\n" + " d = open(n, 'rb').read()\n" + " except OSError:\n" + " d = b''\n" + " h.update(d); h.update(b'\\0')\n" + "print(h.hexdigest())\n" +) + + +def _source_fingerprint(project_dir: Path, names: Optional[List[str]] = None) -> str: + """Stable content hash of ``project_dir``'s ``*.py`` sources. + + When ``names`` (relative filenames) is provided, only those files are + hashed — scoping the fingerprint to the scenario's source modules so that + adding an unrelated auxiliary file does not count as editing the source. + No names falls back to every ``*.py`` under the dir. + + Runs the exact snippet the pytest shim uses, so the run-start / run-end + snapshots are directly comparable to the ``snap`` lines the shim records. + Returns ``""`` if the fingerprint cannot be computed (fail closed upstream). + """ + try: + args = [sys.executable, "-c", _FINGERPRINT_SNIPPET, str(project_dir)] + if names: + args.extend(names) + out = subprocess.run( + args, capture_output=True, text=True, timeout=30, + ) + except (OSError, subprocess.SubprocessError): + return "" + if out.returncode != 0: + return "" + return out.stdout.strip() + + def _run_scenario( scenario: Dict[str, Any], superpowers_dir: Path, @@ -685,7 +919,19 @@ def _run_scenario( audit_log = scenario_home / ".skillopt" / "pytest.log" bin_dir = scenario_home / ".skillopt" / "bin" run_nonce = os.urandom(8).hex() - _write_pytest_shims(bin_dir, audit_log, run_nonce) + # Scope the fingerprint to the scenario's source-under-test: every setup file + # EXCEPT the protected ones (typically the tests). This means adding an + # unrelated auxiliary .py file before reproducing does not count as editing + # the source, while changing the actual code under test still flips the hash. + # Note: a fix that lives entirely in a newly-added module (leaving the original + # source byte-identical) is invisible to this scoped hash and would be rejected. + # The current scenarios cannot trigger that (the test imports the named module); + # widen the scope or include the new file if a future scenario needs it. + setup_files = list(scenario.get("setup", {}).get("files", {}).keys()) + protected_files = list(scenario.get("protected_files", [])) + source_names = [f for f in setup_files if f not in protected_files] + + _write_pytest_shims(bin_dir, audit_log, run_nonce, project_dir, source_names) # Write setup files for filename, content in scenario.get("setup", {}).get("files", {}).items(): @@ -694,6 +940,14 @@ def _run_scenario( project_dir, list(scenario.get("protected_files", [])) ) + # Baseline the source state BEFORE the agent runs. The order judge compares the + # first failing test's snapshot against this to prove reproduce-before-fix (a + # fail whose source already differs from the baseline means the source was + # edited before reproduction). + start_snap = _source_fingerprint(project_dir, source_names) + with open(audit_log, "a", encoding="utf-8") as fh: + fh.write(f"{run_nonce} start {start_snap}\n") + # skill_name becomes a path segment - reject traversal/separators up front if skill_name in ("", ".", "..") or "/" in skill_name or "\\" in skill_name: raise ValueError(f"Invalid skill name: {skill_name!r}") @@ -837,6 +1091,13 @@ def _run_scenario( result.error = str(e) return result + # Reconcile the FINAL source state after the agent exits. The order judge + # requires the last verified snapshot to equal this, so a source edit made + # after the final passing test is still caught instead of silently accepted. + end_snap = _source_fingerprint(project_dir, source_names) + with open(audit_log, "a", encoding="utf-8") as fh: + fh.write(f"{run_nonce} end {end_snap}\n") + # Estimate tokens (rough: ~4 chars per token) result.tokens = (len(prompt) + len(result.output)) // 4 @@ -850,6 +1111,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, diff --git a/tests/test_superpowers_scenarios.py b/tests/test_superpowers_scenarios.py index a724a1c4..a09898fb 100644 --- a/tests/test_superpowers_scenarios.py +++ b/tests/test_superpowers_scenarios.py @@ -274,7 +274,7 @@ def test_shim_counts_real_invocations(self): with tempfile.TemporaryDirectory() as tmpdir: ws = Path(tmpdir) bin_dir, log = ws / "bin", ws / "pytest.log" - _write_pytest_shims(bin_dir, log, "abc123") + _write_pytest_shims(bin_dir, log, "abc123", ws) (ws / "test_ok.py").write_text("def test_ok():\n assert True\n") env = {**os.environ, "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}"} @@ -294,7 +294,7 @@ def test_shim_handles_shell_metacharacters_in_paths(self): ws = Path(tmpdir) / "space $HOME" ws.mkdir() bin_dir, log = ws / "shim bin", ws / "pytest $audit.log" - _write_pytest_shims(bin_dir, log, "abc123") + _write_pytest_shims(bin_dir, log, "abc123", ws) (ws / "test_ok.py").write_text("def test_ok():\n assert True\n") env = {**os.environ, "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}"} @@ -308,7 +308,7 @@ def test_python_shim_matches_module_arguments_not_command_text(self): with tempfile.TemporaryDirectory() as tmpdir: ws = Path(tmpdir) bin_dir, log = ws / "bin", ws / "pytest.log" - _write_pytest_shims(bin_dir, log, "abc123") + _write_pytest_shims(bin_dir, log, "abc123", ws) (ws / "test_ok.py").write_text("def test_ok():\n assert True\n") env = {**os.environ, "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}"} @@ -331,7 +331,7 @@ def test_zero_work_and_skipped_runs_are_not_successes(self): with tempfile.TemporaryDirectory() as tmpdir: ws = Path(tmpdir) bin_dir, log = ws / "bin", ws / "pytest.log" - _write_pytest_shims(bin_dir, log, "abc123") + _write_pytest_shims(bin_dir, log, "abc123", ws) env = {**os.environ, "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}"} version = subprocess.run( @@ -371,7 +371,7 @@ def test_shim_stamps_attempt_number(self): with tempfile.TemporaryDirectory() as tmpdir: ws = Path(tmpdir) bin_dir, log = ws / "bin", ws / "pytest.log" - _write_pytest_shims(bin_dir, log, "abc123") + _write_pytest_shims(bin_dir, log, "abc123", ws) flaky = next(s for s in VERIFICATION_SCENARIOS if s["id"] == "flaky-verify-rerun") (ws / "test_flaky.py").write_text(flaky["setup"]["files"]["test_flaky.py"]) @@ -436,7 +436,7 @@ def test_agent_shim_does_not_reuse_stale_bytecode(self): with tempfile.TemporaryDirectory() as tmpdir: ws = Path(tmpdir) bin_dir, log = ws / "bin", ws / "pytest.log" - _write_pytest_shims(bin_dir, log, "abc123") + _write_pytest_shims(bin_dir, log, "abc123", ws) source = ws / "math_ops.py" source.write_text("def add(a, b):\n return a + b\n") (ws / "test_math.py").write_text( diff --git a/tests/test_systematic_debugging_scenarios.py b/tests/test_systematic_debugging_scenarios.py new file mode 100644 index 00000000..970f1515 --- /dev/null +++ b/tests/test_systematic_debugging_scenarios.py @@ -0,0 +1,235 @@ +"""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, + _FINGERPRINT_SNIPPET, + _get_scenarios, + _pytest_reproduce_fix_order, + _score_check, + _source_fingerprint, +) + +_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 _snap(project, content: str) -> str: + """Write source content and return the real producer fingerprint. + + Every snapshot in these tests comes from the actual ``_source_fingerprint`` + producer on real on-disk edits, so the judge is exercised producer-to-judge + rather than fed hand-authored hashes. + """ + (project / "math_ops.py").write_text(content, encoding="utf-8") + return _source_fingerprint(project) + + +def test_valid_reproduce_edit_pass_accepted(tmp_path): + # Correct discipline: fail on the original source -> edit -> pass on the + # edited source, with no edit after the final verification. + nonce = "abc123" + project = tmp_path / "proj"; project.mkdir() + s0 = _snap(project, "def f():\n return 1\n") + s1 = _snap(project, "def f():\n return 2\n") + log = _audit(nonce, [ + f"{nonce} start {s0}", + f"{nonce} snap {s0}", f"{nonce} result 1: 1", # fail on original + f"{nonce} snap {s1}", f"{nonce} result 2: 0", # pass after fix + f"{nonce} end {s1}", + ], tmp_path) + assert _pytest_reproduce_fix_order(log, nonce) is True + + +def test_adversarial_edit_before_reproduction_rejected(tmp_path): + # Maintainer case 1: edit -> fail -> edit -> pass. The failing run is on an + # ALREADY-EDITED source, so reproduce-before-fix is violated even though the + # last run passes. + nonce = "abc123" + project = tmp_path / "proj"; project.mkdir() + s0 = _snap(project, "def f():\n return 0\n") + sa = _snap(project, "def f():\n return 1\n") # edited before the fail + sb = _snap(project, "def f():\n return 2\n") + log = _audit(nonce, [ + f"{nonce} start {s0}", + f"{nonce} snap {sa}", f"{nonce} result 1: 1", + f"{nonce} snap {sb}", f"{nonce} result 2: 0", + f"{nonce} end {sb}", + ], tmp_path) + assert _pytest_reproduce_fix_order(log, nonce) is False + + +def test_adversarial_final_edit_without_verify_rejected(tmp_path): + # Maintainer case 2: fail -> edit -> pass -> final edit (no test after). The + # final source state was never verified, so verify-after-fix is violated. + nonce = "abc123" + project = tmp_path / "proj"; project.mkdir() + s0 = _snap(project, "def f():\n return 0\n") + s1 = _snap(project, "def f():\n return 1\n") + sc = _snap(project, "def f():\n return 2\n") # made after the pass + log = _audit(nonce, [ + f"{nonce} start {s0}", + f"{nonce} snap {s0}", f"{nonce} result 1: 1", # fail on original + f"{nonce} snap {s1}", f"{nonce} result 2: 0", # pass after fix + f"{nonce} end {sc}", + ], tmp_path) + assert _pytest_reproduce_fix_order(log, nonce) is False + + +def test_aux_file_added_before_reproduce_accepted(tmp_path): + # Regression: creating a NEW auxiliary .py before the failing run must NOT be + # treated as editing the source-under-test. The fingerprint is scoped to the + # scenario's source files (setup minus protected), so an unrelated file is + # invisible while editing the real source still flips the hash. + nonce = "abc123" + project = tmp_path / "proj"; project.mkdir() + names = ["math_ops.py"] + (project / "math_ops.py").write_text("def f():\n return 1\n", encoding="utf-8") + s0 = _source_fingerprint(project, names) + # A new aux file must not change the scoped fingerprint. + (project / "debug_helper.py").write_text("def helper():\n return 1\n", encoding="utf-8") + assert _source_fingerprint(project, names) == s0, "aux .py must be invisible to the scoped fingerprint" + # Editing the real source still changes the hash. + (project / "math_ops.py").write_text("def f():\n return 2\n", encoding="utf-8") + s1 = _source_fingerprint(project, names) + assert s1 != s0 + log = _audit(nonce, [ + f"{nonce} start {s0}", + f"{nonce} snap {s0}", f"{nonce} result 1: 1", # fail on unchanged source + f"{nonce} snap {s1}", f"{nonce} result 2: 0", # pass after fix + f"{nonce} end {s1}", + ], tmp_path) + assert _pytest_reproduce_fix_order(log, nonce) is True + + +def test_pass_before_fail_rejected(tmp_path): + # A passing run with no preceding failing run cannot be a fix. + nonce = "abc123" + project = tmp_path / "proj"; project.mkdir() + s0 = _snap(project, "x = 1\n") + log = _audit(nonce, [ + f"{nonce} start {s0}", + f"{nonce} snap {s0}", f"{nonce} result 1: 0", + f"{nonce} end {s0}", + ], tmp_path) + assert _pytest_reproduce_fix_order(log, nonce) is False + + +def test_no_verify_fails_closed(tmp_path): + # Only a failing run -> no verified fix -> fail closed. + nonce = "abc123" + project = tmp_path / "proj"; project.mkdir() + s0 = _snap(project, "x = 1\n") + log = _audit(nonce, [ + f"{nonce} start {s0}", + f"{nonce} snap {s0}", f"{nonce} result 1: 1", + f"{nonce} end {s0}", + ], tmp_path) + assert _pytest_reproduce_fix_order(log, nonce) is False + + +def test_missing_boundaries_fail_closed(tmp_path): + # No start/end baseline (e.g. early harness error) -> fail closed. + nonce = "abc123" + log = _audit(nonce, [ + f"{nonce} snap aabb", f"{nonce} result 1: 1", + ], tmp_path) + assert _pytest_reproduce_fix_order(log, nonce) is False + + +def test_fingerprint_snippet_matches_source_fingerprint(tmp_path): + # The shim records `snap ` via the same snippet the harness uses for + # start/end, so producer and judge agree on one authoritative hash. + import subprocess + import sys + + import skillopt_sleep.adapters.superpowers as sp + project = tmp_path / "proj"; project.mkdir() + (project / "a.py").write_text("x=1\n", encoding="utf-8") + (project / "b.py").write_text("y=2\n", encoding="utf-8") + via_func = sp._source_fingerprint(project) + via_snippet = subprocess.run( + [sys.executable, "-c", sp._FINGERPRINT_SNIPPET, str(project)], + capture_output=True, text=True, + ).stdout.strip() + assert via_func, "fingerprint must not be empty" + assert via_func == via_snippet, "shim snippet and harness must agree" + + +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"]) + diff --git a/tests/test_systematic_debugging_shim_harness.py b/tests/test_systematic_debugging_shim_harness.py new file mode 100644 index 00000000..145ff37a --- /dev/null +++ b/tests/test_systematic_debugging_shim_harness.py @@ -0,0 +1,111 @@ +"""POSIX-only end-to-end regression: the real bash pytest shim's emitted +``snap``/``result`` lines must be directly consumable by the order judge. + +The offline scenario tests feed hand-authored logs to ``_pytest_reproduce_fix_order``, +so they never drive the actual bash producer (the shim). This test installs the shims, +drives ``pytest`` through them (reproduce -> edit -> verify), and asserts the judge +reaches True on the REAL emitted log. It fails if the shim's ``snap`` hash ever +diverges from the harness ``start``/``end`` snapshot (that divergence is exactly the +producer<->judge contract this guards). Skips on non-POSIX hosts (the shim is bash). +""" +from __future__ import annotations + +import os +import subprocess +from pathlib import Path + +import pytest + +from skillopt_sleep.adapters.superpowers import ( + _pytest_reproduce_fix_order, + _source_fingerprint, + _write_pytest_shims, +) + +posix_only = pytest.mark.skipif(os.name != "posix", reason="test executes POSIX pytest shims") + + +def _run_pytest(project: Path, bin_dir: Path) -> subprocess.CompletedProcess: + env = {**os.environ, "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}"} + return subprocess.run( + ["pytest", "-q", str(project / "test_math.py")], + cwd=project, env=env, capture_output=True, text=True, + ) + + +@posix_only +def test_real_shim_produces_order_the_judge_accepts(tmp_path: Path) -> None: + project = tmp_path / "project"; project.mkdir() + bin_dir = tmp_path / "bin"; bin_dir.mkdir() + audit_log = tmp_path / ".skillopt" / "pytest.log"; audit_log.parent.mkdir() + nonce = "abc123" + # Fingerprint is scoped to the scenario's source-under-test (setup minus protected). + names = ["math_ops.py"] + + (project / "test_math.py").write_text( + "from math_ops import add\n\ndef test_add():\n assert add(2, 3) == 5\n", + encoding="utf-8", + ) + # BUG: should be +. (test_math.py is the protected file, never hashed.) + (project / "math_ops.py").write_text("def add(a, b):\n return a - b\n", encoding="utf-8") + + _write_pytest_shims(bin_dir, audit_log, nonce, project, names) + + start = _source_fingerprint(project, names) + with open(audit_log, "a", encoding="utf-8") as fh: + fh.write(f"{nonce} start {start}\n") + + # 1. Reproduce: failing run on the ORIGINAL source (shim snap must equal start). + r1 = _run_pytest(project, bin_dir) + assert r1.returncode == 1, f"expected reproduce to fail, got {r1.returncode}: {r1.stderr}" + + # 2. Fix the source (the only file the fingerprint hashes). + (project / "math_ops.py").write_text("def add(a, b):\n return a + b\n", encoding="utf-8") + + # 3. Verify: passing run on the edited source. + r2 = _run_pytest(project, bin_dir) + assert r2.returncode == 0, f"expected verify to pass, got {r2.returncode}: {r2.stderr}" + + end = _source_fingerprint(project, names) + with open(audit_log, "a", encoding="utf-8") as fh: + fh.write(f"{nonce} end {end}\n") + + assert _pytest_reproduce_fix_order(audit_log, nonce) is True, ( + "real shim snap/result sequence did not satisfy reproduce-before-fix " + "(producer<->judge contract broken)" + ) + + +@posix_only +def test_real_shim_rejects_edit_before_reproduction(tmp_path: Path) -> None: + """The shim-driven judge must still fail-closed on edit-before-reproduce.""" + project = tmp_path / "project"; project.mkdir() + bin_dir = tmp_path / "bin"; bin_dir.mkdir() + audit_log = tmp_path / ".skillopt" / "pytest.log"; audit_log.parent.mkdir() + nonce = "abc123" + names = ["math_ops.py"] + + (project / "test_math.py").write_text( + "from math_ops import add\n\ndef test_add():\n assert add(2, 3) == 5\n", + encoding="utf-8", + ) + (project / "math_ops.py").write_text("def add(a, b):\n return a - b\n", encoding="utf-8") + _write_pytest_shims(bin_dir, audit_log, nonce, project, names) + + start = _source_fingerprint(project, names) + with open(audit_log, "a", encoding="utf-8") as fh: + fh.write(f"{nonce} start {start}\n") + + env = {**os.environ, "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}"} + + # Edit BEFORE reproducing: the first run is already on a changed source. + (project / "math_ops.py").write_text("def add(a, b):\n return a + b\n", encoding="utf-8") + subprocess.run(["pytest", "-q", str(project / "test_math.py")], cwd=project, env=env, + capture_output=True, text=True) # passes now + + end = _source_fingerprint(project, names) + with open(audit_log, "a", encoding="utf-8") as fh: + fh.write(f"{nonce} end {end}\n") + + # The judge sees only a passing run on the edited source (no fail on `start`). + assert _pytest_reproduce_fix_order(audit_log, nonce) is False