fix(portfolio-truth): derive the security cohort at the plan instant

CHANGELOG.md

@@ -13,6 +13,8 @@ Format: [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 
 ## [Unreleased]
 
+- Derive receipt-backed security cohort and final attention at the cohort plan's
+  instant, preserving wall-clock freshness checks and legacy receipt behavior.
 - Clarify locked contributor setup, optional test dependencies, and credential-free fixture verification without workstation discovery.
 
 ### Added

docs/portfolio-cohort-transition-contract.md

@@ -57,6 +57,17 @@ named.
 producer both call it; `tests/test_portfolio_cohort_plan.py` asserts that, because
 two equivalent algorithms would silently drift apart and reproduce the deadlock.
 
+The collector carries the plan's `generated_at` as the optional
+`cohort.transition.plan_generated_at` receipt field. Publication derives both
+the pre-security candidate (R3) and final attention (R5) at that instant, so a
+recency boundary between planning and publication cannot change membership by
+itself. Current source is still read and exact set equality still rejects a
+catalog or repository membership change. Receipt freshness, provider age
+checks, local observation timestamps, and snapshot generation use the
+publication clock. Older receipts without this field retain wall-clock
+derivation. This is an additive extension of `GitHubSecurityCoverageReceiptV1`;
+the field, when present, must be a timestamp with a timezone.
+
 ## Departure evidence
 
 A repository may leave the published default-attention cohort only with positive

src/github_repo_auditor/github_security_coverage.py

@@ -1894,6 +1894,7 @@ def collect_security_coverage(
             "incoming": list(cohort_plan.incoming_repositories),
             "outgoing": list(cohort_plan.outgoing_repositories),
             "plan_id": cohort_plan.plan_id,
+            "plan_generated_at": cohort_plan.generated_at,
             "prior_truth_sha256": cohort_plan.prior_truth_sha256,
         }
         expected_count = len(cohort)
@@ -2783,6 +2784,11 @@ def _named_set(key: str) -> tuple[str, ...]:
     plan_id = transition.get("plan_id")
     if not isinstance(plan_id, str) or not _RECEIPT_ID_RE.fullmatch(plan_id):
         raise SecurityCoverageError("cohort transition plan_id must be a sha256 identity")
+    if "plan_generated_at" in transition:
+        _parse_datetime(
+            transition["plan_generated_at"],
+            field_name="cohort.transition.plan_generated_at",
+        )
     prior_truth_sha256 = transition.get("prior_truth_sha256")
     if not isinstance(prior_truth_sha256, str) or not _SHA256_RE.fullmatch(
         prior_truth_sha256

src/github_repo_auditor/portfolio_truth_reconcile.py

@@ -644,10 +644,12 @@ class MaterializationContext:
 
     Both callers must derive candidate attention from one implementation or the
     two derivations drift, which is the failure this contract exists to prevent.
+    `now` is the derivation clock; `observed_at` records current local evidence.
     """
 
     workspace_root: Path
     now: datetime
+    observed_at: datetime
     catalog_data: dict[str, Any]
     legacy_rows: dict[str, dict[str, str]]
     notion_context: dict[str, dict[str, str]]
@@ -673,6 +675,7 @@ def materialize_projects(
                 legacy_rows=self.legacy_rows,
                 notion_context=self.notion_context,
                 now=self.now,
+                observed_at=self.observed_at,
                 workspace_root=self.workspace_root,
                 release_count_by_name=self.release_count_by_name,
                 degraded_dimensions_by_name=self.degraded_dimensions_by_name,
@@ -699,6 +702,7 @@ def build_materialization_context(
     include_notion: bool = True,
     notion_context_fallback: dict[str, dict[str, str]] | None = None,
     now: datetime | None = None,
+    observed_at: datetime | None = None,
     release_count_by_name: dict[str, int] | None = None,
     degraded_dimensions_by_name: dict[str, list[str]] | None = None,
     prior_notion_generated_at: str | None = None,
@@ -744,6 +748,7 @@ def build_materialization_context(
     return MaterializationContext(
         workspace_root=workspace_root,
         now=now,
+        observed_at=observed_at or now,
         catalog_data=catalog_data,
         legacy_rows=legacy_rows,
         notion_context=notion_context,
@@ -819,18 +824,28 @@ def build_portfolio_truth_snapshot(
     producer: dict[str, Any] | None = None,
     prior_notion_generated_at: str | None = None,
 ) -> PortfolioTruthBuildResult:
+    now = now or datetime.now(timezone.utc)
+    materialization_now = now
+    if security_coverage_metadata is not None and security_alerts_by_name is not None:
+        plan_generated_at = (security_cohort_transition or {}).get("plan_generated_at")
+        if plan_generated_at is not None:
+            materialization_now = datetime.fromisoformat(
+                plan_generated_at.replace("Z", "+00:00")
+            )
+        # Older receipts lack the plan clock: retain wall-clock derivation for
+        # compatibility. Freshness and observation times always use `now`.
     context = build_materialization_context(
         workspace_root=workspace_root,
         catalog_path=catalog_path,
         legacy_registry_path=legacy_registry_path,
         include_notion=include_notion,
         notion_context_fallback=notion_context_fallback,
-        now=now,
+        now=materialization_now,
+        observed_at=now,
         release_count_by_name=release_count_by_name,
         degraded_dimensions_by_name=degraded_dimensions_by_name,
         prior_notion_generated_at=prior_notion_generated_at,
     )
-    now = context.now
     catalog_data = context.catalog_data
     legacy_rows = context.legacy_rows
     notion_context = context.notion_context
@@ -1180,6 +1195,7 @@ def _build_truth_project(
     notion_context: dict[str, dict[str, str]],
     now: datetime,
     workspace_root: Path,
+    observed_at: datetime | None = None,
     release_count_by_name: dict[str, int] | None = None,
     degraded_dimensions_by_name: dict[str, list[str]] | None = None,
     security_alerts_by_name: dict[str, dict] | None = None,
@@ -1564,14 +1580,14 @@ def _build_truth_project(
     repository_state = (
         observe_repository_state(
             project_path,
-            observed_at=now,
+            observed_at=observed_at or now,
             remote_default_branch=remote_default_branch,
             workspace_root=workspace_root,
         )
         if project_path is not None and has_git
         else {
             "state": "not_a_repository",
-            "observed_at": now.isoformat(),
+            "observed_at": (observed_at or now).isoformat(),
             "remote_default_branch": remote_default_branch
             or {
                 "state": "unknown",

tests/test_portfolio_cohort_transition.py

@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import json
 import subprocess
 from datetime import datetime, timedelta, timezone
@@ -389,12 +390,14 @@ def _transition_receipt_payload(
     producer_commit: str = "a" * 40,
     prior_truth_sha256: str = "b" * 64,
     now: datetime = NOW,
+    plan_generated_at: datetime = NOW,
 ) -> dict:
     plan = _plan(
         prospective=prospective,
         prior=prior,
         producer_commit=producer_commit,
         prior_truth_sha256=prior_truth_sha256,
+        generated_at=plan_generated_at,
     )
     return collect_security_coverage(
         _truth_payload(plan.collection_repositories),
@@ -440,9 +443,10 @@ def _plan(
     prior: tuple[str, ...],
     producer_commit: str = "a" * 40,
     prior_truth_sha256: str = "b" * 64,
+    generated_at: datetime = NOW,
 ):
     payload = build_cohort_plan_payload(
-        generated_at=NOW,
+        generated_at=generated_at,
         producer_commit=producer_commit,
         producer_repository="saagpatel/GithubRepoAuditor",
         catalog_sha256="c" * 64,
@@ -894,6 +898,27 @@ def test_required_and_outgoing_must_be_disjoint() -> None:
         )
 
 
+def test_collector_carries_plan_instant_through_receipt_validation() -> None:
+    plan_at = NOW - timedelta(minutes=30)
+    payload = _transition_receipt_payload(
+        prospective=ELEVEN, prior=ELEVEN, plan_generated_at=plan_at
+    )
+    loaded = validate_security_coverage_receipt(
+        payload, expected_cohort_count=None, max_cohort_size=24, now=NOW
+    )
+    assert loaded.transition["plan_generated_at"] == plan_at.isoformat()
+
+
+@pytest.mark.parametrize("value", [None, "invalid", "2026-10-11T08:30:00", 123])
+def test_receipt_rejects_invalid_plan_instant(value: object) -> None:
+    payload = _transition_receipt_payload(prospective=ELEVEN, prior=ELEVEN)
+    payload["cohort"]["transition"]["plan_generated_at"] = value
+    with pytest.raises(SecurityCoverageError, match="plan_generated_at"):
+        validate_security_coverage_receipt(
+            _sign(payload), expected_cohort_count=None, max_cohort_size=24, now=NOW
+        )
+
+
 # --------------------------------------------------------------------------
 # 12 / 19 / 21 — publication-level transaction cases
 # --------------------------------------------------------------------------
@@ -999,6 +1024,124 @@ def _truth_alerts() -> dict:
     return loaded.entries_by_full_name
 
 
+PLAN_AT = datetime(2026, 10, 11, 8, 30, tzinfo=timezone.utc)
+CHECK_AT = datetime(2026, 10, 11, 9, 0, tzinfo=timezone.utc)
+LAST_ACTIVITY = datetime(2026, 9, 10, 8, 59, 40, tzinfo=timezone.utc)
+
+
+@pytest.fixture
+def recency_transition_workspace(
+    transition_workspace: tuple[Path, Path, Path], monkeypatch: pytest.MonkeyPatch
+) -> tuple[Path, Path, Path]:
+    import github_repo_auditor.portfolio_truth_sources as sources
+
+    workspace, catalog, output_dir = transition_workspace
+    _git_project(workspace, "Beta", "d/Beta")
+    with catalog.open("a") as handle:
+        handle.write(
+            "  Beta:\n    owner: d\n    lifecycle_state: active\n"
+            "    operating_path: finish\n    category: vanity\n"
+        )
+    gather_git_facts = sources._gather_git_facts
+
+    def fixed_git_facts(project_path: Path) -> dict:
+        facts = gather_git_facts(project_path)
+        if project_path.name == "Beta":
+            facts["last_commit_at"] = LAST_ACTIVITY
+        return facts
+
+    monkeypatch.setattr(sources, "_gather_git_facts", fixed_git_facts)
+    _publish(workspace, catalog, output_dir, now=PLAN_AT - timedelta(hours=1))
+    return workspace, catalog, output_dir
+
+
+def _publish_recency_receipt(
+    workspace: Path,
+    catalog: Path,
+    output_dir: Path,
+    *,
+    carry_plan_time: bool = True,
+    check_at: datetime = CHECK_AT,
+) -> None:
+    payload = _transition_receipt_payload(
+        prospective=("d/Alpha",),
+        prior=("d/Alpha",),
+        now=PLAN_AT,
+        plan_generated_at=PLAN_AT,
+        prior_truth_sha256=hashlib.sha256(
+            (output_dir / "portfolio-truth-latest.json").read_bytes()
+        ).hexdigest(),
+    )
+    if not carry_plan_time:
+        del payload["cohort"]["transition"]["plan_generated_at"]
+    loaded = validate_security_coverage_receipt(
+        _sign(payload), expected_cohort_count=None, max_cohort_size=24, now=CHECK_AT
+    )
+    metadata = _metadata(PLAN_AT)
+    metadata["age_hours"] = loaded.age_hours
+    _publish(
+        workspace,
+        catalog,
+        output_dir,
+        security_alerts_by_name=loaded.entries_by_full_name,
+        security_coverage_metadata=metadata,
+        security_cohort_required=loaded.required_repositories,
+        security_cohort_outgoing=loaded.outgoing_repositories,
+        security_cohort_transition=loaded.transition,
+        now=check_at,
+    )
+
+
+def test_plan_clock_keeps_recency_cliff_outside_r3_and_r5(
+    recency_transition_workspace: tuple[Path, Path, Path],
+) -> None:
+    workspace, catalog, output_dir = recency_transition_workspace
+    assert (PLAN_AT - LAST_ACTIVITY).days == 30
+    assert (CHECK_AT - LAST_ACTIVITY).days == 31
+    _publish_recency_receipt(workspace, catalog, output_dir)
+    truth = json.loads((output_dir / "portfolio-truth-latest.json").read_text())
+    beta = next(p for p in truth["projects"] if p["identity"]["display_name"] == "Beta")
+    assert beta["derived"]["activity_status"] == "recent"
+    assert beta["derived"]["attention_state"] == "manual-only"
+    assert beta["repository_state"]["observed_at"] == CHECK_AT.isoformat()
+    assert truth["generated_at"] == CHECK_AT.isoformat()
+    assert truth["inputs"]["github_security"]["age_hours"] == 0.5
+    outcome = json.loads(
+        (output_dir / PORTFOLIO_COHORT_TRANSITION_FILENAME).read_text()
+    )["transition"]
+    assert outcome["required_repositories"] == ["d/Alpha"]
+    assert outcome["final_repositories"] == ["d/Alpha"]
+
+
+def test_receipt_without_plan_clock_retains_wall_clock_r3_failure(
+    recency_transition_workspace: tuple[Path, Path, Path],
+) -> None:
+    with pytest.raises(ValueError, match=r"derived_only=\['d/Beta'\]"):
+        _publish_recency_receipt(*recency_transition_workspace, carry_plan_time=False)
+
+
+def test_plan_clock_does_not_hide_real_catalog_membership_change(
+    recency_transition_workspace: tuple[Path, Path, Path],
+) -> None:
+    workspace, catalog, output_dir = recency_transition_workspace
+    catalog.write_text(
+        catalog.read_text().replace("category: vanity", "category: commercial")
+    )
+    with pytest.raises(ValueError, match=r"derived_only=\['d/Beta'\]"):
+        _publish_recency_receipt(workspace, catalog, output_dir)
+
+
+def test_plan_clock_does_not_extend_receipt_freshness_at_publication(
+    recency_transition_workspace: tuple[Path, Path, Path],
+) -> None:
+    # Evidence validated as fresh at load time must fail once it exceeds the
+    # publication's 24-hour window, even though derivation stays at PLAN_AT.
+    with pytest.raises(ValueError, match="configured freshness window"):
+        _publish_recency_receipt(
+            *recency_transition_workspace, check_at=PLAN_AT + timedelta(hours=25)
+        )
+
+
 def test_case_12_prior_truth_replaced_between_collection_and_production(
     transition_workspace: tuple[Path, Path, Path],
 ) -> None: