test: prepare real local rein acceptance with artifact verification

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0726e-5232-73f2-aaca-2c05ceb62efb

tests/test_local_profile_proof.py

@@ -0,0 +1,172 @@
+import importlib.util
+import json
+import subprocess
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from glas_harness.contract import ExecutionSummary, OperationalReadiness, ToolCall
+from glas_harness.profiles import OperationallyBlockedProfileError, ProfileCatalog
+from glas_harness.reins.rein_aharness import ReinAharness
+
+
+spec = importlib.util.spec_from_file_location(
+    "local_profile_proof", Path(__file__).parents[1] / "scripts/prove-local-profile.py"
+)
+proof = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(proof)
+
+
+def test_blocked_profile_refuses_before_fixture_or_manager_creation():
+    with patch.object(proof.tempfile, "TemporaryDirectory") as temporary, \
+            patch.object(proof, "SandboxManager") as manager:
+        with pytest.raises(OperationallyBlockedProfileError):
+            proof.prove("harness.agent-dev-local@1.0.0", ProfileCatalog())
+    temporary.assert_not_called()
+    manager.assert_not_called()
+
+
+def test_unpinned_profile_is_never_a_proof_candidate():
+    with pytest.raises(ValueError, match="version-pinned"):
+        proof.select_candidate("harness.agent-dev-local", ProfileCatalog())
+
+
+@pytest.mark.parametrize("changes", [
+    {"credential_route_refs": []},
+    {"timeout_seconds": 901},
+    {"timeout_seconds": None},
+    {"budget_tokens": 60001},
+    {"budget_tokens": None},
+])
+def test_candidate_requires_credentials_and_explicit_proof_bounds(changes):
+    catalog = ProfileCatalog()
+    profile, _ = catalog.resolve("harness.agent-dev-local@1.0.0")
+    updates = {
+        "operational_readiness": OperationalReadiness(status="unverified"),
+        "credential_route_refs": ["test-claude-route"],
+        "limits": profile.limits.model_copy(update={
+            key: value for key, value in changes.items() if key != "credential_route_refs"
+        }),
+    }
+    if "credential_route_refs" in changes:
+        updates["credential_route_refs"] = changes["credential_route_refs"]
+    catalog.profiles()[(profile.id, profile.version)] = profile.model_copy(update=updates)
+    with pytest.raises(ValueError):
+        proof.select_candidate(str(profile.ref), catalog)
+
+
+def test_observer_delegates_actual_dispatch_and_cleanup():
+    delegate = MagicMock(spec=ReinAharness)
+    observer = proof.ObservedRein(delegate, Path("/absent-source"))
+    session = {"transport": MagicMock(), "task_file": "/workspace/.git/task.json"}
+    call = ToolCall(name="run_task")
+    assert observer.dispatch_tool(session, call) is delegate.dispatch_tool.return_value
+    delegate.dispatch_tool.assert_called_once_with(session, call)
+    session["transport"].run.return_value.returncode = 0
+    observer.cleanup_session(session)
+    delegate.cleanup_session.assert_called_once_with(session)
+    session["transport"].run.assert_called_once_with(
+        ["test", "!", "-e", session["task_file"]], timeout=15
+    )
+    assert observer.task_removed
+
+
+@pytest.mark.parametrize("report", [
+    {"head": "abc", "checks": {}},
+    {"head": "wrong", "checks": {key: True for key in proof.CHECKS}},
+    {"head": "abc", "checks": {key: "true" for key in proof.CHECKS}},
+])
+def test_observer_rejects_incomplete_mismatched_or_untyped_inspection(report):
+    delegate = MagicMock(spec=ReinAharness)
+    delegate.end_session.return_value = ExecutionSummary(
+        committed=True, outcome="succeeded", commit_sha="abc"
+    )
+    observer = proof.ObservedRein(delegate, Path("/absent-source"))
+    observer.request_id = "proof-request"
+    transport = MagicMock()
+    transport.run.return_value = subprocess.CompletedProcess([], 0, json.dumps(report), "")
+    with pytest.raises(RuntimeError, match="acceptance failed"):
+        observer.end_session({"transport": transport, "head_before": "before",
+                              "task_file": "/workspace/.git/task.json"})
+
+
+@pytest.fixture
+def repository(tmp_path):
+    repo = tmp_path / "sandbox"
+    repo.mkdir()
+    proof.git(repo, "init", "-q")
+    proof.git(repo, "config", "user.name", "Glas Proof Tests")
+    proof.git(repo, "config", "user.email", "proof@example.invalid")
+    (repo / "sentinel").write_text("unchanged\n")
+    proof.git(repo, "add", "sentinel")
+    proof.git(repo, "commit", "-qm", "baseline")
+    baseline = proof.git(repo, "rev-parse", "HEAD").decode().strip()
+    task = repo / ".git/task.json"
+    task.write_text("{}")
+    task.chmod(0o600)
+    (repo / "PROOF.md").write_text("Glas local profile proof.\n")
+    proof.git(repo, "add", "PROOF.md")
+    proof.git(repo, "commit", "-qm", "proof")
+    return repo, baseline, task
+
+
+def inspect(repository, source):
+    repo, baseline, task = repository
+    result = subprocess.run(
+        ["/usr/bin/python3", "-c", proof.INSPECTOR, baseline, str(source), str(task), "req"],
+        cwd=repo, capture_output=True, text=True, timeout=15,
+        env={"PATH": "/usr/bin:/bin", "HOME": str(repo),
+             "SANDBOXER_ACTOR": "agt", "SANDBOXER_PROJECT": "glas-local-proof",
+             "SANDBOXER_RUN_ID": "req"},
+    )
+    assert result.returncode == 0, result.stderr
+    return json.loads(result.stdout)
+
+
+def test_artifact_inspector_validates_content_and_parent_commit(repository, tmp_path):
+    report = inspect(repository, tmp_path / "absent-source")
+    assert set(report["checks"]) == proof.CHECKS
+    assert all(report["checks"].values())
+
+
+@pytest.mark.parametrize("fault,failed_check", [
+    ("extra_commit", "one_commit"),
+    ("wrong_content", "committed_content"),
+    ("extra_path", "only_expected_path"),
+    ("untracked", "clean_worktree"),
+    ("ignored", "clean_worktree"),
+    ("source_visible", "source_absent"),
+    ("task_mode", "task_private"),
+])
+def test_artifact_inspector_rejects_false_positive_commits(repository, tmp_path,
+                                                        fault, failed_check):
+    repo, _, task = repository
+    source = tmp_path / "absent-source"
+    if fault == "extra_commit":
+        proof.git(repo, "commit", "--allow-empty", "-qm", "extra")
+    elif fault == "wrong_content":
+        (repo / "PROOF.md").write_text("wrong\n")
+        proof.git(repo, "commit", "-qam", "wrong", "--amend")
+    elif fault == "extra_path":
+        (repo / "extra").write_text("unwanted\n")
+        proof.git(repo, "add", "extra")
+        proof.git(repo, "commit", "-qm", "extra path", "--amend")
+    elif fault in {"untracked", "ignored"}:
+        (repo / "private-cache").write_text("must not survive\n")
+        if fault == "ignored":
+            (repo / ".git/info/exclude").write_text("private-cache\n")
+    elif fault == "source_visible":
+        source.mkdir()
+    elif fault == "task_mode":
+        task.chmod(0o644)
+    report = inspect(repository, source)
+    assert report["checks"][failed_check] is False
+
+
+def test_cli_never_prints_raw_provider_errors(capsys):
+    with patch.object(proof, "prove", side_effect=RuntimeError("secret-provider-response")):
+        assert proof.main(["--harness-profile", "harness.agent-dev-local@1.0.0"]) == 1
+    output = capsys.readouterr().out
+    assert "secret-provider-response" not in output
+    assert json.loads(output)["error_type"] == "RuntimeError"