feat: harden work-record and SBOM client contracts

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d

tests/test_record_identity.py

@@ -0,0 +1,87 @@
+from pathlib import Path
+
+import pytest
+
+from repo_manager.observe import RecordIdentityCollisionError, observe_repository
+from repo_manager.record_identity import classify_record_id, scan_record_identities
+from repo_manager.standards import check_repository
+
+
+def _write_mason_workplan(repo: Path, second_uuid: str) -> None:
+    path = repo / "workplans" / "MASON-0001-statehub-bootstrap.md"
+    path.parent.mkdir(parents=True)
+    path.write_text(
+        f"""---
+id: MASON-0001
+type: workplan
+title: Bootstrap
+status: finished
+state_hub_workstream_id: "11111111-1111-4111-8111-111111111111"
+---
+
+## First occurrence
+
+```task
+id: MASON-0001-T01
+status: done
+priority: high
+state_hub_task_id: "ede1aa62-fc42-494a-9984-3190a52481af"
+```
+
+## Second occurrence
+
+```task
+id: MASON-0001-T01
+status: done
+priority: high
+state_hub_task_id: "{second_uuid}"
+```
+""",
+        encoding="utf-8",
+    )
+
+
+def test_grandfathered_mason_identity_is_loaded_from_canon_registry() -> None:
+    assert classify_record_id("workplan", "MASON-0001") == "grandfathered"
+    assert classify_record_id("task", "MASON-0001-T01") == "grandfathered"
+
+
+def test_same_id_and_uuid_indexes_once_with_cleanup_diagnostic(tmp_path: Path) -> None:
+    _write_mason_workplan(tmp_path, "ede1aa62-fc42-494a-9984-3190a52481af")
+
+    report = scan_record_identities(tmp_path)
+    assert report["identity_collisions"] == []
+    assert report["duplicate_source_occurrences"][0]["sources"] == [
+        "workplans/MASON-0001-statehub-bootstrap.md#task-block-1",
+        "workplans/MASON-0001-statehub-bootstrap.md#task-block-2",
+    ]
+
+    snapshot, index = observe_repository(tmp_path, slug="ops-mason")
+    tasks = [record for record in index.work_records if record.kind == "task"]
+    assert snapshot["index"]["task_count"] == 1
+    assert len(tasks) == 1
+    assert (
+        tasks[0].extra["source_occurrences"] == report["duplicate_source_occurrences"][0]["sources"]
+    )
+    assert index.events[0]["type"] == "repo.work_record.duplicate_source_occurrence"
+
+    conformance = check_repository(tmp_path, slug="ops-mason")
+    codes = [finding.code for finding in conformance.findings]
+    assert "work-record-duplicate-source-occurrence" in codes
+    assert not any(
+        finding.code == "work-record-id-invalid" and finding.path.startswith("workplans/")
+        for finding in conformance.findings
+    )
+
+
+def test_same_id_with_different_uuids_fails_closed(tmp_path: Path) -> None:
+    _write_mason_workplan(tmp_path, "22222222-2222-4222-8222-222222222222")
+
+    report = scan_record_identities(tmp_path)
+    assert report["duplicate_source_occurrences"] == []
+    assert report["identity_collisions"][0]["id"] == "MASON-0001-T01"
+    with pytest.raises(RecordIdentityCollisionError):
+        observe_repository(tmp_path, slug="ops-mason")
+
+    conformance = check_repository(tmp_path, slug="ops-mason")
+    assert any(finding.code == "work-record-identity-collision" for finding in conformance.findings)

tests/test_registrar_reconcile.py

@@ -120,6 +120,39 @@ status: open
     assert missing["decisions"] == ["DEMO-DEC-0001"]
 
 
+def test_missing_scan_includes_lowercase_top_level_record_files(tmp_path: Path) -> None:
+    repo = _fixture(tmp_path)
+    (repo / "intakes.md").write_text(
+        """# Intakes
+
+```yaml
+id: DEMO-IN-0002
+kind: intake
+title: Standalone intake
+status: open
+```
+""",
+        encoding="utf-8",
+    )
+    (repo / "decisions.md").write_text(
+        """# Decisions
+
+```yaml
+id: DEMO-DEC-0002
+kind: decision
+title: Standalone decision
+status: accepted
+```
+""",
+        encoding="utf-8",
+    )
+
+    missing = rr._missing_identifiers(repo)
+
+    assert missing["intakes"] == ["DEMO-IN-0002"]
+    assert missing["decisions"] == ["DEMO-DEC-0002"]
+
+
 def test_scopes_registrar_env_and_commits_assigned_ids(tmp_path: Path, monkeypatch) -> None:
     repo = _fixture(tmp_path)
     monkeypatch.setattr(rr, "_check_primary", lambda _api: ({"status": "ok", "db": "connected"}, None))

tests/test_sbom.py

@@ -4,10 +4,14 @@ import json
 import subprocess
 from pathlib import Path
 
+import pytest
+
 from repo_manager.cli import main
 from repo_manager.sbom_client import (
+    SBOMContractError,
     licence_report_from_snapshot,
     scan_repository_via_nexus,
+    validate_snapshot_contract,
 )
 
 
@@ -74,6 +78,14 @@ def test_scan_delegates_to_sbom_nexus_without_shell(monkeypatch, tmp_path: Path)
     assert result["schema"] == "sbom-nexus.snapshot.v1"
     assert result["product_owner"] == "sbom-nexus"
     assert result["delegated_by"] == "repo-manager"
+    assert result["repo_manager_context"] == {
+        "mode": "local-preview",
+        "authoritative": False,
+        "persisted": False,
+        "advances_last_attempt_at": False,
+        "advances_last_success_at": False,
+        "creates_snapshot_history": False,
+    }
 
 
 def test_missing_nexus_cli_returns_actionable_error(monkeypatch, tmp_path: Path) -> None:
@@ -98,10 +110,41 @@ def test_licence_report_alias_preserves_shape() -> None:
         "entry_count",
         "licence_report",
         "errors",
+        "repo_manager_context",
         "delegated_by",
         "product_owner",
     }
     assert result["licence_report"]["copyleft_direct_count"] == 0
+    assert result["repo_manager_context"]["authoritative"] is False
+
+
+def test_snapshot_contract_allows_additive_fields_and_rejects_unknown_schema() -> None:
+    validate_snapshot_contract({**_snapshot(), "future_addition": {"accepted": True}})
+
+    with pytest.raises(SBOMContractError, match="unsupported SBOM Nexus schema"):
+        validate_snapshot_contract({**_snapshot(), "schema": "sbom-nexus.snapshot.v2"})
+
+
+def test_scan_turns_unknown_schema_into_deterministic_contract_error(
+    monkeypatch, tmp_path: Path
+) -> None:
+    monkeypatch.setenv("SBOM_NEXUS_CLI", "/opt/sbom-nexus/bin/sbom-nexus")
+    monkeypatch.setattr(
+        subprocess,
+        "run",
+        lambda command, **kwargs: subprocess.CompletedProcess(
+            command,
+            0,
+            json.dumps({**_snapshot(), "schema": "future.snapshot.v9"}),
+            "",
+        ),
+    )
+
+    result = scan_repository_via_nexus(tmp_path)
+
+    assert result["ok"] is False
+    assert result["errors"][0]["reason"] == "sbom-nexus-contract"
+    assert result["repo_manager_context"]["persisted"] is False
 
 
 def test_cli_scan_preserves_output_file_behavior(monkeypatch, tmp_path: Path, capsys) -> None: