Consume verified Notion snapshot context

src/portfolio_truth_reconcile.py

@@ -243,12 +243,18 @@ def build_portfolio_truth_snapshot(
     catalog_data = load_portfolio_catalog(catalog_path)
     legacy_rows = load_legacy_registry_rows(legacy_registry_path)
     notion_context = load_safe_notion_project_context() if include_notion else {}
+    notion_source_mode = str(
+        getattr(notion_context, "source_mode", "live") or "live"
+    )
+    notion_observed_at = getattr(notion_context, "observed_at", None)
     notion_context_carried_forward = False
     if include_notion and not notion_context and notion_context_fallback:
         # Live Notion was unavailable; carry forward the prior published context so
         # a headless refresh updates risk/activity signals without dropping advisory
         # data to zero. The caller opts in via publish_portfolio_truth(allow_empty_notion=True).
         notion_context = notion_context_fallback
+        notion_source_mode = "carried-forward"
+        notion_observed_at = prior_notion_generated_at
         notion_context_carried_forward = True
         logger.warning(
             "Live Notion context unavailable; carrying forward %d project rows "
@@ -346,6 +352,8 @@ def build_portfolio_truth_snapshot(
             notion_context_rows=len(notion_context),
             notion_context_carried_forward=notion_context_carried_forward,
             prior_notion_generated_at=prior_notion_generated_at,
+            notion_source_mode=notion_source_mode,
+            notion_observed_at=notion_observed_at,
             security_coverage_metadata=security_coverage_metadata,
         ),
         coverage=_build_coverage_envelope(
@@ -574,6 +582,8 @@ def _build_input_envelope(
     notion_context_rows: int,
     notion_context_carried_forward: bool,
     prior_notion_generated_at: str | None,
+    notion_source_mode: str,
+    notion_observed_at: str | None,
     security_coverage_metadata: dict[str, Any] | None,
 ) -> dict[str, Any]:
     resolved_catalog = Path(str(catalog_data.get("path") or ""))
@@ -589,8 +599,8 @@ def _build_input_envelope(
         notion_mode = "carried-forward" if prior_notion_generated_at else "unavailable"
         notion_observed_at = prior_notion_generated_at
     else:
-        notion_mode = "live"
-        notion_observed_at = now.isoformat()
+        notion_mode = notion_source_mode
+        notion_observed_at = notion_observed_at or now.isoformat()
     inputs = {
         "catalog": {
             "source_id": "portfolio-catalog",

src/portfolio_truth_sources.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+import hashlib
 import json
+import os
 import re
 import subprocess
 from datetime import datetime, timezone
@@ -105,6 +107,19 @@
 
 
 WORKSPACE_DISCOVERY_POLICY_VERSION = "workspace_discovery.v2"
+MAX_NOTION_SNAPSHOT_AGE_HOURS = 30
+
+
+class NotionProjectContext(dict[str, dict[str, str]]):
+    def __init__(
+        self,
+        *,
+        source_mode: str,
+        observed_at: str | None,
+    ) -> None:
+        super().__init__()
+        self.source_mode = source_mode
+        self.observed_at = observed_at
 
 
 def workspace_exclusion_reason(name: str, *, nested: bool = False) -> str | None:
@@ -306,9 +321,24 @@ def load_legacy_registry_rows(path: Path | None) -> dict[str, dict[str, str]]:
 
 def load_safe_notion_project_context(
     config_dir: Path = Path("config"),
+    snapshot_path: Path | None = None,
 ) -> dict[str, dict[str, str]]:
-    raw_context = load_notion_project_context(config_dir) or {}
-    sanitized: dict[str, dict[str, str]] = {}
+    raw_context = load_notion_project_context(config_dir)
+    source_mode = "live"
+    observed_at: str | None = None
+    if not raw_context:
+        configured_snapshot = snapshot_path or _notion_snapshot_path_from_environment()
+        if configured_snapshot:
+            raw_context, observed_at = _load_verified_notion_snapshot_context(
+                configured_snapshot
+            )
+            source_mode = "verified-snapshot"
+        else:
+            raw_context = {}
+    sanitized = NotionProjectContext(
+        source_mode=source_mode,
+        observed_at=observed_at,
+    )
     for name, context in raw_context.items():
         sanitized[_normalize(name)] = {
             "portfolio_call": str(context.get("portfolio_call", "") or "").strip(),
@@ -322,6 +352,78 @@ def load_safe_notion_project_context(
     return sanitized
 
 
+def _notion_snapshot_path_from_environment() -> Path | None:
+    configured = os.environ.get("GHRA_NOTION_SNAPSHOT_PATH", "").strip()
+    if not configured:
+        return None
+    return Path(configured).expanduser()
+
+
+def _load_verified_notion_snapshot_context(
+    snapshot_path: Path,
+) -> tuple[dict[str, dict[str, str]], str | None]:
+    try:
+        payload = json.loads(snapshot_path.read_text())
+    except (OSError, json.JSONDecodeError):
+        return {}, None
+    if not isinstance(payload, dict):
+        return {}, None
+    if payload.get("schema_version") != "2.0.0":
+        return {}, None
+    projects = payload.get("projects")
+    if (
+        not isinstance(projects, list)
+        or payload.get("project_count") != len(projects)
+        or not projects
+    ):
+        return {}, None
+    live_receipt = payload.get("live_read_receipt")
+    if (
+        not isinstance(live_receipt, dict)
+        or live_receipt.get("state") != "verified"
+        or live_receipt.get("page_count") != len(projects)
+    ):
+        return {}, None
+    authority_receipt = payload.get("attention_authority_receipt")
+    if (
+        not isinstance(authority_receipt, dict)
+        or authority_receipt.get("state") != "verified"
+    ):
+        return {}, None
+    try:
+        generated_at = datetime.fromisoformat(
+            str(payload.get("generated_at", "")).replace("Z", "+00:00")
+        )
+    except ValueError:
+        return {}, None
+    age_hours = (
+        datetime.now(timezone.utc) - generated_at.astimezone(timezone.utc)
+    ).total_seconds() / 3600
+    if age_hours < -(5 / 60) or age_hours > MAX_NOTION_SNAPSHOT_AGE_HOURS:
+        return {}, None
+    content_bytes = json.dumps(
+        projects, separators=(",", ":"), ensure_ascii=False
+    ).encode()
+    if hashlib.sha256(content_bytes).hexdigest() != payload.get("content_sha256"):
+        return {}, None
+
+    context: dict[str, dict[str, str]] = {}
+    for project in projects:
+        if not isinstance(project, dict):
+            continue
+        title = str(project.get("title", "") or "").strip()
+        if not title:
+            continue
+        context[title] = {
+            "portfolio_call": str(project.get("portfolio_call", "") or "").strip(),
+            "momentum": str(
+                project.get("momentum") or project.get("operating_queue") or ""
+            ).strip(),
+            "current_state": str(project.get("current_state", "") or "").strip(),
+        }
+    return context, generated_at.isoformat()
+
+
 def _load_notion_title_aliases(config_dir: Path) -> dict[str, str]:
     path = config_dir / "project-registry-overrides.json"
     if not path.is_file():

src/portfolio_truth_validate.py

@@ -137,12 +137,14 @@ def _validate_contract_envelope(snapshot: PortfolioTruthSnapshot) -> None:
     if not isinstance(notion, dict):
         raise ValueError("Portfolio truth inputs.notion is required.")
     mode = notion.get("mode")
-    if mode not in {"live", "carried-forward", "unavailable"}:
+    if mode not in {"live", "verified-snapshot", "carried-forward", "unavailable"}:
         raise ValueError(f"Invalid Notion input mode: {mode}")
     if mode == "carried-forward" and not notion.get("carried_from_generated_at"):
         raise ValueError("Carried-forward Notion input requires an origin timestamp.")
     if mode == "live" and notion.get("carried_from_generated_at") is not None:
         raise ValueError("Live Notion input cannot declare a carried-forward origin.")
+    if mode == "verified-snapshot" and not notion.get("observed_at"):
+        raise ValueError("Verified Notion snapshot input requires an observation timestamp.")
 
 
 def validate_publish_targets(

tests/test_portfolio_truth.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import hashlib
 import json
 import os
 import subprocess
@@ -225,6 +226,101 @@ def test_notion_context_uses_configured_title_aliases(
     assert context["notion"]["current_state"] == "Shipped"
 
 
+def test_notion_context_uses_fresh_verified_snapshot_when_live_api_unavailable(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    snapshot_path = tmp_path / "project-snapshot.json"
+    projects = [
+        {
+            "title": "operator-scripts",
+            "portfolio_call": "Maintain",
+            "operating_queue": "Watch",
+            "current_state": "Shipped",
+        }
+    ]
+    snapshot_path.write_text(
+        json.dumps(
+            {
+                "schema_version": "2.0.0",
+                "generated_at": datetime.now(timezone.utc).isoformat(),
+                "project_count": 1,
+                "projects": projects,
+                "content_sha256": hashlib.sha256(
+                    json.dumps(
+                        projects, separators=(",", ":"), ensure_ascii=False
+                    ).encode()
+                ).hexdigest(),
+                "live_read_receipt": {"state": "verified", "page_count": 1},
+                "attention_authority_receipt": {"state": "verified"},
+            }
+        )
+    )
+    monkeypatch.setattr(
+        "src.portfolio_truth_sources.load_notion_project_context",
+        lambda _config_dir: None,
+    )
+
+    context = load_safe_notion_project_context(
+        tmp_path / "config", snapshot_path=snapshot_path
+    )
+
+    assert context["operatorscripts"] == {
+        "portfolio_call": "Maintain",
+        "momentum": "Watch",
+        "current_state": "Shipped",
+    }
+    assert context.source_mode == "verified-snapshot"
+    assert context.observed_at is not None
+
+
+def test_notion_context_rejects_snapshot_without_verified_live_receipt(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    snapshot_path = tmp_path / "project-snapshot.json"
+    snapshot_path.write_text(
+        json.dumps(
+            {
+                "schema_version": "2.0.0",
+                "generated_at": datetime.now(timezone.utc).isoformat(),
+                "project_count": 1,
+                "projects": [{"title": "operator-scripts"}],
+                "content_sha256": "not-relevant",
+                "live_read_receipt": {"state": "unverified", "page_count": 1},
+                "attention_authority_receipt": {"state": "verified"},
+            }
+        )
+    )
+    monkeypatch.setattr(
+        "src.portfolio_truth_sources.load_notion_project_context",
+        lambda _config_dir: None,
+    )
+
+    assert (
+        load_safe_notion_project_context(
+            tmp_path / "config", snapshot_path=snapshot_path
+        )
+        == {}
+    )
+
+
+def test_notion_context_rejects_non_object_snapshot_json(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    snapshot_path = tmp_path / "project-snapshot.json"
+    snapshot_path.write_text("[]")
+    monkeypatch.setattr(
+        "src.portfolio_truth_sources.load_notion_project_context",
+        lambda _config_dir: None,
+    )
+
+    assert (
+        load_safe_notion_project_context(
+            tmp_path / "config", snapshot_path=snapshot_path
+        )
+        == {}
+    )
+
+
 @pytest.fixture
 def portfolio_workspace(tmp_path: Path) -> Path:
     workspace = tmp_path / "workspace"