from __future__ import annotations import asyncio import pytest import trading.domain.execution_service as execution_module from trading.adapters.paper_adapter import PaperExchangeAdapter from trading.domain.approvals import create_approval from trading.domain.execution_contracts import ExecuteApprovedPlanCommand from trading.domain.execution_contracts import ExecutionErrorCode from trading.domain.execution_service import ExecutionService, ExecutionServiceError from trading.domain.identity_repo import ensure_paper_exchange_account from trading.domain.plan_revalidation import PlanRevalidationResult from trading.domain.plans_repo import create_plan from trading.domain.risk_limits import MarketRiskContext from trading.domain.schema import apply_migrations, connect @pytest.fixture(autouse=True) def _db(tmp_path, monkeypatch): monkeypatch.setenv("HERMES_HOME", str(tmp_path)) monkeypatch.setenv("TRADING_MODE", "paper") monkeypatch.setenv("HERMES_EXECUTION_ENABLED", "true") ok, messages = apply_migrations() assert ok, messages def _plan(): return { "symbol": "BTCUSDT", "decision": "LONG", "entry": 100.0, "stop_loss": 95.0, "take_profit": 110.0, "leverage": 3, "quantity": 1.0, "risk_profile": "moderate", "risk_percent": 0.01, "futuresVerified": True, "noTradeGuard": False, "trading_readiness": "ready", "executable": True, "expires_at": "2099-01-01T00:00:00Z", } def _accepted(plan_id: str) -> PlanRevalidationResult: return PlanRevalidationResult( revalidation_id=f"rv-{plan_id}", accepted=True, reasons=(), context_hash="context-hash", observed_price=100.0, observed_spread_bps=1.0, observed_slippage_percent=0.0, market_risk=MarketRiskContext( provider_fresh=True, reconciliation_clear=True, spread_bps=1.0, funding_rate=0.0, liquidity_ok=True, ), orderbook={"bids": [[99.9, 100]], "asks": [[100.0, 100]]}, ) def _setup(owner_id: str = "owner-lock"): account = ensure_paper_exchange_account(owner_id) plan = create_plan(_plan(), owner_id=owner_id, account_id=account.account_id) approval_id = create_approval( plan["plan_id"], plan["plan_hash"], owner_id, account_id=account.account_id, ) command = ExecuteApprovedPlanCommand( plan_id=plan["plan_id"], approval_id=approval_id, owner_id=owner_id, account_id=account.account_id, idempotency_key=f"lock:{plan['plan_id']}:{approval_id}", ) return account, plan, command class ObservingAdapter(PaperExchangeAdapter): def __init__(self, *, idem_key: str) -> None: super().__init__(default_price=100.0) self.idem_key = idem_key self.configure_observation = None self.submit_observation = None self.configure_calls = 0 self.create_order_calls = 0 async def configure_account(self, symbol, leverage, margin_mode, position_mode): self.configure_calls += 1 conn = connect() try: idem = conn.execute( "SELECT status FROM idempotency_keys WHERE idem_key=?", (self.idem_key,) ).fetchone() execution = conn.execute( "SELECT execution_id,state FROM execution_history WHERE idempotency_key=?", (self.idem_key,), ).fetchone() events = conn.execute( "SELECT to_state FROM execution_events WHERE execution_id=? ORDER BY id", (execution[0],), ).fetchall() self.configure_observation = { "idem_status": idem[0] if idem else None, "execution_state": execution[1] if execution else None, "events": [row[0] for row in events], } finally: conn.close() return await super().configure_account(symbol, leverage, margin_mode, position_mode) async def create_order(self, req): self.create_order_calls += 1 conn = connect() try: order = conn.execute( "SELECT order_id,state FROM orders WHERE client_order_id=?", (req.client_order_id,) ).fetchone() execution = conn.execute( "SELECT state FROM execution_history WHERE idempotency_key=?", (self.idem_key,) ).fetchone() self.submit_observation = { "order_exists": order is not None, "order_state": order[1] if order else None, "execution_state": execution[0] if execution else None, } finally: conn.close() return await super().create_order(req) def test_durable_idempotency_execution_transitions_and_order_precede_adapter_calls(monkeypatch): _account, plan, command = _setup() async def accepted(plan_id, **_kwargs): return _accepted(plan_id) monkeypatch.setattr(execution_module, "revalidate_plan", accepted) adapter = ObservingAdapter(idem_key=command.idempotency_key or "") result = asyncio.run(ExecutionService(adapter).execute_approved_plan(command)) assert result["status"] == "protected" assert adapter.configure_observation == { "idem_status": "in_progress", "execution_state": "CONFIGURING_ACCOUNT", "events": ["CREATED", "VALIDATING", "LOCKED", "CONFIGURING_ACCOUNT"], } assert adapter.submit_observation == { "order_exists": True, "order_state": "submitted", "execution_state": "SUBMITTING_ENTRY", } conn = connect() try: lock = conn.execute( "SELECT holder,expires_at,fencing_token FROM execution_locks WHERE lock_name=?", (f"exec:{command.account_id}:BTC/USDT:USDT:one_way",), ).fetchone() idem = conn.execute( "SELECT status,result_ref FROM idempotency_keys WHERE idem_key=?", (command.idempotency_key,), ).fetchone() finally: conn.close() assert lock is not None and lock[0] == "" and float(lock[1]) == 0.0 and int(lock[2]) >= 1 assert idem is not None and idem[0] == "completed" and idem[1] def test_concurrent_same_command_has_one_mutation_winner(monkeypatch): _account, _plan_row, command = _setup("owner-concurrent") async def accepted(plan_id, **_kwargs): await asyncio.sleep(0) return _accepted(plan_id) monkeypatch.setattr(execution_module, "revalidate_plan", accepted) adapter = ObservingAdapter(idem_key=command.idempotency_key or "") service = ExecutionService(adapter) async def run(): first, second = await asyncio.gather( service.execute_approved_plan(command), service.execute_approved_plan(command), ) return first, second first, second = asyncio.run(run()) assert sorted([first["status"], second["status"]]) == ["duplicate", "protected"] assert adapter.configure_calls == 1 assert adapter.create_order_calls == 1 conn = connect() try: order_count = conn.execute("SELECT COUNT(*) FROM orders").fetchone()[0] position_count = conn.execute( "SELECT COUNT(*) FROM positions WHERE account_id=? AND status IN ('open','partially_open')", (command.account_id,), ).fetchone()[0] finally: conn.close() assert order_count == 1 assert position_count == 1 def test_execution_record_failure_marks_idempotency_failed_and_prevents_adapter_call(monkeypatch): _account, _plan_row, command = _setup("owner-db-failure") async def accepted(plan_id, **_kwargs): return _accepted(plan_id) monkeypatch.setattr(execution_module, "revalidate_plan", accepted) def explode_begin(**_kwargs): raise RuntimeError("database unavailable before execution record") monkeypatch.setattr(execution_module, "begin_execution", explode_begin) adapter = ObservingAdapter(idem_key=command.idempotency_key or "") with pytest.raises(ExecutionServiceError, match="approved plan execution failed") as caught: asyncio.run(ExecutionService(adapter).execute_approved_plan(command)) assert caught.value.code is ExecutionErrorCode.EXECUTION_FAILED assert adapter.configure_calls == 0 assert adapter.create_order_calls == 0 conn = connect() try: row = conn.execute( "SELECT status,error_class FROM idempotency_keys WHERE idem_key=?", (command.idempotency_key,), ).fetchone() finally: conn.close() assert row is not None and row[0] == "failed" and row[1] == "RuntimeError"