Harden evidence-grade risk assessment gate

src/quant_platform_kit/risk/engine.py

@@ -120,7 +120,13 @@ def assess(
         *,
         market_data: Mapping[str, Any] | None = None,
     ) -> RiskAction:
-        """Assess a decision, rejecting missing or invalid account state."""
+        """Run the legacy runtime sanity check for account state.
+
+        This method is not evidence-grade and its ``approve`` result must not
+        be used as promotion authority. Promotion/evidence callers must use
+        :func:`quant_platform_kit.risk.gate.assess_with_evidence`, which adds
+        the mandate, candidate, snapshot, and exposure contract checks.
+        """
         if portfolio_snapshot is None:
             return RiskAction(
                 action="reject",

src/quant_platform_kit/risk/gate.py

@@ -568,7 +568,7 @@ def _mandate_fields(
             "loss_budget": 0.0,
             "product_leverage_factors": {},
             "allowed_nonzero_assets": None,
-        }, set()
+        }, {"missing_mandate"}
     if not isinstance(mandate_provenance, Mapping):
         return {}, {"invalid_mandate"}
     if mandate_provenance.get("mandate_id") == _RETIRED_GLOBAL_ETF_RESEARCH_MANDATE:
@@ -1316,7 +1316,13 @@ def assess_with_evidence(
     capital_base: CapitalBaseSnapshot | Mapping[str, Any] | None = None,
     capital_base_binding: CapitalBaseBinding | Mapping[str, Any] | None = None,
 ) -> RiskGateResult:
-    """Assess exactly once and fail closed with a redacted canonical receipt."""
+    """Run the sole promotion/evidence-grade risk assessment API.
+
+    ``mandate_provenance`` and a matching ``candidate_identity`` are required
+    for an evidence-grade approval. Missing or invalid authority is rejected;
+    the legacy :meth:`RiskEngine.assess` approval is never sufficient on its
+    own. The returned receipt is redacted and canonical.
+    """
     try:
         risk_action = build_risk_engine().assess(
             decision,

tests/test_risk_gate.py

@@ -1077,6 +1077,95 @@ def test_unmandated_fallback_rejects_stale_snapshot(self) -> None:
         self.assertIn("stale_portfolio_snapshot", result.assessment.reason_codes)
         self.assertEqual(result.decision.positions, ())
 
+    def test_evidence_assessment_requires_mandate(self) -> None:
+        with patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW):
+            result = assess_with_evidence(
+                _decision(positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.10),)),
+                self._snapshot(),
+                scope="MEMBER",
+                mandate_provenance=None,
+                market_data={},
+                candidate_identity=None,
+            )
+
+        self.assertEqual(result.assessment.outcome, "REJECT")
+        self.assertIn("missing_mandate", result.assessment.reason_codes)
+        self.assertFalse(result.assessment.execution_authorized)
+        self.assertEqual(result.decision.positions, ())
+
+    def test_evidence_assessment_rejects_unbounded_target_and_unknown_symbol(self) -> None:
+        cases = (
+            (
+                "10x_target",
+                _decision(positions=(PositionTarget(symbol="BTCUSDT", target_weight=10.0),)),
+                "effective_exposure_cap",
+            ),
+            (
+                "unknown_symbol",
+                _decision(positions=(PositionTarget(symbol="UNKNOWN", target_weight=0.10),)),
+                "asset_not_authorized",
+            ),
+        )
+        for name, decision, reason_code in cases:
+            with (
+                self.subTest(name=name),
+                patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW),
+            ):
+                result = assess_with_evidence(
+                    decision,
+                    self._snapshot(),
+                    scope="MEMBER",
+                    mandate_provenance=self._mandate(),
+                    market_data={},
+                    candidate_identity=self._candidate(),
+                )
+
+            self.assertEqual(result.assessment.outcome, "REJECT")
+            self.assertIn(reason_code, result.assessment.reason_codes)
+            self.assertFalse(result.assessment.execution_authorized)
+            self.assertEqual(result.decision.positions, ())
+
+    def test_evidence_assessment_rejects_expired_mandate(self) -> None:
+        with patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW):
+            result = assess_with_evidence(
+                _decision(positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.10),)),
+                self._snapshot(),
+                scope="MEMBER",
+                mandate_provenance={
+                    **self._mandate(),
+                    "expires_at": "2026-08-04T04:27:59Z",
+                },
+                market_data={},
+                candidate_identity=self._candidate(),
+            )
+
+        self.assertEqual(result.assessment.outcome, "REJECT")
+        self.assertIn("expired_mandate", result.assessment.reason_codes)
+        self.assertFalse(result.assessment.execution_authorized)
+
+    def test_evidence_assessment_rejects_stale_capital_snapshot(self) -> None:
+        with patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW):
+            result = assess_with_evidence(
+                _decision(positions=(PositionTarget(symbol="BTCUSDT", target_value=10_000.0),)),
+                self._snapshot(),
+                scope="MEMBER",
+                mandate_provenance=self._mandate(),
+                market_data={},
+                candidate_identity=self._candidate(),
+                capital_base=_capital_base(
+                    as_of=self._NOW - timedelta(seconds=301),
+                    strategy_scope="crypto_live_pool_rotation",
+                ),
+                capital_base_binding=_capital_base_binding(
+                    strategy_scope="crypto_live_pool_rotation",
+                ),
+            )
+
+        self.assertEqual(result.assessment.outcome, "REJECT")
+        self.assertIn("stale_capital_base", result.assessment.reason_codes)
+        self.assertFalse(result.assessment.execution_authorized)
+        self.assertEqual(result.decision.positions, ())
+
     def test_risk_plugin_exception_rejects_without_exposing_exception(self) -> None:
         class CrashingPlugin:
             plugin_name = "crashing_plugin"
@@ -1377,8 +1466,16 @@ def test_optional_string_nulls_and_nfc_text_remain_canonical(self) -> None:
                 ),
             ),
         )
-        first, first_engine = self._assess(decision)
-        second, second_engine = self._assess(decision)
+        first, first_engine = self._assess(
+            decision,
+            mandate=self._mandate(),
+            candidate=self._candidate(),
+        )
+        second, second_engine = self._assess(
+            decision,
+            mandate=self._mandate(),
+            candidate=self._candidate(),
+        )
 
         self.assertEqual(first.assessment.outcome, "APPROVE")
         self.assertEqual(first.assessment.assessment_sha256, second.assessment.assessment_sha256)
@@ -1568,14 +1665,20 @@ def test_timezone_and_whole_second_timestamp_compatibility(self) -> None:
         first, _engine = self._assess(
             StrategyDecision(),
             snapshot={**self._SNAPSHOT, "as_of": utc.replace(microsecond=1)},
+            mandate=self._mandate(),
+            candidate=self._candidate(),
         )
         second, _engine = self._assess(
             StrategyDecision(),
             snapshot={**self._SNAPSHOT, "as_of": offset.replace(microsecond=999_999)},
+            mandate=self._mandate(),
+            candidate=self._candidate(),
         )
         rejected, _engine = self._assess(
             StrategyDecision(),
             snapshot={**self._SNAPSHOT, "as_of": "2026-08-04T12:27:55+08:00"},
+            mandate=self._mandate(),
+            candidate=self._candidate(),
         )
 
         self.assertEqual(first.assessment.outcome, "APPROVE")