Bind C-25 JSON to the scanned source commit (STATE-WP-0094-T03)

The checker JSON reports per repository repo_slug, source_revision (full
40-hex HEAD SHA) and source_revision_reason; C-25 findings repeat them so a
findings-only consumer can bind. A dirty workplans/ tree, a non-git path,
a missing HEAD or a tree changing mid-scan yields null plus a reason, never
HEAD or a branch name. plan_owner_returns validates and passes the revision
into every candidate and refuses findings bound elsewhere.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 692292@bnt-lap001
Assistant-Session: 89c7d8e8-e1b3-47eb-9411-a0ab227fca2f

tests/test_c25_source_revision.py

@@ -0,0 +1,155 @@
+"""C-25 JSON binds to the scanned source commit (STATE-WP-0094)."""
+from __future__ import annotations
+
+import subprocess
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))
+
+from consistency_check import (  # noqa: E402
+    ConsistencyReport, report_to_dict, scan_source_revision,
+)
+
+
+def _git(repo: Path, *args: str) -> str:
+    return subprocess.run(
+        ["git", "-c", "user.name=t", "-c", "user.email=t@example.invalid", *args],
+        cwd=repo, check=True, capture_output=True, text=True,
+    ).stdout.strip()
+
+
+def _repo(tmp_path: Path) -> Path:
+    repo = tmp_path / "repo"
+    (repo / "workplans").mkdir(parents=True)
+    (repo / "workplans" / "DEMO-WP-0001-x.md").write_text("---\nid: DEMO-WP-0001\n---\n")
+    (repo / "README.md").write_text("demo\n")
+    _git(repo, "init", "-q")
+    _git(repo, "add", ".")
+    _git(repo, "commit", "-qm", "init")
+    return repo
+
+
+def test_clean_tree_reports_full_head_sha(tmp_path):
+    repo = _repo(tmp_path)
+    sha, reason = scan_source_revision(repo)
+    assert sha == _git(repo, "rev-parse", "HEAD")
+    assert len(sha) == 40 and sha == sha.lower()
+    assert reason is None
+
+
+def test_dirt_outside_workplans_does_not_unbind(tmp_path):
+    repo = _repo(tmp_path)
+    (repo / "README.md").write_text("changed\n")
+    (repo / "scratch.txt").write_text("x\n")
+    assert scan_source_revision(repo) == (_git(repo, "rev-parse", "HEAD"), None)
+
+
+def test_modified_staged_or_untracked_workplans_are_not_head(tmp_path):
+    repo = _repo(tmp_path)
+    plan = repo / "workplans" / "DEMO-WP-0001-x.md"
+    plan.write_text("---\nid: DEMO-WP-0001\nstatus: active\n---\n")
+    assert scan_source_revision(repo) == (None, "dirty_workplans")
+    _git(repo, "add", ".")
+    assert scan_source_revision(repo) == (None, "dirty_workplans")
+    _git(repo, "commit", "-qm", "two")
+    assert scan_source_revision(repo)[1] is None
+    (repo / "workplans" / "archived").mkdir()
+    (repo / "workplans" / "archived" / "new.md").write_text("---\n---\n")
+    assert scan_source_revision(repo) == (None, "dirty_workplans")
+
+
+def test_ignored_workplan_file_is_not_head(tmp_path):
+    repo = _repo(tmp_path)
+    (repo / ".gitignore").write_text("*.local.md\n")
+    _git(repo, "add", ".")
+    _git(repo, "commit", "-qm", "ignore")
+    (repo / "workplans" / "secret.local.md").write_text("---\n---\n")
+    assert scan_source_revision(repo) == (None, "dirty_workplans")
+
+
+def test_non_git_path_and_missing_head(tmp_path):
+    plain = tmp_path / "plain"
+    (plain / "workplans").mkdir(parents=True)
+    assert scan_source_revision(plain) == (None, "not_a_git_tree")
+    assert scan_source_revision(tmp_path / "absent") == (None, "not_a_git_tree")
+    empty = tmp_path / "empty"
+    empty.mkdir()
+    _git(empty, "init", "-q")
+    assert scan_source_revision(empty) == (None, "no_head_commit")
+
+
+def _report(sha, reason):
+    report = ConsistencyReport(repo_slug="demo", repo_path="/x",
+                               source_revision=sha, source_revision_reason=reason)
+    report.add(severity="WARN", check_id="C-25", message="m", file_path="workplans/p.md#T01",
+               file_value="message-from:provider",
+               _fix_context={"task_record_id": "DEMO-WP-0001-T01", "message_ids": ["a"]})
+    report.add(severity="WARN", check_id="C-25", message="legacy", file_value="x")
+    report.add(severity="WARN", check_id="C-24", message="other")
+    return report_to_dict(report)
+
+
+def test_json_report_and_c25_findings_carry_revision():
+    sha = "0123456789abcdef0123456789abcdef01234567"
+    data = _report(sha, None)
+    assert (data["repo_slug"], data["source_revision"], data["source_revision_reason"]) == ("demo", sha, None)
+    c25 = [i for i in data["issues"] if i["check_id"] == "C-25"]
+    assert len(c25) == 2
+    for issue in c25:
+        assert (issue["repo_slug"], issue["source_revision"], issue["source_revision_reason"]) == ("demo", sha, None)
+    other = next(i for i in data["issues"] if i["check_id"] == "C-24")
+    assert "source_revision" not in other and "repo_slug" not in other
+
+
+def test_json_unknown_revision_is_null_with_reason():
+    data = _report(None, "dirty_workplans")
+    assert data["source_revision"] is None
+    assert data["source_revision_reason"] == "dirty_workplans"
+    assert all(i["source_revision"] is None for i in data["issues"] if i["check_id"] == "C-25")
+
+
+def test_unscanned_report_defaults_to_null():
+    data = report_to_dict(ConsistencyReport(repo_slug="demo", repo_path="(unknown)"))
+    assert (data["source_revision"], data["source_revision_reason"]) == (None, "not_scanned")
+
+
+def _check(repo, monkeypatch, sweep=None):
+    import socket
+
+    import consistency_check
+
+    def fake_get(_api_base, path, params=None, **_kwargs):
+        if path == "/repos/demo":
+            return {"id": "repo-1", "slug": "demo", "local_path": str(repo),
+                    "host_paths": {socket.gethostname(): str(repo)}, "domain_slug": "infotech"}
+        return []
+
+    monkeypatch.setattr("consistency_check.load_classification_file", lambda _d: ({}, [], []))
+    monkeypatch.setattr("consistency_check._api_get", fake_get)
+    if sweep:
+        monkeypatch.setattr("consistency_check._check_blocked_unblock_sweep", sweep)
+    return consistency_check.check_repo("http://unused", "demo")
+
+
+def test_check_repo_binds_clean_scan_to_head(tmp_path, monkeypatch):
+    repo = _repo(tmp_path)
+    report = _check(repo, monkeypatch)
+    assert (report.source_revision, report.source_revision_reason) == (_git(repo, "rev-parse", "HEAD"), None)
+
+
+def test_check_repo_dirty_workplans_is_null(tmp_path, monkeypatch):
+    repo = _repo(tmp_path)
+    (repo / "workplans" / "DEMO-WP-0001-x.md").write_text("---\nid: DEMO-WP-0001\nstatus: active\n---\n")
+    report = _check(repo, monkeypatch)
+    assert (report.source_revision, report.source_revision_reason) == (None, "dirty_workplans")
+
+
+def test_check_repo_tree_changed_during_scan_is_null(tmp_path, monkeypatch):
+    repo = _repo(tmp_path)
+
+    def sweep(_api, _slug, repo_dir, _infos, _report):
+        (repo_dir / "workplans" / "late.md").write_text("---\n---\n")
+
+    report = _check(repo, monkeypatch, sweep)
+    assert (report.source_revision, report.source_revision_reason) == (None, "changed_during_scan")

tests/test_owner_return_handoff.py

@@ -195,3 +195,40 @@ def test_resolved_decision_does_not_override_human_gate():
     assert plan(report(), activity_definition_id=DEF, current_tasks=tasks)[0]["stop_reason"] == "human_gate"
     row["needs_human"] = False
     assert plan(report(), activity_definition_id=DEF, current_tasks=tasks)[0]["disposition"] == "reviewable"
+
+
+SHA = "0123456789abcdef0123456789abcdef01234567"
+
+
+def test_source_revision_passes_through_to_every_candidate():
+    source = report()
+    source["source_revision"] = SHA
+    source["issues"][0].update(repo_slug="state-hub", source_revision=SHA)
+    rows = plan(source, activity_definition_id=DEF)
+    assert rows and all(row["source_revision"] == SHA for row in rows)
+    # Dedup identity is revision-independent, so a rescan does not re-trigger.
+    assert rows[0]["triggering_event_id"] == plan(report(), activity_definition_id=DEF)[0]["triggering_event_id"]
+
+
+def test_unknown_revision_travels_as_null():
+    source = report()
+    source["source_revision"] = None
+    assert plan(source, activity_definition_id=DEF)[0]["source_revision"] is None
+    assert plan(report(), activity_definition_id=DEF)[0]["source_revision"] is None
+
+
+@pytest.mark.parametrize("bad", ["main", SHA[:12], SHA.upper(), 7])
+def test_non_sha_revision_is_refused(bad):
+    source = report()
+    source["source_revision"] = bad
+    with pytest.raises(ValueError, match="source_revision"):
+        plan(source, activity_definition_id=DEF)
+
+
+@pytest.mark.parametrize("field,value", [("source_revision", "f" * 40), ("repo_slug", "other")])
+def test_finding_bound_elsewhere_is_refused(field, value):
+    source = report()
+    source["source_revision"] = SHA
+    source["issues"][0][field] = value
+    with pytest.raises(ValueError, match="different repository or revision"):
+        plan(source, activity_definition_id=DEF)