From 6521748a65cd52aedfe4b741e2156f7ed11e5ccd Mon Sep 17 00:00:00 2001
From: saagpatel <saagarpatelai@gmail.com>
Date: Wed, 5 Aug 2026 02:58:37 -0700
Subject: [PATCH] Allow replay of the same security receipt

---
 src/portfolio_truth_publish.py | 22 ++++++++++-
 tests/test_portfolio_truth.py  | 69 +++++++++++++++++++++++++++++++++-
 2 files changed, 89 insertions(+), 2 deletions(-)

diff --git a/src/portfolio_truth_publish.py b/src/portfolio_truth_publish.py
index 184b9a7..0548364 100644
--- a/src/portfolio_truth_publish.py
+++ b/src/portfolio_truth_publish.py
@@ -79,6 +79,20 @@ def _parse_bound_datetime(value: object, *, field: str) -> datetime:
     return parsed
 
 
+def _is_same_bound_security_receipt(
+    prior_metadata: dict[str, object],
+    current_metadata: dict[str, object],
+) -> bool:
+    """Return true only when both payloads identify the same immutable receipt."""
+    binding_fields = ("receipt_id", "content_sha256", "producer_commit")
+    return all(
+        isinstance(prior_metadata.get(field), str)
+        and bool(str(prior_metadata[field]).strip())
+        and prior_metadata[field] == current_metadata.get(field)
+        for field in binding_fields
+    )
+
+
 def _load_prior_security_alerts(
     latest_path: Path,
     *,
@@ -145,6 +159,9 @@ def _load_prior_security_alerts(
                 "Prior PortfolioTruth security evidence is not immutably bound."
             )
 
+    prior_security_metadata = (canonical.get("inputs") or {}).get(
+        "github_security"
+    ) or {}
     prior_generated_at = _parse_bound_datetime(
         canonical.get("generated_at"),
         field="Prior PortfolioTruth generated_at",
@@ -153,7 +170,10 @@ def _load_prior_security_alerts(
         current_security_metadata.get("produced_at"),
         field="Current security receipt produced_at",
     )
-    if prior_generated_at > current_produced_at:
+    if prior_generated_at > current_produced_at and not _is_same_bound_security_receipt(
+        prior_security_metadata,
+        current_security_metadata,
+    ):
         raise PortfolioTruthPublishError(
             "Prior PortfolioTruth was generated after the current security receipt."
         )
diff --git a/tests/test_portfolio_truth.py b/tests/test_portfolio_truth.py
index 94f0845..cb0dff8 100644
--- a/tests/test_portfolio_truth.py
+++ b/tests/test_portfolio_truth.py
@@ -591,6 +591,7 @@ def _legacy_prior_security_payload(
     portfolio_workspace: Path,
     portfolio_catalog: Path,
     legacy_registry: Path,
+    snapshot_generated_at: datetime | None = None,
 ) -> tuple[dict, dict, dict]:
     now = datetime(2026, 8, 4, 12, tzinfo=timezone.utc)
     alpha = portfolio_workspace / "Alpha"
@@ -659,7 +660,7 @@ def _legacy_prior_security_payload(
         catalog_path=portfolio_catalog,
         legacy_registry_path=legacy_registry,
         include_notion=False,
-        now=now,
+        now=snapshot_generated_at or now,
         security_alerts_by_name=security,
         security_coverage_metadata=metadata,
     )
@@ -706,6 +707,72 @@ def test_prior_security_loader_accepts_bounded_legacy_truth(
     assert evidence.content_sha256 == hashlib.sha256(latest.read_bytes()).hexdigest()
 
 
+def test_prior_security_loader_allows_same_receipt_truth_generated_after_receipt(
+    tmp_path: Path,
+    portfolio_workspace: Path,
+    portfolio_catalog: Path,
+    legacy_registry: Path,
+) -> None:
+    import src.portfolio_truth_publish as publish_mod
+
+    _, metadata, payload = _legacy_prior_security_payload(
+        portfolio_workspace=portfolio_workspace,
+        portfolio_catalog=portfolio_catalog,
+        legacy_registry=legacy_registry,
+        snapshot_generated_at=datetime(2026, 8, 4, 12, 0, 1, tzinfo=timezone.utc),
+    )
+    latest = tmp_path / "portfolio-truth-latest.json"
+    latest.write_text(json.dumps(payload), encoding="utf-8")
+
+    evidence = publish_mod._load_prior_security_alerts(
+        latest,
+        current_security_metadata=metadata,
+        security_max_age_hours=24,
+    )
+
+    assert evidence.final_cohort_repositories == ("d/Alpha",)
+    assert evidence.alerts_by_full_name["d/Alpha"]["dependabot_high"] == 1
+
+
+@pytest.mark.parametrize(
+    ("binding_field", "replacement"),
+    (
+        ("receipt_id", "sha256:" + "e" * 64),
+        ("content_sha256", "f" * 64),
+        ("producer_commit", "b" * 40),
+    ),
+)
+def test_prior_security_loader_refuses_future_truth_from_different_receipt(
+    binding_field: str,
+    replacement: str,
+    tmp_path: Path,
+    portfolio_workspace: Path,
+    portfolio_catalog: Path,
+    legacy_registry: Path,
+) -> None:
+    import src.portfolio_truth_publish as publish_mod
+
+    _, metadata, payload = _legacy_prior_security_payload(
+        portfolio_workspace=portfolio_workspace,
+        portfolio_catalog=portfolio_catalog,
+        legacy_registry=legacy_registry,
+        snapshot_generated_at=datetime(2026, 8, 4, 12, 0, 1, tzinfo=timezone.utc),
+    )
+    latest = tmp_path / "portfolio-truth-latest.json"
+    latest.write_text(json.dumps(payload), encoding="utf-8")
+    current_metadata = {**metadata, binding_field: replacement}
+
+    with pytest.raises(
+        PortfolioTruthPublishError,
+        match="generated after the current security receipt",
+    ):
+        publish_mod._load_prior_security_alerts(
+            latest,
+            current_security_metadata=current_metadata,
+            security_max_age_hours=24,
+        )
+
+
 @pytest.mark.parametrize(
     ("mutation", "expected_message"),
     (