feat(risk): add exact private observation reader port

docs/qsl_long_horizon_risk_composer_v1.zh-CN.md

@@ -22,6 +22,12 @@
 
 命令行也遵守这条分工：已有 owner-bound 输入可用 `--input`；私有观察件必须同时给出 `--observation` 与 `--risk-preference`。缺少偏好即失败，不会静默选择“均衡”或任何默认档位。
 
+## 受限私有读取口
+
+`python/scripts/long_horizon_risk_observation_ingress.py` 定义了控制面唯一需要的读取能力：按 `long-horizon-risk-observations/v1/<candidate-id>/<p3-evidence-sha256>.json` 精确读取一个不超过 2 MiB 的观察件，核验对象内容的哈希、candidate 与 P3 摘要，再交给 Composer。它只接收调用方注入的 `read_exact` 函数；没有云 SDK、凭据、网络、bucket、列举、猜测最新、重试替代对象、写入、覆盖或删除能力。
+
+这意味着未来任一已授权的私有存储实现都只能获得该精确对象的读权限。存储未配置、读错、超限、JSON/哈希异常或身份不匹配都会闭合为不可用，不会降级到 Actions artifact、公开仓库、控制台或 AI 上下文。当前仍没有真实存储 adapter、运行身份、scheduler 或政策写入。
+
 ## 必要证据和计算方法
 
 输入必须精确绑定 candidate revision 与 P1/P2/P3/plugin 摘要，并至少包含每类一个完整的 252-session 以上路径：

python/scripts/long_horizon_risk_observation_ingress.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+"""Read one exact private long-horizon P3 observation through an injected port.
+
+This module deliberately has no cloud SDK, credential, network, bucket,
+listing, write, delete, broker, account, scheduler, policy-write, or execution
+dependency.  A future runtime may inject a narrowly scoped reader with access
+to one protected storage namespace.  This core only derives an immutable name,
+reads that exact object once, validates the hash-bound observation, and can
+produce a non-sensitive Composer recommendation.
+"""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Callable
+from typing import Any
+
+from long_horizon_risk_composer import (
+    LongHorizonRiskComposerError,
+    build_risk_composer_input_from_observation,
+    compose_long_horizon_risk_recommendation,
+    parse_risk_composer_input_json,
+    validate_long_horizon_risk_observation,
+)
+
+
+PRIVATE_OBSERVATION_OBJECT_PREFIX = "long-horizon-risk-observations/v1"
+MAX_PRIVATE_OBSERVATION_BYTES = 2 * 1024 * 1024
+_IDENTITY_PATTERN = re.compile(r"^[a-z][a-z0-9]*(?:[._-][a-z0-9]+)*$")
+_SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")
+
+
+class LongHorizonRiskObservationIngressError(ValueError):
+    """Fail-closed ingress error that contains no object name or path data."""
+
+
+def _fail() -> None:
+    raise LongHorizonRiskObservationIngressError("private long-horizon risk observation unavailable")
+
+
+def private_observation_object_name(*, candidate_id: str, p3_evidence_sha256: str) -> str:
+    """Return the only readable object name for one candidate/P3 identity."""
+    if not _IDENTITY_PATTERN.fullmatch(candidate_id) or not _SHA256_PATTERN.fullmatch(p3_evidence_sha256):
+        _fail()
+    return f"{PRIVATE_OBSERVATION_OBJECT_PREFIX}/{candidate_id}/{p3_evidence_sha256}.json"
+
+
+def load_private_long_horizon_risk_observation(
+    *,
+    candidate_id: str,
+    p3_evidence_sha256: str,
+    read_exact: Callable[[str], bytes],
+) -> dict[str, Any]:
+    """Read and validate one exact observation; unavailable input never falls back.
+
+    ``read_exact`` must be a capability-scoped dependency.  This function calls
+    it once with the deterministic object name and has no mechanism to list,
+    search, retry with another name, write, overwrite, or delete objects.
+    """
+    if not callable(read_exact):
+        _fail()
+    object_name = private_observation_object_name(
+        candidate_id=candidate_id,
+        p3_evidence_sha256=p3_evidence_sha256,
+    )
+    try:
+        raw = read_exact(object_name)
+    except Exception as exc:  # pragma: no cover - injected I/O boundary
+        raise LongHorizonRiskObservationIngressError(
+            "private long-horizon risk observation unavailable"
+        ) from exc
+    if not isinstance(raw, bytes) or not raw or len(raw) > MAX_PRIVATE_OBSERVATION_BYTES:
+        _fail()
+    try:
+        observation = validate_long_horizon_risk_observation(parse_risk_composer_input_json(raw.decode("utf-8")))
+    except (UnicodeDecodeError, LongHorizonRiskComposerError) as exc:
+        raise LongHorizonRiskObservationIngressError(
+            "private long-horizon risk observation unavailable"
+        ) from exc
+    if (
+        observation["candidate"]["candidate_id"] != candidate_id
+        or observation["source_evidence"]["p3_evidence_sha256"] != p3_evidence_sha256
+    ):
+        _fail()
+    return observation
+
+
+def compose_from_private_long_horizon_risk_observation(
+    *,
+    candidate_id: str,
+    p3_evidence_sha256: str,
+    risk_preference: str,
+    read_exact: Callable[[str], bytes],
+) -> dict[str, Any]:
+    """Return only the Composer's safe recommendation from one private object."""
+    observation = load_private_long_horizon_risk_observation(
+        candidate_id=candidate_id,
+        p3_evidence_sha256=p3_evidence_sha256,
+        read_exact=read_exact,
+    )
+    try:
+        composer_input = build_risk_composer_input_from_observation(
+            observation,
+            risk_preference=risk_preference,
+        )
+        return compose_long_horizon_risk_recommendation(composer_input)
+    except LongHorizonRiskComposerError as exc:
+        raise LongHorizonRiskObservationIngressError(
+            "private long-horizon risk observation unavailable"
+        ) from exc
+
+
+__all__ = [
+    "LongHorizonRiskObservationIngressError",
+    "MAX_PRIVATE_OBSERVATION_BYTES",
+    "PRIVATE_OBSERVATION_OBJECT_PREFIX",
+    "compose_from_private_long_horizon_risk_observation",
+    "load_private_long_horizon_risk_observation",
+    "private_observation_object_name",
+]

python/tests/test_long_horizon_risk_observation_ingress.py

@@ -0,0 +1,141 @@
+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
+
+
+composer = _load_module("long_horizon_risk_composer")
+ingress = _load_module("long_horizon_risk_observation_ingress")
+
+
+class LongHorizonRiskObservationIngressTest(unittest.TestCase):
+    @staticmethod
+    def _returns(gain_bps: int, drawdown_bps: int) -> list[int]:
+        return [gain_bps] * 240 + [-drawdown_bps] * 12
+
+    def _observation(self) -> dict[str, object]:
+        paths = [
+            {
+                "scenario_id": f"soxl_soxx_{kind.lower()}_{index}",
+                "scenario_kind": kind,
+                "session_count": 253,
+                "strategy_returns_bps": self._returns(16 - index, 124 + index),
+                "benchmark_returns_bps": self._returns(10 - index, 100 + index),
+            }
+            for index, kind in enumerate(("WALK_FORWARD", "BOOTSTRAP", "STRESS"), start=1)
+        ]
+        result: dict[str, object] = {
+            "schema": "qsl.long_horizon_risk_observation.v1",
+            "candidate": {
+                "candidate_id": "soxl_soxx_longterm_compounding",
+                "candidate_kind": "individual",
+                "strategy_repository": "QuantStrategyLab/UsEquityStrategies",
+                "strategy_revision": "a" * 40,
+            },
+            "source_evidence": {
+                "p1_input_digest": "1" * 64,
+                "p2_config_digest": "2" * 64,
+                "p3_evidence_sha256": "3" * 64,
+                "plugin_bundle_sha256": "4" * 64,
+            },
+            "benchmark": {
+                "benchmark_id": "soxx",
+                "benchmark_kind": "unlevered_reference",
+                "sessions_per_year": 252,
+            },
+            "scenario_paths": paths,
+            "observation_sha256": "",
+        }
+        result["observation_sha256"] = composer.calculate_risk_observation_sha256(result)
+        return result
+
+    def test_reads_only_the_exact_candidate_and_p3_object_then_returns_a_redacted_recommendation(self):
+        observation = self._observation()
+        raw = json.dumps(observation, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8")
+        calls: list[str] = []
+
+        result = ingress.compose_from_private_long_horizon_risk_observation(
+            candidate_id="soxl_soxx_longterm_compounding",
+            p3_evidence_sha256="3" * 64,
+            risk_preference="BALANCED_COMPOUNDING",
+            read_exact=lambda object_name: calls.append(object_name) or raw,
+        )
+
+        self.assertEqual(
+            calls,
+            [
+                "long-horizon-risk-observations/v1/soxl_soxx_longterm_compounding/"
+                + ("3" * 64)
+                + ".json"
+            ],
+        )
+        self.assertEqual(result["status"], "ADVISORY_RECOMMENDATION_READY")
+        serialized = json.dumps(result, sort_keys=True).lower()
+        self.assertNotIn("strategy_returns", serialized)
+        self.assertNotIn("benchmark_returns", serialized)
+        self.assertNotIn("broker", serialized)
+        self.assertNotIn("account", serialized)
+
+    def test_reader_failure_tampering_and_identity_mismatch_fail_closed_without_fallback(self):
+        observation = self._observation()
+        raw = json.dumps(observation).encode("utf-8")
+        with self.assertRaisesRegex(ingress.LongHorizonRiskObservationIngressError, "unavailable"):
+            ingress.load_private_long_horizon_risk_observation(
+                candidate_id="soxl_soxx_longterm_compounding",
+                p3_evidence_sha256="3" * 64,
+                read_exact=lambda _object_name: (_ for _ in ()).throw(RuntimeError("storage hostname")),
+            )
+
+        tampered = copy.deepcopy(observation)
+        tampered["scenario_paths"][0]["strategy_returns_bps"][0] = 99
+        with self.assertRaisesRegex(ingress.LongHorizonRiskObservationIngressError, "unavailable"):
+            ingress.load_private_long_horizon_risk_observation(
+                candidate_id="soxl_soxx_longterm_compounding",
+                p3_evidence_sha256="3" * 64,
+                read_exact=lambda _object_name: json.dumps(tampered).encode("utf-8"),
+            )
+
+        with self.assertRaisesRegex(ingress.LongHorizonRiskObservationIngressError, "unavailable"):
+            ingress.load_private_long_horizon_risk_observation(
+                candidate_id="soxl_soxx_longterm_compounding",
+                p3_evidence_sha256="4" * 64,
+                read_exact=lambda _object_name: raw,
+            )
+
+    def test_invalid_identities_and_oversized_input_never_reach_the_reader(self):
+        calls: list[str] = []
+        with self.assertRaisesRegex(ingress.LongHorizonRiskObservationIngressError, "unavailable"):
+            ingress.load_private_long_horizon_risk_observation(
+                candidate_id="../latest",
+                p3_evidence_sha256="3" * 64,
+                read_exact=lambda object_name: calls.append(object_name) or b"{}",
+            )
+        self.assertEqual(calls, [])
+
+        with self.assertRaisesRegex(ingress.LongHorizonRiskObservationIngressError, "unavailable"):
+            ingress.load_private_long_horizon_risk_observation(
+                candidate_id="soxl_soxx_longterm_compounding",
+                p3_evidence_sha256="3" * 64,
+                read_exact=lambda _object_name: b"x" * (ingress.MAX_PRIVATE_OBSERVATION_BYTES + 1),
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()