fix(execution): require durable claim before live submission

.github/workflows/ci.yml

@@ -24,44 +24,44 @@ jobs:
             exit 1
           fi
 
-      - name: Resolve QuantPlatformKit ref
-        id: quant-platform-kit-ref
+      - name: Read locked shared dependency refs
+        id: locked-shared-refs
         run: |
           set -euo pipefail
-          ref="main"
-          for candidate in "${GITHUB_HEAD_REF:-}" "${GITHUB_BASE_REF:-}"; do
-            if [ -n "$candidate" ] && git ls-remote --exit-code --heads https://github.com/QuantStrategyLab/QuantPlatformKit.git "$candidate" >/dev/null 2>&1; then
-              ref="$candidate"
-              break
-            fi
-          done
-          echo "ref=${ref}" >> "$GITHUB_OUTPUT"
-
-      - name: Resolve UsEquityStrategies ref
-        id: us-equity-strategies-ref
-        run: |
-          set -euo pipefail
-          ref="main"
-          for candidate in "${GITHUB_HEAD_REF:-}" "${GITHUB_BASE_REF:-}"; do
-            if [ -n "$candidate" ] && git ls-remote --exit-code --heads https://github.com/QuantStrategyLab/UsEquityStrategies.git "$candidate" >/dev/null 2>&1; then
-              ref="$candidate"
-              break
-            fi
-          done
-          echo "ref=${ref}" >> "$GITHUB_OUTPUT"
+          python - <<'PY' >> "$GITHUB_OUTPUT"
+          import re
+          import tomllib
+          from pathlib import Path
+
+          required = {"quant-platform-kit", "us-equity-strategies"}
+          lock = tomllib.loads(Path("uv.lock").read_text(encoding="utf-8"))
+          refs = {}
+          for package in lock.get("package", []):
+              name = package.get("name")
+              if name not in required:
+                  continue
+              source = package.get("source") or {}
+              match = re.search(r"#([0-9a-f]{40})$", str(source.get("git") or ""))
+              if match:
+                  refs[name] = match.group(1)
+          if set(refs) != required:
+              raise SystemExit("uv.lock lacks a full SHA for a shared dependency")
+          for name in sorted(refs):
+              print(f"{name.replace('-', '_')}={refs[name]}")
+          PY
 
       - name: Checkout QuantPlatformKit
         uses: actions/checkout@v6
         with:
           repository: QuantStrategyLab/QuantPlatformKit
-          ref: ${{ steps.quant-platform-kit-ref.outputs.ref }}
+          ref: ${{ steps.locked-shared-refs.outputs.quant_platform_kit }}
           path: external/QuantPlatformKit
 
       - name: Checkout UsEquityStrategies
         uses: actions/checkout@v6
         with:
           repository: QuantStrategyLab/UsEquityStrategies
-          ref: ${{ steps.us-equity-strategies-ref.outputs.ref }}
+          ref: ${{ steps.locked-shared-refs.outputs.us_equity_strategies }}
           path: external/UsEquityStrategies
 
       - name: Setup Python
@@ -74,10 +74,30 @@ jobs:
           set -euo pipefail
           python -m pip install --upgrade pip uv
           uv sync --frozen --extra test
+      - name: Verify shared repository refs
+        run: |
+          set -euo pipefail
+          test "$(git -C external/QuantPlatformKit rev-parse HEAD)" = "${{ steps.locked-shared-refs.outputs.quant_platform_kit }}"
+          test "$(git -C external/UsEquityStrategies rev-parse HEAD)" = "${{ steps.locked-shared-refs.outputs.us_equity_strategies }}"
+
       - name: Smoke import pinned shared packages
         run: |
           set -euo pipefail
           uv run --no-sync python - <<'PY'
+          import importlib.metadata
+          import json
+          import tomllib
+          from pathlib import Path
+
+          lock = tomllib.loads(Path("uv.lock").read_text(encoding="utf-8"))
+          for package in lock["package"]:
+              if package["name"] not in {"quant-platform-kit", "us-equity-strategies"}:
+                  continue
+              expected = package["source"]["git"].rsplit("#", 1)[1]
+              installed = json.loads(importlib.metadata.distribution(package["name"]).read_text("direct_url.json") or "{}")
+              assert installed.get("vcs_info", {}).get("commit_id") == expected, package["name"]
+          PY
+          uv run --no-sync python - <<'PY'
           from quant_platform_kit.common.port_adapters import CallableNotificationPort, CallablePortfolioPort
           from us_equity_strategies import resolve_canonical_profile
 
@@ -86,11 +106,6 @@ jobs:
           assert resolve_canonical_profile("russell_top50_leader_rotation") == "russell_top50_leader_rotation"
           PY
 
-      - name: Install editable shared repositories
-        run: |
-          set -euo pipefail
-          uv pip install --no-deps -e external/QuantPlatformKit -e external/UsEquityStrategies
-
       - name: Verify Python dependencies
         run: uv pip check
 
@@ -102,9 +117,10 @@ jobs:
       - name: Check QPK pin consistency
         run: |
           set -euo pipefail
+          printf '%s\n' "${{ steps.locked-shared-refs.outputs.quant_platform_kit }}" > "$RUNNER_TEMP/firstrade-qpk-pin"
           uv run --no-sync python external/QuantPlatformKit/scripts/check_qpk_pin_consistency.py \
             --root . \
-            --pin-file external/QuantPlatformKit/QPK_PIN
+            --pin-file "$RUNNER_TEMP/firstrade-qpk-pin"
 
       - name: Ensure uv.lock matches pyproject.toml
         run: uv lock --check

application/execution_service.py

@@ -648,7 +648,6 @@ def execute_value_target_plan(
     fetch_order_status=None,
     before_live_submission: Callable[[], bool] | None = None,
 ) -> ExecutionCycleResult:
-    del dry_run_only  # ExecutionPort owns preview vs live submission.
     plan = substitute_small_safe_haven_targets_with_cash(
         plan,
         threshold_usd=safe_haven_cash_substitute_threshold_usd,
@@ -721,7 +720,11 @@ def _submission_claim_unavailable(symbol: str) -> ExecutionCycleResult:
         )
 
     def _may_submit_live_order() -> bool:
-        return before_live_submission is None or bool(before_live_submission())
+        if dry_run_only:
+            return True
+        if before_live_submission is None:
+            raise ValueError("Live submission requires a durable submission claim callback.")
+        return bool(before_live_submission())
 
     tradable_deltas: list[tuple[str, float, float]] = []
     for symbol in sorted(set(targets) | set(market_values)):

application/rebalance_service.py

@@ -642,6 +642,10 @@ def log_message(message: str) -> None:
 
     def acquire_submission_claim() -> bool:
         nonlocal submission_claim_acquired
+        if not persist_strategy_runs:
+            raise ValueError(
+                "Live submission requires enabled strategy-run persistence and a durable state store."
+            )
         if submission_claim_acquired:
             return True
         submission_claim_acquired = claim_live_strategy_run(
@@ -668,7 +672,7 @@ def acquire_submission_claim() -> bool:
         fetch_order_status=lambda broker_order_id: client.get_order_status(account, broker_order_id),
         before_live_submission=(
             acquire_submission_claim
-            if persist_strategy_runs and not settings.dry_run_only
+            if not settings.dry_run_only
             else None
         ),
     )

tests/test_dependency_pin_guard.py

@@ -2,6 +2,7 @@
 
 import importlib.util
 import sys
+import textwrap
 from pathlib import Path
 from types import ModuleType
 
@@ -62,5 +63,57 @@ def test_dependency_pin_guard_is_blocking_in_ci() -> None:
     step = workflow[step_start : next_step if next_step != -1 else len(workflow)]
 
     assert "external/QuantPlatformKit/scripts/check_qpk_pin_consistency.py" in step
-    assert "--pin-file external/QuantPlatformKit/QPK_PIN" in step
+    assert 'steps.locked-shared-refs.outputs.quant_platform_kit' in step
+    assert '--pin-file "$RUNNER_TEMP/firstrade-qpk-pin"' in step
     assert "continue-on-error" not in step
+
+
+def test_ci_shared_checkouts_use_locked_refs() -> None:
+    workflow = CI_WORKFLOW.read_text(encoding="utf-8")
+    assert "name: Read locked shared dependency refs" in workflow
+    for name, directory in (
+        ("quant_platform_kit", "QuantPlatformKit"),
+        ("us_equity_strategies", "UsEquityStrategies"),
+    ):
+        assert f"ref: ${{{{ steps.locked-shared-refs.outputs.{name} }}}}" in workflow
+        assert f'git -C external/{directory} rev-parse HEAD' in workflow
+    assert "git ls-remote" not in workflow
+    assert "uv pip install --no-deps -e" not in workflow
+    assert "uv sync --frozen --extra test" in workflow
+    assert "uv pip check" in workflow
+    assert "uv lock --check" in workflow
+
+
+def test_ci_installed_shared_identity_matches_lock(monkeypatch) -> None:
+    import importlib.metadata
+    import json
+    import tomllib
+
+    import pytest
+
+    workflow = CI_WORKFLOW.read_text(encoding="utf-8")
+    smoke = workflow.split("name: Smoke import pinned shared packages", 1)[1]
+    code = textwrap.dedent(smoke.split("<<'PY'\n", 1)[1].split("\n          PY", 1)[0])
+    refs = {
+        package["name"]: package["source"]["git"].rsplit("#", 1)[1]
+        for package in tomllib.loads(Path("uv.lock").read_text())["package"]
+        if package["name"] in {"quant-platform-kit", "us-equity-strategies"}
+    }
+    identities = {name: {"vcs_info": {"commit_id": ref}} for name, ref in refs.items()}
+
+    class Distribution:
+        def __init__(self, name):
+            self.name = name
+
+        def read_text(self, filename):
+            assert filename == "direct_url.json"
+            return json.dumps(identities[self.name])
+
+    monkeypatch.setattr(importlib.metadata, "distribution", Distribution)
+    exec(code, {})
+    for name in refs:
+        for invalid in ({"vcs_info": {"commit_id": "0" * 40}}, {"dir_info": {"editable": True}}):
+            with monkeypatch.context() as context:
+                context.setitem(identities, name, invalid)
+                with pytest.raises(AssertionError):
+                    exec(code, {})

tests/test_execution_service.py

@@ -3,6 +3,8 @@
 from dataclasses import dataclass
 from datetime import datetime, timezone
 
+import pytest
+
 from application.execution_service import (
     _apply_notional_cash_buffer,
     execute_value_target_plan,
@@ -56,6 +58,44 @@ def submit_order(self, order_intent) -> ExecutionReport:
         )
 
 
+@pytest.mark.parametrize("side,notional", [("sell", False), ("buy", False), ("buy", True)])
+def test_live_order_requires_submission_claim_callback(side, notional):
+    execution_port = SubmittedExecutionPort()
+    with pytest.raises(ValueError, match="submission claim"):
+        execute_value_target_plan(
+            plan={
+                "allocation": {"targets": {"AAA": 0.0 if side == "sell" else 20.0}},
+                "portfolio": {
+                    "market_values": {"AAA": 20.0 if side == "sell" else 0.0},
+                    "sellable_quantities": {"AAA": 2.0},
+                    "liquid_cash": 100.0,
+                },
+                "execution": {"current_min_trade": 1.0},
+            },
+            market_data_port=FakeMarketDataPort({"AAA": 10.0}),
+            execution_port=execution_port,
+            dry_run_only=False,
+            notional_buy_execution=notional,
+        )
+    assert execution_port.orders == []
+
+
+def test_live_noop_does_not_require_submission_claim():
+    execution_port = SubmittedExecutionPort()
+    result = execute_value_target_plan(
+        plan={
+            "allocation": {"targets": {"AAA": 20.0}},
+            "portfolio": {"market_values": {"AAA": 20.0}},
+            "execution": {"current_min_trade": 1.0},
+        },
+        market_data_port=FakeMarketDataPort({"AAA": 10.0}),
+        execution_port=execution_port,
+        dry_run_only=False,
+    )
+    assert execution_port.orders == []
+    assert result.idempotency_blocked is False
+
+
 def test_execute_value_target_plan_sells_before_buys_and_caps_order_notional():
     execution_port = FakeExecutionPort()
     result = execute_value_target_plan(
@@ -99,6 +139,7 @@ def test_execute_value_target_plan_marks_live_submissions_pending_reconciliation
         market_data_port=FakeMarketDataPort({"AAA": 10.0}),
         execution_port=execution_port,
         dry_run_only=False,
+        before_live_submission=lambda: True,  # Simulate an acquired durable claim.
     )
 
     assert result.action_done is False
@@ -666,6 +707,7 @@ def test_notional_buy_keeps_cash_buffer_when_order_would_use_all_available_cash(
         market_data_port=FakeMarketDataPort({"IBIT": 35.0}),
         execution_port=execution_port,
         dry_run_only=False,
+        before_live_submission=lambda: True,  # Simulate an acquired durable claim.
         notional_buy_execution=True,
     )
 
@@ -711,6 +753,7 @@ def submit_order(self, order_intent) -> ExecutionReport:
         market_data_port=FakeMarketDataPort({"IBIT": 35.0}),
         execution_port=execution_port,
         dry_run_only=False,
+        before_live_submission=lambda: True,  # Simulate an acquired durable claim.
         notional_buy_execution=True,
     )