feat(portfolio-truth): --portfolio-truth-allow-empty-notion carry-forward

src/cli.py

@@ -433,6 +433,16 @@ def _build_report_subparser(subparsers: argparse._SubParsersAction) -> None:  #
     p.add_argument(
         "--portfolio-truth", action="store_true", help="Generate canonical portfolio truth snapshot"
     )
+    p.add_argument(
+        "--portfolio-truth-allow-empty-notion",
+        action="store_true",
+        help=(
+            "When live Notion context is unavailable, carry forward the previously "
+            "published Notion context instead of refusing to publish. For headless or "
+            "scheduled refreshes that update risk/activity signals without dropping "
+            "advisory context to zero."
+        ),
+    )
     p.add_argument(
         "--portfolio-context-recovery",
         action="store_true",
@@ -674,6 +684,16 @@ def build_parser() -> argparse.ArgumentParser:
             "risk factor (requires a prior `audit report --ghas-alerts` run)"
         ),
     )
+    parser.add_argument(
+        "--portfolio-truth-allow-empty-notion",
+        action="store_true",
+        help=(
+            "When live Notion context is unavailable, carry forward the previously "
+            "published Notion context instead of refusing to publish. For headless or "
+            "scheduled refreshes that update risk/activity signals without dropping "
+            "advisory context to zero."
+        ),
+    )
     parser.add_argument(
         "--portfolio-context-recovery",
         action="store_true",
@@ -5390,6 +5410,7 @@ def _run_portfolio_truth_mode(args) -> None:
             catalog_path=Path(args.catalog) if args.catalog else None,
             legacy_registry_path=legacy_registry_path,
             include_notion=True,
+            allow_empty_notion=getattr(args, "portfolio_truth_allow_empty_notion", False),
             release_count_by_name=release_count_by_name,
             security_alerts_by_name=security_alerts_by_name,
         )

src/portfolio_truth_publish.py

@@ -5,7 +5,10 @@
 from dataclasses import dataclass
 from pathlib import Path
 
-from src.portfolio_truth_reconcile import build_portfolio_truth_snapshot
+from src.portfolio_truth_reconcile import (
+    build_portfolio_truth_snapshot,
+    load_prior_notion_context,
+)
 from src.portfolio_truth_render import render_portfolio_report_markdown, render_registry_markdown
 from src.portfolio_truth_types import truth_latest_path
 from src.portfolio_truth_validate import (
@@ -82,6 +85,7 @@ def publish_portfolio_truth(
     catalog_path: Path | None = None,
     legacy_registry_path: Path | None = None,
     include_notion: bool = True,
+    allow_empty_notion: bool = False,
     release_count_by_name: dict[str, int] | None = None,
     security_alerts_by_name: dict[str, dict] | None = None,
 ) -> PortfolioTruthPublishResult:
@@ -91,23 +95,28 @@ def publish_portfolio_truth(
         registry_output=registry_output,
         portfolio_report_output=portfolio_report_output,
     )
+    latest_path = truth_latest_path(output_dir)
+    notion_context_fallback = (
+        load_prior_notion_context(latest_path) if allow_empty_notion else None
+    )
     build_result = build_portfolio_truth_snapshot(
         workspace_root=workspace_root,
         catalog_path=catalog_path,
         legacy_registry_path=legacy_registry_path,
         include_notion=include_notion,
+        notion_context_fallback=notion_context_fallback,
         release_count_by_name=release_count_by_name,
         security_alerts_by_name=security_alerts_by_name,
     )
     validate_truth_snapshot(build_result.snapshot)
 
     snapshot_stamp = build_result.snapshot.generated_at.strftime("%Y-%m-%dT%H%M%SZ")
     snapshot_path = output_dir / f"portfolio-truth-{snapshot_stamp}.json"
-    latest_path = truth_latest_path(output_dir)
     _guard_against_notion_context_drop(
         build_result.snapshot.source_summary,
         latest_path=latest_path,
         include_notion=include_notion,
+        allow_empty_notion=allow_empty_notion,
     )
     latest_name = latest_path.name
     snapshot_json = json.dumps(build_result.snapshot.to_dict(), indent=2) + "\n"
@@ -200,8 +209,13 @@ def _guard_against_notion_context_drop(
     *,
     latest_path: Path,
     include_notion: bool,
+    allow_empty_notion: bool = False,
 ) -> None:
     """Avoid overwriting local truth when Notion bootstrap silently disappears."""
+    if allow_empty_notion:
+        # Operator explicitly opted into publishing without live Notion (a headless or
+        # scheduled refresh); prior advisory is carried forward where available.
+        return
     if not include_notion or not _notion_project_context_configured():
         return
     current_rows = _int_value(source_summary.get("notion_context_rows"))

src/portfolio_truth_reconcile.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import logging
 from collections import Counter
 from dataclasses import dataclass
@@ -180,6 +181,7 @@ def build_portfolio_truth_snapshot(
     catalog_path: Path | None = None,
     legacy_registry_path: Path | None = None,
     include_notion: bool = True,
+    notion_context_fallback: dict[str, dict[str, str]] | None = None,
     now: datetime | None = None,
     release_count_by_name: dict[str, int] | None = None,
     security_alerts_by_name: dict[str, dict] | None = None,
@@ -188,6 +190,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_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_context_carried_forward = True
+        logger.warning(
+            "Live Notion context unavailable; carrying forward %d project rows "
+            "from the prior portfolio-truth artifact.",
+            len(notion_context),
+        )
 
     workspace_projects = discover_workspace_projects(
         workspace_root,
@@ -217,6 +231,7 @@ def build_portfolio_truth_snapshot(
         "catalog_warnings": list(catalog_data.get("warnings") or []),
         "legacy_registry_rows": len(legacy_rows),
         "notion_context_rows": len(notion_context),
+        "notion_context_carried_forward": notion_context_carried_forward,
         "context_quality_counts": dict(
             Counter(project.derived.context_quality for project in projects)
         ),
@@ -251,6 +266,46 @@ def build_portfolio_truth_snapshot(
     )
 
 
+def load_prior_notion_context(latest_path: Path) -> dict[str, dict[str, str]]:
+    """Reconstruct a Notion project-context map from a previously published
+    portfolio-truth artifact, keyed identically to live Notion context
+    (``_normalize(display_name)`` -> ``{portfolio_call, momentum, current_state}``).
+
+    Used to carry advisory context forward on a headless refresh when a live
+    Notion token is unavailable, rather than overwriting local truth with zero
+    rows. Only projects that actually carried Notion advisory are returned, so
+    the resulting row count reflects real carried context. Returns an empty map
+    when the artifact is missing or malformed.
+    """
+    try:
+        data = json.loads(latest_path.read_text())
+    except (OSError, json.JSONDecodeError):
+        return {}
+    projects = data.get("projects")
+    if not isinstance(projects, list):
+        return {}
+    context: dict[str, dict[str, str]] = {}
+    for project in projects:
+        if not isinstance(project, dict):
+            continue
+        identity = project.get("identity")
+        advisory = project.get("advisory")
+        if not isinstance(identity, dict) or not isinstance(advisory, dict):
+            continue
+        display_name = str(identity.get("display_name", "")).strip()
+        portfolio_call = str(advisory.get("notion_portfolio_call", "")).strip()
+        momentum = str(advisory.get("notion_momentum", "")).strip()
+        current_state = str(advisory.get("notion_current_state", "")).strip()
+        if not display_name or not (portfolio_call or momentum or current_state):
+            continue
+        context[_normalize(display_name)] = {
+            "portfolio_call": portfolio_call,
+            "momentum": momentum,
+            "current_state": current_state,
+        }
+    return context
+
+
 def _duplicate_display_names(projects: list[PortfolioTruthProject]) -> list[str]:
     return sorted(
         name