fix(lifecycle): require source_revision on daily monitor snapshots

.github/workflows/reusable-drift-check.yml

@@ -276,7 +276,7 @@ jobs:
           if [ -n "${LIFECYCLE_LIVE_STREAM_ID:-}" ]; then
             lifecycle_args+=(--live-stream-id "${LIFECYCLE_LIVE_STREAM_ID}")
           fi
-          quant-lifecycle monitor --domain ${{ inputs.strategy_domain }} "${lifecycle_args[@]}"
+          quant-lifecycle monitor --domain ${{ inputs.strategy_domain }} --source-revision "${GITHUB_SHA}" "${lifecycle_args[@]}"
 
       - name: Validate lifecycle prerequisites
         shell: bash

src/quant_platform_kit/strategy_lifecycle/cli.py

@@ -42,6 +42,9 @@ def _run_monitor(args: argparse.Namespace) -> int:
     live_stream_id = getattr(args, "live_stream_id", None)
     if live_stream_id:
         kwargs["live_stream_id"] = live_stream_id
+    source_revision = getattr(args, "source_revision", None)
+    if source_revision:
+        kwargs["source_revision"] = source_revision
     snapshots = run_monitor(**kwargs)
     _print(f"[monitor] Generated {len(snapshots)} performance snapshots")
     return 0
@@ -381,6 +384,15 @@ def build_parser() -> argparse.ArgumentParser:
         default=None,
         help="Monitor one stable account/runtime telemetry stream; never mix streams.",
     )
+    monitor.add_argument(
+        "--source-revision",
+        default=None,
+        help=(
+            "Observation provenance written onto daily snapshots "
+            "(40-char git SHA preferred). Falls back to LIFECYCLE_SOURCE_REVISION "
+            "or GITHUB_SHA when omitted."
+        ),
+    )
     monitor.add_argument(
         "--benchmark-catalog",
         default=None,

src/quant_platform_kit/strategy_lifecycle/performance_monitor.py

@@ -7,6 +7,9 @@
 from __future__ import annotations
 
 import logging
+import os
+import re
+from dataclasses import replace
 from datetime import date, datetime, timezone
 from pathlib import Path
 from typing import Any, Mapping, Sequence
@@ -28,11 +31,60 @@
     resolve_strategy_benchmark,
 )
 
+_SOURCE_REVISION_SHA = re.compile(r"^[0-9a-f]{40}$")
+_SOURCE_REVISION_SENTINELS = frozenset(
+    {"", "unavailable", "not_available", "legacy_missing", "unknown", "none", "null"}
+)
+
 
 def _now_iso() -> str:
     return datetime.now(timezone.utc).isoformat()
 
 
+def resolve_monitor_source_revision(
+    explicit: str | None = None,
+    *,
+    environ: Mapping[str, str] | None = None,
+) -> str:
+    """Resolve a real observation ``source_revision`` for daily snapshots.
+
+    Order: explicit argument, ``LIFECYCLE_SOURCE_REVISION``, then ``GITHUB_SHA``.
+    Full 40-character lowercase git SHAs are always accepted. Non-SHA labels are
+    allowed only from the explicit argument or ``LIFECYCLE_SOURCE_REVISION``
+    (existing probe/contract non-empty provenance). ``GITHUB_SHA`` must be a
+    full 40-character SHA. Missing or sentinel values raise instead of writing
+    an empty string that would look like a complete observation.
+    """
+    env = os.environ if environ is None else environ
+    candidates: list[tuple[str, bool]] = []
+    if explicit is not None:
+        candidates.append((str(explicit), False))
+    for key, sha_only in (
+        ("LIFECYCLE_SOURCE_REVISION", False),
+        ("GITHUB_SHA", True),
+    ):
+        raw = env.get(key)
+        if raw is None:
+            continue
+        candidates.append((str(raw), sha_only))
+
+    for raw, sha_only in candidates:
+        value = raw.strip()
+        if not value or value.lower() in _SOURCE_REVISION_SENTINELS:
+            continue
+        lowered = value.lower()
+        if _SOURCE_REVISION_SHA.fullmatch(lowered):
+            return lowered
+        if sha_only:
+            continue
+        return value
+
+    raise RuntimeError(
+        "monitor source_revision is required; pass source_revision=..., "
+        "set LIFECYCLE_SOURCE_REVISION, or provide GITHUB_SHA (40-char)"
+    )
+
+
 def resolve_lifecycle_stream_id(
     explicit_stream_id: str = "",
     *,
@@ -45,8 +97,6 @@ def resolve_lifecycle_stream_id(
     synthetic equity curve.  ``LIFECYCLE_STREAM_ID`` is available for
     non-Cloud-Run runtimes that need a stable explicit identity.
     """
-    import os
-
     for candidate in (
         explicit_stream_id,
         os.environ.get("LIFECYCLE_STREAM_ID"),
@@ -83,6 +133,7 @@ def run_monitor(
     strategy_benchmarks: Mapping[str, str] | None = None,
     require_explicit_benchmark: bool = False,
     live_stream_id: str | None = None,
+    source_revision: str | None = None,
 ) -> list[StrategyPerformanceSnapshot]:
     """Run the performance monitor for the given domain.
 
@@ -102,6 +153,9 @@ def run_monitor(
         live_stream_id: Optional stable telemetry stream identity.  When live
             account data is used, this prevents independent broker accounts
             from being merged into one return series.
+        source_revision: Observation provenance. Required when snapshots would be
+            written; resolved from env when omitted (see
+            :func:`resolve_monitor_source_revision`).
 
     Returns:
         List of StrategyPerformanceSnapshot objects generated.
@@ -122,6 +176,9 @@ def run_monitor(
             )
         return []
 
+    # Fail closed before any snapshot write when provenance is missing.
+    resolved_source_revision = resolve_monitor_source_revision(source_revision)
+
     profiles = [strategy_profile] if strategy_profile else sorted(all_returns.keys())
     snapshots: list[StrategyPerformanceSnapshot] = []
 
@@ -160,6 +217,7 @@ def run_monitor(
             as_of=date.today(),
             benchmark_symbol=benchmark_symbol,
             computed_at=_now_iso(),
+            source_revision=resolved_source_revision,
         )
 
         # Compute each window
@@ -177,48 +235,27 @@ def run_monitor(
 
         # Latest return
         if len(series) > 0:
-            snapshot = StrategyPerformanceSnapshot(
-                strategy_profile=snapshot.strategy_profile,
-                domain=snapshot.domain,
-                platform=snapshot.platform,
-                as_of=snapshot.as_of,
+            snapshot = replace(
+                snapshot,
                 windows=windows_dict,
                 latest_return=float(series.iloc[-1]),
-                benchmark_symbol=snapshot.benchmark_symbol,
-                data_freshness_days=(date.today() - series.index[-1].date()).days if hasattr(series.index[-1], "date") else 0,
+                data_freshness_days=(
+                    (date.today() - series.index[-1].date()).days
+                    if hasattr(series.index[-1], "date")
+                    else 0
+                ),
                 source_artifact_path="",
-                computed_at=snapshot.computed_at,
             )
         else:
-            snapshot = StrategyPerformanceSnapshot(
-                strategy_profile=snapshot.strategy_profile,
-                domain=snapshot.domain,
-                platform=snapshot.platform,
-                as_of=snapshot.as_of,
-                windows=windows_dict,
-                benchmark_symbol=snapshot.benchmark_symbol,
-                computed_at=snapshot.computed_at,
-            )
+            snapshot = replace(snapshot, windows=windows_dict)
 
         # Attach drift reference: use 126-day window to compare against backtest
         ref_window = windows_dict.get(126) or windows_dict.get(252)
         if ref_window is not None and latest_backtest is not None:
             deviations = compare_with_backtest(ref_window, latest_backtest)
             if deviations:
                 max_dev = max(deviations.values())
-                snapshot = StrategyPerformanceSnapshot(
-                    strategy_profile=snapshot.strategy_profile,
-                    domain=snapshot.domain,
-                    platform=snapshot.platform,
-                    as_of=snapshot.as_of,
-                    windows=snapshot.windows,
-                    latest_return=snapshot.latest_return,
-                    benchmark_symbol=snapshot.benchmark_symbol,
-                    drift_score=min(max_dev, 1.0),
-                    data_freshness_days=snapshot.data_freshness_days,
-                    source_artifact_path=snapshot.source_artifact_path,
-                    computed_at=snapshot.computed_at,
-                )
+                snapshot = replace(snapshot, drift_score=min(max_dev, 1.0))
 
         # Persist
         store.save_snapshot(snapshot)

tests/test_lifecycle_performance_monitor.py

@@ -11,6 +11,7 @@
 from quant_platform_kit.strategy_lifecycle.performance_monitor import (
     PerformanceMonitor,
     infer_strategy_domain,
+    resolve_monitor_source_revision,
     run_monitor,
     try_record_platform_execution,
 )
@@ -65,15 +66,57 @@ def test_infer_strategy_domain_from_profile_prefix(self) -> None:
     def test_try_record_platform_execution_swallows_errors(self) -> None:
         try_record_platform_execution("", {"status": "ok"})
 
+    def test_resolve_monitor_source_revision_requires_real_provenance(self) -> None:
+        sha = "a" * 40
+        self.assertEqual(resolve_monitor_source_revision(sha), sha)
+        self.assertEqual(
+            resolve_monitor_source_revision(None, environ={"LIFECYCLE_SOURCE_REVISION": "fixture-rev"}),
+            "fixture-rev",
+        )
+        self.assertEqual(
+            resolve_monitor_source_revision(None, environ={"GITHUB_SHA": sha}),
+            sha,
+        )
+        with self.assertRaisesRegex(RuntimeError, "source_revision is required"):
+            resolve_monitor_source_revision(None, environ={})
+        with self.assertRaisesRegex(RuntimeError, "source_revision is required"):
+            resolve_monitor_source_revision("", environ={"GITHUB_SHA": "deadbeef"})
+
     def test_run_monitor_fails_closed_when_no_profiles_found(self) -> None:
         class EmptyCollector:
             def collect(self, _domain: str) -> dict[str, pd.Series]:
                 return {}
 
         with self.assertRaisesRegex(RuntimeError, "No strategy return series found"):
-            run_monitor("us_equity", collector=EmptyCollector())
+            run_monitor(
+                "us_equity",
+                collector=EmptyCollector(),
+                source_revision="a" * 40,
+            )
+
+    def test_run_monitor_fails_closed_without_source_revision(self) -> None:
+        class OneCollector:
+            def collect(self, _domain: str) -> dict[str, pd.Series]:
+                return {
+                    "synthetic_soxl": pd.Series(
+                        [0.01, -0.02],
+                        index=pd.to_datetime(["2026-09-08", "2026-09-09"]),
+                    )
+                }
+
+        with patch.dict("os.environ", {}, clear=False):
+            for key in ("LIFECYCLE_SOURCE_REVISION", "GITHUB_SHA"):
+                # Ensure missing provenance cannot silently write empty revision.
+                pass
+        with patch(
+            "quant_platform_kit.strategy_lifecycle.performance_monitor.resolve_monitor_source_revision",
+            side_effect=RuntimeError("monitor source_revision is required"),
+        ):
+            with self.assertRaisesRegex(RuntimeError, "source_revision is required"):
+                run_monitor("us_equity", collector=OneCollector(), windows=(2,), min_observations=2)
 
     def test_csv_collector_to_monitor_rejects_duplicate_daily_returns(self) -> None:
+        revision = "b" * 40
         for dates, values in (
             (["2026-09-08", "2026-09-08"], [0.01, 0.01]),
             (["2026-09-08T09:00:00", "2026-09-08T16:00:00"], [0.01, -0.02]),
@@ -86,14 +129,30 @@ def test_csv_collector_to_monitor_rejects_duplicate_daily_returns(self) -> None:
                 collector = ReturnCollector(artifact_roots={"us_equity": root}, projects_root=root, store=store)
                 with patch.object(PerformanceStore, "save_snapshot", autospec=True) as save:
                     with self.assertRaisesRegex(RuntimeError, "No strategy return series found"):
-                        run_monitor("us_equity", strategy_profile="synthetic_soxl", collector=collector,
-                                    store=store, windows=(2,), min_observations=2)
-                    self.assertEqual(run_monitor(
-                        "us_equity", collector=collector, store=store, min_observations=2, fail_on_empty=False,
-                    ), [])
+                        run_monitor(
+                            "us_equity",
+                            strategy_profile="synthetic_soxl",
+                            collector=collector,
+                            store=store,
+                            windows=(2,),
+                            min_observations=2,
+                            source_revision=revision,
+                        )
+                    self.assertEqual(
+                        run_monitor(
+                            "us_equity",
+                            collector=collector,
+                            store=store,
+                            min_observations=2,
+                            fail_on_empty=False,
+                            source_revision=revision,
+                        ),
+                        [],
+                    )
                     save.assert_not_called()
 
     def test_csv_collector_to_monitor_preserves_unique_daily_returns(self) -> None:
+        revision = "c" * 40
         with tempfile.TemporaryDirectory() as tmp:
             root = Path(tmp)
             pd.DataFrame({
@@ -107,8 +166,10 @@ def test_csv_collector_to_monitor_preserves_unique_daily_returns(self) -> None:
                 "us_equity", strategy_profile="synthetic_soxl", collector=collector, store=store,
                 windows=(2,), min_observations=2, require_explicit_benchmark=True,
                 strategy_benchmarks={"synthetic_soxl": "SPY"},
+                source_revision=revision,
             )
             self.assertEqual(len(snapshots), 1)
+            self.assertEqual(snapshots[0].source_revision, revision)
             metrics = snapshots[0].windows[2]
             self.assertEqual(metrics.observation_count, 2)
             self.assertAlmostEqual(metrics.total_return, 1.01 * 0.98 - 1.0)
@@ -133,6 +194,7 @@ def test_csv_collector_to_monitor_rejects_duplicate_required_benchmark(self) ->
                         "us_equity", strategy_profile="synthetic_soxl", collector=collector, store=store,
                         windows=(2,), min_observations=2, require_explicit_benchmark=True,
                         strategy_benchmarks={"synthetic_soxl": "SPY"},
+                        source_revision="d" * 40,
                     )
                 save.assert_not_called()
 

tests/test_reusable_drift_workflow.py

@@ -40,7 +40,7 @@ def test_reusable_drift_workflow_enforces_lifecycle_preflight() -> None:
     assert 'lifecycle_args+=(--strategy "${LIFECYCLE_STRATEGY_PROFILE}")' in workflow
     assert 'lifecycle_args+=(--live-stream-id "${LIFECYCLE_LIVE_STREAM_ID}")' in workflow
     assert "not paper/live account PnL" in workflow
-    assert 'quant-lifecycle monitor --domain ${{ inputs.strategy_domain }} "${lifecycle_args[@]}"' in workflow
+    assert 'quant-lifecycle monitor --domain ${{ inputs.strategy_domain }} --source-revision "${GITHUB_SHA}" "${lifecycle_args[@]}"' in workflow
     assert 'quant-lifecycle doctor --domain ${{ inputs.strategy_domain }} --require-snapshot --require-backtest --max-freshness-days 7 "${lifecycle_args[@]}"' in workflow
     assert 'quant-lifecycle drift --domain ${{ inputs.strategy_domain }} --no-alerts "${lifecycle_args[@]}"' in workflow
     assert 'repository: ${{ inputs.snapshot_repository }}' in workflow