feat(portfolio): ingest producer preflight receipts

src/app/portfolio_truth.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 from pathlib import Path
 from typing import Any
 
@@ -19,6 +20,7 @@
     load_security_alerts_by_name,
     warn_if_warehouse_report_stale,
 )
+from src.producer_preflight import load_producer_evidence
 
 
 def run_portfolio_truth_mode(args: Any) -> None:
@@ -35,6 +37,16 @@ def run_portfolio_truth_mode(args: Any) -> None:
         else workspace_root / "PORTFOLIO-AUDIT-REPORT.md"
     )
     legacy_registry_path = Path(args.registry) if args.registry else registry_output
+    producer_evidence_path = os.environ.get("GHRA_PRODUCER_EVIDENCE")
+    producer_evidence = (
+        load_producer_evidence(Path(producer_evidence_path))
+        if producer_evidence_path
+        else None
+    )
+    producer_repo_root_value = os.environ.get("GHRA_PRODUCER_REPO_ROOT")
+    producer_repo_root = (
+        Path(producer_repo_root_value) if producer_repo_root_value else None
+    )
     release_count_by_name: dict[str, int] | None = None
     if getattr(args, "portfolio_truth_include_release_count", False):
         release_count_by_name = load_release_count_by_name(
@@ -70,8 +82,13 @@ def run_portfolio_truth_mode(args: Any) -> None:
             release_count_by_name=release_count_by_name,
             security_alerts_by_name=security_alerts_by_name,
             repo_status_by_name=repo_status_by_name,
+            producer_evidence=producer_evidence,
+            producer_repo_root=producer_repo_root,
+            require_producer_evidence=bool(
+                os.environ.get("GHRA_REQUIRE_PRODUCER_EVIDENCE") == "1"
+            ),
         )
-    except PortfolioTruthPublishError as exc:
+    except (PortfolioTruthPublishError, ValueError) as exc:
         raise SystemExit(str(exc)) from exc
     print_info(f"Portfolio truth snapshot: {result.latest_path}")
     print_info(f"Portfolio truth history snapshot: {result.snapshot_path}")

src/producer_preflight.py

@@ -32,6 +32,42 @@ def to_dict(self) -> dict[str, Any]:
             "verified_at": self.verified_at.isoformat(),
         }
 
+    @classmethod
+    def from_dict(cls, payload: dict[str, Any]) -> ProducerEvidence:
+        required = {
+            "repository",
+            "commit",
+            "ref",
+            "checkout_role",
+            "worktree_clean",
+            "verified_at",
+        }
+        missing = sorted(required - payload.keys())
+        if missing:
+            raise ValueError(f"Producer evidence is missing fields: {missing}")
+        verified_at = str(payload["verified_at"])
+        if verified_at.endswith("Z"):
+            verified_at = f"{verified_at[:-1]}+00:00"
+        try:
+            parsed_verified_at = datetime.fromisoformat(verified_at)
+        except ValueError as exc:
+            raise ValueError("Producer evidence verified_at is not valid ISO-8601.") from exc
+        if parsed_verified_at.tzinfo is None:
+            raise ValueError("Producer evidence verified_at must include a timezone.")
+        commit = str(payload["commit"])
+        if len(commit) != 40 or any(char not in "0123456789abcdef" for char in commit):
+            raise ValueError("Producer evidence commit must be a lowercase 40-character SHA.")
+        if payload["worktree_clean"] is not True:
+            raise ValueError("Producer evidence must declare a clean worktree.")
+        return cls(
+            repository=str(payload["repository"]),
+            commit=commit,
+            ref=str(payload["ref"]),
+            checkout_role=str(payload["checkout_role"]),
+            worktree_clean=True,
+            verified_at=parsed_verified_at.astimezone(UTC),
+        )
+
 
 @dataclass(frozen=True)
 class ProducerPreflightResult:
@@ -103,6 +139,27 @@ def verify_evidence_still_current(repo_root: Path, evidence: ProducerEvidence) -
         )
 
 
+def load_producer_evidence(path: Path) -> ProducerEvidence:
+    try:
+        payload = json.loads(path.read_text())
+    except FileNotFoundError as exc:
+        raise ValueError(f"Producer evidence file does not exist: {path}") from exc
+    except json.JSONDecodeError as exc:
+        raise ValueError(f"Producer evidence file is not valid JSON: {path}") from exc
+    if not isinstance(payload, dict):
+        raise ValueError("Producer evidence root must be a JSON object.")
+    if payload.get("schema_version") != PREFLIGHT_SCHEMA_VERSION:
+        raise ValueError(
+            "Producer evidence schema mismatch: "
+            f"declared={payload.get('schema_version')!r}; expected={PREFLIGHT_SCHEMA_VERSION!r}"
+        )
+    if payload.get("state") != "pass":
+        raise ValueError(
+            f"Producer evidence did not pass preflight: state={payload.get('state')!r}"
+        )
+    return ProducerEvidence.from_dict(payload)
+
+
 def _origin_repository(repo_root: Path) -> str:
     remote = _git(repo_root, "remote", "get-url", "origin")
     if remote.startswith("git@") and ":" in remote:

tests/test_portfolio_truth.py

@@ -1617,6 +1617,78 @@ def test_report_subcommand_parses_allow_empty_notion_flag() -> None:
     assert default.portfolio_truth_allow_empty_notion is False
 
 
+def test_portfolio_truth_app_passes_validated_producer_receipt_to_publisher(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from types import SimpleNamespace
+
+    from src.app.portfolio_truth import run_portfolio_truth_mode
+    from src.producer_preflight import PREFLIGHT_SCHEMA_VERSION
+
+    receipt = tmp_path / "producer.json"
+    receipt.write_text(
+        json.dumps(
+            {
+                "schema_version": PREFLIGHT_SCHEMA_VERSION,
+                "state": "pass",
+                "repository": "saagpatel/GithubRepoAuditor",
+                "commit": "a" * 40,
+                "ref": "refs/remotes/origin/main",
+                "checkout_role": "canonical-automation",
+                "worktree_clean": True,
+                "verified_at": "2026-07-10T12:00:00Z",
+                "checks": {},
+            }
+        )
+    )
+    captured: dict[str, object] = {}
+
+    def fake_publish(**kwargs):
+        captured.update(kwargs)
+        return SimpleNamespace(
+            latest_path=tmp_path / "latest.json",
+            snapshot_path=tmp_path / "history.json",
+            registry_output=tmp_path / "registry.md",
+            portfolio_report_output=tmp_path / "report.md",
+            project_count=0,
+            registry_changed=False,
+            report_changed=False,
+        )
+
+    monkeypatch.setattr("src.app.portfolio_truth.publish_portfolio_truth", fake_publish)
+    monkeypatch.setattr(
+        "src.app.portfolio_truth.load_live_repo_status_by_name", lambda **_kwargs: {}
+    )
+    monkeypatch.setattr(
+        "src.app.portfolio_truth.warn_if_warehouse_report_stale", lambda *_args: None
+    )
+    monkeypatch.setenv("GHRA_REQUIRE_PRODUCER_EVIDENCE", "1")
+    monkeypatch.setenv("GHRA_PRODUCER_EVIDENCE", str(receipt))
+    monkeypatch.setenv("GHRA_PRODUCER_REPO_ROOT", str(tmp_path / "producer-repo"))
+    args = SimpleNamespace(
+        output_dir=str(tmp_path / "output"),
+        workspace_root=str(tmp_path),
+        registry_output=str(tmp_path / "registry.md"),
+        portfolio_report_output=str(tmp_path / "report.md"),
+        registry=None,
+        catalog=None,
+        username="testuser",
+        token=None,
+        no_cache=True,
+        portfolio_truth_include_release_count=False,
+        portfolio_truth_include_security=False,
+        portfolio_truth_allow_empty_notion=False,
+    )
+
+    run_portfolio_truth_mode(args)
+
+    evidence = captured["producer_evidence"]
+    assert evidence.commit == "a" * 40
+    assert captured["producer_repo_root"] == tmp_path / "producer-repo"
+    assert captured["require_producer_evidence"] is True
+
+
 def test_cli_portfolio_truth_allow_empty_notion_carries_forward(
     portfolio_workspace: Path,
     portfolio_catalog: Path,

tests/test_producer_preflight.py

@@ -7,8 +7,10 @@
 import pytest
 
 from src.producer_preflight import (
+    PREFLIGHT_SCHEMA_VERSION,
     ProducerEvidence,
     inspect_canonical_producer,
+    load_producer_evidence,
     verify_evidence_still_current,
 )
 
@@ -91,3 +93,42 @@ def test_evidence_rejects_head_change(tmp_path: Path) -> None:
     _git(repo, "commit", "-m", "move head")
     with pytest.raises(ValueError, match="HEAD changed after preflight"):
         verify_evidence_still_current(repo, evidence)
+
+
+def test_load_producer_evidence_accepts_passing_receipt(tmp_path: Path) -> None:
+    path = tmp_path / "producer.json"
+    path.write_text(
+        __import__("json").dumps(
+            {
+                "schema_version": PREFLIGHT_SCHEMA_VERSION,
+                "state": "pass",
+                "repository": "saagpatel/GithubRepoAuditor",
+                "commit": "a" * 40,
+                "ref": "refs/remotes/origin/main",
+                "checkout_role": "canonical-automation",
+                "worktree_clean": True,
+                "verified_at": "2026-07-10T12:00:00Z",
+                "checks": {},
+            }
+        )
+    )
+
+    evidence = load_producer_evidence(path)
+
+    assert evidence.commit == "a" * 40
+    assert evidence.verified_at.tzinfo is not None
+
+
+def test_load_producer_evidence_rejects_nonpassing_receipt(tmp_path: Path) -> None:
+    path = tmp_path / "producer.json"
+    path.write_text(
+        __import__("json").dumps(
+            {
+                "schema_version": PREFLIGHT_SCHEMA_VERSION,
+                "state": "fail",
+            }
+        )
+    )
+
+    with pytest.raises(ValueError, match="did not pass preflight"):
+        load_producer_evidence(path)