fix(portfolio): preserve carried Notion origin

src/operator_os_seam_linter.py

@@ -11,6 +11,7 @@
 from typing import Any
 
 from src.portfolio_truth_render import GENERATED_MARKDOWN_PROVENANCE_MARKER
+from src.portfolio_truth_lineage import resolve_notion_origin
 from src.portfolio_truth_types import LEGACY_SCHEMA_VERSIONS, SCHEMA_VERSION, truth_latest_path
 from src.portfolio_catalog import load_portfolio_catalog
 from src.portfolio_truth_sources import (
@@ -552,6 +553,16 @@ def _check_carried_freshness(
                 level="unknown",
             )
         ]
+    resolved_origin = resolve_notion_origin(truth_path)
+    if resolved_origin is not None and resolved_origin != origin:
+        return [
+            SeamLintFinding(
+                check="CL-FRESH-002",
+                artifact=str(truth_path),
+                violation="carried-forward origin advanced across artifact history",
+                detail=f"declared={origin}; resolved_origin={resolved_origin}",
+            )
+        ]
     if age > timedelta(hours=48):
         return [
             SeamLintFinding(

src/portfolio_truth_lineage.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+
+def resolve_notion_origin(artifact_path: Path) -> str | None:
+    """Resolve the oldest preserved Notion observation through carry-forward history."""
+    return _resolve_notion_origin(artifact_path, visited=set())
+
+
+def _resolve_notion_origin(
+    artifact_path: Path, *, visited: set[Path]
+) -> str | None:
+    resolved_path = artifact_path.resolve()
+    if resolved_path in visited:
+        return None
+    visited.add(resolved_path)
+    payload = _read_json_object(artifact_path)
+    if payload is None:
+        return None
+    generated_at = _text(payload.get("generated_at"))
+    inputs = payload.get("inputs")
+    notion = inputs.get("notion") if isinstance(inputs, dict) else None
+    if not isinstance(notion, dict):
+        return generated_at
+
+    mode = notion.get("mode")
+    if mode == "live":
+        return _text(notion.get("observed_at")) or generated_at
+    if mode != "carried-forward":
+        return _text(notion.get("observed_at")) or generated_at
+
+    declared_origin = _text(notion.get("carried_from_generated_at")) or _text(
+        notion.get("observed_at")
+    )
+    if declared_origin is None:
+        return None
+    predecessor = _find_artifact_by_generated_at(
+        artifact_path.parent,
+        generated_at=declared_origin,
+        excluded=visited,
+    )
+    if predecessor is None:
+        return declared_origin
+    return (
+        _resolve_notion_origin(predecessor, visited=visited)
+        or declared_origin
+    )
+
+
+def _find_artifact_by_generated_at(
+    directory: Path, *, generated_at: str, excluded: set[Path]
+) -> Path | None:
+    for candidate in sorted(directory.glob("portfolio-truth-*.json"), reverse=True):
+        if candidate.resolve() in excluded:
+            continue
+        payload = _read_json_object(candidate)
+        if payload is not None and _text(payload.get("generated_at")) == generated_at:
+            return candidate
+    return None
+
+
+def _read_json_object(path: Path) -> dict[str, Any] | None:
+    try:
+        payload = json.loads(path.read_text())
+    except (FileNotFoundError, json.JSONDecodeError, OSError):
+        return None
+    return payload if isinstance(payload, dict) else None
+
+
+def _text(value: object) -> str | None:
+    return value if isinstance(value, str) and value.strip() else None

src/portfolio_truth_publish.py

@@ -10,6 +10,7 @@
     load_prior_notion_context,
 )
 from src.portfolio_truth_render import render_portfolio_report_markdown, render_registry_markdown
+from src.portfolio_truth_lineage import resolve_notion_origin
 from src.portfolio_truth_types import truth_latest_path
 from src.producer_preflight import ProducerEvidence, verify_evidence_still_current
 from src.portfolio_truth_validate import (
@@ -117,7 +118,7 @@ def publish_portfolio_truth(
     notion_context_fallback = (
         load_prior_notion_context(latest_path) if allow_empty_notion else None
     )
-    prior_notion_generated_at = _previous_generated_at(latest_path)
+    prior_notion_generated_at = resolve_notion_origin(latest_path)
     build_result = build_portfolio_truth_snapshot(
         workspace_root=workspace_root,
         catalog_path=catalog_path,
@@ -231,15 +232,6 @@ def _content_changed(path: Path, content: str) -> bool:
     return path.read_text() != content
 
 
-def _previous_generated_at(latest_path: Path) -> str | None:
-    try:
-        payload = json.loads(latest_path.read_text())
-    except (FileNotFoundError, json.JSONDecodeError, OSError):
-        return None
-    value = payload.get("generated_at") if isinstance(payload, dict) else None
-    return value if isinstance(value, str) and value else None
-
-
 def _guard_against_notion_context_drop(
     source_summary: dict[str, object],
     *,

tests/test_operator_os_seam_linter.py

@@ -259,6 +259,58 @@ def test_contract_shadow_warns_for_carried_notion_older_than_48_hours(
     )
 
 
+def test_contract_shadow_fails_when_carry_forward_origin_advances(
+    tmp_path: Path,
+) -> None:
+    truth, markdown = _passing_paths(tmp_path)
+    predecessor_generated = "2026-07-02T00:00:00+00:00"
+    oldest = "2026-07-01T00:00:00+00:00"
+    (tmp_path / "portfolio-truth-2026-07-02T000000Z.json").write_text(
+        json.dumps(
+            {
+                "generated_at": predecessor_generated,
+                "inputs": {
+                    "notion": {
+                        "mode": "carried-forward",
+                        "observed_at": oldest,
+                        "carried_from_generated_at": oldest,
+                    }
+                },
+            }
+        )
+    )
+    payload = json.loads(truth.read_text())
+    payload.update(
+        {
+            "producer": {},
+            "inputs": {
+                "notion": {
+                    "mode": "carried-forward",
+                    "observed_at": predecessor_generated,
+                    "carried_from_generated_at": predecessor_generated,
+                },
+                "catalog": {"sha256": None},
+            },
+            "source_summary": {"attention_state_counts": {"decision-needed": 0}},
+            "rollups": {"decision": {"decision_needed_count": 0}},
+        }
+    )
+    truth.write_text(json.dumps(payload))
+
+    result = lint_operator_os_seams(
+        truth_path=truth,
+        markdown_paths=markdown,
+        contract_shadow=True,
+        catalog_path=tmp_path / "missing-catalog.yaml",
+        now=NOW,
+    )
+
+    finding = next(item for item in result.findings if item.check == "CL-FRESH-002")
+    assert finding.level == "fail"
+    assert "advanced" in finding.violation
+    assert oldest in finding.detail
+
+
 def test_contract_shadow_fails_when_excluded_backup_leaks_into_projects(
     tmp_path: Path,
 ) -> None:

tests/test_portfolio_truth.py

@@ -1498,6 +1498,56 @@ def test_load_prior_notion_context_missing_or_malformed_returns_empty(tmp_path:
     assert load_prior_notion_context(malformed) == {}
 
 
+def test_notion_origin_resolution_preserves_oldest_carried_observation(
+    tmp_path: Path,
+) -> None:
+    from src.portfolio_truth_lineage import resolve_notion_origin
+
+    oldest = "2026-07-10T09:00:10+00:00"
+    predecessor_generated = "2026-07-11T03:17:38+00:00"
+    predecessor = tmp_path / "portfolio-truth-2026-07-11T031738Z.json"
+    predecessor.write_text(
+        json.dumps(
+            {
+                "generated_at": predecessor_generated,
+                "inputs": {
+                    "notion": {
+                        "mode": "carried-forward",
+                        "observed_at": oldest,
+                        "carried_from_generated_at": oldest,
+                    }
+                },
+            }
+        )
+    )
+    latest = tmp_path / "portfolio-truth-latest.json"
+    latest.write_text(
+        json.dumps(
+            {
+                "generated_at": "2026-07-11T03:36:36+00:00",
+                "inputs": {
+                    "notion": {
+                        "mode": "carried-forward",
+                        "observed_at": predecessor_generated,
+                        "carried_from_generated_at": predecessor_generated,
+                    }
+                },
+            }
+        )
+    )
+
+    assert resolve_notion_origin(latest) == oldest
+
+
+def test_notion_origin_resolution_uses_legacy_generation_time(tmp_path: Path) -> None:
+    from src.portfolio_truth_lineage import resolve_notion_origin
+
+    artifact = tmp_path / "portfolio-truth-latest.json"
+    artifact.write_text(json.dumps({"generated_at": "2026-07-10T09:00:10+00:00"}))
+
+    assert resolve_notion_origin(artifact) == "2026-07-10T09:00:10+00:00"
+
+
 def test_publish_allow_empty_notion_carries_forward_prior_context(
     portfolio_workspace: Path,
     portfolio_catalog: Path,