diff --git a/docs/m0_research_ledger_contract.md b/docs/m0_research_ledger_contract.md
new file mode 100644
index 0000000..23a56c8
--- /dev/null
+++ b/docs/m0_research_ledger_contract.md
@@ -0,0 +1,61 @@
+# M0 研究台账 v1
+
+`qsl_m0_research_source_snapshot.v1` 和 `qsl_m0_research_ledger.v1` 是
+`QuantAdvisorResearch` 与运行时设置之间的只读边界。
+
+它们只接收已由顾投系统产生且已闭合校验的
+`qsl.m0_research_hypothesis.v1`。这不是策略选择、仓位分配、运行时目标、
+平台路由或订单合约；实现不依赖 QPK selector，也不调用调度、券商或
+执行组件。
+
+## 输入快照
+
+一个 source snapshot 对应一个来源报告摘要：
+
+```text
+schema_version: qsl_m0_research_source_snapshot.v1
+source_id: 稳定的传输来源标识
+source_report_digest: 顾投报告 SHA-256
+generated_at / computed_at / data_status
+hypotheses: [qsl.m0_research_hypothesis.v1, ...]
+errors: [安全错误码, ...]
+```
+
+快照中的每条 hypothesis 必须：
+
+- 精确匹配 M0 的字段闭包，并具有 `authority=research_only`、
+  `no_order=true` 和 `permitted_next_step=research_validation_only`；
+- 具有有效的 7 天有效期、`as_of` 与生成时间关系，以及来源报告/来源条目
+  SHA-256；
+- 与快照的 `source_report_digest` 完全一致；
+- 不含账户、仓位、权重、订单、路由、平台、运行时、密钥或执行语义。
+
+快照不得把失效研究线索重新标为新信号。`ready` 来源必须提供时间与来源
+digest；`unavailable` 来源不得携带 hypothesis。
+
+## 聚合行为
+
+`aggregate_m0_research_sources(snapshots, now=...)` 是确定性的纯函数：
+
+1. 校验每个来源；无效来源只产生安全错误码，不能污染有效台账。
+2. 以 `(subject.kind, subject.identifier, source_report_digest)` 合并完全相同
+   的观测，并保留所有 `source_ids`。
+3. 同一 subject + source digest 出现不同内容时，作为
+   `m0_source_subject_collision` 故障闭合剔除。
+4. 同一 subject 的不同报告保留为独立观测；若 `primary_horizon` 不同，
+   标记 `horizon_conflict.status=conflict`。这是研究队列的人工/后续验证信号，
+   不是交易信号。
+5. 依据 `now` 与 `expires_at` 产生 `fresh`、`stale` 或 `unknown`；来源自身
+   为 `stale` 时不会被提升为 fresh。
+
+输出台账始终固定：
+
+```text
+authority: research_only
+no_order: true
+permitted_next_step: research_validation_only
+```
+
+因此控制台可安全显示 subject、来源证据、有效期和短/中/长期冲突，而不能从
+该对象获得任何策略切换、平台控制或执行动作。未来若要把其中一条线索转成
+研究任务，必须由独立的 P1--P3 入场流程重新绑定数据、回测与审计证据。
diff --git a/python/scripts/m0_research_ledger.py b/python/scripts/m0_research_ledger.py
new file mode 100644
index 0000000..0f2ebe9
--- /dev/null
+++ b/python/scripts/m0_research_ledger.py
@@ -0,0 +1,454 @@
+"""Validate and aggregate closed M0 research hypotheses as a read-only ledger.
+
+This module deliberately has no dependency on a selector, runtime target,
+platform configuration, scheduler, broker, or control-plane dispatcher.  It
+only turns valid ``qsl.m0_research_hypothesis.v1`` source snapshots into a
+bounded ledger suitable for a research console or a later *independent*
+research-task admission step.
+"""
+
+from __future__ import annotations
+
+import copy
+import datetime as dt
+import json
+import re
+from collections import defaultdict
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+
+M0_HYPOTHESIS_SCHEMA = "qsl.m0_research_hypothesis.v1"
+M0_SOURCE_SNAPSHOT_SCHEMA = "qsl_m0_research_source_snapshot.v1"
+M0_LEDGER_SCHEMA = "qsl_m0_research_ledger.v1"
+M0_AUTHORITY = "research_only"
+M0_NEXT_STEP = "research_validation_only"
+
+_M0_FIELDS = frozenset(
+    {
+        "schema_version",
+        "artifact_type",
+        "authority",
+        "no_order",
+        "hypothesis_id",
+        "as_of",
+        "generated_at",
+        "expires_at",
+        "subject",
+        "research_context",
+        "evidence",
+        "provenance",
+        "permitted_next_step",
+    }
+)
+_SUBJECT_FIELDS = frozenset({"kind", "identifier"})
+_RESEARCH_CONTEXT_FIELDS = frozenset(
+    {"state", "primary_horizon", "suitable_horizons", "source_confidence", "source_style", "theme_ids"}
+)
+_EVIDENCE_FIELDS = frozenset({"source_entry_digest", "evidence_ref_count", "risk_note_count"})
+_PROVENANCE_FIELDS = frozenset(
+    {
+        "source_project",
+        "source_schema_version",
+        "source_contract_version",
+        "source_report_digest",
+        "source_input_digest",
+    }
+)
+_SOURCE_SNAPSHOT_FIELDS = frozenset(
+    {
+        "schema_version",
+        "source_id",
+        "source_report_digest",
+        "generated_at",
+        "computed_at",
+        "data_status",
+        "hypotheses",
+        "errors",
+    }
+)
+_ALLOWED_SUBJECT_KINDS = frozenset({"asset_idea", "theme_context", "strategy_hypothesis", "risk_context"})
+_ALLOWED_RESEARCH_STATES = frozenset({"candidate", "source_verification_required", "deferred", "context_only"})
+_ALLOWED_HORIZONS = frozenset({"short", "medium", "long", "not_applicable"})
+_ALLOWED_SOURCE_CONFIDENCE = frozenset({"high", "medium", "low", "mixed", "no_event", "unknown"})
+_ALLOWED_SOURCE_STYLES = frozenset(
+    {"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}$")
+_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$")
+_FORBIDDEN_FIELD_FRAGMENTS = (
+    "account",
+    "allocation",
+    "broker",
+    "canary",
+    "credential",
+    "execution",
+    "live",
+    "order",
+    "paper",
+    "platform",
+    "portfolio",
+    "position",
+    "quantity",
+    "route",
+    "runtime",
+    "secret",
+    "share",
+    "switch",
+    "target",
+    "token",
+    "trade",
+    "weight",
+)
+
+
+class M0ResearchLedgerValidationError(ValueError):
+    """Raised when an M0 source snapshot is malformed or out of scope."""
+
+
+def _exact_mapping(value: object, fields: frozenset[str], label: str) -> dict[str, Any]:
+    if not isinstance(value, Mapping) or set(value) != fields:
+        raise M0ResearchLedgerValidationError(f"{label}_keys_invalid")
+    return dict(value)
+
+
+def _require_identifier(value: object, label: str) -> str:
+    if not isinstance(value, str) or not _IDENTIFIER.fullmatch(value):
+        raise M0ResearchLedgerValidationError(f"{label}_invalid")
+    return value
+
+
+def _require_sha256(value: object, label: str, *, nullable: bool = False) -> str | None:
+    if nullable and value is None:
+        return None
+    if not isinstance(value, str) or not _SHA256.fullmatch(value):
+        raise M0ResearchLedgerValidationError(f"{label}_invalid")
+    return value
+
+
+def _parse_timestamp(value: object, label: str, *, nullable: bool = False) -> dt.datetime | None:
+    if nullable and value is None:
+        return None
+    if not isinstance(value, str) or not _UTC_TIMESTAMP.fullmatch(value):
+        raise M0ResearchLedgerValidationError(f"{label}_invalid")
+    try:
+        parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError as exc:
+        raise M0ResearchLedgerValidationError(f"{label}_invalid") from exc
+    if parsed.tzinfo is None or parsed.utcoffset() != dt.timedelta(0):
+        raise M0ResearchLedgerValidationError(f"{label}_invalid")
+    return parsed
+
+
+def _utc_timestamp(value: dt.datetime) -> str:
+    normalized = value.astimezone(dt.UTC)
+    return normalized.isoformat().replace("+00:00", "Z")
+
+
+def _reject_forbidden_semantic_fields(value: object, *, is_hypothesis_root: bool = True) -> None:
+    """Reject escape hatches even when a future nested object is added."""
+
+    if isinstance(value, Mapping):
+        for key, nested in value.items():
+            if not isinstance(key, str):
+                raise M0ResearchLedgerValidationError("field_name_invalid")
+            normalized = "".join(character for character in key.casefold() if character.isalnum())
+            is_explicit_no_order = is_hypothesis_root and key == "no_order"
+            if not is_explicit_no_order and any(fragment in normalized for fragment in _FORBIDDEN_FIELD_FRAGMENTS):
+                raise M0ResearchLedgerValidationError("forbidden_semantic_field")
+            _reject_forbidden_semantic_fields(nested, is_hypothesis_root=False)
+    elif isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
+        for nested in value:
+            _reject_forbidden_semantic_fields(nested, is_hypothesis_root=False)
+
+
+def _canonical_json(value: object) -> str:
+    try:
+        return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False)
+    except (TypeError, ValueError) as exc:
+        raise M0ResearchLedgerValidationError("m0_record_not_canonicalizable") from exc
+
+
+def validate_m0_research_hypothesis(payload: object) -> dict[str, Any]:
+    """Validate the QAR-owned M0 contract without importing its application code."""
+
+    hypothesis = _exact_mapping(payload, _M0_FIELDS, "hypothesis")
+    _reject_forbidden_semantic_fields(hypothesis)
+    if hypothesis["schema_version"] != M0_HYPOTHESIS_SCHEMA:
+        raise M0ResearchLedgerValidationError("schema_version_invalid")
+    if hypothesis["artifact_type"] != "research_hypothesis":
+        raise M0ResearchLedgerValidationError("artifact_type_invalid")
+    if hypothesis["authority"] != M0_AUTHORITY:
+        raise M0ResearchLedgerValidationError("authority_invalid")
+    if hypothesis["no_order"] is not True:
+        raise M0ResearchLedgerValidationError("no_order_invalid")
+    if hypothesis["permitted_next_step"] != M0_NEXT_STEP:
+        raise M0ResearchLedgerValidationError("permitted_next_step_invalid")
+    _require_identifier(hypothesis["hypothesis_id"], "hypothesis_id")
+
+    if not isinstance(hypothesis["as_of"], str):
+        raise M0ResearchLedgerValidationError("as_of_invalid")
+    try:
+        as_of = dt.date.fromisoformat(hypothesis["as_of"])
+    except ValueError as exc:
+        raise M0ResearchLedgerValidationError("as_of_invalid") from exc
+    generated_at = _parse_timestamp(hypothesis["generated_at"], "generated_at")
+    expires_at = _parse_timestamp(hypothesis["expires_at"], "expires_at")
+    assert generated_at is not None and expires_at is not None
+    if expires_at != generated_at + dt.timedelta(days=7) or as_of > generated_at.date():
+        raise M0ResearchLedgerValidationError("hypothesis_time_invalid")
+
+    subject = _exact_mapping(hypothesis["subject"], _SUBJECT_FIELDS, "subject")
+    if subject["kind"] not in _ALLOWED_SUBJECT_KINDS:
+        raise M0ResearchLedgerValidationError("subject_kind_invalid")
+    _require_identifier(subject["identifier"], "subject_identifier")
+
+    context = _exact_mapping(hypothesis["research_context"], _RESEARCH_CONTEXT_FIELDS, "research_context")
+    if context["state"] not in _ALLOWED_RESEARCH_STATES:
+        raise M0ResearchLedgerValidationError("research_state_invalid")
+    if context["primary_horizon"] not in _ALLOWED_HORIZONS:
+        raise M0ResearchLedgerValidationError("primary_horizon_invalid")
+    suitable_horizons = context["suitable_horizons"]
+    if (
+        not isinstance(suitable_horizons, list)
+        or not suitable_horizons
+        or len(suitable_horizons) > len(_ALLOWED_HORIZONS)
+        or len(set(suitable_horizons)) != len(suitable_horizons)
+        or context["primary_horizon"] not in suitable_horizons
+        or any(horizon not in _ALLOWED_HORIZONS for horizon in suitable_horizons)
+    ):
+        raise M0ResearchLedgerValidationError("suitable_horizons_invalid")
+    if context["source_confidence"] not in _ALLOWED_SOURCE_CONFIDENCE:
+        raise M0ResearchLedgerValidationError("source_confidence_invalid")
+    if context["source_style"] not in _ALLOWED_SOURCE_STYLES:
+        raise M0ResearchLedgerValidationError("source_style_invalid")
+    theme_ids = context["theme_ids"]
+    if not isinstance(theme_ids, list) or len(theme_ids) > 24 or len(set(theme_ids)) != len(theme_ids):
+        raise M0ResearchLedgerValidationError("theme_ids_invalid")
+    for theme_id in theme_ids:
+        _require_identifier(theme_id, "theme_id")
+
+    evidence = _exact_mapping(hypothesis["evidence"], _EVIDENCE_FIELDS, "evidence")
+    _require_sha256(evidence["source_entry_digest"], "source_entry_digest")
+    for key in ("evidence_ref_count", "risk_note_count"):
+        if isinstance(evidence[key], bool) or not isinstance(evidence[key], int) or evidence[key] < 0:
+            raise M0ResearchLedgerValidationError(f"{key}_invalid")
+
+    provenance = _exact_mapping(hypothesis["provenance"], _PROVENANCE_FIELDS, "provenance")
+    if provenance["source_project"] != "QuantAdvisorResearch":
+        raise M0ResearchLedgerValidationError("source_project_invalid")
+    if provenance["source_schema_version"] not in {"5", "6"}:
+        raise M0ResearchLedgerValidationError("source_schema_version_invalid")
+    expected_contract = f"model_recommendations.v{provenance['source_schema_version']}"
+    if provenance["source_contract_version"] != expected_contract:
+        raise M0ResearchLedgerValidationError("source_contract_version_invalid")
+    _require_sha256(provenance["source_report_digest"], "source_report_digest")
+    source_input_digest = _require_sha256(
+        provenance["source_input_digest"],
+        "source_input_digest",
+        nullable=provenance["source_schema_version"] == "5",
+    )
+    if provenance["source_schema_version"] == "5" and source_input_digest is not None:
+        raise M0ResearchLedgerValidationError("source_input_digest_invalid")
+    return copy.deepcopy(hypothesis)
+
+
+def validate_m0_research_source_snapshot(payload: object) -> dict[str, Any]:
+    """Validate a single closed, read-only M0 source snapshot."""
+
+    snapshot = _exact_mapping(payload, _SOURCE_SNAPSHOT_FIELDS, "source_snapshot")
+    if snapshot["schema_version"] != M0_SOURCE_SNAPSHOT_SCHEMA:
+        raise M0ResearchLedgerValidationError("source_snapshot_schema_invalid")
+    _require_identifier(snapshot["source_id"], "source_id")
+    if snapshot["data_status"] not in _SOURCE_STATUSES:
+        raise M0ResearchLedgerValidationError("source_data_status_invalid")
+    generated_at = _parse_timestamp(snapshot["generated_at"], "source_generated_at", nullable=True)
+    computed_at = _parse_timestamp(snapshot["computed_at"], "source_computed_at", nullable=True)
+    if generated_at is not None and computed_at is not None and computed_at < generated_at:
+        raise M0ResearchLedgerValidationError("source_time_invalid")
+    source_report_digest = _require_sha256(
+        snapshot["source_report_digest"], "source_report_digest", nullable=True
+    )
+    hypotheses = snapshot["hypotheses"]
+    if not isinstance(hypotheses, list) or len(hypotheses) > 500:
+        raise M0ResearchLedgerValidationError("source_hypotheses_invalid")
+    errors = snapshot["errors"]
+    if not isinstance(errors, list) or len(errors) > 20 or any(
+        not isinstance(error, str) or not _ERROR_CODE.fullmatch(error) for error in errors
+    ):
+        raise M0ResearchLedgerValidationError("source_errors_invalid")
+
+    if snapshot["data_status"] == "ready" and (
+        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"] == "unavailable" and hypotheses:
+        raise M0ResearchLedgerValidationError("unavailable_source_hypotheses_invalid")
+    if hypotheses and source_report_digest is None:
+        raise M0ResearchLedgerValidationError("source_report_digest_invalid")
+
+    validated_hypotheses: list[dict[str, Any]] = []
+    for hypothesis in hypotheses:
+        validated = validate_m0_research_hypothesis(hypothesis)
+        if validated["provenance"]["source_report_digest"] != source_report_digest:
+            raise M0ResearchLedgerValidationError("source_report_digest_mismatch")
+        if computed_at is not None:
+            hypothesis_generated_at = _parse_timestamp(validated["generated_at"], "generated_at")
+            assert hypothesis_generated_at is not None
+            if hypothesis_generated_at > computed_at:
+                raise M0ResearchLedgerValidationError("source_hypothesis_time_invalid")
+        validated_hypotheses.append(validated)
+    snapshot["hypotheses"] = validated_hypotheses
+    return copy.deepcopy(snapshot)
+
+
+def _freshness(hypothesis: Mapping[str, Any], source_status: str, now: dt.datetime) -> dict[str, Any]:
+    generated_at = _parse_timestamp(hypothesis["generated_at"], "generated_at")
+    expires_at = _parse_timestamp(hypothesis["expires_at"], "expires_at")
+    assert generated_at is not None and expires_at is not None
+    age_seconds = max(0, int((now - generated_at).total_seconds()))
+    if generated_at > now:
+        return {"status": "unknown", "age_seconds": None}
+    if source_status != "ready" or now >= expires_at:
+        return {"status": "stale", "age_seconds": age_seconds}
+    return {"status": "fresh", "age_seconds": age_seconds}
+
+
+def _source_error(error_set: set[str], code: str) -> None:
+    if len(error_set) < 20:
+        error_set.add(code)
+
+
+def aggregate_m0_research_sources(
+    snapshots: Sequence[object], *, now: str | dt.datetime
+) -> dict[str, Any]:
+    """Build a deterministic, no-order ledger from independent source snapshots.
+
+    Exact duplicates are collapsed by ``(subject.kind, subject.identifier,
+    source_report_digest)``.  Differing payloads under that same identity are
+    treated as a source collision and omitted fail-closed.  Different source
+    reports for one subject remain visible and produce a horizon-conflict flag
+    when their primary horizons disagree.
+    """
+
+    if not isinstance(snapshots, Sequence) or isinstance(snapshots, (str, bytes)) or len(snapshots) > 100:
+        raise M0ResearchLedgerValidationError("source_snapshots_invalid")
+    if isinstance(now, dt.datetime) and (now.tzinfo is None or now.utcoffset() is None):
+        raise M0ResearchLedgerValidationError("ledger_now_invalid")
+    now_at = _parse_timestamp(
+        _utc_timestamp(now) if isinstance(now, dt.datetime) else now,
+        "ledger_now",
+    )
+    assert now_at is not None
+
+    error_set: set[str] = set()
+    observations_by_key: dict[tuple[str, str, str], list[tuple[str, dict[str, Any], str]]] = defaultdict(list)
+    for raw_snapshot in snapshots:
+        try:
+            snapshot = validate_m0_research_source_snapshot(raw_snapshot)
+        except M0ResearchLedgerValidationError:
+            _source_error(error_set, "m0_source_invalid")
+            continue
+        if snapshot["data_status"] == "unavailable":
+            _source_error(error_set, "m0_source_unavailable")
+            continue
+        for hypothesis in snapshot["hypotheses"]:
+            subject = hypothesis["subject"]
+            key = (subject["kind"], subject["identifier"], hypothesis["provenance"]["source_report_digest"])
+            observations_by_key[key].append((snapshot["source_id"], hypothesis, snapshot["data_status"]))
+
+    subject_entries: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list)
+    for (subject_kind, subject_identifier, source_report_digest), candidates in observations_by_key.items():
+        canonical_payloads = {_canonical_json(candidate[1]) for candidate in candidates}
+        if len(canonical_payloads) != 1:
+            _source_error(error_set, "m0_source_subject_collision")
+            continue
+        source_ids = sorted({candidate[0] for candidate in candidates})
+        source_statuses = {candidate[2] for candidate in candidates}
+        hypothesis = copy.deepcopy(candidates[0][1])
+        source_status = "ready" if source_statuses == {"ready"} else "stale"
+        fresh = _freshness(hypothesis, source_status, now_at)
+        observation = {
+            "source_ids": source_ids,
+            "source_report_digest": source_report_digest,
+            "source_entry_digest": hypothesis["evidence"]["source_entry_digest"],
+            "hypothesis_id": hypothesis["hypothesis_id"],
+            "as_of": hypothesis["as_of"],
+            "generated_at": hypothesis["generated_at"],
+            "expires_at": hypothesis["expires_at"],
+            "research_context": copy.deepcopy(hypothesis["research_context"]),
+            "freshness": fresh,
+        }
+        subject_entries[(subject_kind, subject_identifier)].append(observation)
+
+    subjects: list[dict[str, Any]] = []
+    fresh_count = 0
+    stale_count = 0
+    unknown_count = 0
+    conflict_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
+        for observation in observations:
+            status = observation["freshness"]["status"]
+            if status == "fresh":
+                fresh_count += 1
+            elif status == "stale":
+                stale_count += 1
+            else:
+                unknown_count += 1
+        subjects.append(
+            {
+                "subject": {"kind": kind, "identifier": identifier},
+                "observations": observations,
+                "horizon_conflict": {
+                    "status": "conflict" if conflict else "none",
+                    "primary_horizons": horizons,
+                },
+            }
+        )
+
+    observation_count = fresh_count + stale_count + unknown_count
+    data_status = "ready" if fresh_count else "stale" if observation_count else "unavailable"
+    ledger = {
+        "schema_version": M0_LEDGER_SCHEMA,
+        "generated_at": _utc_timestamp(now_at),
+        "computed_at": _utc_timestamp(now_at),
+        "data_status": data_status,
+        "summary": {
+            "subject_count": len(subjects),
+            "observation_count": observation_count,
+            "fresh_observation_count": fresh_count,
+            "stale_observation_count": stale_count,
+            "unknown_observation_count": unknown_count,
+            "horizon_conflict_count": conflict_count,
+        },
+        "subjects": subjects,
+        "policy": {
+            "authority": M0_AUTHORITY,
+            "no_order": True,
+            "permitted_next_step": M0_NEXT_STEP,
+            "notice": "Read-only M0 research ledger; it cannot select, route, or execute a strategy.",
+        },
+        "errors": sorted(error_set),
+    }
+    return ledger
+
+
+__all__ = [
+    "M0_AUTHORITY",
+    "M0_HYPOTHESIS_SCHEMA",
+    "M0_LEDGER_SCHEMA",
+    "M0_NEXT_STEP",
+    "M0_SOURCE_SNAPSHOT_SCHEMA",
+    "M0ResearchLedgerValidationError",
+    "aggregate_m0_research_sources",
+    "validate_m0_research_hypothesis",
+    "validate_m0_research_source_snapshot",
+]
diff --git a/python/tests/test_m0_research_ledger.py b/python/tests/test_m0_research_ledger.py
new file mode 100644
index 0000000..c13123b
--- /dev/null
+++ b/python/tests/test_m0_research_ledger.py
@@ -0,0 +1,184 @@
+from __future__ import annotations
+
+import copy
+import importlib.util
+import json
+import sys
+import unittest
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPTS = ROOT / "scripts"
+
+
+def _load_module(name: str):
+    spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py")
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+m0_research_ledger = _load_module("m0_research_ledger")
+
+
+class M0ResearchLedgerTest(unittest.TestCase):
+    def _hypothesis(
+        self,
+        *,
+        report_digest: str = "a" * 64,
+        entry_digest: str = "b" * 64,
+        hypothesis_id: str = "m0r-semiconductor-1",
+        primary_horizon: str = "medium",
+    ) -> dict[str, object]:
+        return {
+            "schema_version": "qsl.m0_research_hypothesis.v1",
+            "artifact_type": "research_hypothesis",
+            "authority": "research_only",
+            "no_order": True,
+            "hypothesis_id": hypothesis_id,
+            "as_of": "2026-08-20",
+            "generated_at": "2026-08-20T12:00:00Z",
+            "expires_at": "2026-08-27T12:00:00Z",
+            "subject": {"kind": "asset_idea", "identifier": "SOXX"},
+            "research_context": {
+                "state": "candidate",
+                "primary_horizon": primary_horizon,
+                "suitable_horizons": [primary_horizon],
+                "source_confidence": "high",
+                "source_style": "mixed_research",
+                "theme_ids": ["semiconductors"],
+            },
+            "evidence": {
+                "source_entry_digest": entry_digest,
+                "evidence_ref_count": 3,
+                "risk_note_count": 1,
+            },
+            "provenance": {
+                "source_project": "QuantAdvisorResearch",
+                "source_schema_version": "6",
+                "source_contract_version": "model_recommendations.v6",
+                "source_report_digest": report_digest,
+                "source_input_digest": "c" * 64,
+            },
+            "permitted_next_step": "research_validation_only",
+        }
+
+    def _snapshot(
+        self,
+        hypothesis: dict[str, object],
+        *,
+        source_id: str = "quant-advisor-research",
+        data_status: str = "ready",
+    ) -> dict[str, object]:
+        return {
+            "schema_version": "qsl_m0_research_source_snapshot.v1",
+            "source_id": source_id,
+            "source_report_digest": hypothesis["provenance"]["source_report_digest"],
+            "generated_at": "2026-08-20T12:01:00Z",
+            "computed_at": "2026-08-20T12:02:00Z",
+            "data_status": data_status,
+            "hypotheses": [hypothesis],
+            "errors": [],
+        }
+
+    def test_source_and_ledger_schemas_remain_closed_and_read_only(self):
+        source_schema = json.loads(
+            (ROOT.parent / "schemas" / "qsl-m0-research-source-snapshot.v1.schema.json").read_text()
+        )
+        ledger_schema = json.loads((ROOT.parent / "schemas" / "qsl-m0-research-ledger.v1.schema.json").read_text())
+        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(ledger_schema["properties"]["policy"]["properties"]["no_order"], {"const": True})
+        self.assertEqual(
+            ledger_schema["properties"]["policy"]["properties"]["permitted_next_step"],
+            {"const": "research_validation_only"},
+        )
+
+    def test_aggregation_deduplicates_subject_and_source_and_flags_horizon_conflict(self):
+        first = self._hypothesis()
+        duplicate = copy.deepcopy(first)
+        second = self._hypothesis(
+            report_digest="d" * 64,
+            entry_digest="e" * 64,
+            hypothesis_id="m0r-semiconductor-2",
+            primary_horizon="long",
+        )
+        ledger = m0_research_ledger.aggregate_m0_research_sources(
+            [
+                self._snapshot(first, source_id="quant-advisor-research"),
+                self._snapshot(duplicate, source_id="research-mirror"),
+                self._snapshot(second, source_id="quant-advisor-research-v2"),
+            ],
+            now="2026-08-21T12:00:00Z",
+        )
+        self.assertEqual(ledger["schema_version"], "qsl_m0_research_ledger.v1")
+        self.assertEqual(ledger["data_status"], "ready")
+        self.assertEqual(ledger["policy"]["authority"], "research_only")
+        self.assertTrue(ledger["policy"]["no_order"])
+        self.assertEqual(ledger["summary"], {
+            "subject_count": 1,
+            "observation_count": 2,
+            "fresh_observation_count": 2,
+            "stale_observation_count": 0,
+            "unknown_observation_count": 0,
+            "horizon_conflict_count": 1,
+        })
+        subject = ledger["subjects"][0]
+        self.assertEqual(subject["horizon_conflict"], {"status": "conflict", "primary_horizons": ["long", "medium"]})
+        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):
+        expired = self._snapshot(self._hypothesis(), data_status="ready")
+        stale = self._snapshot(
+            self._hypothesis(report_digest="d" * 64, entry_digest="e" * 64, hypothesis_id="m0r-stale"),
+            data_status="stale",
+        )
+        ledger = m0_research_ledger.aggregate_m0_research_sources(
+            [expired, stale], now="2026-08-28T12:00:00Z"
+        )
+        self.assertEqual(ledger["data_status"], "stale")
+        self.assertEqual(ledger["summary"]["fresh_observation_count"], 0)
+        self.assertEqual(ledger["summary"]["stale_observation_count"], 2)
+        self.assertEqual(
+            {entry["freshness"]["status"] for item in ledger["subjects"] for entry in item["observations"]},
+            {"stale"},
+        )
+
+    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"),
+            (lambda value: value.update(no_order=False), "no_order_invalid"),
+            (lambda value: value["research_context"].update(targetWeight=1), "forbidden_semantic_field"),
+        ):
+            with self.subTest(mutate=message):
+                hypothesis = self._hypothesis()
+                mutate(hypothesis)
+                with self.assertRaisesRegex(m0_research_ledger.M0ResearchLedgerValidationError, message):
+                    m0_research_ledger.validate_m0_research_hypothesis(hypothesis)
+
+        snapshot = self._snapshot(self._hypothesis())
+        snapshot["source_report_digest"] = "f" * 64
+        with self.assertRaisesRegex(m0_research_ledger.M0ResearchLedgerValidationError, "source_report_digest_mismatch"):
+            m0_research_ledger.validate_m0_research_source_snapshot(snapshot)
+
+    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")
+        ledger = m0_research_ledger.aggregate_m0_research_sources(
+            [
+                self._snapshot(first, source_id="quant-advisor-research"),
+                self._snapshot(collision, source_id="research-mirror"),
+            ],
+            now="2026-08-21T12:00:00Z",
+        )
+        self.assertEqual(ledger["data_status"], "unavailable")
+        self.assertEqual(ledger["summary"]["subject_count"], 0)
+        self.assertEqual(ledger["errors"], ["m0_source_subject_collision"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/schemas/qsl-m0-research-ledger.v1.schema.json b/schemas/qsl-m0-research-ledger.v1.schema.json
new file mode 100644
index 0000000..f74f0ba
--- /dev/null
+++ b/schemas/qsl-m0-research-ledger.v1.schema.json
@@ -0,0 +1,140 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://quantstrategylab.github.io/QuantRuntimeSettings/schemas/qsl-m0-research-ledger.v1.schema.json",
+  "title": "QSL M0 Research Ledger v1",
+  "description": "Read-only aggregation of closed M0 research hypotheses. It is not a strategy selector, allocation, runtime, platform, or execution contract.",
+  "type": "object",
+  "additionalProperties": false,
+  "required": ["schema_version", "generated_at", "computed_at", "data_status", "summary", "subjects", "policy", "errors"],
+  "properties": {
+    "schema_version": { "const": "qsl_m0_research_ledger.v1" },
+    "generated_at": { "type": "string", "format": "date-time", "maxLength": 64 },
+    "computed_at": { "type": "string", "format": "date-time", "maxLength": 64 },
+    "data_status": { "enum": ["ready", "unavailable", "stale"] },
+    "summary": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": ["subject_count", "observation_count", "fresh_observation_count", "stale_observation_count", "unknown_observation_count", "horizon_conflict_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 }
+      }
+    },
+    "subjects": {
+      "type": "array",
+      "maxItems": 50000,
+      "items": {
+        "type": "object",
+        "additionalProperties": false,
+        "required": ["subject", "observations", "horizon_conflict"],
+        "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}$" }
+            }
+          },
+          "observations": {
+            "type": "array",
+            "minItems": 1,
+            "maxItems": 100,
+            "items": { "$ref": "#/$defs/observation" }
+          },
+          "horizon_conflict": {
+            "type": "object",
+            "additionalProperties": false,
+            "required": ["status", "primary_horizons"],
+            "properties": {
+              "status": { "enum": ["none", "conflict"] },
+              "primary_horizons": {
+                "type": "array",
+                "minItems": 1,
+                "maxItems": 4,
+                "uniqueItems": true,
+                "items": { "enum": ["short", "medium", "long", "not_applicable"] }
+              }
+            }
+          }
+        }
+      }
+    },
+    "policy": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": ["authority", "no_order", "permitted_next_step", "notice"],
+      "properties": {
+        "authority": { "const": "research_only" },
+        "no_order": { "const": true },
+        "permitted_next_step": { "const": "research_validation_only" },
+        "notice": { "type": "string", "maxLength": 240 }
+      }
+    },
+    "errors": {
+      "type": "array",
+      "maxItems": 20,
+      "items": { "type": "string", "pattern": "^[a-z][a-z0-9_.-]{0,63}$" }
+    }
+  },
+  "$defs": {
+    "observation": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": ["source_ids", "source_report_digest", "source_entry_digest", "hypothesis_id", "as_of", "generated_at", "expires_at", "research_context", "freshness"],
+      "properties": {
+        "source_ids": {
+          "type": "array",
+          "minItems": 1,
+          "maxItems": 100,
+          "uniqueItems": true,
+          "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}$" },
+        "as_of": { "type": "string", "format": "date" },
+        "generated_at": { "type": "string", "format": "date-time", "maxLength": 64 },
+        "expires_at": { "type": "string", "format": "date-time", "maxLength": 64 },
+        "research_context": {
+          "type": "object",
+          "additionalProperties": false,
+          "required": ["state", "primary_horizon", "suitable_horizons", "source_confidence", "source_style", "theme_ids"],
+          "properties": {
+            "state": { "enum": ["candidate", "source_verification_required", "deferred", "context_only"] },
+            "primary_horizon": { "enum": ["short", "medium", "long", "not_applicable"] },
+            "suitable_horizons": {
+              "type": "array",
+              "minItems": 1,
+              "maxItems": 4,
+              "uniqueItems": true,
+              "items": { "enum": ["short", "medium", "long", "not_applicable"] }
+            },
+            "source_confidence": { "enum": ["high", "medium", "low", "mixed", "no_event", "unknown"] },
+            "source_style": { "enum": ["event_driven", "long_horizon_growth", "value_quality", "macro_context", "mixed_research"] },
+            "theme_ids": {
+              "type": "array",
+              "maxItems": 24,
+              "uniqueItems": true,
+              "items": { "type": "string", "pattern": "^[A-Za-z0-9._:/=-]{1,128}$" }
+            }
+          }
+        },
+        "freshness": {
+          "type": "object",
+          "additionalProperties": false,
+          "required": ["status", "age_seconds"],
+          "properties": {
+            "status": { "enum": ["fresh", "stale", "unknown"] },
+            "age_seconds": { "type": ["integer", "null"], "minimum": 0 }
+          }
+        }
+      }
+    }
+  }
+}
diff --git a/schemas/qsl-m0-research-source-snapshot.v1.schema.json b/schemas/qsl-m0-research-source-snapshot.v1.schema.json
new file mode 100644
index 0000000..a8d7224
--- /dev/null
+++ b/schemas/qsl-m0-research-source-snapshot.v1.schema.json
@@ -0,0 +1,100 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://quantstrategylab.github.io/QuantRuntimeSettings/schemas/qsl-m0-research-source-snapshot.v1.schema.json",
+  "title": "QSL M0 Research Source Snapshot v1",
+  "description": "Closed, read-only transport for QuantAdvisorResearch M0 hypotheses. It carries no strategy, routing, platform, allocation, or execution authority.",
+  "type": "object",
+  "additionalProperties": false,
+  "required": ["schema_version", "source_id", "source_report_digest", "generated_at", "computed_at", "data_status", "hypotheses", "errors"],
+  "properties": {
+    "schema_version": { "const": "qsl_m0_research_source_snapshot.v1" },
+    "source_id": { "type": "string", "pattern": "^[A-Za-z0-9._:/=-]{1,128}$" },
+    "source_report_digest": { "type": ["string", "null"], "pattern": "^[0-9a-f]{64}$" },
+    "generated_at": { "type": ["string", "null"], "format": "date-time", "maxLength": 64 },
+    "computed_at": { "type": ["string", "null"], "format": "date-time", "maxLength": 64 },
+    "data_status": { "enum": ["ready", "unavailable", "stale"] },
+    "hypotheses": {
+      "type": "array",
+      "maxItems": 500,
+      "items": { "$ref": "#/$defs/m0Hypothesis" }
+    },
+    "errors": {
+      "type": "array",
+      "maxItems": 20,
+      "items": { "type": "string", "pattern": "^[a-z][a-z0-9_.-]{0,63}$" }
+    }
+  },
+  "$defs": {
+    "m0Hypothesis": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": ["schema_version", "artifact_type", "authority", "no_order", "hypothesis_id", "as_of", "generated_at", "expires_at", "subject", "research_context", "evidence", "provenance", "permitted_next_step"],
+      "properties": {
+        "schema_version": { "const": "qsl.m0_research_hypothesis.v1" },
+        "artifact_type": { "const": "research_hypothesis" },
+        "authority": { "const": "research_only" },
+        "no_order": { "const": true },
+        "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 },
+        "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}$" }
+          }
+        },
+        "research_context": {
+          "type": "object",
+          "additionalProperties": false,
+          "required": ["state", "primary_horizon", "suitable_horizons", "source_confidence", "source_style", "theme_ids"],
+          "properties": {
+            "state": { "enum": ["candidate", "source_verification_required", "deferred", "context_only"] },
+            "primary_horizon": { "enum": ["short", "medium", "long", "not_applicable"] },
+            "suitable_horizons": {
+              "type": "array",
+              "minItems": 1,
+              "maxItems": 4,
+              "uniqueItems": true,
+              "items": { "enum": ["short", "medium", "long", "not_applicable"] }
+            },
+            "source_confidence": { "enum": ["high", "medium", "low", "mixed", "no_event", "unknown"] },
+            "source_style": { "enum": ["event_driven", "long_horizon_growth", "value_quality", "macro_context", "mixed_research"] },
+            "theme_ids": {
+              "type": "array",
+              "maxItems": 24,
+              "uniqueItems": true,
+              "items": { "type": "string", "pattern": "^[A-Za-z0-9._:/=-]{1,128}$" }
+            }
+          }
+        },
+        "evidence": {
+          "type": "object",
+          "additionalProperties": false,
+          "required": ["source_entry_digest", "evidence_ref_count", "risk_note_count"],
+          "properties": {
+            "source_entry_digest": { "type": "string", "pattern": "^[0-9a-f]{64}$" },
+            "evidence_ref_count": { "type": "integer", "minimum": 0 },
+            "risk_note_count": { "type": "integer", "minimum": 0 }
+          }
+        },
+        "provenance": {
+          "type": "object",
+          "additionalProperties": false,
+          "required": ["source_project", "source_schema_version", "source_contract_version", "source_report_digest", "source_input_digest"],
+          "properties": {
+            "source_project": { "const": "QuantAdvisorResearch" },
+            "source_schema_version": { "enum": ["5", "6"] },
+            "source_contract_version": { "enum": ["model_recommendations.v5", "model_recommendations.v6"] },
+            "source_report_digest": { "type": "string", "pattern": "^[0-9a-f]{64}$" },
+            "source_input_digest": { "type": ["string", "null"], "pattern": "^[0-9a-f]{64}$" }
+          }
+        },
+        "permitted_next_step": { "const": "research_validation_only" }
+      }
+    }
+  }
+}