mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-10-30 06:42:43 +00:00
252a64573e
Validate continuation evidence, sum per-turn tokens and turns, retain final cumulative cost, and include background wait in elapsed time. Add an offline pinned-CLI canary before paid evaluation and allow 60-minute sessions within the existing deadline and cleanup reserve.
1771 lines
68 KiB
Python
1771 lines
68 KiB
Python
"""Unit tests for workflow benchmark sessions, arms, transcripts, and phase boundaries."""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from contextlib import nullcontext
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from workflow_bench import evolve, runner, runner_artifacts, runner_sessions, runtime_mounts
|
|
from workflow_bench.evolution import skill_fingerprint
|
|
from workflow_bench.process_control import ManagedProcessError, ManagedProcessResult
|
|
from workflow_bench.proposer_sandbox import SandboxError, build_claude_settings
|
|
from workflow_bench.runner import snapshot_plan_docs
|
|
|
|
|
|
def fake_cli_result(
|
|
stdout: str,
|
|
*,
|
|
returncode: int = 0,
|
|
stderr: str = "",
|
|
overflow: bool = False,
|
|
):
|
|
return ManagedProcessResult(
|
|
state="exited",
|
|
returncode=returncode,
|
|
stdout_tail=stdout,
|
|
stderr_tail=stderr,
|
|
duration_s=0.1,
|
|
stdout_capture=stdout.encode(),
|
|
stdout_capture_overflow=overflow,
|
|
)
|
|
|
|
|
|
VALID_REPORT = (
|
|
'{"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}}'
|
|
)
|
|
|
|
|
|
def report_variant(**extra):
|
|
data = json.loads(VALID_REPORT)
|
|
data.update(extra)
|
|
return json.dumps(data)
|
|
|
|
|
|
def session_record(**overrides):
|
|
base = {
|
|
"input_tokens": 10,
|
|
"cache_creation_input_tokens": 1,
|
|
"cache_read_input_tokens": 2,
|
|
"output_tokens": 5,
|
|
"cost_usd": 0.1,
|
|
"duration_s": 1.0,
|
|
"num_turns": 2,
|
|
"ok": True,
|
|
"session_id": "sess",
|
|
"error_kind": None,
|
|
"error_detail": None,
|
|
}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
def bench_args(**overrides):
|
|
base = {
|
|
"claude_bin": "claude",
|
|
"timeout": 5,
|
|
"model": None,
|
|
"effort": "xhigh",
|
|
"base_url": None,
|
|
"auth_token": None,
|
|
"permission_mode": None,
|
|
}
|
|
base.update(overrides)
|
|
return argparse.Namespace(**base)
|
|
|
|
|
|
def event_stream(*events: dict, result_overrides: dict | None = None) -> str:
|
|
result = json.loads(VALID_REPORT)
|
|
result.update(result_overrides or {})
|
|
return "\n".join(json.dumps(event) for event in (*events, result)) + "\n"
|
|
|
|
|
|
def skill_events(skill_input: dict, *, tool_id: str = "skill-1", is_error: bool = False) -> list[dict]:
|
|
return [
|
|
{
|
|
"type": "assistant",
|
|
"message": {
|
|
"content": [
|
|
{
|
|
"type": "tool_use",
|
|
"id": tool_id,
|
|
"name": "Skill",
|
|
"input": skill_input,
|
|
}
|
|
]
|
|
},
|
|
},
|
|
{
|
|
"type": "user",
|
|
"message": {
|
|
"content": [
|
|
{
|
|
"type": "tool_result",
|
|
"tool_use_id": tool_id,
|
|
"is_error": is_error,
|
|
"content": "loaded",
|
|
}
|
|
]
|
|
},
|
|
},
|
|
]
|
|
|
|
|
|
def fake_sandbox(root: Path) -> SimpleNamespace:
|
|
# private_root is NOT the clone. Conflating them puts the review artifact
|
|
# directory inside the workspace, which the real sandbox never does and
|
|
# which hides whether the workspace was left untouched.
|
|
private_root = root.parent / f"{root.name}-sandbox-private"
|
|
private_root.mkdir(exist_ok=True)
|
|
return SimpleNamespace(
|
|
backend="test-double",
|
|
claude_bin="claude",
|
|
clone=root,
|
|
private_root=private_root,
|
|
command_prefix=[],
|
|
command_prefix_for=lambda **_kwargs: [],
|
|
settings_json="{}",
|
|
transcript_projects=root / "transcripts",
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("stdout", "expected_ok"),
|
|
[
|
|
(VALID_REPORT, True),
|
|
("", False), # empty output
|
|
("not json", False), # malformed JSON
|
|
('{"session_id": "s", "num_turns": 3}', False), # missing usage entirely
|
|
('{"usage": {"input_tokens": 1}}', False), # usage missing required fields
|
|
],
|
|
)
|
|
def test_run_claude_fails_closed_on_bad_reports(monkeypatch, tmp_path, stdout, expected_ok):
|
|
monkeypatch.setattr(runner_sessions, "run_managed", lambda *a, **k: fake_cli_result(stdout))
|
|
rec = runner.run_claude("task", tmp_path, claude_bin="claude", timeout=5)
|
|
assert rec["ok"] is expected_ok
|
|
|
|
|
|
def test_parent_event_stream_rejects_non_finite_json_constants():
|
|
with pytest.raises(ValueError, match="malformed parent-captured event JSON"):
|
|
runner_sessions._parse_parent_event_stream(b'{"type":"assistant","score":NaN}\n')
|
|
|
|
|
|
def test_run_claude_forwards_the_named_model_to_every_session(monkeypatch, tmp_path):
|
|
captured: list[str] = []
|
|
|
|
def fake_run(command, **kwargs):
|
|
captured.extend(command)
|
|
return fake_cli_result(VALID_REPORT)
|
|
|
|
monkeypatch.setattr(runner_sessions, "run_managed", fake_run)
|
|
runner.run_claude(
|
|
"task",
|
|
tmp_path,
|
|
claude_bin="claude",
|
|
timeout=5,
|
|
model="claude-sonnet-4-20250514",
|
|
)
|
|
assert captured[captured.index("--model") + 1] == "claude-sonnet-4-20250514"
|
|
|
|
|
|
def test_run_claude_forwards_xhigh_effort_to_every_session(monkeypatch, tmp_path):
|
|
captured: list[str] = []
|
|
|
|
def fake_run(command, **kwargs):
|
|
captured.extend(command)
|
|
return fake_cli_result(VALID_REPORT)
|
|
|
|
monkeypatch.setattr(runner_sessions, "run_managed", fake_run)
|
|
runner.run_claude(
|
|
"task",
|
|
tmp_path,
|
|
claude_bin="claude",
|
|
timeout=5,
|
|
model="gpt-5.6-sol",
|
|
effort="xhigh",
|
|
)
|
|
assert captured[captured.index("--effort") + 1] == "xhigh"
|
|
|
|
|
|
def test_run_claude_restricts_tools_via_tools_flag(monkeypatch, tmp_path):
|
|
# The built-in toolset defaults to everything (subagents, WebFetch, Task,
|
|
# ...) and --allowedTools only pre-approves within that -- it does not
|
|
# narrow it. --tools is what actually restricts the set, so an arm session
|
|
# must pass it or it silently gets a far wider toolset than intended.
|
|
captured: list[str] = []
|
|
|
|
def fake_run(command, **kwargs):
|
|
captured.extend(command)
|
|
return fake_cli_result(VALID_REPORT)
|
|
|
|
monkeypatch.setattr(runner_sessions, "run_managed", fake_run)
|
|
runner.run_claude(
|
|
"task",
|
|
tmp_path,
|
|
claude_bin="claude",
|
|
timeout=5,
|
|
allowed_tools=["Read", "Edit", "Bash", "Skill"],
|
|
)
|
|
tools_idx = captured.index("--tools")
|
|
assert captured[tools_idx + 1 : tools_idx + 5] == ["Read", "Edit", "Bash", "Skill"]
|
|
allowed_idx = captured.index("--allowedTools")
|
|
assert captured[allowed_idx + 1 : allowed_idx + 5] == ["Read", "Edit", "Bash", "Skill"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("proc", "expected_kind"),
|
|
[
|
|
(fake_cli_result(VALID_REPORT), None),
|
|
(fake_cli_result(VALID_REPORT, returncode=1, stderr="boom"), "session-error"),
|
|
(fake_cli_result(report_variant(is_error=True)), "session-error"),
|
|
(fake_cli_result(report_variant(subtype="error_max_turns")), "session-error"),
|
|
(fake_cli_result(""), "session-error"), # malformed report
|
|
],
|
|
)
|
|
def test_run_claude_records_error_kind(monkeypatch, tmp_path, proc, expected_kind):
|
|
monkeypatch.setattr(runner_sessions, "run_managed", lambda *a, **k: proc)
|
|
rec = runner.run_claude("task", tmp_path, claude_bin="claude", timeout=5)
|
|
assert rec["error_kind"] == expected_kind
|
|
|
|
|
|
def test_run_claude_keeps_raw_subtype_and_stderr_tail(monkeypatch, tmp_path):
|
|
proc = fake_cli_result(VALID_REPORT, returncode=1, stderr="rate limit hit")
|
|
monkeypatch.setattr(runner_sessions, "run_managed", lambda *a, **k: proc)
|
|
rec = runner.run_claude("task", tmp_path, claude_bin="claude", timeout=5)
|
|
assert rec["error_detail"] == {
|
|
"subtype": None,
|
|
"returncode": 1,
|
|
"process_state": "exited",
|
|
"stderr_tail": "rate limit hit",
|
|
"stdout_tail": VALID_REPORT,
|
|
"process_detail": None,
|
|
"event_stream_error": None,
|
|
}
|
|
|
|
|
|
def test_run_claude_surfaces_stdout_tail_on_empty_stderr(monkeypatch, tmp_path):
|
|
# A session can exit non-zero with an EMPTY stderr (e.g. a pre-flight
|
|
# sandbox failure before any model turn ever runs) -- stdout_tail is then
|
|
# the only place the actual event stream is visible, so it must not be
|
|
# dropped just because stderr had nothing to say.
|
|
proc = fake_cli_result(VALID_REPORT, returncode=1, stderr="")
|
|
monkeypatch.setattr(runner_sessions, "run_managed", lambda *a, **k: proc)
|
|
rec = runner.run_claude("task", tmp_path, claude_bin="claude", timeout=5)
|
|
assert rec["error_detail"]["stderr_tail"] == ""
|
|
assert rec["error_detail"]["stdout_tail"] == VALID_REPORT
|
|
|
|
|
|
def test_run_arm_labels_completed_but_unverified_runs_verify_failed(monkeypatch, tmp_path):
|
|
monkeypatch.setattr(runner, "run_claude", lambda *a, **k: session_record())
|
|
monkeypatch.setattr(runner, "run_verify", lambda *a, **k: (False, "failed"))
|
|
sandbox = fake_sandbox(tmp_path)
|
|
rec = runner.run_arm(
|
|
"baseline",
|
|
{"prompt": "p", "verify": "exit 1"},
|
|
tmp_path,
|
|
bench_args(),
|
|
sandbox=sandbox,
|
|
)
|
|
assert rec["ok"] is True
|
|
assert rec["resolved"] is False
|
|
assert rec["error_kind"] == "verify-failed"
|
|
|
|
|
|
def test_run_arm_keeps_session_error_kind_over_verify(monkeypatch, tmp_path):
|
|
dead = session_record(ok=False, error_kind="session-error", error_detail={"subtype": "error_max_turns"})
|
|
monkeypatch.setattr(runner, "run_claude", lambda *a, **k: dict(dead))
|
|
monkeypatch.setattr(runner, "run_verify", lambda *a, **k: (True, "ok"))
|
|
sandbox = fake_sandbox(tmp_path)
|
|
rec = runner.run_arm(
|
|
"baseline",
|
|
{"prompt": "p", "verify": "exit 0"},
|
|
tmp_path,
|
|
bench_args(),
|
|
sandbox=sandbox,
|
|
)
|
|
assert rec["ok"] is False
|
|
assert rec["resolved"] is False
|
|
assert rec["error_kind"] == "session-error"
|
|
|
|
|
|
def test_agent_tool_grants_are_exact_and_nomcp_has_no_graph_tools(monkeypatch, tmp_path):
|
|
read_only = runner.allowed_agent_tools(implementation=False)
|
|
review_tools = runner.allowed_agent_tools(implementation=False, allow_edit=False)
|
|
implementation = runner.allowed_agent_tools(implementation=True)
|
|
no_mcp = runner.allowed_agent_tools(implementation=True, include_mcp=False)
|
|
|
|
assert read_only == [*runner.BUILTIN_AGENT_TOOLS, *runner.GITNEXUS_READ_ONLY_TOOLS]
|
|
assert review_tools == [
|
|
tool
|
|
for tool in [*runner.BUILTIN_AGENT_TOOLS, *runner.GITNEXUS_READ_ONLY_TOOLS]
|
|
if tool != "Edit"
|
|
]
|
|
assert "Write" in review_tools
|
|
assert "Edit" not in review_tools
|
|
assert implementation == read_only
|
|
assert not (set(runner.GITNEXUS_MUTATING_TOOLS) & set(implementation))
|
|
assert no_mcp == list(runner.BUILTIN_AGENT_TOOLS)
|
|
assert not any(tool.startswith("mcp__") for tool in no_mcp)
|
|
|
|
captured: list[dict[str, object]] = []
|
|
|
|
def fake_run_claude(*args, **kwargs):
|
|
captured.append(dict(kwargs, prompt=args[0]))
|
|
return session_record()
|
|
|
|
monkeypatch.setattr(runner, "run_claude", fake_run_claude)
|
|
monkeypatch.setattr(runner, "run_verify", lambda *args, **kwargs: (True, "ok"))
|
|
sandbox = fake_sandbox(tmp_path)
|
|
for arm in ("workflow", "review", "workflow_direct", "baseline_nomcp"):
|
|
runner.run_arm(
|
|
arm,
|
|
{"prompt": "p", "verify": "true"},
|
|
tmp_path,
|
|
bench_args(),
|
|
sandbox=sandbox,
|
|
)
|
|
|
|
assert captured[0]["allowed_tools"] == read_only # planning
|
|
assert captured[1]["allowed_tools"] == review_tools # review
|
|
assert captured[2]["allowed_tools"] == implementation
|
|
assert captured[3]["allowed_tools"] == [tool for tool in runner.BUILTIN_AGENT_TOOLS if tool != "Skill"]
|
|
assert captured[3]["mcp_config_json"] == '{"mcpServers":{}}'
|
|
assert captured[3]["disallowed_tools"] == ["Skill", "mcp__gitnexus"]
|
|
assert "GitNexus tools are unavailable" in captured[3]["prompt"]
|
|
|
|
# Both tool-enabled and no-MCP arms retain ordinary CLAUDE.md startup
|
|
# context. The latter explicitly disables all skills/commands instead.
|
|
assert all("bare" not in session for session in captured[:4])
|
|
assert captured[3]["disable_slash_commands"] is True
|
|
|
|
|
|
def test_nomcp_and_tool_arms_load_repository_context_with_the_same_startup_mode(monkeypatch, tmp_path):
|
|
commands = []
|
|
|
|
def capture_cli(command, **kwargs):
|
|
commands.append(command)
|
|
return fake_cli_result(VALID_REPORT)
|
|
|
|
monkeypatch.setattr(runner_sessions, "run_managed", capture_cli)
|
|
monkeypatch.setattr(runner, "run_verify", lambda *_a, **_k: (True, "ok"))
|
|
sandbox = fake_sandbox(tmp_path)
|
|
sandbox.settings_json = build_claude_settings()
|
|
for arm in ("baseline", "baseline_nomcp"):
|
|
runner.run_arm(
|
|
arm, {"prompt": "Edit source and run its tests.", "verify": "true"}, tmp_path, bench_args(), sandbox=sandbox
|
|
)
|
|
|
|
assert len(commands) == 2
|
|
assert all("--bare" not in command for command in commands), "ordinary CLAUDE.md context must load in both arms"
|
|
assert "--disable-slash-commands" not in commands[0]
|
|
assert "--disable-slash-commands" in commands[1]
|
|
assert commands[1][commands[1].index("--mcp-config") + 1] == '{"mcpServers":{}}'
|
|
builtin_end = commands[1].index("--allowedTools")
|
|
baseline_tools = commands[1][commands[1].index("--tools") + 1 : builtin_end]
|
|
assert "Skill" not in baseline_tools
|
|
assert not any(tool.startswith("mcp__") for tool in baseline_tools)
|
|
for command in commands:
|
|
settings = json.loads(command[command.index("--settings") + 1])
|
|
assert settings["disableAllHooks"] is True
|
|
|
|
|
|
def _probe_cell_provisioning(monkeypatch, tmp_path):
|
|
clone = tmp_path / "clone"
|
|
clone.mkdir()
|
|
dependencies = runner.ReadOnlyMount(tmp_path / "dependencies", "/workspace/gitnexus/node_modules")
|
|
runtime = runner.ReadOnlyMount(tmp_path / "runtime", runner.SANDBOX_GITNEXUS)
|
|
registry = runner.ReadOnlyMount(tmp_path / "registry", runner.SANDBOX_GITNEXUS_REGISTRY)
|
|
graph_calls = []
|
|
registry_calls = []
|
|
sandbox_calls = []
|
|
|
|
def materialize(worktree, **_kwargs):
|
|
graph_calls.append(worktree)
|
|
index = worktree / ".gitnexus"
|
|
index.mkdir()
|
|
(index / "lbug").write_text("prebuilt graph")
|
|
|
|
def registry_mount(worktree, *_args):
|
|
registry_calls.append(worktree)
|
|
return registry
|
|
|
|
def probe_sandbox(**kwargs):
|
|
sandbox_calls.append(kwargs)
|
|
raise SandboxError("provisioning probe complete")
|
|
|
|
monkeypatch.setattr(runner, "make_worktree", lambda *_a: clone)
|
|
monkeypatch.setattr(runner, "sanitize_clone_for_hidden_oracles", lambda *_a: "b" * 40)
|
|
monkeypatch.setattr(runner, "stage_task_assets", lambda *_a, **_k: [dependencies])
|
|
monkeypatch.setattr(runner, "isolated_gitnexus_registry_mount", registry_mount)
|
|
monkeypatch.setattr(runner, "prepare_sandbox", probe_sandbox)
|
|
monkeypatch.setattr(runner, "remove_clone", lambda *_a: None)
|
|
out = tmp_path / "out"
|
|
out.mkdir()
|
|
ctx = runner.TaskCellContext(
|
|
task={"id": "fixture", "prompt": "Edit the GitNexus source and run its tests."},
|
|
oracle_snapshot=SimpleNamespace(digest="oracle", command_digest="command", manifest_digest="manifest"),
|
|
repo=tmp_path / "repo",
|
|
task_sha="a" * 40,
|
|
graph_snapshot=SimpleNamespace(digest="graph", manifest_digest="graph-manifest", materialize=materialize),
|
|
graph_snapshot_error=None,
|
|
asset_snapshot=SimpleNamespace(
|
|
digest="assets",
|
|
manifest_digest="assets-manifest",
|
|
dependency_content_digest="deps",
|
|
dependency_manifest_digest="deps-manifest",
|
|
),
|
|
asset_snapshot_error=None,
|
|
args=bench_args(proposer_model=None),
|
|
out_dir=out,
|
|
ce_plugin_snapshot=None,
|
|
trees_dir=tmp_path,
|
|
bwrap_bin=tmp_path / "bwrap",
|
|
runtime_mounts=(runtime,),
|
|
candidate_overlay=None,
|
|
overlay_digest=None,
|
|
)
|
|
return ctx, clone, dependencies, runtime, registry, graph_calls, registry_calls, sandbox_calls
|
|
|
|
|
|
def test_nomcp_cell_does_not_provision_a_graph_registry_or_runtime(monkeypatch, tmp_path):
|
|
ctx, clone, dependencies, _, _, graph_calls, registry_calls, sandbox_calls = _probe_cell_provisioning(
|
|
monkeypatch, tmp_path
|
|
)
|
|
record = runner.run_cell(ctx, 0, "baseline_nomcp")
|
|
|
|
assert not graph_calls, "baseline_nomcp was supplied a prebuilt GitNexus graph"
|
|
assert not (clone / ".gitnexus" / "lbug").exists()
|
|
assert not registry_calls, "baseline_nomcp was supplied a repository registry"
|
|
assert sandbox_calls[0]["read_only_mounts"] == [dependencies]
|
|
assert sandbox_calls[0]["gitnexus_available"] is False
|
|
assert record["sanitized_graph_snapshot_digest"] is None
|
|
|
|
|
|
def test_graph_enabled_cell_retains_its_graph_registry_and_runtime(monkeypatch, tmp_path):
|
|
ctx, clone, dependencies, runtime, registry, graph_calls, registry_calls, sandbox_calls = _probe_cell_provisioning(
|
|
monkeypatch, tmp_path
|
|
)
|
|
runner.run_cell(ctx, 0, "workflow_direct")
|
|
|
|
assert graph_calls == [clone]
|
|
assert registry_calls == [clone]
|
|
assert sandbox_calls[0]["read_only_mounts"] == [dependencies, runtime, registry]
|
|
|
|
|
|
def test_nomcp_cell_does_not_require_an_available_graph(monkeypatch, tmp_path):
|
|
ctx, _, dependencies, _, _, _, _, sandbox_calls = _probe_cell_provisioning(monkeypatch, tmp_path)
|
|
ctx = replace(ctx, graph_snapshot=None, graph_snapshot_error=RuntimeError("graph unavailable"))
|
|
runner.run_cell(ctx, 0, "baseline_nomcp")
|
|
assert sandbox_calls[0]["read_only_mounts"] == [dependencies]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"arm",
|
|
["workflow", "workflow_direct", "candidate_workflow", "candidate_workflow_direct", "review", "baseline_nomcp"],
|
|
)
|
|
def test_cell_seeds_selected_release_skills_before_setup_and_fingerprinting(monkeypatch, tmp_path, arm):
|
|
ctx, clone, *_ = _probe_cell_provisioning(monkeypatch, tmp_path)
|
|
ctx.task["setup"] = "fixture setup"
|
|
release = tmp_path / "release"
|
|
ctx.args.gitnexus_root = release
|
|
execution_arm = runner.CANDIDATE_ARMS.get(arm, arm)
|
|
if arm in runner.CANDIDATE_ARMS:
|
|
ctx = replace(ctx, candidate_overlay=tmp_path / "overlay", overlay_digest="candidate")
|
|
events = []
|
|
sandbox = fake_sandbox(clone)
|
|
sandbox.run = lambda *_a, **_k: events.append("setup") or fake_cli_result("")
|
|
monkeypatch.setattr(runner, "prepare_sandbox", lambda **_k: nullcontext(sandbox))
|
|
monkeypatch.setattr(
|
|
runner,
|
|
"seed_evaluated_skills",
|
|
lambda source, worktree, **kwargs: events.append(("seed", source, worktree, kwargs["arm"])),
|
|
)
|
|
monkeypatch.setattr(runner, "skill_fingerprint", lambda *_a: events.append("fingerprint") or "digest")
|
|
monkeypatch.setattr(runner, "require_skill_fingerprint", lambda *_a, **_k: events.append("check"))
|
|
monkeypatch.setattr(runner, "_sandbox_git", lambda *_a: "c" * 40)
|
|
monkeypatch.setattr(runner, "implementation_diff_digest", lambda *_a, **_k: "")
|
|
monkeypatch.setattr(runner, "apply_candidate_overlay", lambda *_a, **_k: events.append("overlay") or "candidate")
|
|
|
|
def finish_probe(*_a, **_k):
|
|
events.append("session")
|
|
raise SandboxError("session probe complete")
|
|
|
|
monkeypatch.setattr(runner, "run_arm", finish_probe)
|
|
runner.run_cell(ctx, 0, arm)
|
|
expected = ["fingerprint", "setup", "check", "session"]
|
|
if arm != "baseline_nomcp":
|
|
expected.insert(0, ("seed", release, clone, execution_arm))
|
|
if arm in runner.CANDIDATE_ARMS:
|
|
expected[-1:-1] = ["overlay", "fingerprint"]
|
|
assert events == expected
|
|
|
|
|
|
def test_mcp_config_uses_only_the_minimal_pinned_harness_runtime(monkeypatch, tmp_path):
|
|
runtime = tmp_path / "gitnexus"
|
|
shared = tmp_path / "gitnexus-shared"
|
|
for directory in (
|
|
runtime / "dist" / "cli",
|
|
runtime / "node_modules",
|
|
runtime / "vendor",
|
|
runtime / "hooks" / "claude",
|
|
shared / "dist",
|
|
):
|
|
directory.mkdir(parents=True)
|
|
(runtime / "dist" / "cli" / "index.js").write_text("")
|
|
(runtime / "hooks" / "claude" / "resolve-analyze-cmd.cjs").write_text("")
|
|
(runtime / "package.json").write_text(json.dumps({"version": "9.9.9-test"}))
|
|
(runtime / "node_modules" / "gitnexus-shared").symlink_to(shared, target_is_directory=True)
|
|
(shared / "package.json").write_text(json.dumps({"name": "gitnexus-shared"}))
|
|
monkeypatch.setattr(runtime_mounts, "HARNESS_ROOT", tmp_path)
|
|
|
|
config = json.loads(runner.sandbox_mcp_config())
|
|
server = config["mcpServers"]["gitnexus"]
|
|
command_line = [server["command"], *server["args"]]
|
|
|
|
assert runner.SANDBOX_GITNEXUS_ENTRYPOINT in command_line
|
|
assert not any(value.startswith("/workspace/") for value in command_line)
|
|
for key, value in (
|
|
("GITNEXUS_HOME", runner.SANDBOX_GITNEXUS_REGISTRY),
|
|
("GITNEXUS_MCP_ALLOWED_REPOS", "/workspace"),
|
|
("GITNEXUS_MCP_DEFAULT_REPO", "/workspace"),
|
|
):
|
|
index = command_line.index(key)
|
|
assert command_line[index - 1:index + 2] == ["--setenv", key, value]
|
|
mounts = runner.trusted_gitnexus_runtime_mounts()
|
|
assert [(mount.source, mount.target) for mount in mounts] == [
|
|
(runtime / "dist", f"{runner.SANDBOX_GITNEXUS}/dist"),
|
|
(runtime / "package.json", f"{runner.SANDBOX_GITNEXUS}/package.json"),
|
|
(runtime / "node_modules", f"{runner.SANDBOX_GITNEXUS}/node_modules"),
|
|
(runtime / "vendor", f"{runner.SANDBOX_GITNEXUS}/vendor"),
|
|
(shared / "dist", f"{runner.SANDBOX_GITNEXUS_SHARED}/dist"),
|
|
(shared / "package.json", f"{runner.SANDBOX_GITNEXUS_SHARED}/package.json"),
|
|
(runtime / "hooks" / "claude", f"{runner.SANDBOX_GITNEXUS}/hooks/claude"),
|
|
]
|
|
mounted_sources = {mount.source for mount in mounts}
|
|
mounted_targets = {mount.target for mount in mounts}
|
|
assert runtime not in mounted_sources
|
|
assert shared not in mounted_sources
|
|
for forbidden in (".env", ".env.example", ".npmrc", ".git", ".gitnexus", "src", "test", "tests", "skills"):
|
|
assert runtime / forbidden not in mounted_sources
|
|
assert f"{runner.SANDBOX_GITNEXUS}/{forbidden}" not in mounted_targets
|
|
assert shared / forbidden not in mounted_sources
|
|
assert f"{runner.SANDBOX_GITNEXUS_SHARED}/{forbidden}" not in mounted_targets
|
|
|
|
# Only hooks/claude is exposed, not the whole hooks/ directory (which also
|
|
# has an unrelated hooks/antigravity/ tree) and not the runtime root itself.
|
|
assert runtime / "hooks" not in mounted_sources
|
|
assert runtime / "hooks" / "antigravity" not in mounted_sources
|
|
assert f"{runner.SANDBOX_GITNEXUS}/hooks" not in mounted_targets
|
|
|
|
|
|
def _install_pinned_runtime(root: Path) -> None:
|
|
runtime = root / "gitnexus"
|
|
shared = root / "gitnexus-shared"
|
|
for directory in (
|
|
runtime / "dist" / "cli",
|
|
runtime / "node_modules",
|
|
runtime / "vendor",
|
|
runtime / "hooks" / "claude",
|
|
shared / "dist",
|
|
):
|
|
directory.mkdir(parents=True)
|
|
(runtime / "dist" / "cli" / "index.js").write_text("")
|
|
(runtime / "hooks" / "claude" / "resolve-analyze-cmd.cjs").write_text("")
|
|
(runtime / "package.json").write_text(json.dumps({"version": "9.9.9-test"}))
|
|
(runtime / "node_modules" / "gitnexus-shared").symlink_to(shared, target_is_directory=True)
|
|
(shared / "package.json").write_text(json.dumps({"name": "gitnexus-shared"}))
|
|
|
|
|
|
def test_runtime_mounts_use_the_explicit_release_checkout(monkeypatch, tmp_path):
|
|
harness = tmp_path / "harness"
|
|
release = tmp_path / "release"
|
|
_install_pinned_runtime(harness)
|
|
_install_pinned_runtime(release)
|
|
monkeypatch.setattr(runtime_mounts, "HARNESS_ROOT", harness)
|
|
|
|
mounts = runner.trusted_gitnexus_runtime_mounts(root=release)
|
|
assert all(mount.source.is_relative_to(release) for mount in mounts)
|
|
args = runner.build_parser().parse_args(
|
|
["--tasks", "fixture.yaml", "--model", "fixture-model", "--gitnexus-root", str(release)]
|
|
)
|
|
assert args.gitnexus_root == release
|
|
|
|
|
|
@pytest.mark.parametrize("explicit_root", [False, True])
|
|
def test_runtime_mounts_reuse_primary_checkout_node_modules_from_a_worktree(
|
|
monkeypatch, tmp_path, explicit_root
|
|
) -> None:
|
|
primary = tmp_path / "primary"
|
|
worktree = tmp_path / "worktree"
|
|
_install_pinned_runtime(primary)
|
|
(primary / ".git" / "worktrees" / "wt").mkdir(parents=True)
|
|
_install_pinned_runtime(worktree)
|
|
shutil.rmtree(worktree / "gitnexus" / "node_modules")
|
|
(worktree / "gitnexus" / "node_modules").symlink_to(
|
|
primary / "gitnexus" / "node_modules",
|
|
target_is_directory=True,
|
|
)
|
|
(worktree / ".git").write_text(f"gitdir: {primary / '.git' / 'worktrees' / 'wt'}\n")
|
|
monkeypatch.setattr(runtime_mounts, "HARNESS_ROOT", tmp_path / "different-harness" if explicit_root else worktree)
|
|
|
|
mounts = runner.trusted_gitnexus_runtime_mounts(root=worktree if explicit_root else None)
|
|
by_target = {mount.target: mount.source for mount in mounts}
|
|
|
|
assert by_target[f"{runner.SANDBOX_GITNEXUS}/node_modules"] == (
|
|
primary / "gitnexus" / "node_modules"
|
|
)
|
|
assert by_target[f"{runner.SANDBOX_GITNEXUS_SHARED}/package.json"] == (
|
|
primary / "gitnexus-shared" / "package.json"
|
|
)
|
|
assert by_target[f"{runner.SANDBOX_GITNEXUS}/dist"] == worktree / "gitnexus" / "dist"
|
|
|
|
|
|
def test_runtime_mounts_reuse_primary_shared_when_only_the_inner_link_points_there(
|
|
monkeypatch, tmp_path
|
|
) -> None:
|
|
primary = tmp_path / "primary"
|
|
worktree = tmp_path / "worktree"
|
|
_install_pinned_runtime(primary)
|
|
(primary / ".git" / "worktrees" / "wt").mkdir(parents=True)
|
|
_install_pinned_runtime(worktree)
|
|
linked = worktree / "gitnexus" / "node_modules" / "gitnexus-shared"
|
|
linked.unlink()
|
|
linked.symlink_to(primary / "gitnexus-shared", target_is_directory=True)
|
|
(worktree / ".git").write_text(f"gitdir: {primary / '.git' / 'worktrees' / 'wt'}\n")
|
|
monkeypatch.setattr(runtime_mounts, "HARNESS_ROOT", worktree)
|
|
|
|
mounts = runner.trusted_gitnexus_runtime_mounts()
|
|
by_target = {mount.target: mount.source for mount in mounts}
|
|
|
|
assert by_target[f"{runner.SANDBOX_GITNEXUS}/node_modules"] == (
|
|
worktree / "gitnexus" / "node_modules"
|
|
)
|
|
assert by_target[f"{runner.SANDBOX_GITNEXUS_SHARED}/package.json"] == (
|
|
primary / "gitnexus-shared" / "package.json"
|
|
)
|
|
|
|
|
|
def test_runtime_mounts_reject_a_node_modules_symlink_outside_the_primary_checkout(
|
|
monkeypatch, tmp_path
|
|
) -> None:
|
|
primary = tmp_path / "primary"
|
|
worktree = tmp_path / "worktree"
|
|
outsider = tmp_path / "outsider"
|
|
_install_pinned_runtime(primary)
|
|
_install_pinned_runtime(outsider)
|
|
(primary / ".git" / "worktrees" / "wt").mkdir(parents=True)
|
|
_install_pinned_runtime(worktree)
|
|
shutil.rmtree(worktree / "gitnexus" / "node_modules")
|
|
(worktree / "gitnexus" / "node_modules").symlink_to(
|
|
outsider / "gitnexus" / "node_modules",
|
|
target_is_directory=True,
|
|
)
|
|
(worktree / ".git").write_text(f"gitdir: {primary / '.git' / 'worktrees' / 'wt'}\n")
|
|
monkeypatch.setattr(runtime_mounts, "HARNESS_ROOT", worktree)
|
|
|
|
with pytest.raises(SandboxError, match="primary checkout"):
|
|
runner.trusted_gitnexus_runtime_mounts()
|
|
|
|
|
|
def test_runtime_mounts_reject_a_node_modules_symlink_in_a_regular_checkout(
|
|
monkeypatch, tmp_path
|
|
) -> None:
|
|
checkout = tmp_path / "checkout"
|
|
other = tmp_path / "other"
|
|
_install_pinned_runtime(checkout)
|
|
_install_pinned_runtime(other)
|
|
(checkout / ".git").mkdir()
|
|
shutil.rmtree(checkout / "gitnexus" / "node_modules")
|
|
(checkout / "gitnexus" / "node_modules").symlink_to(
|
|
other / "gitnexus" / "node_modules",
|
|
target_is_directory=True,
|
|
)
|
|
monkeypatch.setattr(runtime_mounts, "HARNESS_ROOT", checkout)
|
|
|
|
with pytest.raises(SandboxError, match="must be a real directory"):
|
|
runner.trusted_gitnexus_runtime_mounts()
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
os.environ.get("GITNEXUS_REQUIRE_BWRAP_CANARY") != "1",
|
|
reason="real Bubblewrap canary is mandatory in the named Ubuntu CI job",
|
|
)
|
|
def test_real_bubblewrap_runtime_mount_imports_cli_without_exposing_checkout(tmp_path):
|
|
clone = tmp_path / "clone"
|
|
clone.mkdir()
|
|
mounts = runner.trusted_gitnexus_runtime_mounts()
|
|
|
|
required = [
|
|
runner.SANDBOX_GITNEXUS_ENTRYPOINT,
|
|
f"{runner.SANDBOX_GITNEXUS}/package.json",
|
|
f"{runner.SANDBOX_GITNEXUS}/node_modules",
|
|
f"{runner.SANDBOX_GITNEXUS}/vendor",
|
|
f"{runner.SANDBOX_GITNEXUS_SHARED}/dist/index.js",
|
|
f"{runner.SANDBOX_GITNEXUS_SHARED}/package.json",
|
|
f"{runner.SANDBOX_GITNEXUS}/hooks/claude/resolve-analyze-cmd.cjs",
|
|
]
|
|
forbidden = [
|
|
f"{runner.SANDBOX_GITNEXUS}/{relative}"
|
|
for relative in (".env", ".env.example", ".npmrc", ".git", ".gitnexus", "src", "test", "tests", "skills")
|
|
] + [
|
|
f"{runner.SANDBOX_GITNEXUS_SHARED}/{relative}"
|
|
for relative in (".env", ".env.example", ".npmrc", ".git", ".gitnexus", "src", "test", "tests", "skills")
|
|
]
|
|
visibility_script = (
|
|
"const fs=require('fs');"
|
|
f"for(const p of {json.dumps(required)}) fs.accessSync(p,fs.constants.R_OK);"
|
|
f"for(const p of {json.dumps(forbidden)}) "
|
|
"if(fs.existsSync(p))throw new Error('unexpected checkout path: '+p);"
|
|
)
|
|
|
|
with runner.prepare_sandbox(
|
|
clone=clone,
|
|
claude_bin=Path(sys.executable),
|
|
read_only_mounts=mounts,
|
|
preflight=True,
|
|
) as sandbox:
|
|
visibility = sandbox.run(
|
|
[runner.SANDBOX_NODE, "-e", visibility_script],
|
|
timeout=10,
|
|
)
|
|
imported = sandbox.run(
|
|
[runner.SANDBOX_NODE, runner.SANDBOX_GITNEXUS_ENTRYPOINT, "--version"],
|
|
timeout=10,
|
|
)
|
|
# --version never reaches the `analyze` command, which is loaded via a
|
|
# lazy dynamic import and is the only path that pulls in
|
|
# resolve-invocation.ts's module-load-time require of hooks/claude/
|
|
# resolve-analyze-cmd.cjs. Require the compiled analyze module
|
|
# directly so this canary actually exercises that chain.
|
|
analyze_imported = sandbox.run(
|
|
[runner.SANDBOX_NODE, "-e", f"require('{runner.SANDBOX_GITNEXUS}/dist/cli/analyze.js')"],
|
|
timeout=10,
|
|
)
|
|
|
|
assert visibility.ok, visibility.stderr_tail
|
|
assert imported.ok, imported.stderr_tail
|
|
assert analyze_imported.ok, analyze_imported.stderr_tail
|
|
# The runtime the sandbox sees must be the one this checkout built —
|
|
# compared against the checkout itself rather than a constant, so a release
|
|
# bump cannot fail a benchmark that is running exactly what it should.
|
|
built = json.loads((runtime_mounts.HARNESS_ROOT / "gitnexus" / "package.json").read_text())
|
|
assert imported.stdout_tail.strip() == built["version"]
|
|
|
|
|
|
def test_isolated_mcp_registry_contains_only_the_sandbox_clone(tmp_path):
|
|
worktree = tmp_path / "clone"
|
|
metadata = worktree / ".gitnexus" / "gitnexus.json"
|
|
metadata.parent.mkdir(parents=True)
|
|
metadata.write_text(
|
|
json.dumps(
|
|
{
|
|
"indexedAt": "2026-07-18T00:00:00Z",
|
|
"lastCommit": "a" * 40,
|
|
"stats": {"files": 1},
|
|
}
|
|
)
|
|
)
|
|
|
|
mount = runner.isolated_gitnexus_registry_mount(worktree, tmp_path)
|
|
registry_file = mount.source / "registry.json"
|
|
registry = json.loads(registry_file.read_text())
|
|
|
|
assert mount.target == runner.SANDBOX_GITNEXUS_REGISTRY
|
|
assert mount.source.stat().st_mode & 0o777 == 0o700
|
|
assert registry_file.stat().st_mode & 0o777 == 0o600
|
|
assert registry == [
|
|
{
|
|
"indexedAt": "2026-07-18T00:00:00Z",
|
|
"lastCommit": "a" * 40,
|
|
"name": "benchmark-target",
|
|
"path": "/workspace",
|
|
"stats": {"files": 1},
|
|
"storagePath": "/workspace/.gitnexus",
|
|
}
|
|
]
|
|
|
|
|
|
def test_resolved_implementation_without_repository_work_fails_closed():
|
|
rec = {"resolved": True, "error_kind": None, "error_detail": None}
|
|
runner.enforce_work_evidence(
|
|
rec,
|
|
arm="workflow_direct",
|
|
before_digest="same",
|
|
after_digest="same",
|
|
)
|
|
assert rec["resolved"] is False
|
|
assert rec["error_kind"] == "no-work-produced"
|
|
|
|
|
|
@pytest.mark.skipif(os.name == "nt", reason="sandbox patch streaming uses POSIX executable paths")
|
|
def test_capture_patch_materializes_only_the_bounded_prefix(tmp_path, monkeypatch):
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
subprocess.run(["git", "init", "--quiet", str(repo)], check=True)
|
|
changed = repo / "changed.txt"
|
|
changed.write_text("before\n")
|
|
subprocess.run(["git", "-C", str(repo), "add", "changed.txt"], check=True)
|
|
subprocess.run(
|
|
[
|
|
"git",
|
|
"-C",
|
|
str(repo),
|
|
"-c",
|
|
"user.name=test",
|
|
"-c",
|
|
"user.email=test@invalid",
|
|
"commit",
|
|
"--quiet",
|
|
"-m",
|
|
"base",
|
|
],
|
|
check=True,
|
|
)
|
|
orig_sha = subprocess.run(
|
|
["git", "-C", str(repo), "rev-parse", "HEAD"],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
).stdout.strip()
|
|
changed.write_text("changed line\n" * 100_000)
|
|
|
|
class LocalSandbox:
|
|
synthetic_guidance_paths = ()
|
|
|
|
def run(self, command, **kwargs):
|
|
translated = [
|
|
str(repo) + item.removeprefix("/workspace") if item.startswith("/workspace/") else item
|
|
for item in command
|
|
]
|
|
completed = subprocess.run(
|
|
translated,
|
|
cwd=repo,
|
|
env=dict(kwargs["env"]),
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
return ManagedProcessResult(
|
|
state="exited",
|
|
returncode=completed.returncode,
|
|
stdout_tail=completed.stdout,
|
|
stderr_tail=completed.stderr,
|
|
duration_s=0.0,
|
|
)
|
|
|
|
read_bytes = runner_artifacts._bounded_regular_bytes
|
|
materialized_sizes = []
|
|
|
|
def inspect_bounded_sink(path, *, limit):
|
|
materialized_sizes.append(path.stat().st_size)
|
|
return read_bytes(path, limit=limit)
|
|
|
|
monkeypatch.setattr(runner_artifacts, "_bounded_regular_bytes", inspect_bounded_sink)
|
|
patch = runner.capture_patch(LocalSandbox(), repo, orig_sha)
|
|
|
|
assert len(patch) == runner.MAX_PATCH_BYTES
|
|
assert materialized_sizes == [runner.MAX_PATCH_BYTES]
|
|
assert not list(repo.glob(".wfbench-artifact-*"))
|
|
|
|
|
|
def test_workflow_never_starts_work_after_failed_or_invalid_planning(monkeypatch, tmp_path):
|
|
sandbox = fake_sandbox(tmp_path)
|
|
calls: list[str] = []
|
|
|
|
def failed_plan(prompt, *args, **kwargs):
|
|
calls.append(prompt)
|
|
return session_record(ok=False, error_kind="session-error", error_detail="planning failed")
|
|
|
|
monkeypatch.setattr(runner, "run_claude", failed_plan)
|
|
monkeypatch.setattr(runner, "run_verify", lambda *a, **k: (True, "ok"))
|
|
failed = runner.run_arm(
|
|
"workflow",
|
|
{"prompt": "p", "verify": "exit 0"},
|
|
tmp_path,
|
|
bench_args(),
|
|
sandbox=sandbox,
|
|
)
|
|
assert len(calls) == 1
|
|
assert failed["resolved"] is False
|
|
assert failed["plan_produced"] is False
|
|
|
|
calls.clear()
|
|
monkeypatch.setattr(
|
|
runner,
|
|
"run_claude",
|
|
lambda prompt, *args, **kwargs: calls.append(prompt) or session_record(),
|
|
)
|
|
invalid = runner.run_arm(
|
|
"workflow",
|
|
{"prompt": "p", "verify": "exit 0"},
|
|
tmp_path,
|
|
bench_args(),
|
|
sandbox=sandbox,
|
|
)
|
|
assert len(calls) == 1
|
|
assert invalid["resolved"] is False
|
|
assert invalid["error_kind"] == "plan-evidence-invalid"
|
|
|
|
|
|
def test_skill_invocation_is_detected_from_parent_event_stream(monkeypatch, tmp_path):
|
|
stream = event_stream(*skill_events({"command": "gitnexus-work"}))
|
|
monkeypatch.setattr(runner_sessions, "run_managed", lambda *a, **k: fake_cli_result(stream))
|
|
|
|
rec = runner.run_claude(
|
|
"task",
|
|
tmp_path,
|
|
claude_bin="claude",
|
|
timeout=5,
|
|
expected_skill="gitnexus-work",
|
|
)
|
|
|
|
assert rec["ok"] is True
|
|
assert rec["skill_invoked"] is True
|
|
assert rec["error_kind"] is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"skill_input",
|
|
[
|
|
{"skill": "gitnexus-work"},
|
|
{"command": "/gitnexus-work execute the plan"},
|
|
{"name": "gitnexus-work direct-mode"},
|
|
],
|
|
)
|
|
def test_skill_invocation_parses_supported_exact_identifier_fields(skill_input):
|
|
assert (
|
|
runner_sessions.skill_was_invoked_events(
|
|
skill_events(skill_input),
|
|
"gitnexus-work",
|
|
)
|
|
is True
|
|
)
|
|
|
|
|
|
def test_skill_invocation_accepts_plugin_qualified_identifier():
|
|
assert (
|
|
runner_sessions.skill_was_invoked_events(
|
|
skill_events({"skill": "compound-engineering:ce-code-review"}),
|
|
"ce-code-review",
|
|
)
|
|
is True
|
|
)
|
|
assert (
|
|
runner_sessions.skill_was_invoked_events(
|
|
skill_events({"skill": "compound-engineering:ce-plan"}),
|
|
"ce-code-review",
|
|
)
|
|
is False
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"skill_input",
|
|
[
|
|
{"skill": "gitnexus-work-extra"},
|
|
{"skill": "prefix-gitnexus-work"},
|
|
{"command": "/gitnexus-work-extra execute"},
|
|
{"command": "other-skill", "args": "please use gitnexus-work"},
|
|
{"name": "other-skill", "description": "gitnexus-work"},
|
|
{"args": "gitnexus-work"},
|
|
],
|
|
)
|
|
def test_skill_invocation_rejects_prefix_suffix_and_argument_mentions(skill_input):
|
|
assert (
|
|
runner_sessions.skill_was_invoked_events(
|
|
skill_events(skill_input),
|
|
"gitnexus-work",
|
|
)
|
|
is False
|
|
)
|
|
|
|
|
|
def test_skill_invocation_scans_through_eof_and_rejects_later_malformed_json(monkeypatch, tmp_path):
|
|
stream = event_stream(*skill_events({"skill": "gitnexus-work"})) + '{"truncated":'
|
|
monkeypatch.setattr(runner_sessions, "run_managed", lambda *a, **k: fake_cli_result(stream))
|
|
|
|
rec = runner.run_claude(
|
|
"task",
|
|
tmp_path,
|
|
claude_bin="claude",
|
|
timeout=5,
|
|
expected_skill="gitnexus-work",
|
|
)
|
|
|
|
assert rec["ok"] is False
|
|
assert rec["error_kind"] == "session-error"
|
|
assert "malformed parent-captured event JSON" in rec["error_detail"]["event_stream_error"]
|
|
|
|
|
|
def test_matching_skill_requires_one_later_successful_result(monkeypatch, tmp_path):
|
|
failed_stream = event_stream(*skill_events({"skill": "gitnexus-work"}, is_error=True))
|
|
monkeypatch.setattr(runner_sessions, "run_managed", lambda *a, **k: fake_cli_result(failed_stream))
|
|
failed = runner.run_claude(
|
|
"task",
|
|
tmp_path,
|
|
claude_bin="claude",
|
|
timeout=5,
|
|
expected_skill="gitnexus-work",
|
|
)
|
|
assert failed["skill_invoked"] is False
|
|
assert failed["error_kind"] == "skill-not-invoked"
|
|
|
|
missing_result = event_stream(skill_events({"skill": "gitnexus-work"})[0])
|
|
monkeypatch.setattr(runner_sessions, "run_managed", lambda *a, **k: fake_cli_result(missing_result))
|
|
missing = runner.run_claude(
|
|
"task",
|
|
tmp_path,
|
|
claude_bin="claude",
|
|
timeout=5,
|
|
expected_skill="gitnexus-work",
|
|
)
|
|
assert missing["skill_invoked"] is None
|
|
assert missing["error_kind"] == "evidence-unverified"
|
|
assert "no tool result" in missing["error_detail"]
|
|
|
|
|
|
def test_skill_evidence_rejects_duplicate_tool_use_and_result_ids():
|
|
request, result = skill_events({"skill": "gitnexus-work"})
|
|
duplicate_request = json.loads(json.dumps(request))
|
|
with pytest.raises(ValueError, match="duplicate tool-use id"):
|
|
runner_sessions.skill_was_invoked_events(
|
|
[request, duplicate_request, result],
|
|
"gitnexus-work",
|
|
)
|
|
|
|
error_result = json.loads(json.dumps(result))
|
|
error_result["message"]["content"][0]["is_error"] = True
|
|
with pytest.raises(ValueError, match="duplicate tool result"):
|
|
runner_sessions.skill_was_invoked_events(
|
|
[request, result, error_result],
|
|
"gitnexus-work",
|
|
)
|
|
|
|
|
|
def test_agent_writable_home_transcript_cannot_forge_skill_evidence(monkeypatch, tmp_path):
|
|
forged = tmp_path / ".claude" / "projects" / "forged" / "s.jsonl"
|
|
forged.parent.mkdir(parents=True)
|
|
forged.write_text(event_stream(*skill_events({"skill": "gitnexus-work"})))
|
|
monkeypatch.setattr(runner_sessions, "run_managed", lambda *a, **k: fake_cli_result(VALID_REPORT))
|
|
|
|
rec = runner.run_claude(
|
|
"task",
|
|
tmp_path,
|
|
claude_bin="claude",
|
|
timeout=5,
|
|
expected_skill="gitnexus-work",
|
|
transcript_projects=tmp_path / ".claude" / "projects",
|
|
)
|
|
|
|
assert rec["skill_invoked"] is False
|
|
assert rec["error_kind"] == "skill-not-invoked"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"arm",
|
|
["workflow", "workflow_direct", "ce_workflow", "ce_workflow_direct"],
|
|
)
|
|
def test_every_skill_implementation_session_rechecks_fingerprint_immediately(
|
|
monkeypatch,
|
|
tmp_path,
|
|
arm,
|
|
):
|
|
plan = tmp_path / "docs" / "plans" / "plan.md"
|
|
calls: list[tuple[str, str]] = []
|
|
|
|
def fake_run(prompt, *args, **kwargs):
|
|
if "deliverable" in prompt:
|
|
plan.parent.mkdir(parents=True, exist_ok=True)
|
|
plan.write_text("plan")
|
|
return session_record()
|
|
|
|
def fingerprint(*args, **kwargs):
|
|
calls.append((args[1], kwargs["phase"]))
|
|
raise ValueError("implementation changed the evaluated skill fingerprint")
|
|
|
|
monkeypatch.setattr(runner, "run_claude", fake_run)
|
|
monkeypatch.setattr(runner, "require_skill_fingerprint", fingerprint)
|
|
monkeypatch.setattr(runner, "run_verify", lambda *args, **kwargs: (True, "ok"))
|
|
rec = runner.run_arm(
|
|
arm,
|
|
{"prompt": "p", "verify": "true"},
|
|
tmp_path,
|
|
bench_args(),
|
|
sandbox=fake_sandbox(tmp_path),
|
|
expected_skill_digest="trusted",
|
|
)
|
|
|
|
assert calls == [(arm, "implementation")]
|
|
assert rec["ok"] is False
|
|
assert rec["resolved"] is False
|
|
assert rec["error_kind"] == "implementation-evidence-invalid"
|
|
|
|
|
|
def test_direct_implementation_skill_mutation_fails_before_verification(monkeypatch, tmp_path):
|
|
skill = tmp_path / ".claude" / "skills" / "gitnexus-work" / "SKILL.md"
|
|
skill.parent.mkdir(parents=True)
|
|
skill.write_text("trusted prompt")
|
|
expected = skill_fingerprint(tmp_path, "workflow_direct")
|
|
order: list[str] = []
|
|
real_fingerprint_check = runner.require_skill_fingerprint
|
|
|
|
def mutating_session(*args, **kwargs):
|
|
order.append("session")
|
|
skill.write_text("model replaced prompt")
|
|
return session_record()
|
|
|
|
def tracked_fingerprint(*args, **kwargs):
|
|
order.append("fingerprint")
|
|
return real_fingerprint_check(*args, **kwargs)
|
|
|
|
def verify(*args, **kwargs):
|
|
order.append("verify")
|
|
return True, "ok"
|
|
|
|
monkeypatch.setattr(runner, "run_claude", mutating_session)
|
|
monkeypatch.setattr(runner, "require_skill_fingerprint", tracked_fingerprint)
|
|
monkeypatch.setattr(runner, "run_verify", verify)
|
|
rec = runner.run_arm(
|
|
"workflow_direct",
|
|
{"prompt": "p", "verify": "true"},
|
|
tmp_path,
|
|
bench_args(),
|
|
sandbox=fake_sandbox(tmp_path),
|
|
expected_skill_digest=expected,
|
|
)
|
|
|
|
assert order == ["session", "fingerprint", "verify"]
|
|
assert rec["ok"] is False
|
|
assert rec["resolved"] is False
|
|
assert rec["error_kind"] == "implementation-evidence-invalid"
|
|
|
|
|
|
def test_parent_event_stream_is_persisted_private_redacted_and_digest_bound(monkeypatch, tmp_path):
|
|
secret = "sk-ant-transcript-secret"
|
|
bearer = "Authorization: Bearer bearer-postmortem-secret"
|
|
structural_secret = "token-in-authorization-value"
|
|
password = "password-in-structured-field"
|
|
events = skill_events({"command": "gitnexus-work"})
|
|
events.insert(
|
|
1,
|
|
{
|
|
"type": "assistant",
|
|
"message": {"content": [{"type": "text", "text": secret}]},
|
|
},
|
|
)
|
|
events.insert(
|
|
2,
|
|
{
|
|
"type": "assistant",
|
|
"message": {"content": [{"type": "text", "text": bearer}]},
|
|
},
|
|
)
|
|
events.insert(
|
|
3,
|
|
{
|
|
"type": "assistant",
|
|
"message": {
|
|
"content": [
|
|
{
|
|
"type": "text",
|
|
"metadata": {
|
|
"Authorization": f"Bearer {structural_secret}",
|
|
"password": password,
|
|
},
|
|
}
|
|
]
|
|
},
|
|
},
|
|
)
|
|
stream = event_stream(*events)
|
|
monkeypatch.setattr(runner_sessions, "run_managed", lambda *a, **k: fake_cli_result(stream))
|
|
output = tmp_path / "run-output"
|
|
output.mkdir()
|
|
|
|
rec = runner.run_claude(
|
|
"task",
|
|
tmp_path,
|
|
claude_bin="claude",
|
|
timeout=5,
|
|
expected_skill="gitnexus-work",
|
|
transcript_output_dir=output,
|
|
transcript_output_prefix="task-workflow-run0",
|
|
transcript_secrets=(secret,),
|
|
)
|
|
|
|
artifact_meta = rec["transcript_artifact"]
|
|
artifact = output / artifact_meta["path"]
|
|
assert artifact_meta["path"] == "transcripts/task-workflow-run0-s.jsonl"
|
|
assert artifact.stat().st_mode & 0o777 == 0o600
|
|
assert artifact.parent.stat().st_mode & 0o777 == 0o700
|
|
assert secret not in artifact.read_text()
|
|
assert "bearer-postmortem-secret" not in artifact.read_text()
|
|
assert structural_secret not in artifact.read_text()
|
|
assert password not in artifact.read_text()
|
|
assert artifact.read_text().count("[REDACTED]") >= 4
|
|
assert all(isinstance(json.loads(line), dict) for line in artifact.read_text().splitlines())
|
|
assert artifact_meta["bytes"] == artifact.stat().st_size
|
|
assert artifact_meta["sha256"] == hashlib.sha256(artifact.read_bytes()).hexdigest()
|
|
assert artifact_meta["source"] == "parent-captured-stream-json"
|
|
|
|
# PR #2566 P1 regression: sum_sessions forwards the producer's 4-key record
|
|
# (including `source`) into transcript_artifacts, and a --seed-results / gen>=2
|
|
# run JSON round-trips it through the proposer evidence preflight. That preflight
|
|
# once required exactly {path, sha256, bytes} and aborted every seeded run with
|
|
# SandboxError. Round-trip real producer output through the real preflight so the
|
|
# producer/validator schema can never drift apart again.
|
|
seeded = json.loads(json.dumps(runner_sessions.sum_sessions([rec])))
|
|
evolve._preflight_transcript_artifacts([{"transcript_artifacts": seeded["transcript_artifacts"]}])
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("raw", "expected"),
|
|
[
|
|
(0.0, 0.0),
|
|
(1.25, 1.25),
|
|
(3, 3.0),
|
|
(None, None),
|
|
("free", None),
|
|
(True, None),
|
|
(-1.0, None),
|
|
(float("nan"), None),
|
|
(float("inf"), None),
|
|
],
|
|
)
|
|
def test_measured_cost_distinguishes_absent_from_zero(raw, expected):
|
|
# A measured $0 stays 0.0; an absent/garbage cost becomes None so it can
|
|
# never be scored as a real zero the promotion gate ranks on.
|
|
assert runner_sessions.measured_cost(raw) == expected
|
|
|
|
|
|
def test_missing_skill_invocation_fails_closed(monkeypatch, tmp_path):
|
|
read_events = skill_events({"skill": "other-skill"}, tool_id="read-1")
|
|
monkeypatch.setattr(
|
|
runner_sessions,
|
|
"run_managed",
|
|
lambda *a, **k: fake_cli_result(event_stream(*read_events)),
|
|
)
|
|
rec = runner.run_claude("task", tmp_path, claude_bin="claude", timeout=5, expected_skill="gitnexus-work")
|
|
assert rec["ok"] is False
|
|
assert rec["skill_invoked"] is False
|
|
assert rec["error_kind"] == "skill-not-invoked"
|
|
|
|
|
|
def test_overflowed_parent_capture_is_ineligible_evidence(monkeypatch, tmp_path):
|
|
monkeypatch.setattr(
|
|
runner_sessions,
|
|
"run_managed",
|
|
lambda *a, **k: fake_cli_result(VALID_REPORT, overflow=True),
|
|
)
|
|
rec = runner.run_claude(
|
|
"task",
|
|
tmp_path,
|
|
claude_bin="claude",
|
|
timeout=5,
|
|
expected_skill="gitnexus-work",
|
|
)
|
|
assert rec["ok"] is False
|
|
assert rec["skill_invoked"] is None
|
|
assert rec["error_kind"] == "session-error"
|
|
assert "exceeds" in rec["error_detail"]["event_stream_error"]
|
|
|
|
|
|
def test_final_result_event_must_be_last(monkeypatch, tmp_path):
|
|
stream = VALID_REPORT + "\n" + json.dumps({"type": "assistant", "message": {"content": []}}) + "\n"
|
|
monkeypatch.setattr(runner_sessions, "run_managed", lambda *a, **k: fake_cli_result(stream))
|
|
rec = runner.run_claude(
|
|
"task",
|
|
tmp_path,
|
|
claude_bin="claude",
|
|
timeout=5,
|
|
)
|
|
|
|
assert rec["ok"] is False
|
|
assert rec["error_kind"] == "session-error"
|
|
assert "not the last event" in rec["error_detail"]["event_stream_error"]
|
|
|
|
|
|
def background_completion_events():
|
|
first = json.loads(report_variant(uuid="first", subtype="success", is_error=False))
|
|
last = json.loads(report_variant(uuid="last", subtype="success", is_error=False, total_cost_usd=0.15))
|
|
return [
|
|
first,
|
|
{"type": "system", "subtype": "task_notification", "session_id": "s"},
|
|
{"type": "system", "subtype": "init", "session_id": "s"},
|
|
{"type": "assistant", "message": {"content": [{"type": "text", "text": "Task finished"}]}},
|
|
last,
|
|
]
|
|
|
|
|
|
def test_background_completion_sums_turn_usage_but_not_cumulative_cost(monkeypatch, tmp_path):
|
|
events = background_completion_events()
|
|
proc = replace(fake_cli_result("\n".join(map(json.dumps, events))), duration_s=12.5)
|
|
monkeypatch.setattr(runner_sessions, "run_managed", lambda *a, **k: proc)
|
|
rec = runner.run_claude("task", tmp_path, claude_bin="claude", timeout=30)
|
|
|
|
assert rec["ok"] is True, rec["error_detail"]
|
|
assert rec["cost_usd"] == 0.15 # The second result includes the first turn's $0.10.
|
|
assert rec["num_turns"] == 6
|
|
assert [rec[field] for field in runner_sessions.USAGE_FIELDS] == [2, 4, 6, 8]
|
|
assert rec["duration_s"] == 12.5 # Include background wait, not just the two 1s turns.
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mutate",
|
|
[
|
|
lambda es: es[-1].update(session_id="different"),
|
|
lambda es: es[-1].update(uuid="first"),
|
|
lambda es: es[-1].update(total_cost_usd=0.05),
|
|
lambda es: es[0].update(is_error=True),
|
|
lambda es: es[0].update(subtype="error_max_turns"),
|
|
lambda es: es[0]["usage"].pop("input_tokens"),
|
|
lambda es: es[-1]["usage"].update(output_tokens=-1),
|
|
lambda es: es[-1]["usage"].update(output_tokens=True),
|
|
lambda es: es[-1].update(num_turns=-1),
|
|
lambda es: es[-1].update(total_cost_usd=None),
|
|
lambda es: es.pop(1), # No background completion to explain another turn.
|
|
lambda es: es.pop(3), # A second result without a new assistant turn is a replay.
|
|
lambda es: es[2].update(session_id="different"),
|
|
lambda es: es.append({"type": "assistant", "message": {"content": []}}),
|
|
],
|
|
)
|
|
def test_unverifiable_background_continuation_stays_excluded(monkeypatch, tmp_path, mutate):
|
|
events = background_completion_events()
|
|
mutate(events)
|
|
monkeypatch.setattr(
|
|
runner_sessions, "run_managed", lambda *a, **k: fake_cli_result("\n".join(map(json.dumps, events)))
|
|
)
|
|
rec = runner.run_claude("task", tmp_path, claude_bin="claude", timeout=30, transcript_output_dir=tmp_path)
|
|
assert rec["ok"] is False
|
|
assert rec["error_kind"] == "session-error"
|
|
assert rec["cost_usd"] is None
|
|
assert rec["transcript_missing"] is True
|
|
|
|
|
|
def test_real_cli_background_completion_is_one_complete_measurement(monkeypatch, tmp_path):
|
|
"""Only the provider is scripted; the pinned CLI performs a real background Bash task."""
|
|
from workflow_bench.mock_provider import MockProvider, Reply
|
|
|
|
claude = os.environ.get("CLAUDE_CANARY_BIN")
|
|
if not claude or not Path(claude).is_file():
|
|
if os.environ.get("GITNEXUS_REQUIRE_CLAUDE_CANARY") == "1":
|
|
pytest.fail("required pinned CLAUDE_CANARY_BIN is missing")
|
|
pytest.skip("the containment job supplies the pinned CLAUDE_CANARY_BIN")
|
|
home = tmp_path / "home"
|
|
home.mkdir()
|
|
# Release the real background command only after the first result. A fixed
|
|
# sleep could finish before that result on a loaded runner and miss the bug.
|
|
observe_line = runner_sessions.SessionProgress._observe_line
|
|
|
|
def release_background_task(self, line):
|
|
observe_line(self, line)
|
|
if json.loads(line).get("type") == "result":
|
|
(home / "completion-ready").touch()
|
|
|
|
monkeypatch.setattr(runner_sessions.SessionProgress, "_observe_line", release_background_task)
|
|
replies = [
|
|
Reply(
|
|
tools=[
|
|
{
|
|
"name": "Bash",
|
|
"input": {
|
|
"command": "while [ ! -f completion-ready ]; do sleep 0.01; done; printf done",
|
|
"run_in_background": True,
|
|
},
|
|
}
|
|
],
|
|
input_tokens=101,
|
|
output_tokens=11,
|
|
),
|
|
Reply(text="First completion", input_tokens=102, output_tokens=12),
|
|
Reply(text="Background completion acknowledged", input_tokens=103, output_tokens=13),
|
|
]
|
|
with MockProvider(replies) as provider:
|
|
rec = runner.run_claude(
|
|
"Run the command",
|
|
home,
|
|
claude_bin=claude,
|
|
timeout=40,
|
|
env={
|
|
"PATH": os.environ["PATH"],
|
|
"HOME": str(home),
|
|
"ANTHROPIC_API_KEY": "offline-probe",
|
|
"ANTHROPIC_BASE_URL": provider.base_url,
|
|
"CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1",
|
|
"DISABLE_AUTOUPDATER": "1",
|
|
# Only loopback may be contacted; even CLI startup traffic
|
|
# cannot reach an external provider. No host credentials inherit.
|
|
"HTTPS_PROXY": "http://127.0.0.1:9",
|
|
"HTTP_PROXY": "http://127.0.0.1:9",
|
|
"NO_PROXY": "127.0.0.1,localhost",
|
|
},
|
|
allowed_tools=["Bash"],
|
|
strict_mcp_config=True,
|
|
transcript_output_dir=tmp_path,
|
|
progress_label="offline-background-completion",
|
|
)
|
|
assert rec["ok"] is True, rec["error_detail"]
|
|
assert len(provider.requests) == 3
|
|
assert rec["input_tokens"] == 306 and rec["output_tokens"] == 36
|
|
assert rec["num_turns"] == 3
|
|
artifacts = list((tmp_path / "transcripts").glob("*.jsonl"))
|
|
assert len(artifacts) == 1
|
|
results = [e for line in artifacts[0].read_text().splitlines() if (e := json.loads(line))["type"] == "result"]
|
|
assert len(results) == 2
|
|
assert rec["cost_usd"] == results[-1]["total_cost_usd"] > results[0]["total_cost_usd"]
|
|
|
|
|
|
def test_background_task_teardown_after_the_result_stays_valid_evidence(monkeypatch, tmp_path):
|
|
# Claude Code drains background-task bookkeeping after the final result
|
|
# event. Those `system` events carry no tool or usage payload, so they must
|
|
# not invalidate an otherwise complete session (run 29907431284 lost three
|
|
# runs this way, and the gate demands zero excluded runs).
|
|
teardown = [
|
|
{"type": "system", "subtype": "background_tasks_changed", "tasks": []},
|
|
{"type": "system", "subtype": "task_updated", "task_id": "bdw43oy7j", "patch": {"status": "killed"}},
|
|
{"type": "system", "subtype": "task_notification", "task_id": "bdw43oy7j", "status": "stopped"},
|
|
]
|
|
stream = event_stream(*skill_events({"skill": "gitnexus-work"})) + "".join(
|
|
json.dumps(event) + "\n" for event in teardown
|
|
)
|
|
monkeypatch.setattr(runner_sessions, "run_managed", lambda *a, **k: fake_cli_result(stream))
|
|
rec = runner.run_claude(
|
|
"task",
|
|
tmp_path,
|
|
claude_bin="claude",
|
|
timeout=5,
|
|
expected_skill="gitnexus-work",
|
|
)
|
|
|
|
assert rec["ok"] is True
|
|
assert rec["error_kind"] is None
|
|
assert rec["skill_invoked"] is True
|
|
assert rec["transcript_missing"] is False
|
|
assert "evidence_diagnostics" not in rec
|
|
|
|
|
|
def test_snapshot_plan_docs_detects_one_modified_plan_and_rejects_ambiguous_output(tmp_path):
|
|
plans = tmp_path / "docs" / "plans"
|
|
plans.mkdir(parents=True)
|
|
first = plans / "one.md"
|
|
first.write_text("before")
|
|
before = snapshot_plan_docs(tmp_path)
|
|
first.write_text("after")
|
|
assert runner.new_plan_doc(tmp_path, before) == first
|
|
|
|
second = plans / "two.md"
|
|
second.write_text("new")
|
|
first.write_text("changed again")
|
|
with pytest.raises(ValueError, match="exactly one"):
|
|
runner.new_plan_doc(tmp_path, before)
|
|
|
|
|
|
def test_plan_output_rejects_unchanged_deleted_and_symlink_paths(tmp_path):
|
|
plans = tmp_path / "docs" / "plans"
|
|
plans.mkdir(parents=True)
|
|
plan = plans / "one.md"
|
|
plan.write_text("same")
|
|
before = snapshot_plan_docs(tmp_path)
|
|
with pytest.raises(ValueError, match="exactly one"):
|
|
runner.new_plan_doc(tmp_path, before)
|
|
|
|
plan.unlink()
|
|
with pytest.raises(ValueError, match="deleted"):
|
|
runner.new_plan_doc(tmp_path, before)
|
|
|
|
target = tmp_path / "outside.md"
|
|
target.write_text("outside")
|
|
plan.symlink_to(target)
|
|
with pytest.raises(ValueError, match="symlink"):
|
|
runner.new_plan_doc(tmp_path, {})
|
|
|
|
|
|
def test_setup_skill_fingerprint_change_fails_closed(tmp_path):
|
|
for skill in ("gitnexus-plan", "gitnexus-work"):
|
|
path = tmp_path / ".claude" / "skills" / skill / "SKILL.md"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(f"{skill} original")
|
|
expected = skill_fingerprint(tmp_path, "workflow")
|
|
(tmp_path / ".claude" / "skills" / "gitnexus-plan" / "SKILL.md").write_text("replaced by setup")
|
|
|
|
with pytest.raises(ValueError, match="task setup changed the evaluated skill fingerprint"):
|
|
runner.require_skill_fingerprint(tmp_path, "workflow", expected, phase="task setup")
|
|
|
|
|
|
def test_planning_cannot_change_source_tests_or_downstream_skill(monkeypatch, tmp_path):
|
|
for skill in ("gitnexus-plan", "gitnexus-work"):
|
|
path = tmp_path / ".claude" / "skills" / skill / "SKILL.md"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(f"{skill} original")
|
|
source = tmp_path / "source.py"
|
|
source.write_text("original")
|
|
expected = skill_fingerprint(tmp_path, "workflow")
|
|
calls: list[str] = []
|
|
|
|
def adversarial_plan(prompt, *args, **kwargs):
|
|
calls.append(prompt)
|
|
plans = tmp_path / "docs" / "plans"
|
|
plans.mkdir(parents=True)
|
|
(plans / "authorized.md").write_text("plan")
|
|
source.write_text("pre-implemented by planning")
|
|
(tmp_path / ".claude" / "skills" / "gitnexus-work" / "SKILL.md").write_text("weakened")
|
|
return session_record()
|
|
|
|
monkeypatch.setattr(runner, "run_claude", adversarial_plan)
|
|
monkeypatch.setattr(runner, "run_verify", lambda *a, **k: (True, "ok"))
|
|
sandbox = fake_sandbox(tmp_path)
|
|
|
|
rec = runner.run_arm(
|
|
"workflow",
|
|
{"prompt": "p", "verify": "true"},
|
|
tmp_path,
|
|
bench_args(),
|
|
sandbox=sandbox,
|
|
expected_skill_digest=expected,
|
|
enforce_phase_boundary=True,
|
|
)
|
|
|
|
assert len(calls) == 1
|
|
assert rec["ok"] is False
|
|
assert rec["resolved"] is False
|
|
assert rec["error_kind"] == "plan-evidence-invalid"
|
|
assert "unauthorized workspace path" in rec["error_detail"]
|
|
|
|
|
|
@pytest.mark.parametrize("arm", ["review", "ce_review"])
|
|
@pytest.mark.parametrize(
|
|
("attack", "expected_detail"),
|
|
[
|
|
("workspace", "changed the read-only workspace"),
|
|
("skill", "changed the evaluated skill fingerprint"),
|
|
],
|
|
)
|
|
def test_review_phase_rejects_workspace_or_skill_mutation(
|
|
monkeypatch,
|
|
tmp_path,
|
|
arm,
|
|
attack,
|
|
expected_detail,
|
|
):
|
|
source = tmp_path / "source.py"
|
|
source.write_text("original")
|
|
expected_skill_digest = "expected-skill-fingerprint"
|
|
|
|
def adversarial_review(prompt, *args, **kwargs):
|
|
# Write where the contract now says: the artifact directory outside the
|
|
# workspace, which is the only place the agent can write atomically.
|
|
artifact = runner.review_output_path(sandbox, runner.REVIEW_OUTPUT)
|
|
artifact.parent.mkdir(parents=True, exist_ok=True)
|
|
artifact.write_text('{"schema_version":1,"verdict":"approve","findings":[]}')
|
|
if attack == "workspace":
|
|
source.write_text("review silently changed source")
|
|
return session_record()
|
|
|
|
observed_skill_digest = "tampered-skill-fingerprint" if attack == "skill" else expected_skill_digest
|
|
monkeypatch.setattr(runner, "run_claude", adversarial_review)
|
|
monkeypatch.setattr(
|
|
runner,
|
|
"skill_fingerprint",
|
|
lambda worktree, checked_arm: observed_skill_digest,
|
|
)
|
|
monkeypatch.setattr(runner, "run_verify", lambda *a, **k: (True, "ok"))
|
|
sandbox = fake_sandbox(tmp_path)
|
|
|
|
rec = runner.run_arm(
|
|
arm,
|
|
{"prompt": "p", "verify": "true"},
|
|
tmp_path,
|
|
bench_args(),
|
|
sandbox=sandbox,
|
|
expected_skill_digest=expected_skill_digest,
|
|
enforce_phase_boundary=True,
|
|
)
|
|
|
|
assert rec["ok"] is False
|
|
assert rec["resolved"] is False
|
|
assert rec["error_kind"] == "review-evidence-invalid"
|
|
assert expected_detail in rec["error_detail"]
|
|
|
|
|
|
def test_a_cancelled_clone_copy_does_not_fall_back_to_an_uncancellable_copytree(monkeypatch, tmp_path):
|
|
"""The reflink fallback is for a filesystem, not for a teardown.
|
|
|
|
run_managed reports cancellation as a non-OK result rather than raising, so
|
|
the fallback treated it like an unsupported reflink and started a copytree
|
|
that cannot be cancelled — waiting out exactly the full copy the outage
|
|
breaker set the cancellation event to avoid.
|
|
"""
|
|
|
|
source = tmp_path / "template"
|
|
(source / ".git").mkdir(parents=True)
|
|
parent = tmp_path / "clones"
|
|
parent.mkdir()
|
|
copied: list[object] = []
|
|
|
|
monkeypatch.setattr(
|
|
runner_artifacts,
|
|
"run_managed",
|
|
lambda *_a, **_k: ManagedProcessResult(
|
|
state="cancelled",
|
|
returncode=None,
|
|
stdout_tail="",
|
|
stderr_tail="",
|
|
duration_s=0.1,
|
|
),
|
|
)
|
|
monkeypatch.setattr(runner_artifacts.shutil, "copytree", lambda *a, **k: copied.append(a))
|
|
|
|
with pytest.raises(ManagedProcessError):
|
|
runner.copy_isolated_tree(source, parent)
|
|
assert copied == []
|
|
assert list(parent.iterdir()) == [], "the partial target must be cleaned up"
|
|
|
|
|
|
@pytest.mark.parametrize("arm", ["review", "ce_review"])
|
|
def test_run_arm_mounts_the_review_artifact_directory_outside_the_workspace(monkeypatch, tmp_path, arm):
|
|
"""A writable FILE inside a read-only directory is not a writable path.
|
|
|
|
The Write tool creates `<target>.tmp.<n>.<hex>` beside the target and
|
|
renames it, so a read-only parent fails the temp create with EROFS and the
|
|
artifact stays 0 bytes. The mount target must be the directory, and it must
|
|
sit outside the read-only workspace.
|
|
|
|
Driven through run_arm rather than rebuilt here: an expected tuple assembled
|
|
in the test passes whatever run_arm actually mounts, which is the one thing
|
|
this needs to prove.
|
|
"""
|
|
|
|
assert not runner.SANDBOX_REVIEW_OUTPUT.startswith(runner.SANDBOX_WORKSPACE + "/")
|
|
assert runner.SANDBOX_REVIEW_OUTPUT != runner.SANDBOX_WORKSPACE
|
|
|
|
verify_calls: list[dict] = []
|
|
sandbox = fake_sandbox(tmp_path)
|
|
sandbox.command_prefix_for = lambda **kwargs: verify_calls.append(kwargs) or []
|
|
|
|
def review_session(prompt, *args, **kwargs):
|
|
artifact = runner.review_output_path(sandbox, runner.REVIEW_OUTPUT)
|
|
artifact.parent.mkdir(parents=True, exist_ok=True)
|
|
artifact.write_text('{"schema_version":1,"verdict":"approve","findings":[]}')
|
|
return session_record()
|
|
|
|
monkeypatch.setattr(runner, "run_claude", review_session)
|
|
monkeypatch.setattr(runner, "skill_fingerprint", lambda *_a, **_k: "skill-digest")
|
|
monkeypatch.setattr(runner, "run_verify", lambda *a, **k: (True, "ok"))
|
|
|
|
runner.run_arm(
|
|
arm,
|
|
{"prompt": "p", "verify": "true"},
|
|
tmp_path,
|
|
bench_args(),
|
|
sandbox=sandbox,
|
|
expected_skill_digest="skill-digest",
|
|
)
|
|
|
|
review_output = runner.review_output_path(sandbox, runner.REVIEW_OUTPUT)
|
|
expected = (runner.ReadOnlyMount(source=review_output.parent, target=runner.SANDBOX_REVIEW_OUTPUT),)
|
|
# The EROFS bug is about the AGENT's write, so the mount that has to be the
|
|
# directory is the writable one on the review session — not the read-only
|
|
# exposure the verify command gets afterwards. Assert both: they are
|
|
# separate arguments to separate command prefixes.
|
|
writable = [call["extra_writable_mounts"] for call in verify_calls if "extra_writable_mounts" in call]
|
|
assert writable, "the review session must be given a writable artifact mount"
|
|
assert writable[-1] == expected, "mount the directory, not the file"
|
|
read_only = [call["extra_read_only_mounts"] for call in verify_calls if "extra_read_only_mounts" in call]
|
|
assert read_only, "the verify invocation must be given the artifact mount"
|
|
assert read_only[-1] == expected, "mount the directory, not the file"
|
|
assert not expected[0].target.startswith(f"{runner.SANDBOX_WORKSPACE}/")
|
|
# The artifact the harness later reads is the one inside that mount.
|
|
assert review_output.parent in review_output.parents
|
|
|
|
|
|
def _git(repo, *args, check=True):
|
|
return subprocess.run(["git", "-C", str(repo), *args], check=check, capture_output=True, text=True)
|
|
|
|
|
|
def _git_commit(repo, message):
|
|
_git(
|
|
repo,
|
|
"-c",
|
|
"user.name=test",
|
|
"-c",
|
|
"user.email=test@invalid",
|
|
"commit",
|
|
"--quiet",
|
|
"--allow-empty",
|
|
"-m",
|
|
message,
|
|
)
|
|
return _git(repo, "rev-parse", "HEAD").stdout.strip()
|
|
|
|
|
|
def test_make_worktree_clone_has_no_tags_but_keeps_all_branches(tmp_path):
|
|
# oracle_assets.MAX_CLONE_REFS refuses to sanitize a clone with more than
|
|
# 1024 refs; this repo's own history has 1000+ release-candidate tags, so
|
|
# a plain `git clone` of it (inheriting every tag) trips that cap on every
|
|
# benchmark session. make_worktree must not carry tags into its throwaway
|
|
# clone, but callers pass a bare SHA or "HEAD" as `ref` (never a branch
|
|
# name -- see evolve.py:476, runner.py:1037, sanitized_graph.py:345), so
|
|
# branch-fetching itself must stay untouched: a commit reachable only from
|
|
# a non-default branch must still resolve via the existing
|
|
# checkout(ref) -> checkout(origin/{ref}) fallback.
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
_git(repo, "init", "--quiet")
|
|
_git(repo, "checkout", "--quiet", "-b", "main")
|
|
_git_commit(repo, "base")
|
|
_git(repo, "tag", "v1.0.0-rc.1")
|
|
|
|
_git(repo, "checkout", "--quiet", "-b", "other")
|
|
other_sha = _git_commit(repo, "only on other")
|
|
_git(repo, "checkout", "--quiet", "main")
|
|
|
|
clones = tmp_path / "clones"
|
|
clones.mkdir()
|
|
target = runner.make_worktree(repo, other_sha, clones)
|
|
|
|
tags = _git(target, "tag").stdout.split()
|
|
assert tags == [], f"clone must carry no tags, found: {tags}"
|
|
|
|
current = _git(target, "rev-parse", "HEAD").stdout.strip()
|
|
assert current == other_sha
|
|
|
|
|
|
def test_copy_isolated_tree_does_not_share_git_objects_or_refs(tmp_path):
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
_git(repo, "init", "--quiet")
|
|
_git(repo, "checkout", "--quiet", "-b", "main")
|
|
sha = _git_commit(repo, "base")
|
|
clones = tmp_path / "clones"
|
|
clones.mkdir()
|
|
template = runner.make_worktree(repo, sha, clones)
|
|
(template / "marker.txt").write_text("template\n")
|
|
|
|
copy = runner.copy_isolated_tree(template, clones)
|
|
assert copy != template
|
|
assert (copy / "marker.txt").read_text() == "template\n"
|
|
(copy / "marker.txt").write_text("copy\n")
|
|
assert (template / "marker.txt").read_text() == "template\n"
|
|
copy_head = _git(copy, "rev-parse", "HEAD").stdout.strip()
|
|
template_head = _git(template, "rev-parse", "HEAD").stdout.strip()
|
|
assert copy_head == template_head == sha
|
|
# An equal initial HEAD is also what a shared ref namespace looks like, so
|
|
# write a ref and prove the template cannot see it. A linked worktree would
|
|
# pass every assertion above, including the alternates check — its `.git` is
|
|
# a file, so the directory inspected below simply does not exist.
|
|
_git(copy, "branch", "copy-only")
|
|
assert _git(copy, "show-ref", "--verify", "refs/heads/copy-only").returncode == 0
|
|
assert _git(template, "show-ref", "--verify", "refs/heads/copy-only", check=False).returncode != 0
|
|
assert (copy / ".git").is_dir()
|
|
alternates = copy / ".git" / "objects" / "info" / "alternates"
|
|
assert not alternates.exists()
|