fix: harden M0 research ledger validation (#306)

Co-authored-by: Codex <noreply@openai.com>

Author: Pigbibi

docs/m0_research_ledger_contract.md

@@ -31,7 +31,12 @@ errors: [安全错误码, ...]
 - 不含账户、仓位、权重、订单、路由、平台、运行时、密钥或执行语义。
 
 快照不得把失效研究线索重新标为新信号。`ready` 来源必须提供时间与来源
-digest；`unavailable` 来源不得携带 hypothesis。
+digest，且 `errors` 必须为空；`unavailable` 来源不得携带 hypothesis。来源、
+subject、theme 与 hypothesis 标识采用与 `QuantAdvisorResearch` 完全相同的
+字符集，不接受 `=`。M0 v5/v6 的 provenance 也是闭合配对：
+
+- v5 必须为 `model_recommendations.v5` 且 `source_input_digest=null`；
+- v6 必须为 `model_recommendations.v6` 且 `source_input_digest` 为 SHA-256。
 
 ## 聚合行为
 
@@ -43,10 +48,16 @@ digest；`unavailable` 来源不得携带 hypothesis。
 3. 同一 subject + source digest 出现不同内容时，作为
    `m0_source_subject_collision` 故障闭合剔除。
 4. 同一 subject 的不同报告保留为独立观测；若 `primary_horizon` 不同，
-   标记 `horizon_conflict.status=conflict`。这是研究队列的人工/后续验证信号，
-   不是交易信号。
+   仅当**当前 fresh**观测不同才标记 `horizon_conflict.status=conflict`。
+   已失效观测被独立投影为 `historical_stale_horizon_drift`：只有存在当前
+   fresh 基准且历史 horizon 不同时才标记 `drift`；完全 stale 的 subject
+   只标记 `unavailable`，不会伪造当前冲突。这两者都是研究队列的人工/后续
+   验证信息，不是交易信号。
 5. 依据 `now` 与 `expires_at` 产生 `fresh`、`stale` 或 `unknown`；来源自身
    为 `stale` 时不会被提升为 fresh。
+6. source 的 `generated_at` 或 `computed_at` 晚于聚合传入的 `now` 时，整个
+   source 以 `m0_source_future_timestamp` 故障闭合剔除；不会以 `unknown`
+   继续展示或参与去重。
 
 输出台账始终固定：
 

python/scripts/m0_research_ledger.py

@@ -75,7 +75,10 @@
     {"event_driven", "long_horizon_growth", "value_quality", "macro_context", "mixed_research"}
 )
 _SOURCE_STATUSES = frozenset({"ready", "unavailable", "stale"})
-_IDENTIFIER = re.compile(r"^[A-Za-z0-9._:/=-]{1,128}$")
+# Keep this byte-for-byte compatible with QuantAdvisorResearch's M0 contract.
+# In particular, ``=`` is not a valid subject, theme, source, or hypothesis
+# identifier there and must not be accepted by this downstream mirror.
+_IDENTIFIER = re.compile(r"^[A-Za-z0-9._:/-]{1,128}$")
 _ERROR_CODE = re.compile(r"^[a-z][a-z0-9_.-]{0,63}$")
 _SHA256 = re.compile(r"^[0-9a-f]{64}$")
 _UTC_TIMESTAMP = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?Z$")
@@ -285,6 +288,8 @@ def validate_m0_research_source_snapshot(payload: object) -> dict[str, Any]:
         source_report_digest is None or generated_at is None or computed_at is None
     ):
         raise M0ResearchLedgerValidationError("ready_source_metadata_invalid")
+    if snapshot["data_status"] == "ready" and errors:
+        raise M0ResearchLedgerValidationError("ready_source_errors_invalid")
     if snapshot["data_status"] == "unavailable" and hypotheses:
         raise M0ResearchLedgerValidationError("unavailable_source_hypotheses_invalid")
     if hypotheses and source_report_digest is None:
@@ -322,6 +327,52 @@ def _source_error(error_set: set[str], code: str) -> None:
         error_set.add(code)
 
 
+def _snapshot_time_is_future(snapshot: Mapping[str, Any], now: dt.datetime) -> bool:
+    """Return whether source metadata could have been produced after this ledger.
+
+    A future source clock is not merely displayed as an ``unknown`` freshness
+    state.  The entire source is omitted so a clock-skewed or replayed source
+    cannot become a current research input by accident.
+    """
+
+    for field, label in (("generated_at", "source_generated_at"), ("computed_at", "source_computed_at")):
+        value = _parse_timestamp(snapshot[field], label, nullable=True)
+        if value is not None and value > now:
+            return True
+    return False
+
+
+def _horizon_views(observations: Sequence[Mapping[str, Any]]) -> tuple[dict[str, Any], dict[str, Any]]:
+    """Separate present disagreement from non-actionable historical drift."""
+
+    fresh_horizons = sorted(
+        {
+            observation["research_context"]["primary_horizon"]
+            for observation in observations
+            if observation["freshness"]["status"] == "fresh"
+        }
+    )
+    stale_horizons = sorted(
+        {
+            observation["research_context"]["primary_horizon"]
+            for observation in observations
+            if observation["freshness"]["status"] == "stale"
+        }
+    )
+    current = {
+        "status": "conflict" if len(fresh_horizons) > 1 else "none",
+        "primary_horizons": fresh_horizons,
+    }
+    if fresh_horizons:
+        stale_status = "drift" if stale_horizons and stale_horizons != fresh_horizons else "none"
+    else:
+        # A stale-only subject has no current primary horizon against which to
+        # call a drift.  It remains visible for audit, but is not an alert.
+        stale_status = "unavailable" if stale_horizons else "none"
+    historical_stale = {"status": stale_status, "primary_horizons": stale_horizons}
+    return current, historical_stale
+
+
 def aggregate_m0_research_sources(
     snapshots: Sequence[object], *, now: str | dt.datetime
 ) -> dict[str, Any]:
@@ -352,6 +403,9 @@ def aggregate_m0_research_sources(
         except M0ResearchLedgerValidationError:
             _source_error(error_set, "m0_source_invalid")
             continue
+        if _snapshot_time_is_future(snapshot, now_at):
+            _source_error(error_set, "m0_source_future_timestamp")
+            continue
         if snapshot["data_status"] == "unavailable":
             _source_error(error_set, "m0_source_unavailable")
             continue
@@ -388,13 +442,15 @@ def aggregate_m0_research_sources(
     fresh_count = 0
     stale_count = 0
     unknown_count = 0
-    conflict_count = 0
+    current_conflict_count = 0
+    historical_stale_drift_count = 0
     for (kind, identifier), observations in sorted(subject_entries.items()):
         observations.sort(key=lambda entry: (entry["source_report_digest"], entry["source_entry_digest"]))
-        horizons = sorted({entry["research_context"]["primary_horizon"] for entry in observations})
-        conflict = len(horizons) > 1
-        if conflict:
-            conflict_count += 1
+        horizon_conflict, historical_stale_horizon_drift = _horizon_views(observations)
+        if horizon_conflict["status"] == "conflict":
+            current_conflict_count += 1
+        if historical_stale_horizon_drift["status"] == "drift":
+            historical_stale_drift_count += 1
         for observation in observations:
             status = observation["freshness"]["status"]
             if status == "fresh":
@@ -407,10 +463,8 @@ def aggregate_m0_research_sources(
             {
                 "subject": {"kind": kind, "identifier": identifier},
                 "observations": observations,
-                "horizon_conflict": {
-                    "status": "conflict" if conflict else "none",
-                    "primary_horizons": horizons,
-                },
+                "horizon_conflict": horizon_conflict,
+                "historical_stale_horizon_drift": historical_stale_horizon_drift,
             }
         )
 
@@ -427,7 +481,8 @@ def aggregate_m0_research_sources(
             "fresh_observation_count": fresh_count,
             "stale_observation_count": stale_count,
             "unknown_observation_count": unknown_count,
-            "horizon_conflict_count": conflict_count,
+            "horizon_conflict_count": current_conflict_count,
+            "historical_stale_horizon_drift_count": historical_stale_drift_count,
         },
         "subjects": subjects,
         "policy": {

python/tests/test_m0_research_ledger.py

@@ -92,6 +92,17 @@ def test_source_and_ledger_schemas_remain_closed_and_read_only(self):
         self.assertFalse(source_schema["additionalProperties"])
         self.assertFalse(ledger_schema["additionalProperties"])
         self.assertEqual(source_schema["properties"]["schema_version"]["const"], "qsl_m0_research_source_snapshot.v1")
+        self.assertEqual(source_schema["properties"]["source_id"]["pattern"], "^[A-Za-z0-9._:/-]{1,128}$")
+        self.assertEqual(source_schema["allOf"][0]["then"]["properties"]["errors"], {"maxItems": 0})
+        provenance_variants = source_schema["$defs"]["m0Hypothesis"]["properties"]["provenance"]["oneOf"]
+        self.assertEqual(len(provenance_variants), 2)
+        self.assertEqual(provenance_variants[0]["properties"]["source_schema_version"], {"const": "5"})
+        self.assertEqual(provenance_variants[0]["properties"]["source_input_digest"], {"type": "null"})
+        self.assertEqual(provenance_variants[1]["properties"]["source_schema_version"], {"const": "6"})
+        self.assertEqual(
+            provenance_variants[1]["properties"]["source_input_digest"],
+            {"type": "string", "pattern": "^[0-9a-f]{64}$"},
+        )
         self.assertEqual(ledger_schema["properties"]["policy"]["properties"]["no_order"], {"const": True})
         self.assertEqual(
             ledger_schema["properties"]["policy"]["properties"]["permitted_next_step"],
@@ -126,9 +137,11 @@ def test_aggregation_deduplicates_subject_and_source_and_flags_horizon_conflict(
             "stale_observation_count": 0,
             "unknown_observation_count": 0,
             "horizon_conflict_count": 1,
+            "historical_stale_horizon_drift_count": 0,
         })
         subject = ledger["subjects"][0]
         self.assertEqual(subject["horizon_conflict"], {"status": "conflict", "primary_horizons": ["long", "medium"]})
+        self.assertEqual(subject["historical_stale_horizon_drift"], {"status": "none", "primary_horizons": []})
         self.assertEqual(subject["observations"][0]["source_ids"], ["quant-advisor-research", "research-mirror"])
 
     def test_expired_or_stale_source_is_visible_but_cannot_become_fresh(self):
@@ -143,11 +156,36 @@ def test_expired_or_stale_source_is_visible_but_cannot_become_fresh(self):
         self.assertEqual(ledger["data_status"], "stale")
         self.assertEqual(ledger["summary"]["fresh_observation_count"], 0)
         self.assertEqual(ledger["summary"]["stale_observation_count"], 2)
+        self.assertEqual(ledger["summary"]["horizon_conflict_count"], 0)
+        self.assertEqual(ledger["summary"]["historical_stale_horizon_drift_count"], 0)
         self.assertEqual(
             {entry["freshness"]["status"] for item in ledger["subjects"] for entry in item["observations"]},
             {"stale"},
         )
 
+    def test_historical_stale_horizon_drift_does_not_create_a_current_conflict(self):
+        fresh = self._snapshot(self._hypothesis(primary_horizon="medium"), data_status="ready")
+        historical = self._snapshot(
+            self._hypothesis(
+                report_digest="d" * 64,
+                entry_digest="e" * 64,
+                hypothesis_id="m0r-long-history",
+                primary_horizon="long",
+            ),
+            data_status="stale",
+        )
+        ledger = m0_research_ledger.aggregate_m0_research_sources(
+            [fresh, historical], now="2026-08-21T12:00:00Z"
+        )
+        subject = ledger["subjects"][0]
+        self.assertEqual(subject["horizon_conflict"], {"status": "none", "primary_horizons": ["medium"]})
+        self.assertEqual(
+            subject["historical_stale_horizon_drift"],
+            {"status": "drift", "primary_horizons": ["long"]},
+        )
+        self.assertEqual(ledger["summary"]["horizon_conflict_count"], 0)
+        self.assertEqual(ledger["summary"]["historical_stale_horizon_drift_count"], 1)
+
     def test_m0_authority_execution_escape_and_source_digest_mismatch_fail_closed(self):
         for mutate, message in (
             (lambda value: value.update(authority="shadow_only"), "authority_invalid"),
@@ -165,6 +203,53 @@ def test_m0_authority_execution_escape_and_source_digest_mismatch_fail_closed(se
         with self.assertRaisesRegex(m0_research_ledger.M0ResearchLedgerValidationError, "source_report_digest_mismatch"):
             m0_research_ledger.validate_m0_research_source_snapshot(snapshot)
 
+    def test_ready_source_errors_and_non_qar_identifier_fail_closed(self):
+        snapshot = self._snapshot(self._hypothesis())
+        snapshot["errors"] = ["upstream_timeout"]
+        with self.assertRaisesRegex(m0_research_ledger.M0ResearchLedgerValidationError, "ready_source_errors_invalid"):
+            m0_research_ledger.validate_m0_research_source_snapshot(snapshot)
+
+        hypothesis = self._hypothesis()
+        hypothesis["subject"]["identifier"] = "SOXX=leveraged"
+        with self.assertRaisesRegex(m0_research_ledger.M0ResearchLedgerValidationError, "subject_identifier_invalid"):
+            m0_research_ledger.validate_m0_research_hypothesis(hypothesis)
+
+    def test_v5_v6_provenance_pairing_matches_the_closed_schema(self):
+        v5 = self._hypothesis()
+        v5["provenance"].update(
+            source_schema_version="5",
+            source_contract_version="model_recommendations.v5",
+            source_input_digest=None,
+        )
+        m0_research_ledger.validate_m0_research_hypothesis(v5)
+
+        invalid_v6 = self._hypothesis()
+        invalid_v6["provenance"]["source_input_digest"] = None
+        with self.assertRaisesRegex(m0_research_ledger.M0ResearchLedgerValidationError, "source_input_digest_invalid"):
+            m0_research_ledger.validate_m0_research_hypothesis(invalid_v6)
+
+        invalid_v5 = copy.deepcopy(v5)
+        invalid_v5["provenance"]["source_contract_version"] = "model_recommendations.v6"
+        with self.assertRaisesRegex(m0_research_ledger.M0ResearchLedgerValidationError, "source_contract_version_invalid"):
+            m0_research_ledger.validate_m0_research_hypothesis(invalid_v5)
+
+    def test_future_source_metadata_is_omitted_fail_closed(self):
+        for field, value in (
+            ("generated_at", "2026-08-22T12:00:00Z"),
+            ("computed_at", "2026-08-22T12:00:00Z"),
+        ):
+            with self.subTest(field=field):
+                snapshot = self._snapshot(self._hypothesis())
+                snapshot[field] = value
+                if field == "generated_at":
+                    snapshot["computed_at"] = "2026-08-22T12:01:00Z"
+                ledger = m0_research_ledger.aggregate_m0_research_sources(
+                    [snapshot], now="2026-08-21T12:00:00Z"
+                )
+                self.assertEqual(ledger["data_status"], "unavailable")
+                self.assertEqual(ledger["summary"]["observation_count"], 0)
+                self.assertEqual(ledger["errors"], ["m0_source_future_timestamp"])
+
     def test_same_subject_and_source_with_different_payloads_is_omitted_fail_closed(self):
         first = self._hypothesis()
         collision = self._hypothesis(entry_digest="f" * 64, hypothesis_id="m0r-collision")

schemas/qsl-m0-research-ledger.v1.schema.json

@@ -14,14 +14,15 @@
     "summary": {
       "type": "object",
       "additionalProperties": false,
-      "required": ["subject_count", "observation_count", "fresh_observation_count", "stale_observation_count", "unknown_observation_count", "horizon_conflict_count"],
+      "required": ["subject_count", "observation_count", "fresh_observation_count", "stale_observation_count", "unknown_observation_count", "horizon_conflict_count", "historical_stale_horizon_drift_count"],
       "properties": {
         "subject_count": { "type": "integer", "minimum": 0, "maximum": 50000 },
         "observation_count": { "type": "integer", "minimum": 0, "maximum": 50000 },
         "fresh_observation_count": { "type": "integer", "minimum": 0, "maximum": 50000 },
         "stale_observation_count": { "type": "integer", "minimum": 0, "maximum": 50000 },
         "unknown_observation_count": { "type": "integer", "minimum": 0, "maximum": 50000 },
-        "horizon_conflict_count": { "type": "integer", "minimum": 0, "maximum": 50000 }
+        "horizon_conflict_count": { "type": "integer", "minimum": 0, "maximum": 50000 },
+        "historical_stale_horizon_drift_count": { "type": "integer", "minimum": 0, "maximum": 50000 }
       }
     },
     "subjects": {
@@ -30,15 +31,15 @@
       "items": {
         "type": "object",
         "additionalProperties": false,
-        "required": ["subject", "observations", "horizon_conflict"],
+        "required": ["subject", "observations", "horizon_conflict", "historical_stale_horizon_drift"],
         "properties": {
           "subject": {
             "type": "object",
             "additionalProperties": false,
             "required": ["kind", "identifier"],
             "properties": {
               "kind": { "enum": ["asset_idea", "theme_context", "strategy_hypothesis", "risk_context"] },
-              "identifier": { "type": "string", "pattern": "^[A-Za-z0-9._:/=-]{1,128}$" }
+              "identifier": { "type": "string", "pattern": "^[A-Za-z0-9._:/-]{1,128}$" }
             }
           },
           "observations": {
@@ -55,7 +56,22 @@
               "status": { "enum": ["none", "conflict"] },
               "primary_horizons": {
                 "type": "array",
-                "minItems": 1,
+                "minItems": 0,
+                "maxItems": 4,
+                "uniqueItems": true,
+                "items": { "enum": ["short", "medium", "long", "not_applicable"] }
+              }
+            }
+          },
+          "historical_stale_horizon_drift": {
+            "type": "object",
+            "additionalProperties": false,
+            "required": ["status", "primary_horizons"],
+            "properties": {
+              "status": { "enum": ["none", "drift", "unavailable"] },
+              "primary_horizons": {
+                "type": "array",
+                "minItems": 0,
                 "maxItems": 4,
                 "uniqueItems": true,
                 "items": { "enum": ["short", "medium", "long", "not_applicable"] }
@@ -93,11 +109,11 @@
           "minItems": 1,
           "maxItems": 100,
           "uniqueItems": true,
-          "items": { "type": "string", "pattern": "^[A-Za-z0-9._:/=-]{1,128}$" }
+          "items": { "type": "string", "pattern": "^[A-Za-z0-9._:/-]{1,128}$" }
         },
         "source_report_digest": { "type": "string", "pattern": "^[0-9a-f]{64}$" },
         "source_entry_digest": { "type": "string", "pattern": "^[0-9a-f]{64}$" },
-        "hypothesis_id": { "type": "string", "pattern": "^[A-Za-z0-9._:/=-]{1,128}$" },
+        "hypothesis_id": { "type": "string", "pattern": "^[A-Za-z0-9._:/-]{1,128}$" },
         "as_of": { "type": "string", "format": "date" },
         "generated_at": { "type": "string", "format": "date-time", "maxLength": 64 },
         "expires_at": { "type": "string", "format": "date-time", "maxLength": 64 },
@@ -121,7 +137,7 @@
               "type": "array",
               "maxItems": 24,
               "uniqueItems": true,
-              "items": { "type": "string", "pattern": "^[A-Za-z0-9._:/=-]{1,128}$" }
+              "items": { "type": "string", "pattern": "^[A-Za-z0-9._:/-]{1,128}$" }
             }
           }
         },