"""Regression tests for benchmark evidence and phase-boundary hardening.""" import hashlib import json import pytest from workflow_bench import runner, runner_artifacts, runner_sessions from workflow_bench.evolution import skill_fingerprint from workflow_bench.process_control import ManagedProcessError, ManagedProcessResult def _report(**overrides) -> str: payload = { "type": "result", "session_id": "s", "num_turns": 3, "total_cost_usd": 0.1, "duration_ms": 1000, "usage": { "input_tokens": 1, "cache_creation_input_tokens": 2, "cache_read_input_tokens": 3, "output_tokens": 4, }, } payload.update(overrides) return json.dumps(payload) def _stream(*, secret: str = "", **report_overrides: object) -> str: events = [] if secret: events.append( { "type": "assistant", "message": {"content": [{"type": "text", "text": secret}]}, } ) events.append(json.loads(_report(**report_overrides))) return "\n".join(json.dumps(event) for event in events) + "\n" def test_sandboxed_verifier_does_not_execute_candidate_login_profile(tmp_path): home = tmp_path / "home" home.mkdir() profile_sentinel = tmp_path / "profile-ran" (home / ".profile").write_text(f"touch '{profile_sentinel}'\nexit 97\n") passed, output = runner_artifacts.run_verify( "printf verified", tmp_path, 5, command_prefix=["/usr/bin/env"], env={"HOME": str(home), "PATH": "/usr/local/bin:/usr/bin:/bin"}, ) assert passed is True assert output.strip() == "verified" assert not profile_sentinel.exists() @pytest.mark.parametrize( "state", [ "input-failure", "timeout", "forced-kill", "ownership-failure", "spawn-failure", "reap-failure", "cleanup-failure", ], ) def test_verifier_infrastructure_states_are_not_candidate_quality(state): process = ManagedProcessResult( state=state, returncode=None, stdout_tail="", stderr_tail="hidden oracle secret", duration_s=0.1, ) result = runner_artifacts.VerificationResult( command=["verify"], process=process, output="hidden oracle secret", ) with pytest.raises(ManagedProcessError) as caught: runner._verification_outcome(result) assert "hidden oracle secret" not in str(caught.value) def test_verifier_normal_nonzero_exit_remains_candidate_quality(): process = ManagedProcessResult( state="exited", returncode=1, stdout_tail="", stderr_tail="assertion failed", duration_s=0.1, ) result = runner_artifacts.VerificationResult( command=["verify"], process=process, output="assertion failed", ) assert runner._verification_outcome(result) == (False, "assertion failed") def test_review_skill_fingerprint_rejects_setup_and_review_phase_replacement(tmp_path): skill = tmp_path / ".claude" / "skills" / "gitnexus-review" / "SKILL.md" skill.parent.mkdir(parents=True) skill.write_text("trusted review prompt") expected = skill_fingerprint(tmp_path, "review") assert expected is not None skill.write_text("replaced during task setup") with pytest.raises(ValueError, match="task setup changed the evaluated skill fingerprint"): runner_artifacts.require_skill_fingerprint(tmp_path, "review", expected, phase="task setup") skill.write_text("trusted review prompt") expected = skill_fingerprint(tmp_path, "review") skill.write_text("replaced during review") with pytest.raises(ValueError, match="review changed the evaluated skill fingerprint"): runner_artifacts.require_skill_fingerprint(tmp_path, "review", expected, phase="review") @pytest.mark.parametrize( ("state", "returncode", "report_overrides"), [ ("exited", 1, {}), ("timeout", None, {}), ("exited", 0, {"is_error": True}), ], ) def test_failed_session_still_persists_redacted_transcript( monkeypatch, tmp_path, state, returncode, report_overrides, ): secret = "sk-ant-postmortem-secret" output = tmp_path / "output" output.mkdir() stream = _stream(secret=secret, **report_overrides) result = ManagedProcessResult( state=state, returncode=returncode, stdout_tail=stream, stderr_tail="primary failure", duration_s=0.1, timed_out=state == "timeout", stdout_capture=stream.encode(), ) monkeypatch.setattr(runner_sessions, "run_managed", lambda *args, **kwargs: result) record = runner_sessions.run_claude( "task", tmp_path, claude_bin="claude", timeout=5, transcript_output_dir=output, transcript_output_prefix="failed-run", transcript_secrets=(secret,), ) artifact = output / record["transcript_artifact"]["path"] assert record["ok"] is False assert record["error_kind"] == "session-error" assert record["error_detail"]["process_state"] == state assert artifact.is_file() assert secret not in artifact.read_text() assert record["transcript_artifact"]["sha256"] == hashlib.sha256(artifact.read_bytes()).hexdigest() def test_failed_session_keeps_primary_error_when_transcript_persistence_fails(monkeypatch, tmp_path): stream = _stream() result = ManagedProcessResult( state="exited", returncode=1, stdout_tail=stream, stderr_tail="primary failure", duration_s=0.1, stdout_capture=stream.encode(), ) monkeypatch.setattr(runner_sessions, "run_managed", lambda *args, **kwargs: result) record = runner_sessions.run_claude( "task", tmp_path, claude_bin="claude", timeout=5, transcript_output_dir=tmp_path / "missing-output-root", ) assert record["error_kind"] == "session-error" assert record["error_detail"]["stderr_tail"] == "primary failure" assert any("event-stream persistence" in item for item in record["evidence_diagnostics"]) def test_timed_out_session_never_trusts_writable_home_without_parent_result(monkeypatch, tmp_path): projects = tmp_path / "projects" output = tmp_path / "output" output.mkdir() def timeout_after_writing_transcript(*args, **kwargs): forged = projects / "some-slug" / "timeout-session.jsonl" forged.parent.mkdir(parents=True) forged.write_text(_stream()) return ManagedProcessResult( state="timeout", returncode=None, stdout_tail="", stderr_tail="timed out", duration_s=5.0, timed_out=True, stdout_capture=b"", ) monkeypatch.setattr(runner_sessions, "run_managed", timeout_after_writing_transcript) record = runner_sessions.run_claude( "task", tmp_path, claude_bin="claude", timeout=5, transcript_projects=projects, transcript_output_dir=output, transcript_output_prefix="timeout-run", ) assert record["error_kind"] == "session-error" assert record["session_id"] is None assert "transcript_artifact" not in record assert record["transcript_missing"] is True def test_phase_workspace_rejects_unchanged_preseeded_review_output(tmp_path): artifact = tmp_path / "review-output.md" artifact.write_text("preseeded output") before = runner_artifacts.workspace_snapshot(tmp_path) with pytest.raises(ValueError, match="did not create or change"): runner_artifacts.enforce_phase_workspace(tmp_path, before, allowed_artifact=artifact) def test_phase_workspace_rejects_symlink_review_output(tmp_path): before = runner_artifacts.workspace_snapshot(tmp_path) outside = tmp_path.parent / f"{tmp_path.name}-outside-review.md" outside.write_text("outside") artifact = tmp_path / "review-output.md" artifact.symlink_to(outside) with pytest.raises(ValueError, match="regular non-symlink"): runner_artifacts.enforce_phase_workspace(tmp_path, before, allowed_artifact=artifact) def test_phase_workspace_accepts_new_regular_review_output(tmp_path): before = runner_artifacts.workspace_snapshot(tmp_path) artifact = tmp_path / "review-output.md" artifact.write_text("new review") runner_artifacts.enforce_phase_workspace(tmp_path, before, allowed_artifact=artifact)