feat: observe paper command gate [skip-cloud-run-deploy]

application/durable_execution_commands.py

@@ -12,6 +12,12 @@
     ExecutionCommandStore,
     build_execution_command_store_from_env as _build_execution_command_store_from_env,
 )
+from quant_platform_kit.common.runtime_command_gate import (
+    RuntimeCommandExposureEffect,
+    RuntimeCommandGateEnforcement,
+    RuntimeCommandGatePolicy,
+    evaluate_runtime_command_gate,
+)
 
 
 PAPER_EXECUTION_INTENT_SCHEMA_VERSION = "longbridge.paper-execution-intent.v1"
@@ -85,6 +91,8 @@ def enqueue_paper_execution_command(
     strategy_profile: str,
     execution: Mapping[str, Any],
     allocation: Mapping[str, Any],
+    runtime_release_receipt: Mapping[str, Any] | None = None,
+    expected_strategy_release: Any = None,
 ) -> dict[str, object] | None:
     """Create one command only; this phase never claims or routes it."""
     if not enabled:
@@ -101,13 +109,29 @@ def enqueue_paper_execution_command(
         allocation=allocation,
     )
     created = store.enqueue(command)
+    gate_decision = evaluate_runtime_command_gate(
+        action="submit",
+        # A target-allocation command cannot safely infer the net exposure of
+        # each future order. The future consumer must reconcile positions and
+        # re-evaluate per order before it ever switches to enforcement.
+        exposure_effect=RuntimeCommandExposureEffect.UNKNOWN,
+        command=command,
+        command_state="queued",
+        as_of_session=command.effective_date,
+        runtime_release_receipt=runtime_release_receipt,
+        expected_strategy_release=expected_strategy_release,
+        policy=RuntimeCommandGatePolicy(
+            enforcement=RuntimeCommandGateEnforcement.OBSERVE,
+        ),
+    )
     return {
         "schema_version": "longbridge.paper-execution-command-observation.v1",
         "command_id": command.command_id,
         "decision_digest": command.decision_digest,
         "effective_date": command.effective_date,
         "status": "QUEUED" if created else "ALREADY_QUEUED",
         "consumer_authorized": False,
+        "runtime_command_gate": gate_decision.to_receipt(),
     }
 
 

application/rebalance_service.py

@@ -327,6 +327,8 @@ def fetch_replanned_state():
         strategy_profile=str(getattr(config, "strategy_profile", "") or "unknown"),
         execution=execution,
         allocation=allocation,
+        runtime_release_receipt=getattr(config, "runtime_release_receipt", None),
+        expected_strategy_release=getattr(config, "expected_strategy_release", None),
     )
     if paper_command_observation is not None:
         execution["durable_execution_command"] = paper_command_observation

application/runtime_composer.py

@@ -22,6 +22,7 @@
 from quant_platform_kit.common.runtime_assembly import build_runtime_assembly
 from quant_platform_kit.common.runtime_target import build_runtime_context_fields
 from quant_platform_kit.common.runtime_target import RuntimeTarget
+from quant_platform_kit.common.strategy_release import build_runtime_loaded_receipt
 from notifications.telegram import build_prefixer
 from quant_platform_kit.notifications.cycle_channel import build_cycle_sender
 from runtime_execution_policy import FRACTIONAL_BUY_QUANTITY_STEP, dca_compat_mode_enabled, fractional_buy_execution_enabled
@@ -276,6 +277,18 @@ def build_rebalance_config(
                 env_reader=self.env_reader,
                 gcp_project_id=self.project_id,
             ),
+            runtime_release_receipt=build_runtime_loaded_receipt(
+                strategy_release=(
+                    self.runtime_target.strategy_release
+                    if self.runtime_target is not None
+                    else None
+                ),
+            ),
+            expected_strategy_release=(
+                self.runtime_target.strategy_release
+                if self.runtime_target is not None
+                else None
+            ),
         )
 
     def load_strategy_plugin_signals(self, raw_mounts):

application/runtime_dependencies.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from dataclasses import dataclass
 from typing import Any
 
@@ -39,6 +39,8 @@ class LongBridgeRebalanceConfig:
     execution_state_account_scope: str = ""
     durable_execution_command_paper_enabled: bool = False
     execution_command_store: Any = None
+    runtime_release_receipt: Mapping[str, Any] | None = None
+    expected_strategy_release: Any = None
 
 
 @dataclass(frozen=True)

main.py

@@ -315,6 +315,9 @@ def _summarize_cycle_result_for_report(cycle_result, *, dry_run: bool) -> dict:
     ):
         if execution.get(field) not in (None, ""):
             summary[field] = execution[field]
+    durable_command = execution.get("durable_execution_command")
+    if isinstance(durable_command, dict):
+        summary["durable_execution_command"] = dict(durable_command)
     return summary
 
 

pyproject.toml

@@ -18,7 +18,7 @@ dependencies = [
     "google-cloud-storage",
     "google-auth",
     "longport==3.0.23",
-    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@68276cb9507773e2a29ac9579bed3adb7e077daf",
+    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@329fd889ec7a1204d5537a7762a22d215ed35cce",
     "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@4c174934d8a3eaeb3152ea6b11559e015f78ee9a",
     "hk-equity-strategies @ git+https://github.com/QuantStrategyLab/HkEquityStrategies.git@0f3abeb91c4dcd283e4586820da36f3d13343896",
 ]
@@ -61,5 +61,5 @@ include = [
 
 [tool.uv]
 override-dependencies = [
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@68276cb9507773e2a29ac9579bed3adb7e077daf",
+  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@329fd889ec7a1204d5537a7762a22d215ed35cce",
 ]

qsl.toml

@@ -5,7 +5,7 @@ upgrade_ring = "ring_d"
 allow_legacy = false
 
 [qsl.requires]
-quant_platform_kit = "68276cb9507773e2a29ac9579bed3adb7e077daf"
+quant_platform_kit = "329fd889ec7a1204d5537a7762a22d215ed35cce"
 us_equity_strategies = "4c174934d8a3eaeb3152ea6b11559e015f78ee9a"
 hk_equity_strategies = "0f3abeb91c4dcd283e4586820da36f3d13343896"
 

tests/test_durable_execution_commands.py

@@ -13,6 +13,7 @@
     enqueue_paper_execution_command,
     resolve_paper_execution_command_producer_enabled,
 )
+from quant_platform_kit.common.strategy_release import build_runtime_loaded_receipt
 
 
 def _execution() -> dict[str, object]:
@@ -33,6 +34,19 @@ def _allocation() -> dict[str, object]:
     }
 
 
+def _release_identity() -> dict[str, str]:
+    return {
+        "release_id": "soxl-p2-v3.20260824",
+        "manifest_sha256": "a" * 64,
+        "strategy_revision": "soxl-p2-v3",
+        "config_sha256": "b" * 64,
+        "risk_policy_sha256": "c" * 64,
+        "evidence_sha256": "d" * 64,
+        "plugin_bundle_sha256": "e" * 64,
+        "effective_session": "2026-08-25",
+    }
+
+
 def test_paper_command_is_content_addressed_and_excludes_broker_authority() -> None:
     first = build_paper_execution_command(
         platform="longbridge",
@@ -92,6 +106,39 @@ def enqueue(self, command):
     assert len(observed) == 2
 
 
+def test_paper_producer_persists_observation_gate_receipt_without_authorizing_a_consumer() -> None:
+    class Store:
+        cloud_prefix_uri = "gs://paper/commands"
+        local_dir = None
+
+        def enqueue(self, _command):
+            return True
+
+    release = _release_identity()
+    result = enqueue_paper_execution_command(
+        enabled=True,
+        dry_run_only=True,
+        store=Store(),
+        platform="longbridge",
+        account_scope="PAPER",
+        strategy_profile="soxl_soxx_trend_income",
+        execution=_execution(),
+        allocation=_allocation(),
+        runtime_release_receipt=build_runtime_loaded_receipt(strategy_release=release),
+        expected_strategy_release=release,
+    )
+
+    assert result is not None
+    gate = result["runtime_command_gate"]
+    assert isinstance(gate, dict)
+    assert gate["enforcement"] == "observe"
+    assert gate["mode"] == "active"
+    assert gate["policy_allows"] is False
+    assert gate["broker_write_allowed"] is True
+    assert "exposure_effect_unknown" in gate["reasons"]
+    assert result["consumer_authorized"] is False
+
+
 def test_paper_producer_rejects_live_enablement() -> None:
     assert resolve_paper_execution_command_producer_enabled(
         env_reader=lambda _name, _default="": "true",

tests/test_request_handling.py

@@ -857,6 +857,11 @@ def test_cycle_result_summary_counts_dry_run_order_previews(self):
             execution={
                 "signal_date": "2026-07-31",
                 "execution_timing_contract": "monthly_snapshot_window",
+                "durable_execution_command": {
+                    "command_id": "cmd-paper-1",
+                    "consumer_authorized": False,
+                    "runtime_command_gate": {"enforcement": "observe", "would_block": True},
+                },
             },
             dry_run_orders=(
                 {"symbol": "02800.HK", "side": "buy", "quantity": 100, "status": "dry_run"},
@@ -880,6 +885,8 @@ def test_cycle_result_summary_counts_dry_run_order_previews(self):
         self.assertEqual(summary["execution_timing_contract"], "monthly_snapshot_window")
         self.assertEqual(summary["orders_previewed"][0]["symbol"], "02800.HK")
         self.assertEqual(summary["quote_snapshot"]["quotes"][0]["symbol"], "02800.HK")
+        self.assertEqual(summary["durable_execution_command"]["command_id"], "cmd-paper-1")
+        self.assertTrue(summary["durable_execution_command"]["runtime_command_gate"]["would_block"])
 
     def test_cycle_result_summary_keeps_broker_submission_pending_until_reconciled(self):
         module = load_module()