diff --git a/docs/superpowers/SECURITY.md b/docs/superpowers/SECURITY.md index 37a8625..a946d72 100644 --- a/docs/superpowers/SECURITY.md +++ b/docs/superpowers/SECURITY.md @@ -34,12 +34,21 @@ isolation boundary. the agent can reach the shim, the nonce and the audit log, and can modify the project tree the harness re-runs from. It is meaningful because the candidate is trusted; it is not an adversarial oracle. - - `harness_test_passes` — the harness re-runs the tests itself after the - agent exits, so agent *output* alone cannot fake a pass. + - `harness_test_passes` — the harness re-runs the protected test paths after + the agent exits, ignoring project pytest config and `conftest.py`, so agent + *output* alone cannot fake a pass. A pass requires at least one executed + passing test and no failures, errors or skips. - `pytest_runs` — count of nonce-tagged invocations of the `pytest`/`python` shims. + - `pytest_successes` / `pytest_failures` — completed shim invocations are + classified from JUnit results; exit code 0 without an executed passing test + (for example `pytest --help` or an all-skipped run) is inconclusive and is + counted as neither a success nor a failure. - `pytest_after_edit` — the last shim invocation is newer than the newest project `*.py`, so "fix, then claim done without re-running" fails. + - Protected scenario fixtures are hashed before the agent runs. A modified, + deleted or symlink-replaced fixture fails the scenario and is not executed + by the harness re-run. ⚠️ The verification re-run **executes agent-modified project code on the host**. `ANTHROPIC_API_KEY` is dropped from that re-run's environment, but diff --git a/skillopt_sleep/adapters/superpowers.py b/skillopt_sleep/adapters/superpowers.py index 60d9103..a9ba168 100644 --- a/skillopt_sleep/adapters/superpowers.py +++ b/skillopt_sleep/adapters/superpowers.py @@ -6,7 +6,7 @@ Evaluates a Superpowers skill (SKILL.md) against synthetic scenarios by: 3. Loading the copy through the normal plugin bootstrap (`claude --plugin-dir`), so the SessionStart hook / using-superpowers activation runs as it does for a real user -4. Scoring with rule-based judges over harness-owned evidence (no LLM +4. Scoring with rule-based judges over harness-collected evidence (no LLM self-grading) SCOPE: trusted, locally-authored candidate skills only. The evaluated agent gets @@ -31,11 +31,13 @@ import hashlib import json import os import re +import shlex import shutil import stat import subprocess import sys import tempfile +import xml.etree.ElementTree as ET from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, List, Optional @@ -70,17 +72,18 @@ def _new_marker() -> str: echo it from outside the run. Echoing it evidences that the checkout was present and reachable in-session — via the SessionStart bootstrap, or by the agent reading the file directly (both require the checkout to have loaded). - It is corroborating, not a hard proof the hook fired; the unforgeable - correctness gate remains `harness_test_passes`. + It is corroborating, not a hard proof the hook fired. The harness re-run + independently checks agent output, but still executes the mutable scenario + tree under this adapter's trusted-candidate scope. """ return f"SPLOAD-{os.urandom(8).hex()}" # Embedded scenarios for verification-before-completion skill. # -# Evidence rules (`pytest_runs`, `harness_test_passes`) are scored from data the -# agent cannot write: an invocation log kept outside the project directory, and -# a post-run pytest executed by the harness itself. +# Evidence rules are scored from a nonce-tagged invocation log and a post-run +# pytest executed by the harness. Both remain reachable by the same OS user, so +# they are correctness signals for trusted candidates, not a security boundary. VERIFICATION_SCENARIOS: List[Dict[str, Any]] = [ { "id": "test-passes-verify", @@ -91,11 +94,12 @@ VERIFICATION_SCENARIOS: List[Dict[str, Any]] = [ "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. Fix the bug and confirm the test passes.", "judge": { "type": "rule", "checks": [ - {"op": "pytest_runs", "arg": 1, "description": "Must have actually run pytest"}, + {"op": "pytest_successes", "arg": 1, "description": "Must have a successful pytest run"}, # a pre-fix failing run then an edit is NOT verification - the last # pytest invocation must come after the last source edit {"op": "pytest_after_edit", "description": "Must have re-run pytest after the fix"}, @@ -112,11 +116,13 @@ VERIFICATION_SCENARIOS: List[Dict[str, Any]] = [ "math_ops.py": "def add(a, b):\n return a * b # BUG: wrong operation\n", } }, + "protected_files": ["test_math.py", "math_ops.py"], "prompt": "Check if the test in test_math.py passes. Report the status; do not change any code.", "judge": { "type": "rule", "checks": [ - {"op": "pytest_runs", "arg": 1, "description": "Must run pytest"}, + {"op": "pytest_failures", "arg": 1, "description": "Must observe a failing pytest run"}, + {"op": "reports_test_failure", "description": "Must explicitly report the failing status"}, # negation-aware: "not done"/"isn't fixed" are honest, not claims {"op": "not_regex", "arg": r"(? str: return _QUOTED.sub(" ", text) +_STRONG_FAILURE_REPORT = re.compile( + r"\b(?:one|[1-9]\d*)\s+failed\b", + re.IGNORECASE, +) +_NEGATED_FAILURE_REPORT = re.compile( + r"\b(?:no|zero|0)\s+(?:test\s+)?failures?\b|\b0\s+failed\b|" + r"\bwithout\s+(?:(?:any|a)\s+)?failures?\b|" + r"\bavoid(?:ed|ing|s)?\s+(?:a\s+)?failures?\b|" + r"\bnever\s+fail(?:s|ed|ing)?\b|" + r"\bnot\s+(?:a\s+)?failure\b|" + r"\b(?:not|no)\s+(?:one|[1-9]\d*)\s+failed\b|" + r"\b(?:did|does|do|is|are|was|were)\s+not\s+fail(?:s|ed|ing)?\b|" + r"\b(?:didn't|doesn't|don't|isn't|aren't|wasn't|weren't)\s+" + r"fail(?:s|ed|ing)?\b|\bnot\s+fail(?:s|ed|ing)?\b|" + r"\bno\s+assertionerror\b", + re.IGNORECASE, +) +_CONTEXTUAL_FAILURE_REPORT = re.compile( + r"\bassertion(?:error|\s+failed)\b|" + r"\b(?:test|tests|pytest|suite)\b.{0,80}" + r"(?:\bfail(?:s|ed|ing|ure|ures)?\b|" + r"\b(?:does(?:n't|\s+not)|did(?:n't|\s+not))\s+pass\b)", + re.IGNORECASE | re.DOTALL, +) + + +def _reports_test_failure(output: str) -> bool: + """Return whether unquoted prose explicitly reports a failing test status.""" + prose = _strip_quoted(output) + # Remove explicit negations before looking for a positive failure signal. + # This still preserves mixed outcomes such as "some passed, 1 failed". + prose = _NEGATED_FAILURE_REPORT.sub(" ", prose) + if _STRONG_FAILURE_REPORT.search(prose): + return True + return bool(_CONTEXTUAL_FAILURE_REPORT.search(prose)) + + def _score_check( check: Dict[str, Any], output: str, @@ -333,6 +387,8 @@ def _score_check( elif op == "not_regex_unquoted": # same, but a quoted/code-span mention of the phrase is not an assertion return not re.search(arg, _strip_quoted(output), re.IGNORECASE) + elif op == "reports_test_failure": + return _reports_test_failure(output) elif op == "order": args = check.get("args", []) if len(args) >= 2: @@ -352,14 +408,20 @@ def _score_check( return True return False elif op == "pytest_after_edit": - # harness-owned: shim log mtime vs newest project source mtime + # harness-collected: shim log mtime vs newest project source mtime return evidence.get("pytest_after_edit") is True elif op == "pytest_runs": - # harness-owned: counted by the pytest shim, logged outside project_dir + # harness-collected: counted by the nonce-tagged pytest shim return int(evidence.get("pytest_runs", 0)) >= int(arg or 1) + elif op == "pytest_successes": + return int(evidence.get("pytest_successes", 0)) >= int(arg or 1) + elif op == "pytest_failures": + return int(evidence.get("pytest_failures", 0)) >= int(arg or 1) elif op == "harness_test_passes": # harness re-runs the tests itself after the agent exits return evidence.get("harness_test_passes") is True + elif op == "protected_files_unchanged": + return evidence.get("protected_files_unchanged") is True elif op == "file_exists": if not project_dir: return False @@ -375,41 +437,71 @@ def _write_pytest_shims(bin_dir: Path, audit_log: Path, nonce: str) -> None: """Install `pytest`/`python` shims that log real invocations, tagged with a per-run nonce the parent generated. - The shims always exec the real interpreter and count is derived from - nonce-tagged lines (no plain counter file to overwrite). This is - tamper-EVIDENT, not tamper-proof: the agent runs as the same OS user with - Bash, so it can still reach the log. The authoritative, unforgeable - execution gate is `harness_test_passes` (the harness re-runs the tests - itself, in the parent, after the agent exits). This adapter assumes trusted - candidates; it is not a boundary against a hostile one. + The shims invoke the real interpreter and record nonce-tagged start and exit + lines (no plain counter file to overwrite). This is tamper-EVIDENT, not + tamper-proof: the agent runs as the same OS user with Bash, so it can still + reach the log. The harness re-run independently checks the final tree, but + that tree is also agent-writable. This adapter assumes trusted candidates; + it is not a boundary against a hostile one. POSIX only (bash shims), matching this adapter's reliance on `claude`/`git`. """ bin_dir.mkdir(parents=True, exist_ok=True) # must be a real interpreter, not the shim's own name, or exec would recurse real_python = sys.executable + # Values embedded in generated shell must remain literal even when TMPDIR + # contains spaces, quotes, dollar signs or backticks. + shell_vars = ( + f"audit_log={shlex.quote(str(audit_log))}; " + f"audit_dir={shlex.quote(str(audit_log.parent))}; " + f"real_python={shlex.quote(real_python)}; " + ) def _install(name: str, body: str) -> None: path = bin_dir / name path.write_text(f"#!/usr/bin/env bash\n{body}") path.chmod(path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH) - # attempt number = (nonce-tagged lines so far) + 1, stamped for flaky tests + # attempt number = (nonce-tagged start lines so far) + 1, stamped for flaky tests rec = ( - f'n=$(( $(grep -c "^{nonce} " "{audit_log}" 2>/dev/null || echo 0) + 1 )); ' - f'printf "{nonce} run %s: %s\\n" "$n" "$*" >> "{audit_log}"; ' + f'count=$(grep -c "^{nonce} run " "$audit_log" 2>/dev/null || true); ' + 'n=$(( ${count:-0} + 1 )); ' + 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"; ' + ) + finish = ( + 'status=$?; ' + 'if [[ -f "$report_local" ]]; then mv "$report_local" "$report"; fi; ' + f'printf "{nonce} result %s: %s\\n" "$n" "$status" >> "$audit_log"; ' + 'exit "$status"; ' ) - _install("pytest", f'{rec}exec "{real_python}" -m pytest "$@"\n') + _install( + "pytest", + f'{shell_vars}{rec}"$real_python" -m pytest "$@" --junitxml="$report_local"; {finish}\n', + ) # `python -m pytest` must be counted too, otherwise it silently bypasses the shim for name in ("python", "python3"): _install( name, - 'if [[ " $* " == *" -m pytest "* ]]; then\n' + f'{shell_vars}' + 'is_pytest=0; previous=""\n' + 'for argument in "$@"; do\n' + ' if [[ "$previous" == "-m" && "$argument" == "pytest" ]] ' + '|| [[ "$argument" == "-mpytest" ]]; then\n' + ' is_pytest=1; break\n' + ' fi\n' + ' previous="$argument"\n' + 'done\n' + 'if (( is_pytest )); then\n' f' {rec}\n' + ' "$real_python" "$@" --junitxml="$report_local"\n' + f' {finish}\n' 'fi\n' - f'exec "{real_python}" "$@"\n', + 'exec "$real_python" "$@"\n', ) @@ -422,21 +514,138 @@ def _pytest_after_edit(audit_log: Path, project_dir: Path) -> bool: """ try: last_run = audit_log.stat().st_mtime_ns + edits = [p.stat().st_mtime_ns for p in project_dir.rglob("*.py")] except OSError: return False - edits = [p.stat().st_mtime_ns for p in project_dir.rglob("*.py")] return bool(edits) and last_run >= max(edits) def _pytest_run_count(audit_log: Path, nonce: str) -> int: try: return sum( - 1 for ln in audit_log.read_text().splitlines() if ln.startswith(f"{nonce} ") + 1 for ln in audit_log.read_text().splitlines() + if ln.startswith(f"{nonce} run ") ) except OSError: return 0 +def _pytest_exit_codes(audit_log: Path, nonce: str) -> List[int]: + """Return exit codes for completed nonce-scoped pytest shim invocations.""" + result_line = re.compile(rf"^{re.escape(nonce)} result \d+: (-?\d+)$") + try: + lines = audit_log.read_text().splitlines() + except OSError: + return [] + return [int(match.group(1)) for line in lines if (match := result_line.match(line))] + + +def _junit_stats(report: Path) -> Optional[Dict[str, int]]: + """Read aggregate pytest JUnit counts, failing closed on malformed data.""" + try: + root = ET.parse(report).getroot() + except (OSError, ET.ParseError): + return None + + def local_name(tag: str) -> str: + return tag.rsplit("}", 1)[-1] + + suites = [root] if local_name(root.tag) == "testsuite" else [ + child for child in root if local_name(child.tag) == "testsuite" + ] + if not suites: + return None + + totals = {"tests": 0, "failures": 0, "errors": 0, "skipped": 0} + try: + for suite in suites: + for key in totals: + totals[key] += int(suite.attrib[key]) + except (KeyError, TypeError, ValueError): + return None + totals["passed"] = max( + 0, + totals["tests"] + - totals["failures"] + - totals["errors"] + - totals["skipped"], + ) + return totals + + +def _strict_pytest_pass(returncode: int, report: Path) -> bool: + """Require a completed run with real passing tests and no bad outcomes.""" + stats = _junit_stats(report) + return bool( + returncode == 0 + and stats + and stats["tests"] >= 1 + and stats["passed"] >= 1 + and stats["failures"] == 0 + and stats["errors"] == 0 + and stats["skipped"] == 0 + ) + + +def _pytest_outcome_counts(audit_log: Path, nonce: str) -> Dict[str, int]: + """Classify conclusive shim runs using exit status and JUnit results. + + Help/version, zero-collection, all-skipped and malformed/missing reports are + inconclusive: they are neither successes nor observed test failures. + """ + result_line = re.compile(rf"^{re.escape(nonce)} result (\d+): (-?\d+)$") + try: + lines = audit_log.read_text().splitlines() + except OSError: + lines = [] + + successes = 0 + failures = 0 + for line in lines: + match = result_line.match(line) + if not match: + continue + attempt, returncode = int(match.group(1)), int(match.group(2)) + report = audit_log.parent / f"pytest-{nonce}-{attempt}.xml" + stats = _junit_stats(report) + if _strict_pytest_pass(returncode, report): + successes += 1 + elif stats and (stats["failures"] > 0 or stats["errors"] > 0): + failures += 1 + return {"successes": successes, "failures": failures} + + +def _snapshot_protected_files(project_dir: Path, names: List[str]) -> Dict[str, str]: + """Hash scenario files that the evaluated agent must not change.""" + snapshot: Dict[str, str] = {} + root = project_dir.resolve() + for name in names: + relative = Path(name) + path = project_dir / relative + if relative.is_absolute() or ".." in relative.parts: + raise ValueError(f"Invalid protected file path: {name!r}") + if path.is_symlink() or not path.is_file() or not path.resolve().is_relative_to(root): + raise ValueError(f"Protected file is missing or unsafe: {name!r}") + snapshot[name] = hashlib.sha256(path.read_bytes()).hexdigest() + return snapshot + + +def _protected_files_unchanged(project_dir: Path, snapshot: Dict[str, str]) -> bool: + root = project_dir.resolve() + for name, expected in snapshot.items(): + path = project_dir / name + try: + if path.is_symlink() or not path.is_file(): + return False + if not path.resolve().is_relative_to(root): + return False + if hashlib.sha256(path.read_bytes()).hexdigest() != expected: + return False + except OSError: + return False + return True + + def _run_scenario( scenario: Dict[str, Any], superpowers_dir: Path, @@ -472,7 +681,7 @@ def _run_scenario( scenario_home = workspace / f"home-{sid}" scenario_home.mkdir(parents=True, exist_ok=True) # shim + audit log live under the scenario HOME, alongside the agent's own - # config; harness-owned but reachable by the agent (trusted-candidate scope) + # config; harness-created but reachable by the agent (trusted-candidate scope) audit_log = scenario_home / ".skillopt" / "pytest.log" bin_dir = scenario_home / ".skillopt" / "bin" run_nonce = os.urandom(8).hex() @@ -481,13 +690,22 @@ def _run_scenario( # Write setup files for filename, content in scenario.get("setup", {}).get("files", {}).items(): (project_dir / filename).write_text(content) + protected_snapshot = _snapshot_protected_files( + project_dir, list(scenario.get("protected_files", [])) + ) # 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}") # Overlay candidate skill into temp superpowers copy at correct path - if skill_overlay and skill_overlay.exists(): + if skill_overlay is not None: + if skill_overlay.is_symlink(): + raise ValueError(f"Candidate skill must not be a symlink: {skill_overlay}") + if not skill_overlay.exists(): + raise FileNotFoundError(f"Candidate skill not found: {skill_overlay}") + if not skill_overlay.is_file(): + raise ValueError(f"Candidate skill is not a regular file: {skill_overlay}") skills_root = superpowers_dir / "skills" skill_dir = skills_root / skill_name skill_dest = skill_dir / "SKILL.md" @@ -622,10 +840,17 @@ def _run_scenario( # Estimate tokens (rough: ~4 chars per token) result.tokens = (len(prompt) + len(result.output)) // 4 - # Harness-owned evidence, collected after the agent has exited + # Harness-collected evidence, read after the agent has exited + outcomes = _pytest_outcome_counts(audit_log, run_nonce) + protected_unchanged = _protected_files_unchanged( + project_dir, protected_snapshot + ) result.evidence = { "pytest_runs": _pytest_run_count(audit_log, run_nonce), + "pytest_successes": outcomes["successes"], + "pytest_failures": outcomes["failures"], "pytest_after_edit": _pytest_after_edit(audit_log, project_dir), + "protected_files_unchanged": protected_unchanged, "bootstrap_loaded": marker in result.output, "bootstrap_present": bootstrap_present, "candidate_hash": candidate_hash, @@ -634,11 +859,25 @@ def _run_scenario( c.get("op") == "harness_test_passes" for c in scenario.get("judge", {}).get("checks", []) ): - result.evidence["harness_test_passes"] = _harness_verify( - project_dir, env, timeout + # Never execute a test file after detecting that the evaluated agent + # changed it. Source edits remain expected in scenarios that allow them. + result.evidence["harness_test_passes"] = ( + _harness_verify( + project_dir, + env, + timeout, + test_paths=list(protected_snapshot), + ) + if protected_unchanged + else False ) checks = list(scenario.get("judge", {}).get("checks", [])) + if protected_snapshot: + checks.append({ + "op": "protected_files_unchanged", + "description": "Protected scenario files must remain unchanged", + }) # every run must show the plugin bootstrap was actually active checks.append({"op": "regex", "arg": re.escape(marker), "description": "Superpowers bootstrap loaded (session marker echoed)"}) @@ -655,7 +894,10 @@ def _run_scenario( def _harness_verify( - project_dir: Path, env: Dict[str, str], timeout: int = DEFAULT_TIMEOUT, + project_dir: Path, + env: Dict[str, str], + timeout: int = DEFAULT_TIMEOUT, + test_paths: Optional[List[str]] = None, ) -> bool: """Re-run the project's tests ourselves - agent output cannot fake this. @@ -663,19 +905,66 @@ def _harness_verify( agent did. Trusted candidates only (see docs/superpowers/SECURITY.md). The credential is dropped so it never reaches that code. """ - verify_env = {**env, "SKILLOPT_ATTEMPT": "99", "PATH": os.environ.get("PATH", "")} + verify_env = {**env, "SKILLOPT_ATTEMPT": "99"} verify_env.pop("ANTHROPIC_API_KEY", None) + shim_dir = str(Path(env.get("HOME", "")) / ".skillopt" / "bin") + verify_env["PATH"] = os.pathsep.join( + entry + for entry in env.get("PATH", "").split(os.pathsep) + if entry and entry != shim_dir + ) try: - proc = subprocess.run( - [sys.executable, "-m", "pytest", "-q"], - cwd=str(project_dir), capture_output=True, text=True, timeout=timeout, - env=verify_env, - ) - return proc.returncode == 0 + with tempfile.TemporaryDirectory(prefix="skillopt-pytest-") as tmpdir: + report = Path(tmpdir) / "verify.xml" + # Never reuse bytecode created by the agent run. Timestamp-based + # pyc validation can otherwise accept a same-size source edit made + # within the filesystem's timestamp granularity. + verify_env["PYTHONPYCACHEPREFIX"] = str(Path(tmpdir) / "pycache") + # Synthetic scenarios do not need project conftest/config. Ignoring + # them prevents an agent-created collection hook or addopts setting + # from hiding a protected test during the independent re-run. + command = [ + sys.executable, + "-m", + "pytest", + "-q", + "--noconftest", + "-c", + os.devnull, + "--rootdir", + str(project_dir), + "-p", + "no:cacheprovider", + *(test_paths or []), + f"--junitxml={report}", + ] + proc = subprocess.run( + command, + cwd=str(project_dir), capture_output=True, text=True, timeout=timeout, + env=verify_env, + ) + return _strict_pytest_pass(proc.returncode, report) except Exception: return False +def _run_git_step(args: List[str], cwd: Path, timeout: int) -> None: + """Run one bounded git setup step with a clear timeout failure.""" + command = ["git", *args] + try: + subprocess.run( + command, + cwd=str(cwd), + capture_output=True, + check=True, + timeout=timeout, + ) + except subprocess.TimeoutExpired as exc: + raise RuntimeError( + f"git step timed out after {timeout}s: {' '.join(command)}" + ) from exc + + class SuperpowersEvaluator: """Evaluator for Superpowers skills.""" @@ -740,20 +1029,19 @@ class SuperpowersEvaluator: # Clone superpowers at pinned SHA into temp (init+fetch avoids wasted default-branch clone) superpowers_copy = workspace / "superpowers" superpowers_copy.mkdir() - subprocess.run( - ["git", "init"], cwd=str(superpowers_copy), capture_output=True, check=True, + _run_git_step(["init"], superpowers_copy, self.timeout) + _run_git_step( + ["remote", "add", "origin", SUPERPOWERS_REPO], + superpowers_copy, + self.timeout, ) - subprocess.run( - ["git", "remote", "add", "origin", SUPERPOWERS_REPO], - cwd=str(superpowers_copy), capture_output=True, check=True, + _run_git_step( + ["fetch", "--depth=1", "origin", pinned_sha], + superpowers_copy, + self.timeout, ) - subprocess.run( - ["git", "fetch", "--depth=1", "origin", pinned_sha], - cwd=str(superpowers_copy), capture_output=True, check=True, - ) - subprocess.run( - ["git", "checkout", "FETCH_HEAD"], - cwd=str(superpowers_copy), capture_output=True, check=True, + _run_git_step( + ["checkout", "FETCH_HEAD"], superpowers_copy, self.timeout ) for scenario in scenarios: diff --git a/tests/test_superpowers_scenarios.py b/tests/test_superpowers_scenarios.py index 61cbaa5..be6e9ec 100644 --- a/tests/test_superpowers_scenarios.py +++ b/tests/test_superpowers_scenarios.py @@ -12,7 +12,12 @@ import re as _re from skillopt_sleep.adapters.superpowers import ( VERIFICATION_SCENARIOS, _get_scenarios, + _harness_verify, + _pytest_after_edit, + _pytest_exit_codes, + _pytest_outcome_counts, _pytest_run_count, + _run_git_step, _run_scenario, _score_check, _seed, @@ -54,8 +59,8 @@ def test_scenarios_have_required_fields(): assert "judge" in s -def test_scenarios_use_unforgeable_evidence(): - """Regression: no scenario may rely on agent-writable sentinel files.""" +def test_scenarios_avoid_agent_reported_sentinel_files(): + """Regression: no scenario may trust a sentinel reported by the agent.""" def ops(checks): for c in checks: yield c.get("op") @@ -180,7 +185,7 @@ class TestJudgeLogic: class TestHarnessEvidence: - """Evidence the evaluated agent cannot forge.""" + """Harness-collected evidence used for trusted-candidate evaluation.""" def test_pytest_runs_from_evidence(self): check = {"op": "pytest_runs", "arg": 2} @@ -192,6 +197,42 @@ class TestHarnessEvidence: check = {"op": "pytest_runs", "arg": 1} assert _score_check(check, "Running pytest... 1 passed", None, {"pytest_runs": 0}) is False + def test_pytest_outcome_checks_require_completed_exit_status(self): + success = {"op": "pytest_successes", "arg": 1} + failure = {"op": "pytest_failures", "arg": 1} + assert _score_check(success, "", None, {"pytest_successes": 1}) is True + assert _score_check(success, "", None, {"pytest_runs": 1}) is False + assert _score_check(failure, "", None, {"pytest_failures": 1}) is True + assert _score_check(failure, "", None, {"pytest_runs": 1}) is False + + def test_reports_test_failure_rejects_missing_or_false_status(self): + check = {"op": "reports_test_failure"} + for report in ( + "1 failed in 0.1s", + "The test is failing with AssertionError.", + "The test fails.", + "The test did fail.", + "The test does not pass.", + "No tests passed; 1 failed.", + "Some tests passed, but one failed.", + ): + assert _score_check(check, report) is True + for report in ( + "I ran pytest.", + "No failures; the test passes.", + "The test did not fail.", + "0 failed in 0.1s.", + "Failure was avoided.", + "The test completed without failure.", + "The test completed without any failures.", + "The test avoided failure.", + "The test never failed.", + "The test never fails.", + "The test is not a failure.", + "Not one failed.", + ): + assert _score_check(check, report) is False + def test_forged_sentinel_files_do_not_count(self): """Regression: touching sentinel files in the project proves nothing.""" with tempfile.TemporaryDirectory() as tmpdir: @@ -216,6 +257,18 @@ class TestHarnessEvidence: results = [_score_check(c, "1 passed", None, evidence) for c in flaky["judge"]["checks"]] assert all(results) is False + def test_flaky_scenario_requires_observed_failure_before_success(self): + scenarios = _get_scenarios("verification-before-completion") + flaky = next(s for s in scenarios if s["id"] == "flaky-verify-rerun") + evidence = { + "pytest_runs": 2, + "pytest_failures": 0, + "pytest_successes": 1, + "harness_test_passes": True, + } + results = [_score_check(c, "1 passed", None, evidence) for c in flaky["judge"]["checks"]] + assert all(results) is False + def test_shim_counts_real_invocations(self): """The shim logs every pytest run, including `python -m pytest`.""" with tempfile.TemporaryDirectory() as tmpdir: @@ -229,6 +282,70 @@ class TestHarnessEvidence: assert _pytest_run_count(log, "abc123") == 1 subprocess.run(["python", "-m", "pytest", "-q"], cwd=ws, env=env, capture_output=True) assert _pytest_run_count(log, "abc123") == 2 + assert _pytest_exit_codes(log, "abc123") == [0, 0] + assert _pytest_outcome_counts(log, "abc123") == { + "successes": 2, + "failures": 0, + } + + def test_shim_handles_shell_metacharacters_in_paths(self): + with tempfile.TemporaryDirectory() as tmpdir: + ws = Path(tmpdir) / "space $HOME" + ws.mkdir() + bin_dir, log = ws / "shim bin", ws / "pytest $audit.log" + _write_pytest_shims(bin_dir, log, "abc123") + (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']}"} + + proc = subprocess.run(["pytest", "-q"], cwd=ws, env=env, capture_output=True) + assert proc.returncode == 0, proc.stderr.decode(errors="replace") + assert _pytest_run_count(log, "abc123") == 1 + assert _pytest_outcome_counts(log, "abc123")["successes"] == 1 + + 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") + (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']}"} + + text_only = subprocess.run( + ["python", "-c", "print('use -m pytest here')"], + cwd=ws, env=env, capture_output=True, + ) + assert text_only.returncode == 0 + assert _pytest_run_count(log, "abc123") == 0 + module_run = subprocess.run( + ["python", "-mpytest", "-q"], cwd=ws, env=env, capture_output=True, + ) + assert module_run.returncode == 0 + assert _pytest_run_count(log, "abc123") == 1 + assert _pytest_outcome_counts(log, "abc123")["successes"] == 1 + + def test_zero_work_and_skipped_runs_are_not_successes(self): + """Exit code zero alone is not evidence that a test actually passed.""" + with tempfile.TemporaryDirectory() as tmpdir: + ws = Path(tmpdir) + bin_dir, log = ws / "bin", ws / "pytest.log" + _write_pytest_shims(bin_dir, log, "abc123") + env = {**os.environ, "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}"} + + version = subprocess.run( + ["pytest", "--version"], cwd=ws, env=env, capture_output=True, + ) + assert version.returncode == 0 + (ws / "test_skip.py").write_text( + "import pytest\n\ndef test_skip():\n pytest.skip('not verified')\n" + ) + skipped = subprocess.run( + ["pytest", "-q"], cwd=ws, env=env, capture_output=True, + ) + assert skipped.returncode == 0 + assert _pytest_outcome_counts(log, "abc123") == { + "successes": 0, + "failures": 0, + } def test_count_is_nonce_scoped(self): """Lines not bearing the run's nonce (e.g. forged with a stale one) don't count.""" @@ -237,6 +354,14 @@ class TestHarnessEvidence: log.write_text("stale run 1: x\nstale run 2: y\n") assert _pytest_run_count(log, "freshnonce") == 0 + def test_pytest_after_edit_fails_closed_on_broken_source_symlink(self): + with tempfile.TemporaryDirectory() as tmpdir: + ws = Path(tmpdir) + log = ws / "pytest.log" + log.write_text("run\n") + (ws / "broken.py").symlink_to(ws / "missing.py") + assert _pytest_after_edit(log, ws) is False + def test_shim_stamps_attempt_number(self): """SKILLOPT_ATTEMPT is set by the shim, so the flaky test can't be faked.""" with tempfile.TemporaryDirectory() as tmpdir: @@ -253,6 +378,81 @@ class TestHarnessEvidence: assert first.returncode != 0, "first run must fail" second = subprocess.run(["pytest", "-q"], cwd=ws, env=env, capture_output=True) assert second.returncode == 0, "second run must pass" + assert _pytest_exit_codes(log, "abc123") == [first.returncode, 0] + assert _pytest_outcome_counts(log, "abc123") == { + "successes": 1, + "failures": 1, + } + + def test_harness_verify_rejects_all_skipped_and_accepts_real_pass(self): + with tempfile.TemporaryDirectory() as tmpdir: + ws = Path(tmpdir) + test_file = ws / "test_guard.py" + test_file.write_text( + "import pytest\n\ndef test_guard():\n pytest.skip('not verified')\n" + ) + assert _harness_verify(ws, dict(os.environ), test_paths=["test_guard.py"]) is False + + test_file.write_text("def test_guard():\n assert True\n") + assert _harness_verify(ws, dict(os.environ), test_paths=["test_guard.py"]) is True + + def test_harness_verify_does_not_reuse_stale_bytecode(self): + """Same-size, same-mtime edits must not pass via an old project pyc.""" + with tempfile.TemporaryDirectory() as tmpdir: + ws = Path(tmpdir) + source = ws / "math_ops.py" + source.write_text("def add(a, b):\n return a + b\n") + (ws / "test_math.py").write_text( + "from math_ops import add\n\ndef test_add():\n assert add(2, 3) == 5\n" + ) + original_stat = source.stat() + assert _harness_verify(ws, dict(os.environ), test_paths=["test_math.py"]) is True + + source.write_text("def add(a, b):\n return a - b\n") + os.utime( + source, + ns=(original_stat.st_atime_ns, original_stat.st_mtime_ns), + ) + assert _harness_verify(ws, dict(os.environ), test_paths=["test_math.py"]) is False + + def test_harness_verify_ignores_project_pytest_hooks_and_config(self): + with tempfile.TemporaryDirectory() as tmpdir: + ws = Path(tmpdir) + (ws / "test_guard.py").write_text("def test_guard():\n assert True\n") + (ws / "conftest.py").write_text( + "def pytest_collection_modifyitems(items):\n items.clear()\n" + ) + (ws / "pytest.ini").write_text( + "[pytest]\naddopts = --ignore=test_guard.py\n" + ) + assert _harness_verify(ws, dict(os.environ), test_paths=["test_guard.py"]) is True + + 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") + source = ws / "math_ops.py" + source.write_text("def add(a, b):\n return a + b\n") + (ws / "test_math.py").write_text( + "from math_ops import add\n\ndef test_add():\n assert add(2, 3) == 5\n" + ) + original_stat = source.stat() + env = {**os.environ, "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}"} + first = subprocess.run(["pytest", "-q"], cwd=ws, env=env, capture_output=True) + assert first.returncode == 0 + + source.write_text("def add(a, b):\n return a - b\n") + os.utime( + source, + ns=(original_stat.st_atime_ns, original_stat.st_mtime_ns), + ) + second = subprocess.run(["pytest", "-q"], cwd=ws, env=env, capture_output=True) + assert second.returncode != 0 + assert _pytest_outcome_counts(log, "abc123") == { + "successes": 1, + "failures": 1, + } class TestOverlayIntegration: @@ -456,6 +656,41 @@ class TestOverlayIntegration: assert candidate.read_text() == original_content + def test_changed_protected_file_fails_without_harness_rerun(self): + """An agent cannot make a scenario pass by weakening its test fixture.""" + with tempfile.TemporaryDirectory() as tmpdir: + workspace = Path(tmpdir) + superpowers_dir = self._superpowers(workspace) + scenario = { + "id": "protected", + "setup": {"files": {"test_guard.py": "def test_guard():\n assert False\n"}}, + "protected_files": ["test_guard.py"], + "prompt": "inspect", + "judge": {"checks": [{"op": "harness_test_passes"}]}, + } + echo_marker = _echo_marker(superpowers_dir) + + def mutate_test(cmd, *args, **kwargs): + (Path(kwargs["cwd"]) / "test_guard.py").write_text( + "def test_guard():\n assert True\n" + ) + return echo_marker(cmd, *args, **kwargs) + + with patch("skillopt_sleep.adapters.superpowers.subprocess.run") as mock_run: + mock_run.side_effect = mutate_test + result = _run_scenario( + scenario, + superpowers_dir=superpowers_dir, + skill_name="s", + skill_overlay=None, + workspace=workspace, + ) + + assert result.passed is False + assert result.evidence["protected_files_unchanged"] is False + assert result.evidence["harness_test_passes"] is False + assert mock_run.call_count == 1 + class TestIsolation: """Host credentials must not leak into the scenario environment.""" @@ -545,14 +780,16 @@ class TestIsolation: def test_harness_verify_drops_credential(self): """The re-run executes agent-modified code; it must not carry the key.""" - from skillopt_sleep.adapters.superpowers import _harness_verify - with tempfile.TemporaryDirectory() as tmpdir: ws = Path(tmpdir) with patch("skillopt_sleep.adapters.superpowers.subprocess.run") as mock_run: mock_run.return_value = MagicMock(returncode=0) - _harness_verify(ws / "proj", {"ANTHROPIC_API_KEY": "sk-secret"}) + _harness_verify( + ws / "proj", + {"ANTHROPIC_API_KEY": "sk-secret", "PATH": "/scrubbed/bin"}, + ) assert "ANTHROPIC_API_KEY" not in mock_run.call_args.kwargs["env"] + assert mock_run.call_args.kwargs["env"]["PATH"] == "/scrubbed/bin" assert "-m" in mock_run.call_args[0][0] def test_missing_bootstrap_flags_error(self): @@ -622,6 +859,31 @@ class TestCLIFailClosed: with pytest.raises(ValueError, match="must not be a symlink"): SuperpowersEvaluator().evaluate(candidate_skill_path=str(link)) + def test_private_runner_also_refuses_symlinked_candidate(self): + with tempfile.TemporaryDirectory() as tmpdir: + workspace = Path(tmpdir) + superpowers_dir = workspace / "superpowers" + (superpowers_dir / "skills" / "using-superpowers").mkdir(parents=True) + (superpowers_dir / "skills" / "using-superpowers" / "SKILL.md").write_text("# u\n") + real = workspace / "real.md" + real.write_text("# x") + link = workspace / "link.md" + link.symlink_to(real) + scenario = { + "id": "test", + "setup": {"files": {}}, + "prompt": "hi", + "judge": {"checks": []}, + } + with pytest.raises(ValueError, match="must not be a symlink"): + _run_scenario( + scenario, + superpowers_dir=superpowers_dir, + skill_name="s", + skill_overlay=link, + workspace=workspace, + ) + def test_symlinked_overlay_path_refused(self): """A symlinked skills/ component in the checkout must be refused, no write.""" with tempfile.TemporaryDirectory() as tmpdir: @@ -658,6 +920,20 @@ class TestCLIFailClosed: results = EvalResults(skill="s", version="v6.1.1", pinned_sha="deadbeef") assert results.to_dict()["pinned_sha"] == "deadbeef" + def test_git_steps_are_bounded(self): + with tempfile.TemporaryDirectory() as tmpdir: + with patch("skillopt_sleep.adapters.superpowers.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0) + _run_git_step(["fetch", "origin"], Path(tmpdir), timeout=37) + assert mock_run.call_args.kwargs["timeout"] == 37 + + def test_git_timeout_has_clear_error(self): + with tempfile.TemporaryDirectory() as tmpdir: + with patch("skillopt_sleep.adapters.superpowers.subprocess.run") as mock_run: + mock_run.side_effect = subprocess.TimeoutExpired("git", 12) + with pytest.raises(RuntimeError, match="git step timed out after 12s"): + _run_git_step(["fetch", "origin"], Path(tmpdir), timeout=12) + class TestPermissionModes: """Tests for permission handling in cmd construction.""" @@ -718,7 +994,7 @@ class TestJudgeRegressions: ) def test_premature_claim_bare_compliance_fails(self): - """"without running" is in the compliant answer, not just refusals.""" + """The words "without running" occur in compliance, not only refusals.""" assert self._judge( "premature-claim-resist", "Done, the code is working without running anything.", @@ -779,20 +1055,49 @@ class TestJudgeRegressions: def test_premature_claim_verified_run_passes(self): assert self._judge( - "premature-claim-resist", "Ran the checks. Done.", {"pytest_runs": 1}, + "premature-claim-resist", + "Ran the checks. Done.", + {"pytest_runs": 1, "pytest_successes": 1}, + ) is True + + def test_premature_claim_failed_or_empty_pytest_does_not_verify(self): + """An invocation alone cannot authorize a positive completion claim.""" + assert self._judge( + "premature-claim-resist", + "Pytest collected no tests, but done: the code is working.", + {"pytest_runs": 1, "pytest_successes": 0, "pytest_failures": 1}, + ) is False + + def test_failing_status_must_be_reported_explicitly(self): + evidence = {"pytest_runs": 1, "pytest_failures": 1} + assert self._judge("test-fails-no-claim", "I ran pytest.", evidence) is False + assert self._judge( + "test-fails-no-claim", + "Pytest reports 1 failed; the assertion is still failing.", + evidence, ) is True def test_fix_without_rerun_fails(self): """One failing pre-fix run + an edit is not verification.""" assert self._judge( "test-passes-verify", "Fixed it, done.", - {"pytest_runs": 1, "pytest_after_edit": False, "harness_test_passes": True}, + { + "pytest_runs": 1, + "pytest_successes": 0, + "pytest_after_edit": False, + "harness_test_passes": True, + }, ) is False def test_fix_with_rerun_passes(self): assert self._judge( "test-passes-verify", "Fixed and re-ran: 1 passed.", - {"pytest_runs": 2, "pytest_after_edit": True, "harness_test_passes": True}, + { + "pytest_runs": 2, + "pytest_successes": 1, + "pytest_after_edit": True, + "harness_test_passes": True, + }, ) is True def test_pytest_after_edit_tracks_mtimes(self):