diff --git a/tests/test_cursor_cli_install_surface.py b/tests/test_cursor_cli_install_surface.py new file mode 100644 index 0000000..d4c1b13 --- /dev/null +++ b/tests/test_cursor_cli_install_surface.py @@ -0,0 +1,221 @@ +# Test classification: distribution +"""Regression tests for the Cursor CLI entry added in #73. + +The installer change touched cursor_plugin.py, hosts/base.py, and +inspection.py. These tests lock the behaviors that a content-only README +assertion does not: the top-level Skill is part of the install contract, +Doctor fails closed when it is missing, and the CLI behavior check stays an +unverified best-effort recommendation. +""" + +from __future__ import annotations + +from pathlib import Path +import shutil +import sys +import tempfile +import unittest + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from installer.cursor_plugin import install_cursor_user_plugin_assets +from installer.hosts.codex import CODEX_ADAPTER +from installer.hosts.cursor import CURSOR_ADAPTER +from installer.inspection import build_doctor_payload +from installer.models import InstallError + + +def _cursor_checks(home_root: Path) -> dict[str, dict[str, object]]: + doctor = build_doctor_payload(home_root=home_root, workspace_root=None) + return { + check["check_id"]: check + for check in doctor["checks"] + if check.get("host_id") == "cursor" + } + + +def _snapshot(root: Path) -> dict[str, bytes]: + if not root.exists(): + return {} + return { + path.relative_to(root).as_posix(): path.read_bytes() + for path in sorted(root.rglob("*")) + if path.is_file() + } + + +class CursorCliInstallSurfaceTests(unittest.TestCase): + def test_cursor_contract_includes_top_level_cli_skill(self) -> None: + """Dropping SKILL.md from the Cursor path list must fail this test. + + hosts/base.py added that file to global_skill_paths, and the user + Plugin expected paths are built from the same list. Codex does not + install this entry. + """ + home = Path("/tmp/sopify-cursor-contract-home") + cli_entry = home / ".cursor" / "skills" / "sopify" / "SKILL.md" + self.assertIn(cli_entry, CURSOR_ADAPTER.global_skill_paths(home)) + self.assertIn(cli_entry, CURSOR_ADAPTER.expected_paths(home)) + self.assertNotIn( + home / ".codex" / "skills" / "sopify" / "SKILL.md", + CODEX_ADAPTER.expected_paths(home), + ) + + def test_doctor_fails_when_only_the_cli_entry_is_missing(self) -> None: + """A Cursor install without the top-level Skill is not a healthy skill tree. + + Phase skills can all be present while SKILL.md is gone. Doctor must + report that hole instead of treating the tree as installed. + """ + with tempfile.TemporaryDirectory() as home_dir: + home_root = Path(home_dir) + install_cursor_user_plugin_assets( + CURSOR_ADAPTER, + repo_root=REPO_ROOT, + home_root=home_root, + language_directory="CN", + ) + cli_entry = home_root / ".cursor" / "skills" / "sopify" / "SKILL.md" + cli_entry.unlink() + + skill_tree = _cursor_checks(home_root)["global_skill_tree_present"] + + self.assertEqual(skill_tree["status"], "fail") + self.assertEqual(skill_tree["reason_code"], "MISSING_REQUIRED_FILE") + self.assertTrue( + any(str(cli_entry) in item for item in skill_tree.get("evidence", [])) + ) + + def test_cli_behavior_recommendation_stays_best_effort(self) -> None: + """The CLI check must not go back to claiming the entry is unsupported. + + inspection.py now tells the operator to run a managed-entry black box + and to keep Skill selection separate from file installation. File + presence still must not flip this check to pass. + """ + with tempfile.TemporaryDirectory() as home_dir: + home_root = Path(home_dir) + install_cursor_user_plugin_assets( + CURSOR_ADAPTER, + repo_root=REPO_ROOT, + home_root=home_root, + language_directory="CN", + ) + checks = _cursor_checks(home_root) + cli = checks["cursor_cli_behavior"] + ide = checks["cursor_ide_behavior"] + + self.assertEqual(cli["status"], "skip") + self.assertEqual(cli["reason_code"], "BLACK_BOX_NOT_VERIFIED") + recommendation = str(cli.get("recommendation") or "") + self.assertIn("managed-entry", recommendation) + self.assertIn("best-effort", recommendation) + self.assertIn("separate from file installation evidence", recommendation) + self.assertNotIn("not a supported CLI entry", recommendation) + self.assertIn( + "file presence alone is not behavior evidence", + str(ide.get("recommendation") or ""), + ) + + def test_language_switch_replaces_cli_entry(self) -> None: + """A later cursor:en-US install must not leave the zh-CN CLI entry in place.""" + with tempfile.TemporaryDirectory() as home_dir: + home_root = Path(home_dir) + install_cursor_user_plugin_assets( + CURSOR_ADAPTER, + repo_root=REPO_ROOT, + home_root=home_root, + language_directory="CN", + ) + updated = install_cursor_user_plugin_assets( + CURSOR_ADAPTER, + repo_root=REPO_ROOT, + home_root=home_root, + language_directory="EN", + ) + installed = ( + home_root / ".cursor" / "skills" / "sopify" / "SKILL.md" + ).read_text(encoding="utf-8") + canonical = ( + REPO_ROOT / "skills" / "en" / "cursor-cli-skill.md.template" + ).read_text(encoding="utf-8") + + self.assertEqual(updated.action, "updated") + self.assertEqual(installed, canonical) + self.assertNotIn("顶层入口", installed) + + def test_missing_cli_skill_source_does_not_remove_existing_install(self) -> None: + """Source checks happen before the plugin directory is replaced. + + A checkout that lost cursor-cli-skill.md.template must fail closed + and leave the previous Plugin and Skill bytes untouched. + """ + with tempfile.TemporaryDirectory() as home_dir, tempfile.TemporaryDirectory() as repo_dir: + home_root = Path(home_dir) + install_cursor_user_plugin_assets( + CURSOR_ADAPTER, + repo_root=REPO_ROOT, + home_root=home_root, + language_directory="CN", + ) + before = _snapshot(home_root / ".cursor") + + source = Path(repo_dir) / "skills" / "zh" + source.mkdir(parents=True) + for name in ( + "cursor-plugin-rule.mdc.template", + "cursor-plugin-readme.md.template", + ): + shutil.copy(REPO_ROOT / "skills" / "zh" / name, source / name) + + with self.assertRaisesRegex(InstallError, "Missing source Cursor CLI Skill"): + install_cursor_user_plugin_assets( + CURSOR_ADAPTER, + repo_root=Path(repo_dir), + home_root=home_root, + language_directory="CN", + ) + + self.assertEqual(_snapshot(home_root / ".cursor"), before) + + @pytest.mark.xfail( + strict=True, + reason=( + "AUDIT-1: reinstall reports skipped while a required phase skill " + "still differs from the packaged source tree" + ), + ) + def test_reinstall_restores_phase_skill_that_differs_from_source(self) -> None: + """Rerunning the Cursor install must not claim the tree is current when it is not. + + analyze/SKILL.md is a required Doctor path. If its bytes diverge from + the source tree, a repair install has to rewrite it. Skipping leaves + the divergence in place while the host action is skipped. + """ + with tempfile.TemporaryDirectory() as home_dir: + home_root = Path(home_dir) + install_cursor_user_plugin_assets( + CURSOR_ADAPTER, + repo_root=REPO_ROOT, + home_root=home_root, + language_directory="CN", + ) + phase_skill = home_root / ".cursor" / "skills" / "sopify" / "analyze" / "SKILL.md" + source = ( + REPO_ROOT / "skills" / "zh" / "skills" / "sopify" / "analyze" / "SKILL.md" + ).read_text(encoding="utf-8") + phase_skill.write_text("truncated\n", encoding="utf-8") + + repaired = install_cursor_user_plugin_assets( + CURSOR_ADAPTER, + repo_root=REPO_ROOT, + home_root=home_root, + language_directory="CN", + ) + + self.assertNotEqual(repaired.action, "skipped") + self.assertEqual(phase_skill.read_text(encoding="utf-8"), source) diff --git a/tests/test_cursor_hook_branches.py b/tests/test_cursor_hook_branches.py new file mode 100644 index 0000000..4c86982 --- /dev/null +++ b/tests/test_cursor_hook_branches.py @@ -0,0 +1,286 @@ +# Test classification: distribution +"""Branches in installer/cursor_hook.py that tests/test_cursor_hooks.py does not lock. + +The helper copies the plan-package rules instead of calling inspect_plan_package. +These tests keep sessionStart aligned with that contract, and keep the write +guard on the tool and shell shapes the existing file does not execute. +""" + +from __future__ import annotations + +import io +import json +from pathlib import Path +import sys +import tempfile +import unittest +from unittest.mock import patch + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from installer.cursor_hook import handle, main +from sopify_contracts import PLAN_FILES_BY_LEVEL, inspect_plan_package + + +def _enable_workspace(root: Path) -> Path: + (root / ".sopify").mkdir(parents=True, exist_ok=True) + return root + + +def _write_package(plan_dir: Path, level: str, *, plan_md: str | None = None) -> None: + plan_dir.mkdir(parents=True, exist_ok=True) + if plan_md is None: + plan_md = f"---\nlevel: {level}\n---\n# Plan\n" + (plan_dir / "plan.md").write_text(plan_md, encoding="utf-8") + for name in PLAN_FILES_BY_LEVEL[level][1:]: + (plan_dir / name).write_text(f"# {name}\n", encoding="utf-8") + + +def _point_at(workspace: Path, plan_id: str) -> None: + state = workspace / ".sopify" / "state" + state.mkdir(parents=True, exist_ok=True) + (state / "active_plan.json").write_text( + json.dumps({"plan_id": plan_id}), + encoding="utf-8", + ) + + +def _session(workspace: Path) -> dict[str, object]: + return handle( + { + "hook_event_name": "sessionStart", + "workspace_roots": [str(workspace)], + } + ) + + +class CursorHookUncoveredBranchTests(unittest.TestCase): + def test_session_start_follows_plan_package_contract(self) -> None: + """sessionStart must accept exactly the packages inspect_plan_package accepts. + + The hook reimplements the level and semantic-file rules. A standard or + architecture package, or a light package that also has background.md, + must not drift from the writer contract. + """ + for level in PLAN_FILES_BY_LEVEL: + with self.subTest(level=level): + with tempfile.TemporaryDirectory() as workspace_dir: + workspace = _enable_workspace(Path(workspace_dir)) + plan_id = f"plan_{level}" + plan_dir = workspace / ".sopify" / "plan" / plan_id + _write_package(plan_dir, level) + _point_at(workspace, plan_id) + + self.assertTrue(inspect_plan_package(plan_dir).valid) + self.assertIn( + f"active_plan: {plan_id}", + _session(workspace)["additional_context"], + ) + + (plan_dir / "background.md").write_text("legacy\n", encoding="utf-8") + self.assertFalse(inspect_plan_package(plan_dir).valid) + self.assertEqual(_session(workspace), {}) + + def test_session_start_rejects_incomplete_or_unclosed_frontmatter(self) -> None: + """A standard plan without tasks.md, or an unclosed frontmatter, is not active.""" + with tempfile.TemporaryDirectory() as workspace_dir: + workspace = _enable_workspace(Path(workspace_dir)) + plan_dir = workspace / ".sopify" / "plan" / "plan_std" + _write_package(plan_dir, "standard") + (plan_dir / "tasks.md").unlink() + _point_at(workspace, "plan_std") + + self.assertFalse(inspect_plan_package(plan_dir).valid) + self.assertEqual(_session(workspace), {}) + + (plan_dir / "tasks.md").write_text("# Tasks\n", encoding="utf-8") + (plan_dir / "plan.md").write_text("---\nlevel: standard\n# no close\n", encoding="utf-8") + self.assertEqual(_session(workspace), {}) + + def test_session_start_accepts_one_quoted_level_only(self) -> None: + """Quoted level tokens are part of the package contract; two levels are not.""" + with tempfile.TemporaryDirectory() as workspace_dir: + workspace = _enable_workspace(Path(workspace_dir)) + plan_dir = workspace / ".sopify" / "plan" / "plan_std" + _write_package( + plan_dir, + "standard", + plan_md='---\nlevel: "standard"\n---\n# Plan\n', + ) + _point_at(workspace, "plan_std") + + self.assertIn("active_plan: plan_std", _session(workspace)["additional_context"]) + + (plan_dir / "plan.md").write_text( + "---\nlevel: light\nlevel: standard\n---\n# Plan\n", + encoding="utf-8", + ) + self.assertEqual(_session(workspace), {}) + + def test_session_start_ignores_plan_id_outside_the_token_pattern(self) -> None: + """The active pointer has to be the same token the writer accepts. + + A value that is not letters, digits, and underscores must not be + injected as an active plan. + """ + with tempfile.TemporaryDirectory() as workspace_dir: + workspace = _enable_workspace(Path(workspace_dir)) + _point_at(workspace, "plan.std") + plan_dir = workspace / ".sopify" / "plan" / "plan.std" + _write_package(plan_dir, "light") + + self.assertEqual(_session(workspace), {}) + + def test_session_start_orders_zulu_timestamps_ahead_of_older_offsets(self) -> None: + """A Zulu timestamp is a real receipt time, not a missing timestamp.""" + with tempfile.TemporaryDirectory() as workspace_dir: + workspace = _enable_workspace(Path(workspace_dir)) + plan_dir = workspace / ".sopify" / "plan" / "plan_a" + receipts = plan_dir / "receipts" + _write_package(plan_dir, "light") + receipts.mkdir() + (receipts / "exec_001.json").write_text( + json.dumps({"timestamp": "2026-08-19T09:00:00+00:00", "provenance": {"receipt_id": "exec_001"}}), + encoding="utf-8", + ) + (receipts / "exec_002.json").write_text( + json.dumps({"timestamp": "2026-08-19T10:00:00Z", "provenance": {"receipt_id": "exec_002"}}), + encoding="utf-8", + ) + _point_at(workspace, "plan_a") + + self.assertIn("latest_receipt: exec_002", _session(workspace)["additional_context"]) + + (receipts / "exec_002.json").write_text( + json.dumps({"timestamp": "not-a-date", "provenance": {"receipt_id": "exec_002"}}), + encoding="utf-8", + ) + self.assertIn("latest_receipt: exec_001", _session(workspace)["additional_context"]) + + def test_pre_tool_use_denies_delete_and_notebook_edits_of_machine_truth(self) -> None: + """Delete and notebook edits are mutating tools; Read is not. + + Dropping delete or notebook_path from the guard would let those tools + rewrite handoff or receipt files while StrReplace stayed blocked. + """ + with tempfile.TemporaryDirectory() as workspace_dir: + workspace = _enable_workspace(Path(workspace_dir)) + deleted = handle( + { + "hook_event_name": "preToolUse", + "cwd": str(workspace), + "tool_name": "Delete", + "tool_input": {"path": ".sopify/state/current_handoff.json"}, + } + ) + notebook = handle( + { + "hook_event_name": "preToolUse", + "cwd": str(workspace), + "tool_name": "EditNotebook", + "tool_input": { + "notebook_path": ".sopify/plan/plan_a/receipts/exec_001.json", + "new_source": "x", + }, + } + ) + read = handle( + { + "hook_event_name": "preToolUse", + "cwd": str(workspace), + "tool_name": "Read", + "tool_input": {"path": ".sopify/state/active_plan.json"}, + } + ) + + self.assertEqual(deleted["permission"], "deny") + self.assertEqual(notebook["permission"], "deny") + self.assertEqual(read, {"permission": "allow"}) + + def test_pre_tool_use_uses_tool_input_cwd_for_a_relative_path(self) -> None: + """Some events put cwd only inside tool_input. The relative path still resolves.""" + with tempfile.TemporaryDirectory() as workspace_dir: + workspace = _enable_workspace(Path(workspace_dir)) + result = handle( + { + "hook_event_name": "preToolUse", + "tool_name": "Write", + "tool_input": { + "path": ".sopify/state/active_plan.json", + "cwd": str(workspace), + "contents": "{}", + }, + } + ) + + self.assertEqual(result["permission"], "deny") + + def test_before_shell_denies_rm_tee_and_absolute_redirects(self) -> None: + """Obvious deletes and redirects stay denied beyond the echo-with-space case. + + rm, tee, an absolute path that contains a space, and a stderr redirect + are write shapes the existing shell test does not execute. + """ + with tempfile.TemporaryDirectory() as parent_dir: + workspace = _enable_workspace(Path(parent_dir) / "ws dir") + absolute = f"{workspace}/.sopify/state/active_plan.json" + cases = { + "rm": "rm -f .sopify/state/active_plan.json", + "tee": "echo '{}' | tee .sopify/state/current_handoff.json", + "absolute": f"echo '{{}}' > '{absolute}'", + "stderr": "cmd 2>.sopify/state/current_handoff.json", + } + for name, command in cases.items(): + with self.subTest(command=name): + result = handle( + { + "hook_event_name": "beforeShellExecution", + "cwd": str(workspace), + "command": command, + } + ) + self.assertEqual(result["permission"], "deny") + + self.assertEqual( + handle( + { + "hook_event_name": "beforeShellExecution", + "cwd": str(workspace), + "command": "rm -f README.md", + } + ), + {"permission": "allow"}, + ) + self.assertEqual( + handle( + { + "hook_event_name": "beforeShellExecution", + "command": "rm -f .sopify/state/active_plan.json", + } + ), + {"permission": "allow"}, + ) + + def test_main_emits_one_json_line_for_a_valid_session(self) -> None: + """The success path must print the handler result, not the fail-open object.""" + with tempfile.TemporaryDirectory() as workspace_dir: + workspace = _enable_workspace(Path(workspace_dir)) + plan_dir = workspace / ".sopify" / "plan" / "plan_a" + _write_package(plan_dir, "light") + _point_at(workspace, "plan_a") + raw = json.dumps( + { + "hook_event_name": "sessionStart", + "workspace_roots": [str(workspace)], + } + ) + with patch("sys.stdin", io.StringIO(raw)), patch("sys.stdout", io.StringIO()) as out: + exit_code = main() + + rendered = out.getvalue() + self.assertEqual(exit_code, 0) + self.assertEqual(rendered.count("\n"), 1) + payload = json.loads(rendered) + self.assertIn("active_plan: plan_a", payload["additional_context"]) diff --git a/tests/test_install_user_summary.py b/tests/test_install_user_summary.py new file mode 100644 index 0000000..14089aa --- /dev/null +++ b/tests/test_install_user_summary.py @@ -0,0 +1,97 @@ +# Test classification: distribution +"""User-visible install summary gaps found by running the documented installer. + +These are the summary strings a zh-CN Cursor install prints. The assertions +are the behavior the summary should have; current output fails them. +""" + +from __future__ import annotations + +from pathlib import Path +import sys +import tempfile +import unittest + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from installer.distribution import ( + DistributionRequest, + DistributionSourceMetadata, + _friendly_smoke_summary, + render_distribution_user_result, + run_distribution_install, +) +from scripts.install_sopify import run_install + + +def _cursor_zh_summary(home_root: Path) -> str: + report = run_distribution_install( + request=DistributionRequest( + target="cursor:zh-CN", + workspace=None, + ref_override=None, + interactive=False, + source_channel="repo-local", + source_metadata=DistributionSourceMetadata( + resolved_ref="working-tree", + asset_name="scripts/install_sopify.py", + ), + ), + repo_root=REPO_ROOT, + home_root=home_root, + install_executor=run_install, + ) + return render_distribution_user_result(report) + + +class InstallUserSummaryTests(unittest.TestCase): + def test_passed_runtime_smoke_stays_translated(self) -> None: + """A real passing smoke line is already translated; do not drop that branch.""" + self.assertEqual( + _friendly_smoke_summary("Runtime smoke check passed: bundle ok\n", "zh-CN"), + "运行时自检已通过。", + ) + + @pytest.mark.xfail( + strict=True, + reason=( + "AUDIT-2: zh-CN success summary prints the English placeholder " + "(no smoke output) when smoke did not run" + ), + ) + def test_success_summary_does_not_print_empty_smoke_placeholder(self) -> None: + """A finished install with no smoke output must not look like a missing check. + + Cursor has no smoke target, so every successful zh-CN install currently + puts `(no smoke output)` under 诊断. The summary should not surface + that English placeholder. + """ + with tempfile.TemporaryDirectory() as home_dir: + rendered = _cursor_zh_summary(Path(home_dir)) + + self.assertIn("Sopify 安装完成。", rendered) + self.assertNotIn("(no smoke output)", rendered) + + @pytest.mark.xfail( + strict=True, + reason=( + "AUDIT-3: cursor is missing from the host display-name map, " + "so the zh-CN summary says 打开 cursor" + ), + ) + def test_cursor_summary_uses_the_product_display_name(self) -> None: + """The next-step line should use the same Cursor name as the restart step. + + Codex and Claude are mapped to product names. cursor falls through to + the raw host id, so the summary says 打开 cursor next to 重启 Cursor. + """ + with tempfile.TemporaryDirectory() as home_dir: + rendered = _cursor_zh_summary(Path(home_dir)) + + self.assertIn("重启 Cursor", rendered) + self.assertIn("打开 Cursor", rendered) + self.assertNotIn("打开 cursor", rendered) diff --git a/tests/test_sopify_writer_state_guards.py b/tests/test_sopify_writer_state_guards.py new file mode 100644 index 0000000..54b6e8e --- /dev/null +++ b/tests/test_sopify_writer_state_guards.py @@ -0,0 +1,213 @@ +# Test classification: distribution +"""Protocol-state regressions for sopify_writer. + +Covers active_plan, current_handoff, and receipt writes that the existing +writer tests do not execute: handoff/plan disagreement, an occupied history +slot, and a provenance plan_version that does not match the package. +""" + +from __future__ import annotations + +import json +from pathlib import Path +import tempfile +import unittest + +import pytest + +from sopify_contracts import RuntimeHandoff, inspect_plan_package +from sopify_writer import InvariantViolationError, ProtocolStore + + +def _write_light_plan(sopify_root: Path, plan_id: str) -> Path: + plan_dir = sopify_root / "plan" / plan_id + plan_dir.mkdir(parents=True, exist_ok=True) + (plan_dir / "plan.md").write_text( + "---\nlevel: light\n---\n\n# Test plan\n", + encoding="utf-8", + ) + return plan_dir + + +def _handoff(plan_id: str, **overrides: object) -> RuntimeHandoff: + fields: dict[str, object] = { + "schema_version": "1", + "plan_id": plan_id, + "required_host_action": "continue_host_develop", + } + fields.update(overrides) + return RuntimeHandoff(**fields) + + +class ProtocolStoreStateGuardTests(unittest.TestCase): + def test_set_active_plan_does_not_retarget_handoff(self) -> None: + """Moving the active pointer must not silently rewrite the handoff plan. + + Resume reads both files. Coupling the writes would make a pointer + switch look like the old handoff belongs to the new plan. + """ + with tempfile.TemporaryDirectory() as temp_dir: + store = ProtocolStore(Path(temp_dir)) + store.set_active_plan(plan_id="plan_a") + store.set_current_handoff(_handoff("plan_a", notes=("先问再写",))) + store.set_active_plan(plan_id="plan_b") + + self.assertEqual(store.get_active_plan(), {"plan_id": "plan_b"}) + handoff = store.get_current_handoff() + self.assertIsNotNone(handoff) + self.assertEqual(handoff.plan_id, "plan_a") + self.assertEqual(handoff.notes, ("先问再写",)) + raw = store.current_handoff_path.read_text(encoding="utf-8") + self.assertIn("先问再写", raw) + self.assertNotIn("\\u5148", raw) + + def test_handoff_preserves_caller_observability_and_overwrites_writer(self) -> None: + """Caller observability keys survive; the writer identity does not. + + The store injects state_kind, writer, and written_at. Replacing the + whole observability object would drop host-supplied keys, and leaving + a caller-supplied writer would hide who wrote the file. + """ + with tempfile.TemporaryDirectory() as temp_dir: + store = ProtocolStore(Path(temp_dir)) + store.set_current_handoff( + _handoff( + "plan_obs", + observability={"source": "host", "writer": "not-the-writer"}, + ) + ) + + observability = json.loads( + store.current_handoff_path.read_text(encoding="utf-8") + )["observability"] + self.assertEqual(observability["source"], "host") + self.assertEqual(observability["writer"], "sopify_writer") + self.assertEqual(observability["state_kind"], "current_handoff") + self.assertIn("written_at", observability) + + def test_provenance_plan_version_conflict_writes_no_receipt(self) -> None: + """A stale provenance plan_version must not create a receipt file. + + expected_plan_version is already rejected. provenance.plan_version is + a second hash gate and has to fail before the exclusive create. + """ + with tempfile.TemporaryDirectory() as temp_dir: + store = ProtocolStore(Path(temp_dir)) + plan_dir = _write_light_plan(store.root, "plan_001") + current = inspect_plan_package(plan_dir).version + + with self.assertRaises(InvariantViolationError): + store.write_plan_receipt( + plan_id="plan_001", + receipt_id="exec_001", + verdict="pass", + provenance={"plan_version": "sha256:stale"}, + ) + + self.assertFalse((plan_dir / "receipts" / "exec_001.json").exists()) + + written = store.write_plan_receipt( + plan_id="plan_001", + receipt_id="exec_001", + verdict="pass", + provenance={"plan_version": current}, + evidence={"note": "收据"}, + ) + payload = json.loads(written.read_text(encoding="utf-8")) + self.assertEqual(payload["provenance"]["plan_version"], current) + self.assertEqual(payload["evidence"]["note"], "收据") + self.assertIn("收据", written.read_text(encoding="utf-8")) + + def test_finalize_rejects_mismatched_handoff_without_archiving(self) -> None: + """Finalize must stop when the handoff still names another plan. + + The active pointer can match while the handoff does not. Archiving + in that state would clear a handoff that does not belong to the plan. + """ + with tempfile.TemporaryDirectory() as temp_dir: + store = ProtocolStore(Path(temp_dir)) + _write_light_plan(store.root, "target_001") + store.set_active_plan(plan_id="target_001") + store.set_current_handoff(_handoff("other_001")) + + with self.assertRaises(InvariantViolationError): + store.finalize_plan( + plan_id="target_001", + outcome="completed", + summary="Done.", + key_decisions=["A"], + month="2026-06", + ) + + self.assertTrue((store.root / "plan" / "target_001").is_dir()) + self.assertFalse((store.root / "history").exists()) + self.assertEqual(store.get_active_plan(), {"plan_id": "target_001"}) + self.assertEqual(store.get_current_handoff().plan_id, "other_001") + self.assertFalse( + (store.root / "plan" / "target_001" / "receipts" / "final.json").exists() + ) + + def test_finalize_does_not_replace_an_existing_history_plan(self) -> None: + """An occupied history slot must stay byte-for-byte, including its marker. + + Finalize checks the archive directory before writing the final receipt. + A retry against an existing archive must not delete that archive or + clear the live pointer. + """ + with tempfile.TemporaryDirectory() as temp_dir: + store = ProtocolStore(Path(temp_dir)) + _write_light_plan(store.root, "target_001") + store.set_active_plan(plan_id="target_001") + store.set_current_handoff(_handoff("target_001")) + archive = store.root / "history" / "2026-06" / "target_001" + archive.mkdir(parents=True) + (archive / "keep.txt").write_text("keep\n", encoding="utf-8") + + with self.assertRaises(FileExistsError): + store.finalize_plan( + plan_id="target_001", + outcome="completed", + summary="Done.", + key_decisions=["A"], + month="2026-06", + ) + + self.assertEqual((archive / "keep.txt").read_text(encoding="utf-8"), "keep\n") + self.assertTrue((store.root / "plan" / "target_001").is_dir()) + self.assertFalse( + (store.root / "plan" / "target_001" / "receipts" / "final.json").exists() + ) + self.assertEqual(store.get_active_plan(), {"plan_id": "target_001"}) + self.assertIsNotNone(store.get_current_handoff()) + + @pytest.mark.xfail( + strict=True, + reason=( + "AUDIT-4: non-object active_plan.json makes finalize raise " + "AttributeError instead of a protocol error" + ), + ) + def test_finalize_rejects_non_object_active_plan(self) -> None: + """A non-object active pointer must be a protocol error, and must not archive. + + get_active_plan returns the raw JSON value. Finalize then calls .get + on it. A list payload currently escapes as AttributeError instead of + InvariantViolationError. The plan directory still has to stay put. + """ + with tempfile.TemporaryDirectory() as temp_dir: + store = ProtocolStore(Path(temp_dir)) + _write_light_plan(store.root, "target_001") + store.set_active_plan(plan_id="target_001") + store.active_plan_path.write_text("[]\n", encoding="utf-8") + + with self.assertRaises(InvariantViolationError): + store.finalize_plan( + plan_id="target_001", + outcome="completed", + summary="Done.", + key_decisions=["A"], + month="2026-06", + ) + + self.assertTrue((store.root / "plan" / "target_001").is_dir()) + self.assertFalse((store.root / "history").exists())