diff --git a/.github/workflows/portability-offline.yml b/.github/workflows/portability-offline.yml index 7ae7275e..ef763bb6 100644 --- a/.github/workflows/portability-offline.yml +++ b/.github/workflows/portability-offline.yml @@ -40,6 +40,7 @@ jobs: tests/test_evidence_durability.py \ tests/test_surface_cache.py \ tests/test_evidence_settlement.py \ + tests/test_evidence_index_resilience.py \ tests/test_evidence_receipts.py \ tests/test_select_checks.py \ tests/test_scaffold_verification_skill.py \ diff --git a/scripts/select_checks.py b/scripts/select_checks.py index 6ba99adf..eab28682 100644 --- a/scripts/select_checks.py +++ b/scripts/select_checks.py @@ -165,7 +165,16 @@ def offline_contract_tests(root: Path = ROOT) -> List[str]: ("tests/test_verify_portable.py", "tests/test_verdict_reduction.py"), (), ), - ("scripts/verify-harness.py", ("tests/test_verify_harness.py",), ()), + ( + "scripts/verify-harness.py", + ( + "tests/test_verify_harness.py", + "tests/test_verify_harness_archive.py", + "tests/test_verify_harness_guides.py", + "tests/test_verify_harness_playbooks.py", + ), + (), + ), ("scripts/verify-decision-contract.py", ("tests/test_verify_decision_contract.py",), ()), ("scripts/verify_rollback_equivalence.py", ("tests/test_rollback_equivalence.py",), ()), ("scripts/scan-host-boundary.py", ("tests/test_scan_host_boundary.py",), ()), @@ -204,7 +213,12 @@ def offline_contract_tests(root: Path = ROOT) -> List[str]: ), ( "scripts/evidence_durability.py", - ("tests/test_evidence_durability.py", "tests/test_evidence_settlement.py", "tests/test_portability_schemas.py"), + ( + "tests/test_evidence_durability.py", + "tests/test_evidence_index_resilience.py", + "tests/test_evidence_settlement.py", + "tests/test_portability_schemas.py", + ), (), ), ( @@ -219,7 +233,11 @@ def offline_contract_tests(root: Path = ROOT) -> List[str]: ), ( "scripts/evidence_index.py", - ("tests/test_evidence_durability.py", "tests/test_verify_portable.py"), + ( + "tests/test_evidence_durability.py", + "tests/test_evidence_index_resilience.py", + "tests/test_verify_portable.py", + ), (), ), ( diff --git a/tests/evidence_test_helpers.py b/tests/evidence_test_helpers.py new file mode 100644 index 00000000..fcf1445c --- /dev/null +++ b/tests/evidence_test_helpers.py @@ -0,0 +1,362 @@ +from __future__ import annotations + +from dataclasses import asdict +import importlib.util +import json +import os +from pathlib import Path +import shutil +import subprocess +import sys + +from scripts.portability_types import ( + Binding, + ChildSettlement, + CONFORMANCE_PLANES, + Evidence, + HarnessProfile, + ScenarioResult, + VerificationReceipt, + reduce_verdicts, + settle_planes, +) +from scripts.surface_revisions import compute_surface_revisions + +ROOT = Path(__file__).resolve().parents[1] +EVIDENCE_DIR = ROOT / ".audit" / "evidence" + + +def _declared_hosts() -> tuple: + data = json.loads((ROOT / "pstack.package.json").read_text(encoding="utf-8")) + return tuple(data["host_targets"]) + + +def _omp_profile(data: dict | None = None) -> HarnessProfile: + d = data if data is not None else json.loads((ROOT / "profiles" / "omp.json").read_text(encoding="utf-8")) + return HarnessProfile( + host=d["host"], + support_state=d["support_state"], + bindings=[Binding(**x) for x in d["bindings"]], + skills_dir=d.get("skills_dir"), + plugins_dir=d.get("plugins_dir"), + plugin_manifest=d.get("plugin_manifest"), + evidence_ledger=[Evidence(**x) for x in d.get("evidence_ledger", [])], + tool_mappings=d.get("tool_mappings"), + skill_order=d.get("skill_order"), + runtime_conventions=d.get("runtime_conventions"), + ) + + +def _package_ignored_paths() -> frozenset: + listed = subprocess.run( + [ + "git", "-C", str(ROOT), "ls-files", + "--others", "--ignored", "--exclude-standard", "--directory", "-z", + ], + capture_output=True, + check=True, + ).stdout + return frozenset(os.fsdecode(path.rstrip(b"/")) for path in listed.split(b"\0") if path) + + +_PACKAGE_IGNORED = _package_ignored_paths() + + +def _disposable_copy(tmp_path: Path) -> Path: + copy_root = tmp_path / "extracted-package" + dropped_by_name = shutil.ignore_patterns(".git", "__pycache__", "node_modules", ".venv") + + def ignore(src: str, names: list) -> set: + skipped = set(dropped_by_name(src, names)) + skipped.update( + name for name in names + if Path(src, name).relative_to(ROOT).as_posix() in _PACKAGE_IGNORED + ) + return skipped + + shutil.copytree(ROOT, copy_root, ignore=ignore) + return copy_root + + +def _load_verify_portable_module(): + loader = importlib.util.spec_from_file_location( + "verify_portable_refresh", ROOT / "scripts" / "verify-portable.py" + ) + assert loader is not None and loader.loader is not None + module = importlib.util.module_from_spec(loader) + loader.loader.exec_module(module) + return module + + +def _stat_denied_copy(tmp_path: Path, relative: str) -> tuple: + copy_root = _disposable_copy(tmp_path) + link = copy_root / relative + denied = tmp_path / "denied" + denied.mkdir() + real = denied / Path(relative).name + real.write_bytes(link.read_bytes()) + link.unlink() + link.symlink_to(real) + denied.chmod(0o000) + return copy_root, link, denied + + +def _run_gate(copy_root: Path, *arguments: str) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, str(copy_root / "scripts" / "verify-portable.py"), *arguments], + capture_output=True, + text=True, + cwd=copy_root, + ) + + +def _declared_from(scenarios: list) -> tuple: + planes = settle_planes(scenarios) + return planes, reduce_verdicts(planes.values()) + + +def _plane_rows(covered: set, verdict: str = "PASS") -> list: + return [ + ScenarioResult( + id=f"synthetic-{plane}-plane-row", + description=f"synthetic {plane} plane row", + plane=plane, + command="true", + exit_code=0 if verdict == "PASS" else 1, + stdout_snippet=f"{plane} fixture row", + verdict=verdict, + duration_s=0.01, + evidence_level="offline", + ) + for plane in CONFORMANCE_PLANES + if plane not in covered + ] + + +def _synthetic_evidence_root( + tmp_path: Path, + evidence_levels: list, + planes: list | None = None, + host: str = "omp", + snippets: list | None = None, + settlements: list | None = None, + force_verified: bool = False, +) -> tuple: + tmp_path.mkdir(parents=True, exist_ok=True) + (tmp_path / "pstack.package.json").write_text("{}", encoding="utf-8") + manifest = tmp_path / ".omp-plugin" / "package.json" + manifest.parent.mkdir(parents=True, exist_ok=True) + manifest.write_text("{}", encoding="utf-8") + + revisions = compute_surface_revisions(tmp_path, host) + ref = ".audit/evidence/synthetic-receipt.json" + scenarios = [ + ScenarioResult( + id=f"synthetic-{i}", + description="synthetic scenario", + plane=(planes[i] if planes else "runtime"), + command="true", + exit_code=0, + stdout_snippet=(snippets[i] if snippets else ""), + verdict="PASS", + duration_s=0.01, + evidence_level=level, + settlement=(settlements[i] if settlements else None), + ) + for i, level in enumerate(evidence_levels) + ] + scenarios += _plane_rows({s.plane for s in scenarios}) + declared, overall = _declared_from(scenarios) + receipt = VerificationReceipt( + run_id="evidence-level-gate", + host=host, + host_version="test", + timestamp="2026-10-11T00:00:00Z", + overall_verdict=overall, + planes=declared, + surface_revisions=revisions, + scenarios=scenarios, + ) + receipt_path = tmp_path / ref + receipt_path.parent.mkdir(parents=True, exist_ok=True) + receipt_path.write_text(json.dumps(asdict(receipt)), encoding="utf-8") + + d = json.loads((ROOT / "profiles" / f"{host}.json").read_text(encoding="utf-8")) + for binding in d["bindings"]: + binding["evidence_ref"] = ref + if force_verified: + binding["verification_status"] = "verified" + prof = HarnessProfile( + host=d["host"], + support_state="verified", + bindings=[Binding(**x) for x in d["bindings"]], + skills_dir=d.get("skills_dir"), + plugins_dir=d.get("plugins_dir"), + plugin_manifest=d.get("plugin_manifest"), + tool_mappings=d.get("tool_mappings"), + skill_order=d.get("skill_order"), + runtime_conventions=d.get("runtime_conventions"), + ) + return tmp_path, prof + + +def _omp_spawn_scenario(evidence_level: str, verdict: str = "PASS", settlement=None) -> ScenarioResult: + return ScenarioResult( + id="runtime-child-spawn-omp", + description="bounded omp rpc v2 child spawn and join", + plane="runtime", + command="PSTACK_RUN_OMP_LIVE=1 python -m pytest tests/test_omp_rpc.py", + exit_code=0 if verdict == "PASS" else 1, + stdout_snippet="child=subagent-1 outcome=success" if settlement else "static evidence", + verdict=verdict, + duration_s=19.1, + evidence_level=evidence_level, + settlement=settlement, + ) + + +def _omp_live_root(tmp_path: Path) -> tuple: + root = tmp_path / "repo" + (root / ".omp-plugin").mkdir(parents=True) + (root / "pstack.package.json").write_text("{}", encoding="utf-8") + (root / ".omp-plugin" / "package.json").write_text("{}", encoding="utf-8") + evidence_root = root / ".audit" / "evidence" + evidence_root.mkdir(parents=True) + revisions = compute_surface_revisions(root, "omp") + scenarios = [ + _omp_spawn_scenario( + "observed_live", + settlement=ChildSettlement( + child_id="subagent-1", + terminal_outcome="success", + model="deepseek-v4-pro", + transport="rpc", + effective_rpc="rpc-v2:profile=disposable:depth=1", + ), + ) + ] + _plane_rows({"runtime"}) + declared, overall = _declared_from(scenarios) + live = VerificationReceipt( + run_id="previous-live-run", + host="omp", + host_version="18.8.9", + timestamp="2026-10-11T20:00:00Z", + overall_verdict=overall, + planes=declared, + surface_revisions=revisions, + scenarios=scenarios, + ) + (evidence_root / "omp-receipt.json").write_text(json.dumps(asdict(live)), encoding="utf-8") + + profile_data = json.loads((ROOT / "profiles" / "omp.json").read_text(encoding="utf-8")) + for binding in profile_data["bindings"]: + binding["evidence_ref"] = ".audit/evidence/omp-receipt.json" + binding["verification_status"] = "verified" + prof = HarnessProfile( + host="omp", + support_state="verified", + bindings=[Binding(**x) for x in profile_data["bindings"]], + skills_dir=profile_data.get("skills_dir"), + plugins_dir=profile_data.get("plugins_dir"), + plugin_manifest=profile_data.get("plugin_manifest"), + tool_mappings=profile_data.get("tool_mappings"), + skill_order=profile_data.get("skill_order"), + runtime_conventions=profile_data.get("runtime_conventions"), + ) + return root, prof, evidence_root, revisions + + +def _portable_verifier(evidence_root: Path, root: Path | None = None): + loader = importlib.util.spec_from_file_location("verify_portable_preserve", ROOT / "scripts" / "verify-portable.py") + assert loader is not None and loader.loader is not None + module = importlib.util.module_from_spec(loader) + loader.loader.exec_module(module) + if root is not None: + module.ROOT = root + verifier = module.PortableVerifier(host="omp", evidence_root=evidence_root) + verifier.launch() + return verifier + + +def _proof_bar_refresh(root: Path, evidence_root: Path, *, verdict: str = "PASS") -> tuple: + verifier = _portable_verifier(evidence_root, root=root) + verifier.scenarios = [ + ScenarioResult( + id=f"doctor-{plane}-stub", + description=f"static {plane} plane", + plane=plane, + command="stub", + exit_code=0, + stdout_snippet="stub", + verdict="PASS", + duration_s=0.0, + ) + for plane in ("canonical", "adapter", "package") + ] + verifier.scenarios.append(_omp_spawn_scenario("static", verdict=verdict)) + return verifier, verifier.proof_bar() + + +def _observed_scenario( + scenario_id: str, settlement: ChildSettlement | None = None +) -> ScenarioResult: + return ScenarioResult( + id=scenario_id, + description="adversarial observed_live runtime row", + plane="runtime", + command="omp --mode rpc", + exit_code=0, + stdout_snippet="observed_live row", + verdict="PASS", + duration_s=0.01, + evidence_level="observed_live", + settlement=settlement, + ) + + +def _success_settlement() -> ChildSettlement: + return ChildSettlement( + child_id="subagent-7", + terminal_outcome="success", + model="test-model", + transport="rpc", + effective_rpc="rpc-v2:profile=disposable:depth=1", + ) + + +def _adversarial_receipt(scenario: ScenarioResult) -> VerificationReceipt: + planes, overall = _declared_from([scenario]) + receipt = VerificationReceipt( + run_id="p8-adversarial", + host="omp", + host_version="test", + timestamp="2026-10-12T00:00:00Z", + overall_verdict=overall, + planes=planes, + surface_revisions=compute_surface_revisions(ROOT, "omp"), + scenarios=[scenario], + ) + receipt.validate() + return receipt + + +def _profile_pinning( + receipt_ref: str, scenario_id: str, capability: str = "agent.resume" +) -> HarnessProfile: + d = json.loads((ROOT / "profiles" / "omp.json").read_text(encoding="utf-8")) + for binding in d["bindings"]: + binding["evidence_ref"] = receipt_ref + if binding["capability"] == capability: + binding["verification_status"] = "verified" + binding["evidence_ref"] = {"receipt": receipt_ref, "scenario": scenario_id} + return HarnessProfile( + host=d["host"], + support_state=d["support_state"], + bindings=[Binding(**x) for x in d["bindings"]], + skills_dir=d.get("skills_dir"), + plugins_dir=d.get("plugins_dir"), + plugin_manifest=d.get("plugin_manifest"), + tool_mappings=d.get("tool_mappings"), + skill_order=d.get("skill_order"), + runtime_conventions=d.get("runtime_conventions"), + ) diff --git a/tests/test_evidence_durability.py b/tests/test_evidence_durability.py index 1de443c2..3ae45eaa 100644 --- a/tests/test_evidence_durability.py +++ b/tests/test_evidence_durability.py @@ -23,11 +23,12 @@ check_evidence_durability, compute_surface_revisions, ) - - -def _declared_hosts() -> tuple: - data = json.loads((ROOT / "pstack.package.json").read_text(encoding="utf-8")) - return tuple(data["host_targets"]) +from tests.evidence_test_helpers import ( + _declared_hosts, + _disposable_copy, + _load_verify_portable_module, + _omp_profile, +) SIX_HARNESSES = _declared_hosts() @@ -162,22 +163,6 @@ def test_dangling_evidence_path_fails(tmp_path: Path) -> None: check_evidence_durability(prof, ROOT) -def _omp_profile(data: dict | None = None) -> HarnessProfile: - d = data if data is not None else json.loads((ROOT / "profiles" / "omp.json").read_text(encoding="utf-8")) - return HarnessProfile( - host=d["host"], - support_state=d["support_state"], - bindings=[Binding(**x) for x in d["bindings"]], - skills_dir=d.get("skills_dir"), - plugins_dir=d.get("plugins_dir"), - plugin_manifest=d.get("plugin_manifest"), - evidence_ledger=[Evidence(**x) for x in d.get("evidence_ledger", [])], - tool_mappings=d.get("tool_mappings"), - skill_order=d.get("skill_order"), - runtime_conventions=d.get("runtime_conventions"), - ) - - def test_omp_profile_claims_only_what_its_receipt_settles() -> None: prof = _omp_profile() assert prof.support_state == "packaged" @@ -394,37 +379,6 @@ def test_hand_authored_support_state_cannot_outrun_evidence() -> None: check_evidence_durability(prof, ROOT, check_freshness=True, current_revisions=tampered_revs) -def _package_ignored_paths() -> frozenset: - listed = subprocess.run( - [ - "git", "-C", str(ROOT), "ls-files", - "--others", "--ignored", "--exclude-standard", "--directory", "-z", - ], - capture_output=True, - check=True, - ).stdout - return frozenset(os.fsdecode(path.rstrip(b"/")) for path in listed.split(b"\0") if path) - - -_PACKAGE_IGNORED = _package_ignored_paths() - - -def _disposable_copy(tmp_path: Path) -> Path: - copy_root = tmp_path / "extracted-package" - dropped_by_name = shutil.ignore_patterns(".git", "__pycache__", "node_modules", ".venv") - - def ignore(src: str, names: list) -> set: - skipped = set(dropped_by_name(src, names)) - skipped.update( - name for name in names - if Path(src, name).relative_to(ROOT).as_posix() in _PACKAGE_IGNORED - ) - return skipped - - shutil.copytree(ROOT, copy_root, ignore=ignore) - return copy_root - - def test_check_staleness_cli_json_output() -> None: res = subprocess.run( [sys.executable, str(ROOT / "scripts" / "verify-portable.py"), "check-staleness", "--host", "all", "--json"], @@ -534,15 +488,6 @@ def test_a_row_less_host_with_a_durable_receipt_reports_rebuild_required(tmp_pat assert "refresh-static --host all" in deleted.stdout -def _load_verify_portable_module(): - loader = importlib.util.spec_from_file_location( - "verify_portable_refresh", ROOT / "scripts" / "verify-portable.py" - ) - module = importlib.util.module_from_spec(loader) - loader.loader.exec_module(module) - return module - - def test_refresh_static_is_partial_and_never_rounds_a_failure_up( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -638,379 +583,3 @@ def failing_doctor(self, check_durability: bool = True, allow_stale: bool = Fals assert failed["overall_verdict"] == "FAIL" -@pytest.mark.parametrize("shape", ["[]", '{"version": 1, "hosts": []}']) -def test_a_wrong_shaped_evidence_index_is_refused_not_crashed(tmp_path: Path, shape: str) -> None: - from tests.test_evidence_settlement import _omp_live_root, _proof_bar_refresh - - root, prof, evidence_root, _ = _omp_live_root(tmp_path) - durable = evidence_root / "omp-receipt.json" - index_file = evidence_root / "index.json" - first, first_receipt = _proof_bar_refresh(root, evidence_root) - first.evidence(first_receipt, update_durable=True) - durable_before = durable.read_bytes() - index_file.write_text(shape, encoding="utf-8") - - second, second_receipt = _proof_bar_refresh(root, evidence_root) - with pytest.raises(SystemExit) as exit_info: - second.evidence(second_receipt, update_durable=True) - - assert "REFUSING to rewrite the unreadable evidence index" in str(exit_info.value.code) - assert str(index_file) in str(exit_info.value.code) - assert durable.read_bytes() == durable_before - assert index_file.read_text(encoding="utf-8") == shape - run_receipt = json.loads((second.run_dir / "receipt.json").read_text(encoding="utf-8")) - assert run_receipt["run_id"] == second_receipt.run_id != first_receipt.run_id - - -def test_a_corrupt_evidence_index_is_refused_not_reset(tmp_path: Path) -> None: - from tests.test_evidence_settlement import _omp_live_root, _proof_bar_refresh - - root, prof, evidence_root, _ = _omp_live_root(tmp_path) - index_file = evidence_root / "index.json" - index_file.write_text( - json.dumps({"version": 1, "hosts": {"grok": {"run_id": "grok-run"}}}), encoding="utf-8" - ) - corrupt_bytes = b'{ "version": 1, "hosts": { "grok": { "run_id": "grok-run"' - index_file.write_bytes(corrupt_bytes) - verifier, receipt = _proof_bar_refresh(root, evidence_root) - - with pytest.raises(SystemExit) as exit_info: - verifier.evidence(receipt, update_durable=True) - - assert str(index_file) in str(exit_info.value.code) - assert index_file.read_bytes() == corrupt_bytes - - -def test_a_stat_denied_evidence_index_is_refused_not_crashed(tmp_path: Path) -> None: - from tests.test_evidence_settlement import _omp_live_root, _proof_bar_refresh - - root, prof, evidence_root, _ = _omp_live_root(tmp_path) - durable = evidence_root / "omp-receipt.json" - index_file = evidence_root / "index.json" - first, first_receipt = _proof_bar_refresh(root, evidence_root) - first.evidence(first_receipt, update_durable=True) - durable_before = durable.read_bytes() - index_before = index_file.read_bytes() - - denied = tmp_path / "denied" - denied.mkdir() - real_index = denied / "index.json" - real_index.write_bytes(index_before) - index_file.unlink() - index_file.symlink_to(real_index) - denied.chmod(0o000) - - second, second_receipt = _proof_bar_refresh(root, evidence_root) - try: - with pytest.raises(SystemExit) as exit_info: - second.evidence(second_receipt, update_durable=True) - assert "REFUSING to rewrite the unreadable evidence index" in str(exit_info.value.code) - assert str(index_file) in str(exit_info.value.code) - assert durable.read_bytes() == durable_before - assert (second.run_dir / "receipt.json").is_file() - finally: - denied.chmod(0o755) - - assert real_index.read_bytes() == index_before - - -def test_a_stat_denied_durable_receipt_is_refused_not_crashed(tmp_path: Path) -> None: - from tests.test_evidence_settlement import _omp_live_root, _proof_bar_refresh - - root, prof, evidence_root, _ = _omp_live_root(tmp_path) - durable = evidence_root / "omp-receipt.json" - index_file = evidence_root / "index.json" - first, first_receipt = _proof_bar_refresh(root, evidence_root) - first.evidence(first_receipt, update_durable=True) - durable_before = durable.read_bytes() - index_before = index_file.read_bytes() - - denied = tmp_path / "denied" - denied.mkdir() - real_durable = denied / "omp-receipt.json" - real_durable.write_bytes(durable_before) - durable.unlink() - durable.symlink_to(real_durable) - denied.chmod(0o000) - - second, second_receipt = _proof_bar_refresh(root, evidence_root) - try: - second.evidence(second_receipt, update_durable=True) - finally: - denied.chmod(0o755) - - assert durable.read_bytes() == durable_before - assert index_file.read_bytes() == index_before - assert (second.run_dir / "receipt.json").is_file() - - -def _stat_denied_copy(tmp_path: Path, relative: str) -> tuple: - copy_root = _disposable_copy(tmp_path) - link = copy_root / relative - denied = tmp_path / "denied" - denied.mkdir() - real = denied / Path(relative).name - real.write_bytes(link.read_bytes()) - link.unlink() - link.symlink_to(real) - denied.chmod(0o000) - return copy_root, link, denied - - -def _run_gate(copy_root: Path, *arguments: str) -> subprocess.CompletedProcess: - return subprocess.run( - [sys.executable, str(copy_root / "scripts" / "verify-portable.py"), *arguments], - capture_output=True, - text=True, - cwd=copy_root, - ) - - -def test_a_stat_denied_profile_is_refused_not_reported_absent(tmp_path: Path) -> None: - copy_root, _, denied = _stat_denied_copy(tmp_path, "profiles/grok.json") - try: - res = _run_gate(copy_root, "check-evidence", "--host", "grok") - finally: - denied.chmod(0o755) - - assert res.returncode == 1, res.stdout + res.stderr - assert "Traceback" not in res.stderr - assert "Profile not found" not in res.stdout - assert "profiles/grok.json" in res.stdout and "Permission denied" in res.stdout - - -def test_a_stat_denied_staleness_receipt_is_refused_not_reported_absent(tmp_path: Path) -> None: - copy_root, _, denied = _stat_denied_copy(tmp_path, ".audit/evidence/grok-receipt.json") - try: - res = _run_gate(copy_root, "check-staleness", "--host", "grok") - finally: - denied.chmod(0o755) - - assert res.returncode == 1, res.stdout + res.stderr - assert "Traceback" not in res.stderr - assert "not found" not in res.stdout - assert "grok-receipt.json" in res.stdout and "Permission denied" in res.stdout - - -def test_a_stat_denied_recorded_receipt_is_refused_not_reported_missing(tmp_path: Path) -> None: - copy_root, _, denied = _stat_denied_copy(tmp_path, ".audit/evidence/grok-receipt.json") - try: - res = _run_gate(copy_root, "check-evidence", "--host", "grok") - finally: - denied.chmod(0o755) - - assert res.returncode == 1, res.stdout + res.stderr - assert "Traceback" not in res.stderr - assert "is missing" not in res.stdout - assert "cannot be read" in res.stdout and "grok-receipt.json" in res.stdout - assert "Permission denied" in res.stdout - - -def test_a_missing_evidence_index_is_initialized_for_this_host(tmp_path: Path) -> None: - """A fresh evidence root has no inventory to protect, so the first host creates it.""" - from tests.test_evidence_settlement import _omp_live_root, _proof_bar_refresh - - root, prof, evidence_root, _ = _omp_live_root(tmp_path) - index_file = evidence_root / "index.json" - assert not index_file.is_file() - verifier, receipt = _proof_bar_refresh(root, evidence_root) - - verifier.evidence(receipt, update_durable=True) - - index = json.loads(index_file.read_text(encoding="utf-8")) - assert sorted(index["hosts"]) == ["omp"] - assert index["hosts"]["omp"]["run_id"] == receipt.run_id - - -def test_repeated_refreshes_keep_the_run_that_observed(tmp_path: Path) -> None: - from tests.test_evidence_settlement import _omp_live_root, _proof_bar_refresh - - root, prof, evidence_root, _ = _omp_live_root(tmp_path) - durable_file = evidence_root / "omp-receipt.json" - index_file = evidence_root / "index.json" - - first, first_receipt = _proof_bar_refresh(root, evidence_root) - first.evidence(first_receipt, update_durable=True) - first_durable = json.loads(durable_file.read_text(encoding="utf-8")) - assert first_durable["run_id"] == first_receipt.run_id - - second, second_receipt = _proof_bar_refresh(root, evidence_root) - second.evidence(second_receipt, update_durable=True) - second_durable = json.loads(durable_file.read_text(encoding="utf-8")) - assert second_durable["run_id"] == second_receipt.run_id - second_scenario = next(s for s in second_durable["scenarios"] if s["id"] == "runtime-child-spawn-omp") - assert second_scenario["preserved_from"] == "previous-live-run" - - -def test_check_evidence_names_the_omp_live_remedy(tmp_path: Path) -> None: - copy_root = _disposable_copy(tmp_path) - profile = json.loads((copy_root / "profiles" / "omp.json").read_text(encoding="utf-8")) - spawn = next(b for b in profile["bindings"] if b["capability"] == "agent.spawn") - assert spawn["verification_status"] == "verified" - assert spawn["evidence_ref"]["scenario"] == "runtime-child-spawn-omp" - - receipt_path = copy_root / ".audit" / "evidence" / "omp-receipt.json" - receipt = json.loads(receipt_path.read_text(encoding="utf-8")) - live_rows = [s for s in receipt["scenarios"] if s["evidence_level"] == "observed_live"] - assert {s["id"] for s in live_rows} == { - "runtime-child-spawn-omp", - "runtime-child-cancel-omp", - "runtime-isolation-omp", - } - for scenario in live_rows: - scenario["evidence_level"] = "static" - scenario["settlement"] = None - receipt["planes"]["runtime"] = "UNTESTED" - receipt["overall_verdict"] = "UNTESTED" - receipt_path.write_text(json.dumps(receipt), encoding="utf-8") - - proc = subprocess.run( - [sys.executable, str(copy_root / "scripts" / "verify-portable.py"), "check-evidence", "--host", "omp"], - capture_output=True, - text=True, - cwd=copy_root, - ) - assert proc.returncode == 1, proc.stdout + proc.stderr - assert "carries no observed_live settlement" in proc.stdout - assert "PSTACK_RUN_OMP_LIVE=1" in proc.stdout - - -def test_a_corrupt_index_refusal_leaves_the_durable_receipt_untouched(tmp_path: Path) -> None: - from tests.test_evidence_settlement import _omp_live_root, _proof_bar_refresh - - root, prof, evidence_root, _ = _omp_live_root(tmp_path) - verifier, receipt = _proof_bar_refresh(root, evidence_root) - verifier.evidence(receipt, update_durable=True) - - durable_file = evidence_root / "omp-receipt.json" - index_file = evidence_root / "index.json" - durable_before = durable_file.read_bytes() - assert json.loads(index_file.read_text(encoding="utf-8"))["hosts"]["omp"]["run_id"] == json.loads( - durable_before - )["run_id"] - - corrupt = b'{ "version": 1, "hosts": { BAD' - index_file.write_bytes(corrupt) - second, second_receipt = _proof_bar_refresh(root, evidence_root) - with pytest.raises(SystemExit): - second.evidence(second_receipt, update_durable=True) - - assert durable_file.read_bytes() == durable_before - assert index_file.read_bytes() == corrupt - assert json.loads((second.run_dir / "receipt.json").read_text(encoding="utf-8"))["run_id"] == second_receipt.run_id - - -def test_check_evidence_cli_refuses_a_profile_that_violates_a_domain_rule(tmp_path: Path) -> None: - copy_root = _disposable_copy(tmp_path) - command = [ - sys.executable, - str(copy_root / "scripts" / "verify-portable.py"), - "check-evidence", - "--host", - "grok", - ] - clean = subprocess.run(command, capture_output=True, text=True, cwd=copy_root) - assert clean.returncode == 0, clean.stdout + clean.stderr - - profile_file = copy_root / "profiles" / "grok.json" - d = json.loads(profile_file.read_text(encoding="utf-8")) - d["support_state"] = "supported" - for binding in d["bindings"]: - binding["verification_status"] = "verified" - if binding["capability"] == "workspace.shared": - binding["implementation_status"] = "gap" - profile_file.write_text(json.dumps(d, indent=2) + "\n", encoding="utf-8") - - res = subprocess.run(command, capture_output=True, text=True, cwd=copy_root) - assert res.returncode == 1, res.stdout + res.stderr - assert "[grok] FAIL" in res.stdout - assert "cannot have gap on mandatory capability workspace.shared" in res.stdout - - -def test_a_missing_index_is_rebuilt_from_the_durable_receipts(tmp_path: Path) -> None: - copy_root = _disposable_copy(tmp_path) - command = [sys.executable, str(copy_root / "scripts" / "verify-portable.py")] - index_file = copy_root / ".audit" / "evidence" / "index.json" - - index_file.unlink() - named = subprocess.run(command + ["check-evidence", "--host", "grok"], capture_output=True, text=True, cwd=copy_root) - assert named.returncode == 1, named.stdout + named.stderr - assert "is missing" in named.stdout and "refresh-static --host all" in named.stdout - - rebuilt = subprocess.run(command + ["refresh-static", "--host", "grok"], capture_output=True, text=True, cwd=copy_root) - assert "Traceback" not in (rebuilt.stdout + rebuilt.stderr), rebuilt.stdout + rebuilt.stderr - hosts = json.loads(index_file.read_text(encoding="utf-8"))["hosts"] - assert sorted(hosts) == ["antigravity", "codex", "grok", "omp", "opencode"] - for host, row in hosts.items(): - receipt = json.loads((copy_root / row["receipt"]).read_text(encoding="utf-8")) - assert row["run_id"] == receipt["run_id"], host - assert row["timestamp"] == receipt["timestamp"], host - assert row["planes"] == receipt["planes"], host - assert row["surface_revisions"] == receipt["surface_revisions"], host - - all_hosts = subprocess.run(command + ["check-evidence", "--host", "all"], capture_output=True, text=True, cwd=copy_root) - assert all_hosts.returncode == 0, all_hosts.stdout + all_hosts.stderr - - -def test_the_projection_skips_receipt_bytes_a_row_cannot_stand_on(tmp_path: Path) -> None: - copy_root = _disposable_copy(tmp_path) - command = [sys.executable, str(copy_root / "scripts" / "verify-portable.py")] - index_file = copy_root / ".audit" / "evidence" / "index.json" - - index_file.unlink() - (copy_root / ".audit" / "evidence" / "codex-receipt.json").write_text('{"run_id": "x"}\n', encoding="utf-8") - rebuilt = subprocess.run(command + ["refresh-static", "--host", "grok"], capture_output=True, text=True, cwd=copy_root) - assert "Traceback" not in (rebuilt.stdout + rebuilt.stderr), rebuilt.stdout + rebuilt.stderr - hosts = sorted(json.loads(index_file.read_text(encoding="utf-8"))["hosts"]) - assert hosts == ["antigravity", "grok", "omp", "opencode"] - - orphan = subprocess.run(command + ["check-evidence", "--host", "codex"], capture_output=True, text=True, cwd=copy_root) - assert orphan.returncode == 1, orphan.stdout + orphan.stderr - assert "codex-receipt.json" in orphan.stdout - - -def test_a_missing_row_comes_back_from_the_receipt_on_the_next_run(tmp_path: Path) -> None: - copy_root = _disposable_copy(tmp_path) - command = [sys.executable, str(copy_root / "scripts" / "verify-portable.py")] - index_file = copy_root / ".audit" / "evidence" / "index.json" - - index = json.loads(index_file.read_text(encoding="utf-8")) - del index["hosts"]["grok"] - del index["hosts"]["codex"] - index_file.write_text(json.dumps(index, indent=2) + "\n", encoding="utf-8") - - reported = subprocess.run(command + ["check-evidence", "--host", "grok"], capture_output=True, text=True, cwd=copy_root) - assert reported.returncode == 1, reported.stdout + reported.stderr - assert "STALE/REBUILD-REQUIRED" in reported.stdout - assert json.loads(index_file.read_text(encoding="utf-8"))["hosts"].keys() == {"omp", "opencode", "antigravity"} - - rebuilt = subprocess.run(command + ["refresh-static", "--host", "grok"], capture_output=True, text=True, cwd=copy_root) - assert "Traceback" not in (rebuilt.stdout + rebuilt.stderr), rebuilt.stdout + rebuilt.stderr - hosts = json.loads(index_file.read_text(encoding="utf-8"))["hosts"] - assert sorted(hosts) == ["antigravity", "codex", "grok", "omp", "opencode"] - for host, row in hosts.items(): - receipt = json.loads((copy_root / row["receipt"]).read_text(encoding="utf-8")) - assert row["run_id"] == receipt["run_id"] and row["planes"] == receipt["planes"], host - - green = subprocess.run(command + ["check-evidence", "--host", "all"], capture_output=True, text=True, cwd=copy_root) - assert green.returncode == 0, green.stdout + green.stderr - - -def test_a_denied_index_write_is_named_not_raised(tmp_path: Path) -> None: - copy_root = _disposable_copy(tmp_path) - command = [sys.executable, str(copy_root / "scripts" / "verify-portable.py")] - index_file = copy_root / ".audit" / "evidence" / "index.json" - before = index_file.read_bytes() - - index_file.chmod(0o444) - try: - denied = subprocess.run(command + ["refresh-static", "--host", "grok"], capture_output=True, text=True, cwd=copy_root) - finally: - index_file.chmod(0o644) - - output = denied.stdout + denied.stderr - assert denied.returncode == 1, output - assert "FAILED to write the evidence index" in output - assert "Traceback" not in output - assert index_file.read_bytes() == before - assert "grok-receipt.json" in output diff --git a/tests/test_evidence_index_resilience.py b/tests/test_evidence_index_resilience.py new file mode 100644 index 00000000..eb25e246 --- /dev/null +++ b/tests/test_evidence_index_resilience.py @@ -0,0 +1,358 @@ +from __future__ import annotations + +import json +from pathlib import Path +import subprocess +import sys + +import pytest + +from tests.evidence_test_helpers import ( + _disposable_copy, + _omp_live_root, + _proof_bar_refresh, + _run_gate, + _stat_denied_copy, +) + + +@pytest.mark.parametrize("shape", ["[]", '{"version": 1, "hosts": []}']) +def test_a_wrong_shaped_evidence_index_is_refused_not_crashed(tmp_path: Path, shape: str) -> None: + root, prof, evidence_root, _ = _omp_live_root(tmp_path) + durable = evidence_root / "omp-receipt.json" + index_file = evidence_root / "index.json" + first, first_receipt = _proof_bar_refresh(root, evidence_root) + first.evidence(first_receipt, update_durable=True) + durable_before = durable.read_bytes() + index_file.write_text(shape, encoding="utf-8") + + second, second_receipt = _proof_bar_refresh(root, evidence_root) + with pytest.raises(SystemExit) as exit_info: + second.evidence(second_receipt, update_durable=True) + + assert "REFUSING to rewrite the unreadable evidence index" in str(exit_info.value.code) + assert str(index_file) in str(exit_info.value.code) + assert durable.read_bytes() == durable_before + assert index_file.read_text(encoding="utf-8") == shape + run_receipt = json.loads((second.run_dir / "receipt.json").read_text(encoding="utf-8")) + assert run_receipt["run_id"] == second_receipt.run_id != first_receipt.run_id + + +def test_a_corrupt_evidence_index_is_refused_not_reset(tmp_path: Path) -> None: + root, prof, evidence_root, _ = _omp_live_root(tmp_path) + index_file = evidence_root / "index.json" + index_file.write_text( + json.dumps({"version": 1, "hosts": {"grok": {"run_id": "grok-run"}}}), encoding="utf-8" + ) + corrupt_bytes = b'{ "version": 1, "hosts": { "grok": { "run_id": "grok-run"' + index_file.write_bytes(corrupt_bytes) + verifier, receipt = _proof_bar_refresh(root, evidence_root) + + with pytest.raises(SystemExit) as exit_info: + verifier.evidence(receipt, update_durable=True) + + assert str(index_file) in str(exit_info.value.code) + assert index_file.read_bytes() == corrupt_bytes + + +def test_a_stat_denied_evidence_index_is_refused_not_crashed(tmp_path: Path) -> None: + root, prof, evidence_root, _ = _omp_live_root(tmp_path) + durable = evidence_root / "omp-receipt.json" + index_file = evidence_root / "index.json" + first, first_receipt = _proof_bar_refresh(root, evidence_root) + first.evidence(first_receipt, update_durable=True) + durable_before = durable.read_bytes() + index_before = index_file.read_bytes() + + denied = tmp_path / "denied" + denied.mkdir() + real_index = denied / "index.json" + real_index.write_bytes(index_before) + index_file.unlink() + index_file.symlink_to(real_index) + denied.chmod(0o000) + + second, second_receipt = _proof_bar_refresh(root, evidence_root) + try: + with pytest.raises(SystemExit) as exit_info: + second.evidence(second_receipt, update_durable=True) + assert "REFUSING to rewrite the unreadable evidence index" in str(exit_info.value.code) + assert str(index_file) in str(exit_info.value.code) + assert durable.read_bytes() == durable_before + assert (second.run_dir / "receipt.json").is_file() + finally: + denied.chmod(0o755) + + assert real_index.read_bytes() == index_before + + +def test_a_stat_denied_durable_receipt_is_refused_not_crashed(tmp_path: Path) -> None: + root, prof, evidence_root, _ = _omp_live_root(tmp_path) + durable = evidence_root / "omp-receipt.json" + index_file = evidence_root / "index.json" + first, first_receipt = _proof_bar_refresh(root, evidence_root) + first.evidence(first_receipt, update_durable=True) + durable_before = durable.read_bytes() + index_before = index_file.read_bytes() + + denied = tmp_path / "denied" + denied.mkdir() + real_durable = denied / "omp-receipt.json" + real_durable.write_bytes(durable_before) + durable.unlink() + durable.symlink_to(real_durable) + denied.chmod(0o000) + + second, second_receipt = _proof_bar_refresh(root, evidence_root) + try: + second.evidence(second_receipt, update_durable=True) + finally: + denied.chmod(0o755) + + assert durable.read_bytes() == durable_before + assert index_file.read_bytes() == index_before + assert (second.run_dir / "receipt.json").is_file() + + +def test_a_stat_denied_profile_is_refused_not_reported_absent(tmp_path: Path) -> None: + copy_root, _, denied = _stat_denied_copy(tmp_path, "profiles/grok.json") + try: + res = _run_gate(copy_root, "check-evidence", "--host", "grok") + finally: + denied.chmod(0o755) + + assert res.returncode == 1, res.stdout + res.stderr + assert "Traceback" not in res.stderr + assert "Profile not found" not in res.stdout + assert "profiles/grok.json" in res.stdout and "Permission denied" in res.stdout + + +def test_a_stat_denied_staleness_receipt_is_refused_not_reported_absent(tmp_path: Path) -> None: + copy_root, _, denied = _stat_denied_copy(tmp_path, ".audit/evidence/grok-receipt.json") + try: + res = _run_gate(copy_root, "check-staleness", "--host", "grok") + finally: + denied.chmod(0o755) + + assert res.returncode == 1, res.stdout + res.stderr + assert "Traceback" not in res.stderr + assert "not found" not in res.stdout + assert "grok-receipt.json" in res.stdout and "Permission denied" in res.stdout + + +def test_a_stat_denied_recorded_receipt_is_refused_not_reported_missing(tmp_path: Path) -> None: + copy_root, _, denied = _stat_denied_copy(tmp_path, ".audit/evidence/grok-receipt.json") + try: + res = _run_gate(copy_root, "check-evidence", "--host", "grok") + finally: + denied.chmod(0o755) + + assert res.returncode == 1, res.stdout + res.stderr + assert "Traceback" not in res.stderr + assert "is missing" not in res.stdout + assert "cannot be read" in res.stdout and "grok-receipt.json" in res.stdout + assert "Permission denied" in res.stdout + + +def test_a_missing_evidence_index_is_initialized_for_this_host(tmp_path: Path) -> None: + """A fresh evidence root has no inventory to protect, so the first host creates it.""" + root, prof, evidence_root, _ = _omp_live_root(tmp_path) + index_file = evidence_root / "index.json" + assert not index_file.is_file() + verifier, receipt = _proof_bar_refresh(root, evidence_root) + + verifier.evidence(receipt, update_durable=True) + + index = json.loads(index_file.read_text(encoding="utf-8")) + assert sorted(index["hosts"]) == ["omp"] + assert index["hosts"]["omp"]["run_id"] == receipt.run_id + + +def test_repeated_refreshes_keep_the_run_that_observed(tmp_path: Path) -> None: + root, prof, evidence_root, _ = _omp_live_root(tmp_path) + durable_file = evidence_root / "omp-receipt.json" + index_file = evidence_root / "index.json" + + first, first_receipt = _proof_bar_refresh(root, evidence_root) + first.evidence(first_receipt, update_durable=True) + first_durable = json.loads(durable_file.read_text(encoding="utf-8")) + assert first_durable["run_id"] == first_receipt.run_id + + second, second_receipt = _proof_bar_refresh(root, evidence_root) + second.evidence(second_receipt, update_durable=True) + second_durable = json.loads(durable_file.read_text(encoding="utf-8")) + assert second_durable["run_id"] == second_receipt.run_id + second_scenario = next(s for s in second_durable["scenarios"] if s["id"] == "runtime-child-spawn-omp") + assert second_scenario["preserved_from"] == "previous-live-run" + + +def test_check_evidence_names_the_omp_live_remedy(tmp_path: Path) -> None: + copy_root = _disposable_copy(tmp_path) + profile = json.loads((copy_root / "profiles" / "omp.json").read_text(encoding="utf-8")) + spawn = next(b for b in profile["bindings"] if b["capability"] == "agent.spawn") + assert spawn["verification_status"] == "verified" + assert spawn["evidence_ref"]["scenario"] == "runtime-child-spawn-omp" + + receipt_path = copy_root / ".audit" / "evidence" / "omp-receipt.json" + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + live_rows = [s for s in receipt["scenarios"] if s["evidence_level"] == "observed_live"] + assert {s["id"] for s in live_rows} == { + "runtime-child-spawn-omp", + "runtime-child-cancel-omp", + "runtime-isolation-omp", + } + for scenario in live_rows: + scenario["evidence_level"] = "static" + scenario["settlement"] = None + receipt["planes"]["runtime"] = "UNTESTED" + receipt["overall_verdict"] = "UNTESTED" + receipt_path.write_text(json.dumps(receipt), encoding="utf-8") + + proc = subprocess.run( + [sys.executable, str(copy_root / "scripts" / "verify-portable.py"), "check-evidence", "--host", "omp"], + capture_output=True, + text=True, + cwd=copy_root, + ) + assert proc.returncode == 1, proc.stdout + proc.stderr + assert "carries no observed_live settlement" in proc.stdout + assert "PSTACK_RUN_OMP_LIVE=1" in proc.stdout + + +def test_a_corrupt_index_refusal_leaves_the_durable_receipt_untouched(tmp_path: Path) -> None: + root, prof, evidence_root, _ = _omp_live_root(tmp_path) + verifier, receipt = _proof_bar_refresh(root, evidence_root) + verifier.evidence(receipt, update_durable=True) + + durable_file = evidence_root / "omp-receipt.json" + index_file = evidence_root / "index.json" + durable_before = durable_file.read_bytes() + assert json.loads(index_file.read_text(encoding="utf-8"))["hosts"]["omp"]["run_id"] == json.loads( + durable_before + )["run_id"] + + corrupt = b'{ "version": 1, "hosts": { BAD' + index_file.write_bytes(corrupt) + second, second_receipt = _proof_bar_refresh(root, evidence_root) + with pytest.raises(SystemExit): + second.evidence(second_receipt, update_durable=True) + + assert durable_file.read_bytes() == durable_before + assert index_file.read_bytes() == corrupt + assert json.loads((second.run_dir / "receipt.json").read_text(encoding="utf-8"))["run_id"] == second_receipt.run_id + + +def test_check_evidence_cli_refuses_a_profile_that_violates_a_domain_rule(tmp_path: Path) -> None: + copy_root = _disposable_copy(tmp_path) + command = [ + sys.executable, + str(copy_root / "scripts" / "verify-portable.py"), + "check-evidence", + "--host", + "grok", + ] + clean = subprocess.run(command, capture_output=True, text=True, cwd=copy_root) + assert clean.returncode == 0, clean.stdout + clean.stderr + + profile_file = copy_root / "profiles" / "grok.json" + d = json.loads(profile_file.read_text(encoding="utf-8")) + d["support_state"] = "supported" + for binding in d["bindings"]: + binding["verification_status"] = "verified" + if binding["capability"] == "workspace.shared": + binding["implementation_status"] = "gap" + profile_file.write_text(json.dumps(d, indent=2) + "\n", encoding="utf-8") + + res = subprocess.run(command, capture_output=True, text=True, cwd=copy_root) + assert res.returncode == 1, res.stdout + res.stderr + assert "[grok] FAIL" in res.stdout + assert "cannot have gap on mandatory capability workspace.shared" in res.stdout + + +def test_a_missing_index_is_rebuilt_from_the_durable_receipts(tmp_path: Path) -> None: + copy_root = _disposable_copy(tmp_path) + command = [sys.executable, str(copy_root / "scripts" / "verify-portable.py")] + index_file = copy_root / ".audit" / "evidence" / "index.json" + + index_file.unlink() + named = subprocess.run(command + ["check-evidence", "--host", "grok"], capture_output=True, text=True, cwd=copy_root) + assert named.returncode == 1, named.stdout + named.stderr + assert "is missing" in named.stdout and "refresh-static --host all" in named.stdout + + rebuilt = subprocess.run(command + ["refresh-static", "--host", "grok"], capture_output=True, text=True, cwd=copy_root) + assert "Traceback" not in (rebuilt.stdout + rebuilt.stderr), rebuilt.stdout + rebuilt.stderr + hosts = json.loads(index_file.read_text(encoding="utf-8"))["hosts"] + assert sorted(hosts) == ["antigravity", "codex", "grok", "omp", "opencode"] + for host, row in hosts.items(): + receipt = json.loads((copy_root / row["receipt"]).read_text(encoding="utf-8")) + assert row["run_id"] == receipt["run_id"], host + assert row["timestamp"] == receipt["timestamp"], host + assert row["planes"] == receipt["planes"], host + assert row["surface_revisions"] == receipt["surface_revisions"], host + + all_hosts = subprocess.run(command + ["check-evidence", "--host", "all"], capture_output=True, text=True, cwd=copy_root) + assert all_hosts.returncode == 0, all_hosts.stdout + all_hosts.stderr + + +def test_the_projection_skips_receipt_bytes_a_row_cannot_stand_on(tmp_path: Path) -> None: + copy_root = _disposable_copy(tmp_path) + command = [sys.executable, str(copy_root / "scripts" / "verify-portable.py")] + index_file = copy_root / ".audit" / "evidence" / "index.json" + + index_file.unlink() + (copy_root / ".audit" / "evidence" / "codex-receipt.json").write_text('{"run_id": "x"}\n', encoding="utf-8") + rebuilt = subprocess.run(command + ["refresh-static", "--host", "grok"], capture_output=True, text=True, cwd=copy_root) + assert "Traceback" not in (rebuilt.stdout + rebuilt.stderr), rebuilt.stdout + rebuilt.stderr + hosts = sorted(json.loads(index_file.read_text(encoding="utf-8"))["hosts"]) + assert hosts == ["antigravity", "grok", "omp", "opencode"] + + orphan = subprocess.run(command + ["check-evidence", "--host", "codex"], capture_output=True, text=True, cwd=copy_root) + assert orphan.returncode == 1, orphan.stdout + orphan.stderr + assert "codex-receipt.json" in orphan.stdout + + +def test_a_missing_row_comes_back_from_the_receipt_on_the_next_run(tmp_path: Path) -> None: + copy_root = _disposable_copy(tmp_path) + command = [sys.executable, str(copy_root / "scripts" / "verify-portable.py")] + index_file = copy_root / ".audit" / "evidence" / "index.json" + + index = json.loads(index_file.read_text(encoding="utf-8")) + del index["hosts"]["grok"] + del index["hosts"]["codex"] + index_file.write_text(json.dumps(index, indent=2) + "\n", encoding="utf-8") + + reported = subprocess.run(command + ["check-evidence", "--host", "grok"], capture_output=True, text=True, cwd=copy_root) + assert reported.returncode == 1, reported.stdout + reported.stderr + assert "STALE/REBUILD-REQUIRED" in reported.stdout + assert json.loads(index_file.read_text(encoding="utf-8"))["hosts"].keys() == {"omp", "opencode", "antigravity"} + + rebuilt = subprocess.run(command + ["refresh-static", "--host", "grok"], capture_output=True, text=True, cwd=copy_root) + assert "Traceback" not in (rebuilt.stdout + rebuilt.stderr), rebuilt.stdout + rebuilt.stderr + hosts = json.loads(index_file.read_text(encoding="utf-8"))["hosts"] + assert sorted(hosts) == ["antigravity", "codex", "grok", "omp", "opencode"] + for host, row in hosts.items(): + receipt = json.loads((copy_root / row["receipt"]).read_text(encoding="utf-8")) + assert row["run_id"] == receipt["run_id"] and row["planes"] == receipt["planes"], host + + green = subprocess.run(command + ["check-evidence", "--host", "all"], capture_output=True, text=True, cwd=copy_root) + assert green.returncode == 0, green.stdout + green.stderr + + +def test_a_denied_index_write_is_named_not_raised(tmp_path: Path) -> None: + copy_root = _disposable_copy(tmp_path) + command = [sys.executable, str(copy_root / "scripts" / "verify-portable.py")] + index_file = copy_root / ".audit" / "evidence" / "index.json" + before = index_file.read_bytes() + + index_file.chmod(0o444) + try: + denied = subprocess.run(command + ["refresh-static", "--host", "grok"], capture_output=True, text=True, cwd=copy_root) + finally: + index_file.chmod(0o644) + + output = denied.stdout + denied.stderr + assert denied.returncode == 1, output + assert "FAILED to write the evidence index" in output + assert "Traceback" not in output + assert index_file.read_bytes() == before + assert "grok-receipt.json" in output diff --git a/tests/test_evidence_settlement.py b/tests/test_evidence_settlement.py index 7a0048ed..203d45bb 100644 --- a/tests/test_evidence_settlement.py +++ b/tests/test_evidence_settlement.py @@ -37,99 +37,19 @@ RECEIPT_SCHEMA = ROOT / "schemas" / "portability" / "receipt.schema.json" EVIDENCE_DIR = ROOT / ".audit" / "evidence" - -def _declared_from(scenarios: list) -> tuple: - """The planes and overall verdict a run's own scenarios reduce to. - - Fixtures declare planes the way a driver writes them — from the reduction — because - `VerificationReceipt.validate` refuses a declaration that floats free of its scenarios - (#473). A receipt that observed nothing declares UNTESTED, not PASS. - """ - planes = settle_planes(scenarios) - return planes, reduce_verdicts(planes.values()) - - -def _plane_rows(covered: set, verdict: str = "PASS") -> list: - """One `verdict` scenario per conformance plane not already covered by a fixture row, - so a fixture that claims a whole-receipt shape proves each plane it declares.""" - return [ - ScenarioResult( - id=f"synthetic-{plane}-plane-row", - description=f"synthetic {plane} plane row", - plane=plane, - command="true", - exit_code=0 if verdict == "PASS" else 1, - stdout_snippet=f"{plane} fixture row", - verdict=verdict, - duration_s=0.01, - evidence_level="offline", - ) - for plane in CONFORMANCE_PLANES - if plane not in covered - ] - - -def _synthetic_evidence_root(tmp_path: Path, evidence_levels: list, planes: list | None = None, host: str = "omp", snippets: list | None = None, settlements: list | None = None, force_verified: bool = False) -> tuple: - """Self-consistent minimal root: every OMP binding points at one synthetic receipt - whose scenario (evidence_level, plane) pairs are exactly zipped `evidence_levels`, - plus one row per remaining plane so the receipt's declared planes are its own - reduction instead of a hand-written PASS (#473).""" - tmp_path.mkdir(parents=True, exist_ok=True) - (tmp_path / "pstack.package.json").write_text("{}", encoding="utf-8") - manifest = tmp_path / ".omp-plugin" / "package.json" - manifest.parent.mkdir(parents=True, exist_ok=True) - manifest.write_text("{}", encoding="utf-8") - - revisions = compute_surface_revisions(tmp_path, host) - ref = ".audit/evidence/synthetic-receipt.json" - scenarios = [ - ScenarioResult( - id=f"synthetic-{i}", - description="synthetic scenario", - plane=(planes[i] if planes else "runtime"), - command="true", - exit_code=0, - stdout_snippet=(snippets[i] if snippets else ""), - verdict="PASS", - duration_s=0.01, - evidence_level=level, - settlement=(settlements[i] if settlements else None), - ) - for i, level in enumerate(evidence_levels) - ] - scenarios += _plane_rows({s.plane for s in scenarios}) - declared, overall = _declared_from(scenarios) - receipt = VerificationReceipt( - run_id="evidence-level-gate", - host=host, - host_version="test", - timestamp="2026-10-11T00:00:00Z", - overall_verdict=overall, - planes=declared, - surface_revisions=revisions, - scenarios=scenarios, - ) - receipt_path = tmp_path / ref - receipt_path.parent.mkdir(parents=True, exist_ok=True) - receipt_path.write_text(json.dumps(asdict(receipt)), encoding="utf-8") - - d = json.loads((ROOT / "profiles" / f"{host}.json").read_text(encoding="utf-8")) - for binding in d["bindings"]: - binding["evidence_ref"] = ref - if force_verified: - binding["verification_status"] = "verified" - prof = HarnessProfile( - host=d["host"], - support_state="verified", - bindings=[Binding(**x) for x in d["bindings"]], - skills_dir=d.get("skills_dir"), - plugins_dir=d.get("plugins_dir"), - plugin_manifest=d.get("plugin_manifest"), - tool_mappings=d.get("tool_mappings"), - skill_order=d.get("skill_order"), - runtime_conventions=d.get("runtime_conventions"), - ) - return tmp_path, prof +from tests.evidence_test_helpers import ( + _adversarial_receipt, + _declared_from, + _observed_scenario, + _omp_live_root, + _omp_spawn_scenario, + _plane_rows, + _portable_verifier, + _profile_pinning, + _proof_bar_refresh, + _success_settlement, + _synthetic_evidence_root, +) def test_offline_only_receipt_derives_packaged_not_verified(tmp_path: Path) -> None: @@ -372,102 +292,6 @@ def test_driver_settlement_survives_receipt_serialization(tmp_path: Path) -> Non assert parsed["settlement"][key] -def _omp_spawn_scenario(evidence_level: str, verdict: str = "PASS", settlement=None) -> ScenarioResult: - return ScenarioResult( - id="runtime-child-spawn-omp", - description="bounded omp rpc v2 child spawn and join", - plane="runtime", - command="PSTACK_RUN_OMP_LIVE=1 python -m pytest tests/test_omp_rpc.py", - exit_code=0 if verdict == "PASS" else 1, - stdout_snippet="child=subagent-1 outcome=success" if settlement else "static evidence", - verdict=verdict, - duration_s=19.1, - evidence_level=evidence_level, - settlement=settlement, - ) - - -def _omp_live_root(tmp_path: Path) -> tuple: - root = tmp_path / "repo" - (root / ".omp-plugin").mkdir(parents=True) - (root / "pstack.package.json").write_text("{}", encoding="utf-8") - (root / ".omp-plugin" / "package.json").write_text("{}", encoding="utf-8") - evidence_root = root / ".audit" / "evidence" - evidence_root.mkdir(parents=True) - revisions = compute_surface_revisions(root, "omp") - scenarios = [ - _omp_spawn_scenario( - "observed_live", - settlement=ChildSettlement( - child_id="subagent-1", - terminal_outcome="success", - model="deepseek-v4-pro", - transport="rpc", - effective_rpc="rpc-v2:profile=disposable:depth=1", - ), - ) - ] + _plane_rows({"runtime"}) - declared, overall = _declared_from(scenarios) - live = VerificationReceipt( - run_id="previous-live-run", - host="omp", - host_version="18.8.9", - timestamp="2026-10-11T20:00:00Z", - overall_verdict=overall, - planes=declared, - surface_revisions=revisions, - scenarios=scenarios, - ) - (evidence_root / "omp-receipt.json").write_text(json.dumps(asdict(live)), encoding="utf-8") - - profile_data = json.loads((ROOT / "profiles" / "omp.json").read_text(encoding="utf-8")) - for binding in profile_data["bindings"]: - binding["evidence_ref"] = ".audit/evidence/omp-receipt.json" - binding["verification_status"] = "verified" - prof = HarnessProfile( - host="omp", - support_state="verified", - bindings=[Binding(**x) for x in profile_data["bindings"]], - skills_dir=profile_data.get("skills_dir"), - plugins_dir=profile_data.get("plugins_dir"), - plugin_manifest=profile_data.get("plugin_manifest"), - tool_mappings=profile_data.get("tool_mappings"), - skill_order=profile_data.get("skill_order"), - runtime_conventions=profile_data.get("runtime_conventions"), - ) - return root, prof, evidence_root, revisions - - -def _portable_verifier(evidence_root: Path, root: Path | None = None): - loader = importlib.util.spec_from_file_location("verify_portable_preserve", ROOT / "scripts" / "verify-portable.py") - module = importlib.util.module_from_spec(loader) - loader.loader.exec_module(module) - if root is not None: - module.ROOT = root - verifier = module.PortableVerifier(host="omp", evidence_root=evidence_root) - verifier.launch() - return verifier - - -def _proof_bar_refresh(root: Path, evidence_root: Path, *, verdict: str = "PASS") -> tuple: - verifier = _portable_verifier(evidence_root, root=root) - verifier.scenarios = [ - ScenarioResult( - id=f"doctor-{plane}-stub", - description=f"static {plane} plane", - plane=plane, - command="stub", - exit_code=0, - stdout_snippet="stub", - verdict="PASS", - duration_s=0.0, - ) - for plane in ("canonical", "adapter", "package") - ] - verifier.scenarios.append(_omp_spawn_scenario("static", verdict=verdict)) - return verifier, verifier.proof_bar() - - def test_durable_refresh_preserves_an_unchanged_observed_settlement(tmp_path: Path) -> None: root, prof, evidence_root, _ = _omp_live_root(tmp_path) verifier, receipt = _proof_bar_refresh(root, evidence_root) @@ -581,71 +405,6 @@ def test_observed_settlement_requires_a_passing_successful_scenario() -> None: ) -def _observed_scenario( - scenario_id: str, settlement: ChildSettlement | None = None -) -> ScenarioResult: - return ScenarioResult( - id=scenario_id, - description="adversarial observed_live runtime row", - plane="runtime", - command="omp --mode rpc", - exit_code=0, - stdout_snippet="observed_live row", - verdict="PASS", - duration_s=0.01, - evidence_level="observed_live", - settlement=settlement, - ) - - -def _success_settlement() -> ChildSettlement: - return ChildSettlement( - child_id="subagent-7", - terminal_outcome="success", - model="test-model", - transport="rpc", - effective_rpc="rpc-v2:profile=disposable:depth=1", - ) - - -def _adversarial_receipt(scenario: ScenarioResult) -> VerificationReceipt: - planes, overall = _declared_from([scenario]) - receipt = VerificationReceipt( - run_id="p8-adversarial", - host="omp", - host_version="test", - timestamp="2026-10-12T00:00:00Z", - overall_verdict=overall, - planes=planes, - surface_revisions=compute_surface_revisions(ROOT, "omp"), - scenarios=[scenario], - ) - receipt.validate() - return receipt - - -def _profile_pinning( - receipt_ref: str, scenario_id: str, capability: str = "agent.resume" -) -> HarnessProfile: - d = json.loads((ROOT / "profiles" / "omp.json").read_text(encoding="utf-8")) - for binding in d["bindings"]: - binding["evidence_ref"] = receipt_ref - if binding["capability"] == capability: - binding["verification_status"] = "verified" - binding["evidence_ref"] = {"receipt": receipt_ref, "scenario": scenario_id} - return HarnessProfile( - host=d["host"], - support_state=d["support_state"], - bindings=[Binding(**x) for x in d["bindings"]], - skills_dir=d.get("skills_dir"), - plugins_dir=d.get("plugins_dir"), - plugin_manifest=d.get("plugin_manifest"), - tool_mappings=d.get("tool_mappings"), - skill_order=d.get("skill_order"), - runtime_conventions=d.get("runtime_conventions"), - ) - - def test_a_session_pickup_never_satisfies_a_child_resume_pin() -> None: assert CONDITIONAL_CAPABILITIES["session.resume"] != CONDITIONAL_CAPABILITIES["agent.resume"] diff --git a/tests/test_select_checks.py b/tests/test_select_checks.py index 0bbec0c7..b424276a 100644 --- a/tests/test_select_checks.py +++ b/tests/test_select_checks.py @@ -118,18 +118,28 @@ def test_host_dependent_module_carries_the_documented_opt_in(self) -> None: def test_specific_script_rule_beats_the_scripts_fallback(self) -> None: result = vp.select_checks(["scripts/verify-harness.py"]) - assert _ids(result) == ["pytest:tests/test_verify_harness.py"] - assert _commands(result) == [[PY, "-m", "pytest", "-q", "tests/test_verify_harness.py"]] + assert _ids(result) == [ + "pytest:tests/test_verify_harness.py", + "pytest:tests/test_verify_harness_archive.py", + "pytest:tests/test_verify_harness_guides.py", + "pytest:tests/test_verify_harness_playbooks.py", + ] + assert _commands(result) == [ + [PY, "-m", "pytest", "-q", "tests/test_verify_harness.py"], + [PY, "-m", "pytest", "-q", "tests/test_verify_harness_archive.py"], + [PY, "-m", "pytest", "-q", "tests/test_verify_harness_guides.py"], + [PY, "-m", "pytest", "-q", "tests/test_verify_harness_playbooks.py"], + ] def test_decomposed_scripts_select_owning_modules_rather_than_scripts_fallback(self) -> None: decomposed = [ ("scripts/surface_revisions.py", ["pytest:tests/test_portability_schemas.py", "pytest:tests/test_surface_cache.py"]), - ("scripts/evidence_durability.py", ["pytest:tests/test_evidence_durability.py", "pytest:tests/test_evidence_settlement.py", "pytest:tests/test_portability_schemas.py"]), + ("scripts/evidence_durability.py", ["pytest:tests/test_evidence_durability.py", "pytest:tests/test_evidence_index_resilience.py", "pytest:tests/test_evidence_settlement.py", "pytest:tests/test_portability_schemas.py"]), ("scripts/portability_types.py", ["pytest:tests/test_evidence_durability.py", "pytest:tests/test_portability_schemas.py", "pytest:tests/test_verdict_reduction.py"]), ("scripts/harness_scenarios.py", ["pytest:tests/test_verify_portable.py"]), ("scripts/select_checks.py", ["pytest:tests/test_select_checks.py", "pytest:tests/test_verify_portable.py"]), ("scripts/openspec_change.py", ["pytest:tests/test_verify_portable.py"]), - ("scripts/evidence_index.py", ["pytest:tests/test_evidence_durability.py", "pytest:tests/test_verify_portable.py"]), + ("scripts/evidence_index.py", ["pytest:tests/test_evidence_durability.py", "pytest:tests/test_evidence_index_resilience.py", "pytest:tests/test_verify_portable.py"]), ("scripts/model_role_resolver.py", ["pytest:tests/test_model_role_policy.py", "pytest:tests/test_setup_pstack.py"]), ("scripts/model_role_policy.py", ["pytest:tests/test_model_role_policy.py", "pytest:tests/test_setup_pstack.py"]), ("scripts/setup_pstack.py", ["pytest:tests/test_model_role_policy.py", "pytest:tests/test_setup_pstack.py"]), diff --git a/tests/test_verify_harness.py b/tests/test_verify_harness.py index 8b7b0165..6a2d48c3 100644 --- a/tests/test_verify_harness.py +++ b/tests/test_verify_harness.py @@ -2,11 +2,10 @@ import importlib.util import json +from pathlib import Path import re import subprocess import sys -import tempfile -from pathlib import Path ROOT = Path(__file__).resolve().parents[1] SCANNER = ROOT / "scripts" / "verify-harness.py" @@ -48,13 +47,13 @@ def test_verify_harness_passes_on_this_tree() -> None: ) assert proc.returncode == 0, proc.stderr + proc.stdout val = subprocess.run( - ["grok", "plugin", "validate", str(ROOT)], + ["openspec", "validate", "--changes", "--strict"], + cwd=ROOT, capture_output=True, text=True, check=False, ) - assert val.returncode == 0, val.stderr + val.stdout - assert "Plugin manifest is valid" in val.stdout + assert val.returncode == 0, val.stderr or val.stdout def test_babysit_and_shipping_do_not_use_cursor_dynamic_loop() -> None: @@ -78,12 +77,6 @@ def test_babysit_and_shipping_do_not_use_cursor_dynamic_loop() -> None: assert "timeout_ms" in babysit -def test_verify_guide_names_monitor_boundary() -> None: - guide = (ROOT / "docs/guide/06-verify-and-ship.md").read_text(encoding="utf-8") - assert "bundled watcher" not in guide - assert "`monitor`" in guide - - def test_orchestrate_uses_canonical_host_state() -> None: orchestrate = ( ROOT / "skills/poteto-mode/playbooks/orchestrate.md" @@ -118,648 +111,6 @@ def test_host_manifests_keep_grok_parity_and_adapter_asymmetry() -> None: assert "./automations/benny-grok/skills/" not in json.dumps(claude) -def test_archivable_intent_chain_guard_distinguishes_changes() -> None: - loader = importlib.util.spec_from_file_location("verify_harness", SCANNER) - assert loader is not None - scanner = importlib.util.module_from_spec(loader) - assert loader.loader is not None - loader.loader.exec_module(scanner) - with tempfile.TemporaryDirectory() as tmp: - changes = Path(tmp) - complete = changes / "complete" - complete.mkdir() - (complete / "tasks.md").write_text("- [x] 1.1 done\n", encoding="utf-8") - incomplete = changes / "incomplete" - incomplete.mkdir() - (incomplete / "tasks.md").write_text("- [ ] 1.1 later\n", encoding="utf-8") - archived = changes / "archive" / "historical" - archived.mkdir(parents=True) - (archived / "tasks.md").write_text("- [x] 1.1 done\n", encoding="utf-8") - assert scanner.archivable_changes_missing_artifacts(changes) == [ - "complete/design.md", - "complete/adr.md", - ] - -def test_archived_intent_chains_are_complete() -> None: - loader = importlib.util.spec_from_file_location("verify_harness", SCANNER) - assert loader is not None - scanner = importlib.util.module_from_spec(loader) - assert loader.loader is not None - loader.loader.exec_module(scanner) - missing = scanner.archived_changes_missing_artifacts(ROOT / "openspec/changes/archive") - assert not missing, missing - - -def test_archived_change_missing_adr_artifact_is_flagged() -> None: - loader = importlib.util.spec_from_file_location("verify_harness", SCANNER) - assert loader is not None - scanner = importlib.util.module_from_spec(loader) - assert loader.loader is not None - loader.loader.exec_module(scanner) - with tempfile.TemporaryDirectory() as tmp: - archive = Path(tmp) - change = archive / "2026-10-13-pstack-fixture" - (change / "specs" / "capability").mkdir(parents=True) - (change / "specs" / "capability" / "spec.md").write_text( - "## Purpose\nfixture\n", encoding="utf-8" - ) - for artifact in ("proposal.md", "design.md", "tasks.md", ".openspec.yaml"): - (change / artifact).write_text("x\n", encoding="utf-8") - assert scanner.archived_changes_missing_artifacts(archive) == [ - "2026-10-13-pstack-fixture/adr.md" - ] - (change / "adr.md").write_text("x\n", encoding="utf-8") - assert scanner.archived_changes_missing_artifacts(archive) == [] - - -def test_archived_change_skip_specs_exempts_specs_only() -> None: - loader = importlib.util.spec_from_file_location("verify_harness", SCANNER) - assert loader is not None - scanner = importlib.util.module_from_spec(loader) - assert loader.loader is not None - loader.loader.exec_module(scanner) - with tempfile.TemporaryDirectory() as tmp: - archive = Path(tmp) - for name, manifest in ( - ("skipped", "schema: intent-driven\nskip_specs: true\n"), - ("enforced", "schema: intent-driven\n"), - ): - change = archive / f"2026-10-13-{name}" - change.mkdir(parents=True) - for artifact in ("proposal.md", "design.md", "adr.md", "tasks.md"): - (change / artifact).write_text("x\n", encoding="utf-8") - (change / ".openspec.yaml").write_text(manifest, encoding="utf-8") - assert scanner.archived_changes_missing_artifacts(archive) == [ - "2026-10-13-enforced/specs/**/*.md" - ] - # skip_specs exempts specs only; other required artifacts stay enforced. - (archive / "2026-10-13-skipped" / "adr.md").unlink() - assert scanner.archived_changes_missing_artifacts(archive) == [ - "2026-10-13-enforced/specs/**/*.md", - "2026-10-13-skipped/adr.md", - ] - - -def test_durable_specs_have_non_placeholder_purposes() -> None: - for spec in sorted((ROOT / "openspec/specs").glob("*/spec.md")): - text = spec.read_text(encoding="utf-8") - assert "## Purpose\n" in text - purpose = text.split("## Purpose\n", 1)[1].split("\n\n", 1)[0].strip() - assert purpose and not purpose.startswith("TBD -"), spec - - -def test_benny_guard_denies_compound_merge_and_plain_push() -> None: - script = ROOT / "automations/benny-grok/bin/fail-closed.sh" - for command in ( - "git merge feature && git push origin main", - "git push origin main && git merge feature", - ): - deny = subprocess.run( - ["sh", str(script)], - input=json.dumps({"toolInput": {"command": command}}), - capture_output=True, - text=True, - check=False, - ) - assert deny.returncode == 2, command - assert json.loads(deny.stdout)["decision"] == "deny", command - - force_lease = subprocess.run( - ["sh", str(script)], - input=json.dumps( - {"toolInput": {"command": "git push --force-with-lease origin main"}} - ), - capture_output=True, - text=True, - check=False, - ) - assert force_lease.returncode == 2 - assert json.loads(force_lease.stdout)["decision"] == "deny" - - -def test_autopilot_is_parent_fanout_and_skips_goal() -> None: - full = (ROOT / "skills/poteto-mode/playbooks/autopilot-full.md").read_text( - encoding="utf-8" - ) - stack = (ROOT / "skills/poteto-mode/playbooks/autopilot-stack.md").read_text( - encoding="utf-8" - ) - for text in (full, stack): - assert "arm a `/goal`" not in text - assert "depth is 1" in text.lower() or "depth" in text.lower() - assert "pstack:comment-sicko" in text - spec = ( - ROOT - / "openspec/specs/pstack-depth-1-overnight/spec.md" - ) - assert spec.is_file() - assert "MAX_SUBAGENT_DEPTH" in spec.read_text(encoding="utf-8") - - -def test_poteto_mode_first_todo_requires_host_map() -> None: - skill = (ROOT / "skills/poteto-mode/SKILL.md").read_text(encoding="utf-8") - assert "grok-tools.md" in skill - assert "codex-tools.md" in skill - assert "current host" in skill - head = skill.split("## Principles", 1)[0] - assert "mapping file for the current host" in head - - -def test_codex_map_matches_grok_call_sites() -> None: - mapping = ( - ROOT / "skills/poteto-mode/references/codex-tools.md" - ).read_text(encoding="utf-8") - assert "skills retain Claude Code tool language" not in mapping - assert "`task`" in mapping or "task" in mapping - assert "spawn_agent" in mapping - assert "ask_user_question" in mapping - assert "Codex compatibility utilities" in mapping - assert "not a Grok durable orchestration surface" in mapping - - -def test_poteto_mode_copies_tui_spawn_names() -> None: - skill = (ROOT / "skills/poteto-mode/SKILL.md").read_text(encoding="utf-8") - harness = (ROOT / "skills" / "poteto-mode" / "references" / "grok-tools.md").read_text(encoding="utf-8") - grok_tools = (ROOT / "skills/poteto-mode/references/grok-tools.md").read_text( - encoding="utf-8" - ) - assert "agent.spawn" in skill or "subagents" in skill.lower() - assert "spawn_subagent" in grok_tools - assert "get_command_or_subagent_output" in grok_tools - assert "spawn_subagent" in harness - assert "get_command_or_subagent_output" in harness - assert "scheduler_create" in harness - assert "EROFS" in harness - assert "config.toml" in harness - assert "herdr integration install grok" in harness - assert "HERDR_AGENT=grok" in harness - assert "herdr:grok" in harness - assert "agent_status" in harness - assert "workspace" in harness - assert "read-only" in harness - assert "strict" in harness - assert "herdr-install" not in harness - assert "HERDR_SOCKET_PATH" in harness - - -def test_herdr_workspace_gate_is_documented() -> None: - plan = (ROOT / "TEST-PLAN.md").read_text(encoding="utf-8") - assert "Gate 0a. Herdr callback from the built-in workspace sandbox" in plan - assert "source-managed Bash/Zsh helper" in plan - assert 'test -r "$HOME/.bashrc.d/30-herdr-grok.sh"' in plan - assert 'herdr pane run "$HINTED_PANE" grok --sandbox workspace --no-alt-screen' in plan - assert 'herdr pane run "$NOHINT_PANE" command grok --sandbox workspace --no-alt-screen' in plan - assert "HERDR_AGENT=grok" in plan - assert "herdr:grok" in plan - assert "agent_status == \"unknown\"" in plan - assert "herdr agent list" in plan - assert "herdr workspace close" in plan - assert "[ ] Gate 0a PASS" in plan - assert "Required: 0, 0a" in plan - assert "TUI `spawn_subagent`" in plan - assert 'remaining="$(jq --arg hinted' in plan - assert 'workspace-list-before-close.json' in plan - assert 'workspace-list-after-close.json' in plan - assert 'test "$remaining" = 0' in plan - assert 'HINTED_WORKSPACE="$(jq -r' in plan - assert '"$HERDR_EVIDENCE/hinted-create.json")"' in plan - assert 'NOHINT_WORKSPACE="$(jq -r' in plan - assert '"$HERDR_EVIDENCE/nohint-create.json")"' in plan - assert 'test -n "$HINTED_WORKSPACE"' in plan - assert 'test -n "$NOHINT_WORKSPACE"' in plan - - -def test_visual_parity_and_bug_fix_drive_real_surface() -> None: - visual = ( - ROOT / "skills/poteto-mode/playbooks/visual-parity.md" - ).read_text(encoding="utf-8") - bug = (ROOT / "skills/poteto-mode/playbooks/bug-fix.md").read_text( - encoding="utf-8" - ) - grok_tools = (ROOT / "skills/poteto-mode/references/grok-tools.md").read_text( - encoding="utf-8" - ) - assert "control skill" not in visual - assert "control-cli" not in visual - assert "schedule" in visual - assert "scheduler_create" in grok_tools - assert "control skill" not in bug - - -def test_make_bot_ui_is_not_invocable() -> None: - assert not (ROOT / "skills/make-bot-ui").exists() - plugin = (ROOT / "plugin.json").read_text(encoding="utf-8") - assert "make-bot-ui" not in plugin - - -def test_benny_is_source_and_has_grok_remap() -> None: - plugin = json.loads((ROOT / "plugin.json").read_text(encoding="utf-8")) - assert "hooks" not in plugin - assert "fail-closed" not in json.dumps(plugin) - skills_field = plugin.get("skills") - skill_paths = ( - [skills_field] if isinstance(skills_field, str) else list(skills_field) - ) - assert "./automations/benny-grok/skills/" in skill_paths - assert "./skills/" in skill_paths - assert not (ROOT / "skills/setup-benny").exists() - assert not (ROOT / "skills/triage-issue-reports").exists() - assert not (ROOT / "skills/reproduce-and-fix-issues").exists() - pack = ROOT / "automations/benny" - assert (pack / "FOR_AGENTS.md").is_file() - triage = (pack / "skills/triage-issue-reports/SKILL.md").read_text( - encoding="utf-8" - ) - assert "[benny:bug]" in triage - assert not (pack / "grok").exists() - live_root = ROOT / "automations/benny-grok" - assert live_root.is_dir() - readme = (live_root / "README.md").read_text(encoding="utf-8") - assert "grok plugin enable pstack" in readme - assert "grok -p" in readme - assert "mkdir -p .grok/hooks" not in readme - assert "/automate" not in readme - assert ".cursor/automations" not in readme - assert "scheduler_create" in readme - script = live_root / "bin/fail-closed.sh" - deny = subprocess.run( - ["sh", str(script)], - input=json.dumps( - {"toolInput": {"command": "gh pr merge 1 --merge"}} - ), - capture_output=True, - text=True, - check=False, - ) - assert deny.returncode == 2 - assert json.loads(deny.stdout)["decision"] == "deny" - for command in ( - "git merge feature && git push origin main", - "git push origin main && git merge feature", - ): - compound_deny = subprocess.run( - ["sh", str(script)], - input=json.dumps({"toolInput": {"command": command}}), - capture_output=True, - text=True, - check=False, - ) - assert compound_deny.returncode == 2, command - assert json.loads(compound_deny.stdout)["decision"] == "deny", command - force_lease = subprocess.run( - ["sh", str(script)], - input=json.dumps( - {"toolInput": {"command": "git push --force-with-lease origin main"}} - ), - capture_output=True, - text=True, - check=False, - ) - assert force_lease.returncode == 2 - assert json.loads(force_lease.stdout)["decision"] == "deny" - - allow = subprocess.run( - ["sh", str(script)], - input=json.dumps({"toolInput": {"command": "gh pr view 1"}}), - capture_output=True, - text=True, - check=False, - ) - assert allow.returncode == 0 - assert json.loads(allow.stdout)["decision"] == "allow" - slackish = subprocess.run( - ["sh", str(script)], - input=json.dumps( - {"toolInput": {"command": "use_tool slack__post_message"}} - ), - capture_output=True, - text=True, - check=False, - ) - assert slackish.returncode == 0 - assert json.loads(slackish.stdout)["decision"] == "allow" - safety = ( - ROOT - / "openspec/specs/benny-safety/spec.md" - ).read_text(encoding="utf-8") - assert "prompt-enforced" in safety - remap = ( - ROOT - / "openspec/specs/benny-grok-remap/spec.md" - ).read_text(encoding="utf-8") - assert "prompt-enforced" in remap - assert "upstream reference" in remap - assert "plugin enable" in remap - live_triage = ( - live_root / "skills/benny-triage/SKILL.md" - ).read_text(encoding="utf-8") - live_repro = ( - live_root / "skills/benny-repro/SKILL.md" - ).read_text(encoding="utf-8") - harness = (ROOT / "skills" / "poteto-mode" / "references" / "grok-tools.md").read_text(encoding="utf-8") - assert "automations/benny-grok" in harness - assert "automations/benny/grok" not in harness - live = f"{readme}\n{live_triage}\n{live_repro}" - assert "/automate" not in live - assert "trigger.thread_ts" not in live - assert "trigger.ts" not in live - assert ".cursor/automations" not in live - assert "spawn_subagent" in live_triage - assert "pstack:" in live_triage - assert "scheduler_create" in live - assert "spawn_subagent" in live_repro - assert "pstack:" in live_repro - assert "name: benny-triage" in live_triage - assert "name: benny-repro" in live_repro - design = ( - ROOT - / "openspec/changes/archive/2026-08-31-pstack-benny-atomic-blocks/design.md" - ).read_text(encoding="utf-8") - assert "automations/benny-grok/skills" in design - assert "Installing Benny SKILL.md files under plugin `skills/`." not in design - assert "grok/triage.md" not in design - assert "grok/repro.md" not in design - adr4 = ROOT / "adr/0004-benny-live-path-is-plugin-skills.md" - assert adr4.is_file() - assert "Supersedes: ADR-0002" in adr4.read_text(encoding="utf-8") - pack_spec = ( - ROOT - / "openspec/specs/benny-pack/spec.md" - ).read_text(encoding="utf-8") - assert "./automations/benny-grok/skills/" in pack_spec - assert "THEN `skills` is `./skills/`" not in pack_spec - - -def test_playbooks_are_not_plugin_rhai_workflows() -> None: - plugin = json.loads((ROOT / "plugin.json").read_text(encoding="utf-8")) - assert "workflows" not in plugin - assert not (ROOT / ".grok/workflows").exists() - rhai = list((ROOT / "skills/poteto-mode/playbooks").glob("*.rhai")) - assert rhai == [] - harness = (ROOT / "skills" / "poteto-mode" / "references" / "grok-tools.md").read_text(encoding="utf-8") - assert "not a plugin component" in harness - spec = ( - ROOT - / "openspec/specs/pstack-playbooks-not-rhai/spec.md" - ) - assert spec.is_file() - assert "MUST NOT" in spec.read_text(encoding="utf-8") - adr = ROOT / "adr/0005-playbooks-are-not-rhai-workflows.md" - assert adr.is_file() - assert "/poteto-mode" in adr.read_text(encoding="utf-8") - guide = (ROOT / "docs/guide/11-grok-workflows.md").read_text(encoding="utf-8") - assert "agent_type" in guide - assert "pstack:how-explorer" in guide - rhai_example = re.search(r"```rhai\n(.*?)```", guide, re.S) - assert rhai_example, "docs/guide/11-grok-workflows.md missing rhai example" - example = rhai_example.group(1) - assert 'agent_type: "pstack:how-explorer"' in example - assert "capability_mode" not in example - assert ".grok/workflows" in guide - assert "PluginManifest" in guide - assert "/workflow" in guide - assert "not a plugin" in guide.lower() or "not a plugin.json" in guide - index = (ROOT / "docs/guide/README.md").read_text(encoding="utf-8") - assert "11-grok-workflows.md" in index - spec = ( - ROOT - / "openspec/specs/pstack-grok-workflows/spec.md" - ) - assert spec.is_file() - assert "agent_type" in spec.read_text(encoding="utf-8") - - -def test_openspec_intent_driven_schema_resolves() -> None: - schema = ROOT / "openspec/schemas/intent-driven/schema.yaml" - assert schema.is_file() - text = schema.read_text(encoding="utf-8") - assert "name: intent-driven" in text - assert "id: adr" in text - cfg = (ROOT / "openspec/config.yaml").read_text(encoding="utf-8") - assert "schema: intent-driven" in cfg - change = ( - ROOT - / "openspec/changes/archive/2026-08-31-pstack-benny-atomic-blocks" - ) - assert (change / "design.md").is_file() - assert (change / "adr.md").is_file() - proc = subprocess.run( - ["openspec", "schema", "which", "intent-driven"], - cwd=ROOT, - capture_output=True, - text=True, - check=False, - ) - assert proc.returncode == 0, proc.stderr or proc.stdout - combined = (proc.stderr or "") + (proc.stdout or "") - assert "intent-driven" in combined.lower() - assert "not found" not in combined.lower() - - -def test_guide_teaches_sync_then_adapt() -> None: - guide = (ROOT / "docs/guide/09-make-it-yours.md").read_text(encoding="utf-8") - setup = (ROOT / "docs/guide/01-setup.md").read_text(encoding="utf-8") - harness = (ROOT / "skills" / "poteto-mode" / "references" / "grok-tools.md").read_text(encoding="utf-8") - upstream = (ROOT / "UPSTREAM").read_text(encoding="utf-8") - assert "atomic building blocks" in guide - assert "adapt-harness.py" in guide - assert "verify-harness.py" in guide - assert "sync-from-upstream.py" in guide - index = (ROOT / "docs/guide/README.md").read_text(encoding="utf-8") - assert "does not auto-enter" in index - natives = (ROOT / "docs/guide/13-grok-natives.md").read_text(encoding="utf-8") - assert "grok --worktree" in natives - assert "grok -p" in natives - assert "enter_plan_mode" in natives - assert "skip" in natives.lower() - assert "13-grok-natives.md" in index - spec_n = ( - ROOT - / "openspec/specs/pstack-grok-natives/spec.md" - ) - assert spec_n.is_file() - assert "adopt" in spec_n.read_text(encoding="utf-8").lower() - assert "spawn_subagent" in guide - assert "scheduler_create" in guide - assert "make-bot-ui" in guide - assert "control-cli" in guide - assert "tommy-ca/pstack" in setup - assert "aa2246740/pstack-grokbuild --trust" not in setup - assert "grok plugin install pstack --trust" not in setup - assert "cursor/plugins" in setup - assert "spawn_subagent" in setup - assert "automations/benny-grok" in setup - assert "skills/` and `agents/` only" not in setup - assert "grok-build-plugins" in guide - assert "host mapping" in setup - assert "not a plugin.json field" in setup - assert "MAX_SUBAGENT_DEPTH" in harness - assert "Skill order" in setup - assert "pstack:independent-verifier" in setup - assert "scheduler_create" in setup - assert "Essential entries" in setup - assert "/tdd" in setup - assert "/how" in setup - assert "/workflow" in setup - assert "workspace" in setup - assert "Daily driver" in setup - spec_sb = ( - ROOT - / "openspec/specs/pstack-sandbox-daily/spec.md" - ) - assert spec_sb.is_file() - assert "workspace" in spec_sb.read_text(encoding="utf-8") - spec_q = ( - ROOT - / "openspec/specs/pstack-quickstart/spec.md" - ) - assert spec_q.is_file() - assert "MAX_SUBAGENT_DEPTH" in spec_q.read_text(encoding="utf-8") - spec = ROOT / "openspec/specs/pstack-harness-md/spec.md" - assert spec.is_file() - assert "not a PluginManifest field" in spec.read_text(encoding="utf-8") - plugin = json.loads((ROOT / "plugin.json").read_text(encoding="utf-8")) - assert "HARNESS" not in json.dumps(plugin) - port = (ROOT / "docs/guide/12-porting.md").read_text(encoding="utf-8") - assert "24 principles" in port - assert "host map" in port - assert "Laziness Protocol" in port - assert "Capability checklist" in port - assert "Spawn a child" in port - assert "Join / wait" in port - assert "Overnight loop" in port - assert "Write `gap`" in port or "write `gap`" in port.lower() - assert "not loc" in port.lower() or "less code" in port.lower() - assert "12-porting.md" in index - spec_port = ( - ROOT / "openspec/specs/pstack-reference-port/spec.md" - ) - assert spec_port.is_file() - assert "host map" in spec_port.read_text(encoding="utf-8") - assert "adapt-harness.py" in upstream - assert "sync-from-upstream.py" in upstream - pin = subprocess.run( - [sys.executable, str(ROOT / "scripts/sync-from-upstream.py"), "--pin"], - cwd=ROOT, - capture_output=True, - text=True, - check=False, - ) - assert pin.returncode == 0, pin.stderr - sha = pin.stdout.strip() - assert re.fullmatch(r"[0-9a-f]{40}", sha), sha - plan = (ROOT / "TEST-PLAN.md").read_text(encoding="utf-8") - assert f"tree `{sha}`" in plan, sha - assert "46125561306434d8a1d7745d540d8932ab0cd2a2" not in plan - recipe = subprocess.run( - [sys.executable, str(ROOT / "scripts/sync-from-upstream.py"), "--recipe"], - cwd=ROOT, - capture_output=True, - text=True, - check=False, - ) - assert recipe.returncode == 0, recipe.stderr - assert sha in recipe.stdout - assert "adapt-harness.py" in recipe.stdout - assert "verify-harness.py" in recipe.stdout - assert "verify.py run" in recipe.stdout - assert "sync-from-upstream.py --pin" in recipe.stdout - assert "Full sweep" in recipe.stdout - assert "leftover-scanner" in recipe.stdout - assert "upstream-pin" in recipe.stdout - assert "absorbed-intent" in recipe.stdout - assert "upstream-recipe" in recipe.stdout - assert "refresh-hygiene" in recipe.stdout - assert "release-tag" in recipe.stdout - assert "verify-harness.py && python3 tests/test_verify_harness.py" not in recipe.stdout - assert "make-bot-ui" in recipe.stdout - assert "pstack:" in recipe.stdout - assert "upstream-cursor-plugins/pstack" in recipe.stdout - assert "classification.tsv" in recipe.stdout - assert "partition.py" in recipe.stdout - assert "apply.py" in recipe.stdout - assert "apply-check" in recipe.stdout - assert "print --coverage" in recipe.stdout - cache_table = ( - "--cache .worktrees/upstream-cursor-plugins --table " - "/tmp/pstack-refresh/overlay.tsv" - ) - assert cache_table in recipe.stdout - assert recipe.stdout.count(cache_table) >= 3 - assert "seed --cache .worktrees/upstream-cursor-plugins" in recipe.stdout - assert "--cache --table" not in recipe.stdout - assert not (ROOT / ".grok/workflows").exists() - rhai = ROOT / "skills/swarm/references/pstack-upstream-refresh.rhai" - assert rhai.is_file() - rhai_text = rhai.read_text(encoding="utf-8") - assert "capability_mode" not in rhai_text - assert 'agent_type: "pstack:swarm-workers"' in rhai_text - assert 'agent_type: "pstack:interrogate-reviewers"' in rhai_text - assert 'agent_type: "pstack:independent-verifier"' in rhai_text - assert "pack.py" in recipe.stdout - assert "scripts/orch/" in recipe.stdout - assert "advisor" in recipe.stdout - assert "0.15.29-grokbuild.0" in (ROOT / "plugin.json").read_text( - encoding="utf-8" - ) - default = subprocess.run( - [sys.executable, str(ROOT / "scripts/sync-from-upstream.py")], - cwd=ROOT, - capture_output=True, - text=True, - check=False, - ) - assert default.returncode == 0, default.stderr - assert default.stdout == recipe.stdout - logged = subprocess.run( - [sys.executable, str(ROOT / "scripts/sync-from-upstream.py"), "--log"], - cwd=ROOT, - capture_output=True, - text=True, - check=False, - ) - assert logged.returncode == 0, logged.stderr - assert f"pin {sha}" in logged.stdout - assert "up to date" in logged.stdout or re.search( - r"^[0-9a-f]{7} ", logged.stdout, re.MULTILINE - ) - sync_loader = importlib.util.spec_from_file_location( - "sync_from_upstream_log", ROOT / "scripts/sync-from-upstream.py" - ) - assert sync_loader is not None and sync_loader.loader is not None - sync = importlib.util.module_from_spec(sync_loader) - sync_loader.loader.exec_module(sync) - cache = sync.remote_cache() - if (cache / ".git").is_dir() or (cache / "HEAD").is_file(): - head = subprocess.run( - ["git", "-C", str(cache), "rev-parse", "HEAD"], - capture_output=True, - text=True, - check=True, - ).stdout.strip() - tip = subprocess.run( - ["git", "-C", str(cache), "rev-parse", "origin/main"], - capture_output=True, - text=True, - check=True, - ).stdout.strip() - assert head == tip, (head, tip) - assert ".cursor/skills" not in guide - assert "Cursor's built-in `create-skill`" not in guide - readme = (ROOT / "README.md").read_text(encoding="utf-8") - verify = (ROOT / "docs/guide/06-verify-and-ship.md").read_text( - encoding="utf-8" - ) - assert "tommy-ca/pstack --trust" in readme - assert "aa2246740/pstack-grokbuild --trust" not in readme - assert "spawn_subagent" in readme - assert "xAI Official also lists" in readme - assert "grok plugin install pstack --trust" not in readme - assert ".cursor/skills" not in verify - assert ".grok/skills/verify-" in verify - assert ".cursor/skills" not in setup - - def test_grok_spawn_types_are_plugin_qualified() -> None: effort = ( ROOT / "skills/setup-pstack/references/resolve-effort.md" @@ -802,62 +153,6 @@ def test_grok_spawn_types_are_plugin_qualified() -> None: assert "pstack:poteto-agent" in poteto -SWITCHER = "[English](README.md) · [简体中文](README.zh-CN.md)" -CJK = re.compile(r"[\u4e00-\u9fff]") - - -def test_readme_locale_split() -> None: - en = (ROOT / "README.md").read_text(encoding="utf-8") - zh_path = ROOT / "README.zh-CN.md" - assert zh_path.is_file(), zh_path - zh = zh_path.read_text(encoding="utf-8") - assert SWITCHER in en - assert SWITCHER in zh - en_body = en.replace(SWITCHER, "") - assert CJK.search(en_body) is None, "English README has CJK outside the switcher" - assert CJK.search(zh) is not None - for text in (en, zh): - assert "tommy-ca/pstack --trust" in text - assert "grok plugin install pstack --trust" not in text - assert "spawn_subagent" in text - assert "xAI Official" in text - assert "cursor/plugins" in text - assert "aa2246740/pstack-grokbuild --trust" not in text - assert "pstack:poteto-agent" in text - assert "pstack:comment-sicko" in text - assert "pstack:independent-verifier" in text - assert "plugins/pstack" in text - assert "grok-build-plugins" in text - - -def test_first_session_names_sandbox_reload_and_slash() -> None: - en = (ROOT / "README.md").read_text(encoding="utf-8") - zh = (ROOT / "README.zh-CN.md").read_text(encoding="utf-8") - setup = (ROOT / "docs/guide/01-setup.md").read_text(encoding="utf-8") - assert not (ROOT / "commands").exists() - for text in (en, zh, setup): - assert "## First session" in text or "## 第一次会话" in text - assert "grok --sandbox off plugin enable pstack" in text - assert "pstack:how-explorer" in text - assert "/poteto-mode" in text - assert "[plugins].enabled" in text - assert "new session" in text or "新会话" in text - assert "grok plugin install pstack --trust" not in text - assert "sticky mode" not in text.lower() - assert "粘滞模式" not in text - poteto = (ROOT / "skills/poteto-mode/SKILL.md").read_text(encoding="utf-8") - fm = poteto.split("---", 2)[1] - assert "mode: true" not in fm - guide = (ROOT / "docs/guide/02-poteto-mode.md").read_text(encoding="utf-8") - assert "sticky" not in guide.lower() - assert "does not auto-enter" in guide - assert "a mode mid-Feature" not in guide - setup = (ROOT / "docs/guide/01-setup.md").read_text(encoding="utf-8") - assert "sticky" not in setup.lower() - - -# grok 1.0.13 pager builtins from 04-slash-commands.md. Skills with these -# names lose /name to the builtin (qualified as pstack:name). GROK_BUILTIN_SLASH = { "new", "resume", @@ -948,38 +243,7 @@ def test_overlay_stems_and_adapter_are_plugin_qualified() -> None: overlay = json.loads( (ROOT / ".grok-plugin/plugin.json").read_text(encoding="utf-8") ) - root_manifest = json.loads((ROOT / "plugin.json").read_text(encoding="utf-8")) - assert "displayName" not in overlay - assert overlay["version"] == root_manifest["version"] - assert overlay["skills"] == root_manifest["skills"] - assert overlay["agents"] == root_manifest["agents"] - assert "hooks" not in overlay - harness = (ROOT / "skills" / "poteto-mode" / "references" / "grok-tools.md").read_text(encoding="utf-8") - assert "~/.grok/roles/.toml" not in harness - assert "~/.grok/roles/pstack:.toml" in harness - setup = (ROOT / "skills/setup-pstack/SKILL.md").read_text(encoding="utf-8") - en = (ROOT / "README.md").read_text(encoding="utf-8") - zh = (ROOT / "README.zh-CN.md").read_text(encoding="utf-8") - for text in (en, zh, setup): - assert "~/.grok/roles/*.toml" not in text - assert "~/.grok/roles/pstack:" in text or "pstack:<" in text - assert "files under `~/.grok/roles/`" not in setup - assert "matching files under `~/.grok/roles/`" not in setup - checklist = [ln for ln in plan.splitlines() if "Gate 4a PASS" in ln and "feature.toml" in ln] - assert checklist, "missing Gate 4a checklist line" - assert "pstack:feature.toml" in checklist[0] - assert "bare or `pstack:`" not in plan - assert "(`how-explorer`, `swarm-workers`" not in plan - assert "`pstack:how-explorer`" in plan - root_ver = root_manifest["version"] - claude_mkt = json.loads( - (ROOT / ".claude-plugin/marketplace.json").read_text(encoding="utf-8") - ) - assert claude_mkt["plugins"][0]["version"] == root_ver - upstream = (ROOT / "UPSTREAM").read_text(encoding="utf-8") - this_port = [ln.strip() for ln in upstream.splitlines() if ln.startswith("version ")] - assert this_port, "UPSTREAM missing this-port version line" - assert this_port[-1] == f"version {root_ver}", this_port + assert overlay["name"] == "pstack" def test_effort_frontmatter_matches_ladder() -> None: @@ -1042,8 +306,6 @@ def test_effort_frontmatter_matches_ladder() -> None: def test_plugin_manifest_matches_grok_parsed_fields() -> None: - import json - manifest = json.loads((ROOT / "plugin.json").read_text(encoding="utf-8")) assert "displayName" not in manifest assert manifest["name"] == "pstack" @@ -1068,59 +330,6 @@ def test_plugin_manifest_matches_grok_parsed_fields() -> None: assert "~/.grok/roles/pstack:" in text, path.name -NO_WRITE_AGENT_STEMS = frozenset( - { - "how-explorer", - "how-explainer", - "how-critics", - "interrogate-reviewers", - "arena-cross-judge-pool", - "independent-verifier", - "comment-sicko", - } -) -POSTURE_NOWRITE_STEMS = frozenset( - { - "why-investigators", - "why-synthesizer", - "reflect-judgment", - "reflect-tooling", - } -) - - -def test_nowrite_agents_use_capability_mode_execute() -> None: - agents = list((ROOT / "agents").glob("*.md")) - assert agents, "agents/ is empty" - found = {path.stem for path in agents} - missing_execute = NO_WRITE_AGENT_STEMS - found - missing_posture = POSTURE_NOWRITE_STEMS - found - assert not missing_execute, ( - f"execute-sandbox stems missing from agents/: {sorted(missing_execute)}" - ) - assert not missing_posture, ( - f"prompt-posture stems missing from agents/: {sorted(missing_posture)}" - ) - execute_re = re.compile(r"^capabilityMode:\s*execute\s*$", re.MULTILINE) - for path in agents: - fm = path.read_text(encoding="utf-8").split("---", 2)[1] - if path.stem in NO_WRITE_AGENT_STEMS: - assert execute_re.search(fm), ( - f"{path.relative_to(ROOT)}: no-write agent must set " - "capabilityMode: execute" - ) - continue - if path.stem in POSTURE_NOWRITE_STEMS: - assert "capabilityMode:" not in fm, ( - f"{path.relative_to(ROOT)}: prompt posture, not a sandbox; " - "do not set capabilityMode" - ) - continue - assert "capabilityMode:" not in fm, ( - f"{path.relative_to(ROOT)}: writer must omit capabilityMode" - ) - - def test_agent_and_droid_schema_parity() -> None: agent_files = list((ROOT / "agents").glob("*.md")) droid_files = list((ROOT / "droids").glob("*.md")) @@ -1131,7 +340,6 @@ def test_agent_and_droid_schema_parity() -> None: assert agent_stems == droid_stems, f"mismatch: {agent_stems ^ droid_stems}" - def test_audit_dir_skipped_by_harness_scanners() -> None: spec = importlib.util.spec_from_file_location("verify_harness", SCANNER) verify = importlib.util.module_from_spec(spec) @@ -1201,44 +409,6 @@ def test_harness_skill_order_is_pstack_then_user_then_native() -> None: assert "pstack:how-explorer" in adapt -def test_forge_neutral_pr_path_without_graphite() -> None: - harness = (ROOT / "skills" / "poteto-mode" / "references" / "grok-tools.md").read_text(encoding="utf-8") - ref = ROOT / "skills/poteto-mode/references/github-pr-fallback.md" - assert ref.is_file() - text = ref.read_text(encoding="utf-8").lower() - harness_lower = harness.lower() - assert "github-pr-fallback.md" in harness - assert "gh pr" in harness_lower - assert "origin pr" in harness_lower - assert "command -v gt" not in text - assert "gh pr create" in text - assert "gh pr view" in text - assert "gh pr checks" in text - assert "gh pr merge" in text - assert "--auto" in text - assert "protected trunk" in text - assert "--base" in text - opening = (ROOT / "skills/poteto-mode/playbooks/opening-a-pr.md").read_text( - encoding="utf-8" - ).lower() - shipping = (ROOT / "skills/poteto-mode/playbooks/shipping.md").read_text( - encoding="utf-8" - ).lower() - assert "graphite-first" not in opening - assert "gt submit" not in shipping - assert "origin" in opening - orchestrate = ( - ROOT / "skills/poteto-mode/playbooks/orchestrate.md" - ).read_text(encoding="utf-8") - orch = orchestrate.lower() - assert "recompute it from `gt`" not in orch - assert "may run `gt`" not in orch - assert "gt tracking" not in orch - assert "where gt knows" not in orch - assert "github-pr-fallback.md" in orchestrate - assert "origin pr" in orch or "gh pr" in orch - - def test_upstream_metadata_contract() -> None: upstream = (ROOT / "UPSTREAM").read_text(encoding="utf-8") assert "tree 88d47514b82e7c44f00f2b2a72ff6dd12158038c" in upstream @@ -1409,29 +579,17 @@ def test_all_skills_have_matching_kebab_case_frontmatter_name() -> None: test_verify_harness_passes_on_this_tree() test_all_skills_have_matching_kebab_case_frontmatter_name() test_babysit_and_shipping_do_not_use_cursor_dynamic_loop() - test_verify_guide_names_monitor_boundary() - test_autopilot_is_parent_fanout_and_skips_goal() - test_poteto_mode_first_todo_requires_host_map() - test_codex_map_matches_grok_call_sites() - test_poteto_mode_copies_tui_spawn_names() - test_visual_parity_and_bug_fix_drive_real_surface() - test_make_bot_ui_is_not_invocable() - test_benny_is_source_and_has_grok_remap() - test_playbooks_are_not_plugin_rhai_workflows() - test_openspec_intent_driven_schema_resolves() - test_guide_teaches_sync_then_adapt() + test_orchestrate_uses_canonical_host_state() + test_host_manifests_keep_grok_parity_and_adapter_asymmetry() test_grok_spawn_types_are_plugin_qualified() - test_readme_locale_split() - test_first_session_names_sandbox_reload_and_slash() test_pstack_slash_names_do_not_collide_with_grok_builtins() test_overlay_stems_and_adapter_are_plugin_qualified() test_effort_frontmatter_matches_ladder() test_plugin_manifest_matches_grok_parsed_fields() - test_nowrite_agents_use_capability_mode_execute() test_agent_and_droid_schema_parity() test_audit_dir_skipped_by_harness_scanners() test_harness_skill_order_is_pstack_then_user_then_native() - test_forge_neutral_pr_path_without_graphite() test_upstream_metadata_contract() test_harness_grok_profile_capability_parity() + test_harness_grok_tools_restored_scheduler_resume_semantics() print("PASS tests/test_verify_harness.py") diff --git a/tests/test_verify_harness_archive.py b/tests/test_verify_harness_archive.py new file mode 100644 index 00000000..31c8a0b6 --- /dev/null +++ b/tests/test_verify_harness_archive.py @@ -0,0 +1,157 @@ +from __future__ import annotations + +import importlib.util +import json +from pathlib import Path +import subprocess +import tempfile + +ROOT = Path(__file__).resolve().parents[1] +SCANNER = ROOT / "scripts" / "verify-harness.py" + + +def test_archivable_intent_chain_guard_distinguishes_changes() -> None: + loader = importlib.util.spec_from_file_location("verify_harness", SCANNER) + assert loader is not None + scanner = importlib.util.module_from_spec(loader) + assert loader.loader is not None + loader.loader.exec_module(scanner) + with tempfile.TemporaryDirectory() as tmp: + changes = Path(tmp) + complete = changes / "complete" + complete.mkdir() + (complete / "tasks.md").write_text("- [x] 1.1 done\n", encoding="utf-8") + incomplete = changes / "incomplete" + incomplete.mkdir() + (incomplete / "tasks.md").write_text("- [ ] 1.1 later\n", encoding="utf-8") + archived = changes / "archive" / "historical" + archived.mkdir(parents=True) + (archived / "tasks.md").write_text("- [x] 1.1 done\n", encoding="utf-8") + assert scanner.archivable_changes_missing_artifacts(changes) == [ + "complete/design.md", + "complete/adr.md", + ] + + +def test_archived_intent_chains_are_complete() -> None: + loader = importlib.util.spec_from_file_location("verify_harness", SCANNER) + assert loader is not None + scanner = importlib.util.module_from_spec(loader) + assert loader.loader is not None + loader.loader.exec_module(scanner) + missing = scanner.archived_changes_missing_artifacts(ROOT / "openspec/changes/archive") + assert not missing, missing + + +def test_archived_change_missing_adr_artifact_is_flagged() -> None: + loader = importlib.util.spec_from_file_location("verify_harness", SCANNER) + assert loader is not None + scanner = importlib.util.module_from_spec(loader) + assert loader.loader is not None + loader.loader.exec_module(scanner) + with tempfile.TemporaryDirectory() as tmp: + archive = Path(tmp) + change = archive / "2026-10-13-pstack-fixture" + (change / "specs" / "capability").mkdir(parents=True) + (change / "specs" / "capability" / "spec.md").write_text( + "## Purpose\nfixture\n", encoding="utf-8" + ) + for artifact in ("proposal.md", "design.md", "tasks.md", ".openspec.yaml"): + (change / artifact).write_text("x\n", encoding="utf-8") + assert scanner.archived_changes_missing_artifacts(archive) == [ + "2026-10-13-pstack-fixture/adr.md" + ] + (change / "adr.md").write_text("x\n", encoding="utf-8") + assert scanner.archived_changes_missing_artifacts(archive) == [] + + +def test_archived_change_skip_specs_exempts_specs_only() -> None: + loader = importlib.util.spec_from_file_location("verify_harness", SCANNER) + assert loader is not None + scanner = importlib.util.module_from_spec(loader) + assert loader.loader is not None + loader.loader.exec_module(scanner) + with tempfile.TemporaryDirectory() as tmp: + archive = Path(tmp) + for name, manifest in ( + ("skipped", "schema: intent-driven\nskip_specs: true\n"), + ("enforced", "schema: intent-driven\n"), + ): + change = archive / f"2026-10-13-{name}" + change.mkdir(parents=True) + for artifact in ("proposal.md", "design.md", "adr.md", "tasks.md"): + (change / artifact).write_text("x\n", encoding="utf-8") + (change / ".openspec.yaml").write_text(manifest, encoding="utf-8") + assert scanner.archived_changes_missing_artifacts(archive) == [ + "2026-10-13-enforced/specs/**/*.md" + ] + # skip_specs exempts specs only; other required artifacts stay enforced. + (archive / "2026-10-13-skipped" / "adr.md").unlink() + assert scanner.archived_changes_missing_artifacts(archive) == [ + "2026-10-13-enforced/specs/**/*.md", + "2026-10-13-skipped/adr.md", + ] + + +def test_durable_specs_have_non_placeholder_purposes() -> None: + for spec in sorted((ROOT / "openspec/specs").glob("*/spec.md")): + text = spec.read_text(encoding="utf-8") + assert "## Purpose\n" in text + purpose = text.split("## Purpose\n", 1)[1].split("\n\n", 1)[0].strip() + assert purpose and not purpose.startswith("TBD -"), spec + + +def test_benny_guard_denies_compound_merge_and_plain_push() -> None: + script = ROOT / "automations/benny-grok/bin/fail-closed.sh" + for command in ( + "git merge feature && git push origin main", + "git push origin main && git merge feature", + ): + deny = subprocess.run( + ["sh", str(script)], + input=json.dumps({"toolInput": {"command": command}}), + capture_output=True, + text=True, + check=False, + ) + assert deny.returncode == 2, command + assert json.loads(deny.stdout)["decision"] == "deny", command + + force_lease = subprocess.run( + ["sh", str(script)], + input=json.dumps( + {"toolInput": {"command": "git push --force-with-lease origin main"}} + ), + capture_output=True, + text=True, + check=False, + ) + assert force_lease.returncode == 2 + assert json.loads(force_lease.stdout)["decision"] == "deny" + + +def test_openspec_intent_driven_schema_resolves() -> None: + schema = ROOT / "openspec/schemas/intent-driven/schema.yaml" + assert schema.is_file() + text = schema.read_text(encoding="utf-8") + assert "name: intent-driven" in text + assert "id: adr" in text + cfg = (ROOT / "openspec/config.yaml").read_text(encoding="utf-8") + assert "schema: intent-driven" in cfg + change = ( + ROOT + / "openspec/changes/archive/2026-08-31-pstack-benny-atomic-blocks" + ) + assert (change / "design.md").is_file() + assert (change / "adr.md").is_file() + proc = subprocess.run( + ["openspec", "schema", "which", "intent-driven"], + cwd=ROOT, + capture_output=True, + text=True, + check=False, + ) + assert proc.returncode == 0, proc.stderr or proc.stdout + combined = (proc.stderr or "") + (proc.stdout or "") + assert "intent-driven" in combined.lower() + assert "not found" not in combined.lower() diff --git a/tests/test_verify_harness_guides.py b/tests/test_verify_harness_guides.py new file mode 100644 index 00000000..84561de3 --- /dev/null +++ b/tests/test_verify_harness_guides.py @@ -0,0 +1,311 @@ +from __future__ import annotations + +import importlib.util +import json +from pathlib import Path +import re +import subprocess +import sys + +ROOT = Path(__file__).resolve().parents[1] + +SWITCHER = "[English](README.md) · [简体中文](README.zh-CN.md)" +CJK = re.compile(r"[\u4e00-\u9fff]") + + +def test_verify_guide_names_monitor_boundary() -> None: + guide = (ROOT / "docs/guide/06-verify-and-ship.md").read_text(encoding="utf-8") + assert "bundled watcher" not in guide + assert "`monitor`" in guide + + +def test_guide_teaches_sync_then_adapt() -> None: + guide = (ROOT / "docs/guide/09-make-it-yours.md").read_text(encoding="utf-8") + setup = (ROOT / "docs/guide/01-setup.md").read_text(encoding="utf-8") + harness = (ROOT / "skills" / "poteto-mode" / "references" / "grok-tools.md").read_text(encoding="utf-8") + upstream = (ROOT / "UPSTREAM").read_text(encoding="utf-8") + assert "atomic building blocks" in guide + assert "adapt-harness.py" in guide + assert "verify-harness.py" in guide + assert "sync-from-upstream.py" in guide + index = (ROOT / "docs/guide/README.md").read_text(encoding="utf-8") + assert "does not auto-enter" in index + natives = (ROOT / "docs/guide/13-grok-natives.md").read_text(encoding="utf-8") + assert "grok --worktree" in natives + assert "grok -p" in natives + assert "enter_plan_mode" in natives + assert "skip" in natives.lower() + assert "13-grok-natives.md" in index + spec_n = ( + ROOT + / "openspec/specs/pstack-grok-natives/spec.md" + ) + assert spec_n.is_file() + assert "adopt" in spec_n.read_text(encoding="utf-8").lower() + assert "spawn_subagent" in guide + assert "scheduler_create" in guide + assert "make-bot-ui" in guide + assert "control-cli" in guide + assert "tommy-ca/pstack" in setup + assert "aa2246740/pstack-grokbuild --trust" not in setup + assert "grok plugin install pstack --trust" not in setup + assert "cursor/plugins" in setup + assert "spawn_subagent" in setup + assert "automations/benny-grok" in setup + assert "skills/` and `agents/` only" not in setup + assert "grok-build-plugins" in guide + assert "host mapping" in setup + assert "not a plugin.json field" in setup + assert "MAX_SUBAGENT_DEPTH" in harness + assert "Skill order" in setup + assert "pstack:independent-verifier" in setup + assert "scheduler_create" in setup + assert "Essential entries" in setup + assert "/tdd" in setup + assert "/how" in setup + assert "/workflow" in setup + assert "workspace" in setup + assert "Daily driver" in setup + spec_sb = ( + ROOT + / "openspec/specs/pstack-sandbox-daily/spec.md" + ) + assert spec_sb.is_file() + assert "workspace" in spec_sb.read_text(encoding="utf-8") + spec_q = ( + ROOT + / "openspec/specs/pstack-quickstart/spec.md" + ) + assert spec_q.is_file() + assert "MAX_SUBAGENT_DEPTH" in spec_q.read_text(encoding="utf-8") + spec = ROOT / "openspec/specs/pstack-harness-md/spec.md" + assert spec.is_file() + assert "not a PluginManifest field" in spec.read_text(encoding="utf-8") + plugin = json.loads((ROOT / "plugin.json").read_text(encoding="utf-8")) + assert "HARNESS" not in json.dumps(plugin) + port = (ROOT / "docs/guide/12-porting.md").read_text(encoding="utf-8") + assert "24 principles" in port + assert "host map" in port + assert "Laziness Protocol" in port + assert "Capability checklist" in port + assert "Spawn a child" in port + assert "Join / wait" in port + assert "Overnight loop" in port + assert "Write `gap`" in port or "write `gap`" in port.lower() + assert "not loc" in port.lower() or "less code" in port.lower() + assert "12-porting.md" in index + spec_port = ( + ROOT / "openspec/specs/pstack-reference-port/spec.md" + ) + assert spec_port.is_file() + assert "host map" in spec_port.read_text(encoding="utf-8") + assert "adapt-harness.py" in upstream + assert "sync-from-upstream.py" in upstream + pin = subprocess.run( + [sys.executable, str(ROOT / "scripts/sync-from-upstream.py"), "--pin"], + cwd=ROOT, + capture_output=True, + text=True, + check=False, + ) + assert pin.returncode == 0, pin.stderr + sha = pin.stdout.strip() + assert re.fullmatch(r"[0-9a-f]{40}", sha), sha + plan = (ROOT / "TEST-PLAN.md").read_text(encoding="utf-8") + assert f"tree `{sha}`" in plan, sha + assert "46125561306434d8a1d7745d540d8932ab0cd2a2" not in plan + recipe = subprocess.run( + [sys.executable, str(ROOT / "scripts/sync-from-upstream.py"), "--recipe"], + cwd=ROOT, + capture_output=True, + text=True, + check=False, + ) + assert recipe.returncode == 0, recipe.stderr + assert sha in recipe.stdout + assert "adapt-harness.py" in recipe.stdout + assert "verify-harness.py" in recipe.stdout + assert "verify.py run" in recipe.stdout + assert "sync-from-upstream.py --pin" in recipe.stdout + assert "Full sweep" in recipe.stdout + assert "leftover-scanner" in recipe.stdout + assert "upstream-pin" in recipe.stdout + assert "absorbed-intent" in recipe.stdout + assert "upstream-recipe" in recipe.stdout + assert "refresh-hygiene" in recipe.stdout + assert "release-tag" in recipe.stdout + assert "verify-harness.py && python3 tests/test_verify_harness.py" not in recipe.stdout + assert "make-bot-ui" in recipe.stdout + assert "pstack:" in recipe.stdout + assert "upstream-cursor-plugins/pstack" in recipe.stdout + assert "classification.tsv" in recipe.stdout + assert "partition.py" in recipe.stdout + assert "apply.py" in recipe.stdout + assert "apply-check" in recipe.stdout + assert "print --coverage" in recipe.stdout + cache_table = ( + "--cache .worktrees/upstream-cursor-plugins --table " + "/tmp/pstack-refresh/overlay.tsv" + ) + assert cache_table in recipe.stdout + assert recipe.stdout.count(cache_table) >= 3 + assert "seed --cache .worktrees/upstream-cursor-plugins" in recipe.stdout + assert "--cache --table" not in recipe.stdout + assert not (ROOT / ".grok/workflows").exists() + rhai = ROOT / "skills/swarm/references/pstack-upstream-refresh.rhai" + assert rhai.is_file() + rhai_text = rhai.read_text(encoding="utf-8") + assert "capability_mode" not in rhai_text + assert 'agent_type: "pstack:swarm-workers"' in rhai_text + assert 'agent_type: "pstack:interrogate-reviewers"' in rhai_text + assert 'agent_type: "pstack:independent-verifier"' in rhai_text + assert "pack.py" in recipe.stdout + assert "scripts/orch/" in recipe.stdout + assert "advisor" in recipe.stdout + assert "0.15.29-grokbuild.0" in (ROOT / "plugin.json").read_text( + encoding="utf-8" + ) + default = subprocess.run( + [sys.executable, str(ROOT / "scripts/sync-from-upstream.py")], + cwd=ROOT, + capture_output=True, + text=True, + check=False, + ) + assert default.returncode == 0, default.stderr + assert default.stdout == recipe.stdout + logged = subprocess.run( + [sys.executable, str(ROOT / "scripts/sync-from-upstream.py"), "--log"], + cwd=ROOT, + capture_output=True, + text=True, + check=False, + ) + assert logged.returncode == 0, logged.stderr + assert f"pin {sha}" in logged.stdout + assert "up to date" in logged.stdout or re.search( + r"^[0-9a-f]{7} ", logged.stdout, re.MULTILINE + ) + sync_loader = importlib.util.spec_from_file_location( + "sync_from_upstream_log", ROOT / "scripts/sync-from-upstream.py" + ) + assert sync_loader is not None and sync_loader.loader is not None + sync = importlib.util.module_from_spec(sync_loader) + sync_loader.loader.exec_module(sync) + cache = sync.remote_cache() + if (cache / ".git").is_dir() or (cache / "HEAD").is_file(): + head = subprocess.run( + ["git", "-C", str(cache), "rev-parse", "HEAD"], + capture_output=True, + text=True, + check=True, + ).stdout.strip() + tip = subprocess.run( + ["git", "-C", str(cache), "rev-parse", "origin/main"], + capture_output=True, + text=True, + check=True, + ).stdout.strip() + assert head == tip, (head, tip) + assert ".cursor/skills" not in guide + assert "Cursor's built-in `create-skill`" not in guide + readme = (ROOT / "README.md").read_text(encoding="utf-8") + verify = (ROOT / "docs/guide/06-verify-and-ship.md").read_text( + encoding="utf-8" + ) + assert "tommy-ca/pstack --trust" in readme + assert "aa2246740/pstack-grokbuild --trust" not in readme + assert "spawn_subagent" in readme + assert "xAI Official" in readme + assert "grok plugin install pstack --trust" not in readme + assert ".cursor/skills" not in verify + assert ".grok/skills/verify-" in verify + assert ".cursor/skills" not in setup + + +def test_readme_locale_split() -> None: + en = (ROOT / "README.md").read_text(encoding="utf-8") + zh_path = ROOT / "README.zh-CN.md" + assert zh_path.is_file(), zh_path + zh = zh_path.read_text(encoding="utf-8") + assert SWITCHER in en + assert SWITCHER in zh + en_body = en.replace(SWITCHER, "") + assert CJK.search(en_body) is None, "English README has CJK outside the switcher" + assert CJK.search(zh) is not None + for text in (en, zh): + assert "tommy-ca/pstack --trust" in text + assert "grok plugin install pstack --trust" not in text + assert "spawn_subagent" in text + assert "xAI Official" in text + assert "cursor/plugins" in text + assert "aa2246740/pstack-grokbuild --trust" not in text + assert "pstack:poteto-agent" in text + assert "pstack:comment-sicko" in text + assert "pstack:independent-verifier" in text + assert "plugins/pstack" in text + assert "grok-build-plugins" in text + + +def test_first_session_names_sandbox_reload_and_slash() -> None: + en = (ROOT / "README.md").read_text(encoding="utf-8") + zh = (ROOT / "README.zh-CN.md").read_text(encoding="utf-8") + setup = (ROOT / "docs/guide/01-setup.md").read_text(encoding="utf-8") + assert not (ROOT / "commands").exists() + for text in (en, zh, setup): + assert "## First session" in text or "## 第一次会话" in text + assert "grok --sandbox off plugin enable pstack" in text + assert "pstack:how-explorer" in text + assert "/poteto-mode" in text + assert "[plugins].enabled" in text + assert "new session" in text or "新会话" in text + assert "grok plugin install pstack --trust" not in text + assert "sticky mode" not in text.lower() + assert "粘滞模式" not in text + poteto = (ROOT / "skills/poteto-mode/SKILL.md").read_text(encoding="utf-8") + fm = poteto.split("---", 2)[1] + assert "mode: true" not in fm + guide = (ROOT / "docs/guide/02-poteto-mode.md").read_text(encoding="utf-8") + assert "sticky" not in guide.lower() + assert "does not auto-enter" in guide + assert "a mode mid-Feature" not in guide + setup = (ROOT / "docs/guide/01-setup.md").read_text(encoding="utf-8") + assert "sticky" not in setup.lower() + + +def test_forge_neutral_pr_path_without_graphite() -> None: + harness = (ROOT / "skills" / "poteto-mode" / "references" / "grok-tools.md").read_text(encoding="utf-8") + ref = ROOT / "skills/poteto-mode/references/github-pr-fallback.md" + assert ref.is_file() + text = ref.read_text(encoding="utf-8").lower() + harness_lower = harness.lower() + assert "github-pr-fallback.md" in harness + assert "gh pr" in harness_lower + assert "origin pr" in harness_lower + assert "command -v gt" not in text + assert "gh pr create" in text + assert "gh pr view" in text + assert "gh pr checks" in text + assert "gh pr merge" in text + assert "--auto" in text + assert "protected trunk" in text + assert "--base" in text + opening = (ROOT / "skills/poteto-mode/playbooks/opening-a-pr.md").read_text( + encoding="utf-8" + ).lower() + shipping = (ROOT / "skills/poteto-mode/playbooks/shipping.md").read_text( + encoding="utf-8" + ).lower() + assert "graphite-first" not in opening + assert "gt submit" not in shipping + assert "origin" in opening + orchestrate = ( + ROOT / "skills/poteto-mode/playbooks/orchestrate.md" + ).read_text(encoding="utf-8") + orch = orchestrate.lower() + assert "recompute it from `gt`" not in orch + assert "may run `gt`" not in orch + assert "gt tracking" not in orch + assert "where gt knows" not in orch + assert "github-pr-fallback.md" in orchestrate + assert "origin pr" in orch or "gh pr" in orch diff --git a/tests/test_verify_harness_playbooks.py b/tests/test_verify_harness_playbooks.py new file mode 100644 index 00000000..c5be7faa --- /dev/null +++ b/tests/test_verify_harness_playbooks.py @@ -0,0 +1,352 @@ +from __future__ import annotations + +import json +from pathlib import Path +import re +import subprocess + +ROOT = Path(__file__).resolve().parents[1] + +NO_WRITE_AGENT_STEMS = frozenset( + { + "how-explorer", + "how-explainer", + "how-critics", + "interrogate-reviewers", + "arena-cross-judge-pool", + "independent-verifier", + "comment-sicko", + } +) +POSTURE_NOWRITE_STEMS = frozenset( + { + "why-investigators", + "why-synthesizer", + "reflect-judgment", + "reflect-tooling", + } +) + + +def test_autopilot_is_parent_fanout_and_skips_goal() -> None: + full = (ROOT / "skills/poteto-mode/playbooks/autopilot-full.md").read_text( + encoding="utf-8" + ) + stack = (ROOT / "skills/poteto-mode/playbooks/autopilot-stack.md").read_text( + encoding="utf-8" + ) + for text in (full, stack): + assert "arm a `/goal`" not in text + assert "depth is 1" in text.lower() or "depth" in text.lower() + assert "pstack:comment-sicko" in text + spec = ( + ROOT + / "openspec/specs/pstack-depth-1-overnight/spec.md" + ) + assert spec.is_file() + assert "MAX_SUBAGENT_DEPTH" in spec.read_text(encoding="utf-8") + + +def test_poteto_mode_first_todo_requires_host_map() -> None: + skill = (ROOT / "skills/poteto-mode/SKILL.md").read_text(encoding="utf-8") + assert "grok-tools.md" in skill + assert "codex-tools.md" in skill + assert "current host" in skill + head = skill.split("## Principles", 1)[0] + assert "mapping file for the current host" in head + + +def test_codex_map_matches_grok_call_sites() -> None: + mapping = ( + ROOT / "skills/poteto-mode/references/codex-tools.md" + ).read_text(encoding="utf-8") + assert "skills retain Claude Code tool language" not in mapping + assert "`task`" in mapping or "task" in mapping + assert "spawn_agent" in mapping + assert "ask_user_question" in mapping + assert "Codex compatibility utilities" in mapping + assert "not a Grok durable orchestration surface" in mapping + + +def test_poteto_mode_copies_tui_spawn_names() -> None: + skill = (ROOT / "skills/poteto-mode/SKILL.md").read_text(encoding="utf-8") + harness = (ROOT / "skills" / "poteto-mode" / "references" / "grok-tools.md").read_text(encoding="utf-8") + grok_tools = (ROOT / "skills/poteto-mode/references/grok-tools.md").read_text( + encoding="utf-8" + ) + assert "agent.spawn" in skill or "subagents" in skill.lower() + assert "spawn_subagent" in grok_tools + assert "get_command_or_subagent_output" in grok_tools + assert "spawn_subagent" in harness + assert "get_command_or_subagent_output" in harness + assert "scheduler_create" in harness + assert "EROFS" in harness + assert "config.toml" in harness + assert "herdr integration install grok" in harness + assert "HERDR_AGENT=grok" in harness + assert "herdr:grok" in harness + assert "agent_status" in harness + assert "workspace" in harness + assert "read-only" in harness + assert "strict" in harness + assert "herdr-install" not in harness + assert "HERDR_SOCKET_PATH" in harness + + +def test_herdr_workspace_gate_is_documented() -> None: + plan = (ROOT / "TEST-PLAN.md").read_text(encoding="utf-8") + assert "Gate 0a. Herdr callback from the built-in workspace sandbox" in plan + assert "source-managed Bash/Zsh helper" in plan + assert 'test -r "$HOME/.bashrc.d/30-herdr-grok.sh"' in plan + assert 'herdr pane run "$HINTED_PANE" grok --sandbox workspace --no-alt-screen' in plan + assert 'herdr pane run "$NOHINT_PANE" command grok --sandbox workspace --no-alt-screen' in plan + assert "HERDR_AGENT=grok" in plan + assert "herdr:grok" in plan + assert "agent_status == \"unknown\"" in plan + assert "herdr agent list" in plan + assert "herdr workspace close" in plan + assert "[ ] Gate 0a PASS" in plan + assert "Required: 0, 0a" in plan + assert "TUI `spawn_subagent`" in plan + assert 'remaining="$(jq --arg hinted' in plan + assert 'workspace-list-before-close.json' in plan + assert 'workspace-list-after-close.json' in plan + assert 'test "$remaining" = 0' in plan + assert 'HINTED_WORKSPACE="$(jq -r' in plan + assert '"$HERDR_EVIDENCE/hinted-create.json")"' in plan + assert 'NOHINT_WORKSPACE="$(jq -r' in plan + assert '"$HERDR_EVIDENCE/nohint-create.json")"' in plan + assert 'test -n "$HINTED_WORKSPACE"' in plan + assert 'test -n "$NOHINT_WORKSPACE"' in plan + + +def test_visual_parity_and_bug_fix_drive_real_surface() -> None: + visual = ( + ROOT / "skills/poteto-mode/playbooks/visual-parity.md" + ).read_text(encoding="utf-8") + bug = (ROOT / "skills/poteto-mode/playbooks/bug-fix.md").read_text( + encoding="utf-8" + ) + grok_tools = (ROOT / "skills/poteto-mode/references/grok-tools.md").read_text( + encoding="utf-8" + ) + assert "control skill" not in visual + assert "control-cli" not in visual + assert "schedule" in visual + assert "scheduler_create" in grok_tools + assert "control skill" not in bug + + +def test_make_bot_ui_is_not_invocable() -> None: + assert not (ROOT / "skills/make-bot-ui").exists() + plugin = (ROOT / "plugin.json").read_text(encoding="utf-8") + assert "make-bot-ui" not in plugin + + +def test_benny_is_source_and_has_grok_remap() -> None: + plugin = json.loads((ROOT / "plugin.json").read_text(encoding="utf-8")) + assert "hooks" not in plugin + assert "fail-closed" not in json.dumps(plugin) + skills_field = plugin.get("skills") + skill_paths = ( + [skills_field] if isinstance(skills_field, str) else list(skills_field) + ) + assert "./automations/benny-grok/skills/" in skill_paths + assert "./skills/" in skill_paths + assert not (ROOT / "skills/setup-benny").exists() + assert not (ROOT / "skills/triage-issue-reports").exists() + assert not (ROOT / "skills/reproduce-and-fix-issues").exists() + pack = ROOT / "automations/benny" + assert (pack / "FOR_AGENTS.md").is_file() + triage = (pack / "skills/triage-issue-reports/SKILL.md").read_text( + encoding="utf-8" + ) + assert "[benny:bug]" in triage + assert not (pack / "grok").exists() + live_root = ROOT / "automations/benny-grok" + assert live_root.is_dir() + readme = (live_root / "README.md").read_text(encoding="utf-8") + assert "grok plugin enable pstack" in readme + assert "grok -p" in readme + assert "mkdir -p .grok/hooks" not in readme + assert "/automate" not in readme + assert ".cursor/automations" not in readme + assert "scheduler_create" in readme + script = live_root / "bin/fail-closed.sh" + deny = subprocess.run( + ["sh", str(script)], + input=json.dumps( + {"toolInput": {"command": "gh pr merge 1 --merge"}} + ), + capture_output=True, + text=True, + check=False, + ) + assert deny.returncode == 2 + assert json.loads(deny.stdout)["decision"] == "deny" + for command in ( + "git merge feature && git push origin main", + "git push origin main && git merge feature", + ): + compound_deny = subprocess.run( + ["sh", str(script)], + input=json.dumps({"toolInput": {"command": command}}), + capture_output=True, + text=True, + check=False, + ) + assert compound_deny.returncode == 2, command + assert json.loads(compound_deny.stdout)["decision"] == "deny", command + force_lease = subprocess.run( + ["sh", str(script)], + input=json.dumps( + {"toolInput": {"command": "git push --force-with-lease origin main"}} + ), + capture_output=True, + text=True, + check=False, + ) + assert force_lease.returncode == 2 + assert json.loads(force_lease.stdout)["decision"] == "deny" + + allow = subprocess.run( + ["sh", str(script)], + input=json.dumps({"toolInput": {"command": "gh pr view 1"}}), + capture_output=True, + text=True, + check=False, + ) + assert allow.returncode == 0 + assert json.loads(allow.stdout)["decision"] == "allow" + slackish = subprocess.run( + ["sh", str(script)], + input=json.dumps( + {"toolInput": {"command": "use_tool slack__post_message"}} + ), + capture_output=True, + text=True, + check=False, + ) + assert slackish.returncode == 0 + assert json.loads(slackish.stdout)["decision"] == "allow" + safety = ( + ROOT + / "openspec/specs/benny-safety/spec.md" + ).read_text(encoding="utf-8") + assert "prompt-enforced" in safety + remap = ( + ROOT + / "openspec/specs/benny-grok-remap/spec.md" + ).read_text(encoding="utf-8") + assert "prompt-enforced" in remap + assert "upstream reference" in remap + assert "plugin enable" in remap + live_triage = ( + live_root / "skills/benny-triage/SKILL.md" + ).read_text(encoding="utf-8") + live_repro = ( + live_root / "skills/benny-repro/SKILL.md" + ).read_text(encoding="utf-8") + harness = (ROOT / "skills" / "poteto-mode" / "references" / "grok-tools.md").read_text(encoding="utf-8") + assert "automations/benny-grok" in harness + assert "automations/benny/grok" not in harness + live = f"{readme}\n{live_triage}\n{live_repro}" + assert "/automate" not in live + assert "trigger.thread_ts" not in live + assert "trigger.ts" not in live + assert ".cursor/automations" not in live + assert "spawn_subagent" in live_triage + assert "pstack:" in live_triage + assert "scheduler_create" in live + assert "spawn_subagent" in live_repro + assert "pstack:" in live_repro + assert "name: benny-triage" in live_triage + assert "name: benny-repro" in live_repro + design = ( + ROOT + / "openspec/changes/archive/2026-08-31-pstack-benny-atomic-blocks/design.md" + ).read_text(encoding="utf-8") + assert "automations/benny-grok/skills" in design + assert "Installing Benny SKILL.md files under plugin `skills/`." not in design + assert "grok/triage.md" not in design + assert "grok/repro.md" not in design + adr4 = ROOT / "adr/0004-benny-live-path-is-plugin-skills.md" + assert adr4.is_file() + assert "Supersedes: ADR-0002" in adr4.read_text(encoding="utf-8") + pack_spec = ( + ROOT + / "openspec/specs/benny-pack/spec.md" + ).read_text(encoding="utf-8") + assert "./automations/benny-grok/skills/" in pack_spec + assert "THEN `skills` is `./skills/`" not in pack_spec + + +def test_playbooks_are_not_plugin_rhai_workflows() -> None: + plugin = json.loads((ROOT / "plugin.json").read_text(encoding="utf-8")) + assert "workflows" not in plugin + assert not (ROOT / ".grok/workflows").exists() + rhai = list((ROOT / "skills/poteto-mode/playbooks").glob("*.rhai")) + assert rhai == [] + harness = (ROOT / "skills" / "poteto-mode" / "references" / "grok-tools.md").read_text(encoding="utf-8") + assert "not a plugin component" in harness + spec = ( + ROOT + / "openspec/specs/pstack-playbooks-not-rhai/spec.md" + ) + assert spec.is_file() + assert "MUST NOT" in spec.read_text(encoding="utf-8") + adr = ROOT / "adr/0005-playbooks-are-not-rhai-workflows.md" + assert adr.is_file() + assert "/poteto-mode" in adr.read_text(encoding="utf-8") + guide = (ROOT / "docs/guide/11-grok-workflows.md").read_text(encoding="utf-8") + assert "agent_type" in guide + assert "pstack:how-explorer" in guide + rhai_example = re.search(r"```rhai\n(.*?)```", guide, re.S) + assert rhai_example, "docs/guide/11-grok-workflows.md missing rhai example" + example = rhai_example.group(1) + assert 'agent_type: "pstack:how-explorer"' in example + assert "capability_mode" not in example + assert ".grok/workflows" in guide + assert "PluginManifest" in guide + assert "/workflow" in guide + assert "not a plugin" in guide.lower() or "not a plugin.json" in guide + index = (ROOT / "docs/guide/README.md").read_text(encoding="utf-8") + assert "11-grok-workflows.md" in index + spec = ( + ROOT + / "openspec/specs/pstack-grok-workflows/spec.md" + ) + assert spec.is_file() + assert "agent_type" in spec.read_text(encoding="utf-8") + + +def test_nowrite_agents_use_capability_mode_execute() -> None: + agents = list((ROOT / "agents").glob("*.md")) + assert agents, "agents/ is empty" + found = {path.stem for path in agents} + missing_execute = NO_WRITE_AGENT_STEMS - found + missing_posture = POSTURE_NOWRITE_STEMS - found + assert not missing_execute, ( + f"execute-sandbox stems missing from agents/: {sorted(missing_execute)}" + ) + assert not missing_posture, ( + f"prompt-posture stems missing from agents/: {sorted(missing_posture)}" + ) + execute_re = re.compile(r"^capabilityMode:\s*execute\s*$", re.MULTILINE) + for path in agents: + fm = path.read_text(encoding="utf-8").split("---", 2)[1] + if path.stem in NO_WRITE_AGENT_STEMS: + assert execute_re.search(fm), ( + f"{path.relative_to(ROOT)}: no-write agent must set " + "capabilityMode: execute" + ) + continue + if path.stem in POSTURE_NOWRITE_STEMS: + assert "capabilityMode:" not in fm, ( + f"{path.relative_to(ROOT)}: prompt posture, not a sandbox; " + "do not set capabilityMode" + ) + continue + assert "capabilityMode:" not in fm, ( + f"{path.relative_to(ROOT)}: writer must omit capabilityMode" + )