hermes_overlay/tests/test_phase4_execution_state_machine.py

from __future__ import annotations

import ast
from pathlib import Path

import pytest

from trading.domain.execution_history import (
    begin_execution,
    execution_events,
    finish_execution,
    transition_execution,
)
from trading.domain.execution_state import (
    CLOSE_UNKNOWN,
    CLOSED,
    CLOSING,
    CONFIGURING_ACCOUNT,
    CREATED,
    ENTRY_FILLED,
    ENTRY_OPEN,
    ENTRY_PARTIALLY_FILLED,
    ENTRY_UNKNOWN,
    FAILED_SAFE,
    FAILED_UNPROTECTED,
    InvalidExecutionTransition,
    LOCKED,
    MANUAL_INTERVENTION_REQUIRED,
    PARTIALLY_CLOSED,
    PLACING_PROTECTION,
    PROTECTED,
    PROTECTION_PARTIAL,
    SUBMITTING_ENTRY,
    TRANSITIONS,
    VALIDATING,
    transition,
)
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 _begin() -> str:
    return begin_execution(
        owner_id="state-owner",
        symbol="BTC/USDT:USDT",
        action="entry",
        mode="paper",
        correlation_id="corr-state",
        payload={"safe": True},
        actor="state-owner",
    )


def _history(execution_id: str):
    conn = connect()
    try:
        return conn.execute(
            "SELECT state,version,completed_at FROM execution_history WHERE execution_id=?",
            (execution_id,),
        ).fetchone()
    finally:
        conn.close()


def test_complete_state_graph_and_service_transition_targets_are_declared():
    required = {
        CREATED,
        VALIDATING,
        LOCKED,
        CONFIGURING_ACCOUNT,
        SUBMITTING_ENTRY,
        ENTRY_UNKNOWN,
        ENTRY_OPEN,
        ENTRY_PARTIALLY_FILLED,
        ENTRY_FILLED,
        PLACING_PROTECTION,
        PROTECTION_PARTIAL,
        PROTECTED,
        CLOSING,
        CLOSE_UNKNOWN,
        PARTIALLY_CLOSED,
        CLOSED,
        FAILED_SAFE,
        FAILED_UNPROTECTED,
        MANUAL_INTERVENTION_REQUIRED,
    }
    assert set(TRANSITIONS) == required
    assert set().union(*TRANSITIONS.values()) <= required

    service_path = Path(__file__).parents[1] / "trading" / "domain" / "execution_service.py"
    tree = ast.parse(service_path.read_text())
    transition_names = {
        call.args[1].id
        for call in ast.walk(tree)
        if isinstance(call, ast.Call)
        and isinstance(call.func, ast.Name)
        and call.func.id == "transition_execution"
        and len(call.args) >= 2
        and isinstance(call.args[1], ast.Name)
    }
    assert transition_names <= required
    assert {
        VALIDATING,
        LOCKED,
        CONFIGURING_ACCOUNT,
        SUBMITTING_ENTRY,
        ENTRY_UNKNOWN,
        ENTRY_OPEN,
        ENTRY_PARTIALLY_FILLED,
        ENTRY_FILLED,
        PLACING_PROTECTION,
        PROTECTION_PARTIAL,
        PROTECTED,
        CLOSING,
        CLOSE_UNKNOWN,
        PARTIALLY_CLOSED,
        CLOSED,
        FAILED_SAFE,
        FAILED_UNPROTECTED,
    } <= transition_names


def test_transition_is_atomic_and_records_actor_reason_and_payload():
    execution_id = _begin()
    assert transition_execution(
        execution_id,
        VALIDATING,
        actor="operator-1",
        reason="validated_snapshot",
        payload={"risk": "accepted"},
    ) == VALIDATING

    row = _history(execution_id)
    assert row[0] == VALIDATING
    assert int(row[1]) == 2
    events = execution_events(execution_id)
    assert [(event["from_state"], event["to_state"]) for event in events] == [
        (None, CREATED),
        (CREATED, VALIDATING),
    ]
    assert events[-1]["actor"] == "operator-1"
    assert events[-1]["reason"] == "validated_snapshot"
    assert '"risk": "accepted"' in events[-1]["payload"]


def test_illegal_transition_rejection_keeps_prior_durable_state():
    execution_id = _begin()
    with pytest.raises(InvalidExecutionTransition, match="not allowed"):
        transition_execution(
            execution_id,
            PROTECTED,
            actor="operator-1",
            reason="illegal_skip",
        )

    row = _history(execution_id)
    assert row[0] == CREATED
    assert int(row[1]) == 1
    assert [event["to_state"] for event in execution_events(execution_id)] == [CREATED]


def test_crash_at_event_boundary_rolls_back_state_and_event_together():
    execution_id = _begin()
    conn = connect()
    try:
        conn.execute(
            "CREATE TRIGGER abort_crash_event BEFORE INSERT ON execution_events "
            "WHEN NEW.reason='crash_boundary' BEGIN "
            "SELECT RAISE(ABORT, 'simulated transition crash'); END"
        )
        conn.commit()
    finally:
        conn.close()

    with pytest.raises(Exception, match="simulated transition crash"):
        transition_execution(
            execution_id,
            VALIDATING,
            actor="operator-1",
            reason="crash_boundary",
        )

    row = _history(execution_id)
    assert row[0] == CREATED
    assert int(row[1]) == 1
    assert [event["to_state"] for event in execution_events(execution_id)] == [CREATED]


def test_finish_execution_cannot_bypass_a_missing_transition_event():
    execution_id = _begin()
    with pytest.raises(InvalidExecutionTransition, match="cannot bypass"):
        finish_execution(execution_id, state=PROTECTED, result_ref="order-1")

    row = _history(execution_id)
    assert row[0] == CREATED
    assert row[2] is None

    transition_execution(
        execution_id,
        FAILED_SAFE,
        actor="operator-1",
        reason="safe_terminal_failure",
    )
    finish_execution(execution_id, state=FAILED_SAFE, error=RuntimeError("safe failure"))
    row = _history(execution_id)
    assert row[0] == FAILED_SAFE
    assert row[2] is not None


def test_representative_recovery_paths_follow_declared_edges():
    entry_path = [
        CREATED,
        VALIDATING,
        LOCKED,
        CONFIGURING_ACCOUNT,
        SUBMITTING_ENTRY,
        ENTRY_FILLED,
        PLACING_PROTECTION,
        PROTECTED,
        CLOSING,
        CLOSED,
    ]
    for current, following in zip(entry_path, entry_path[1:]):
        assert transition(current, following) == following

    recovery_path = [
        ENTRY_UNKNOWN,
        MANUAL_INTERVENTION_REQUIRED,
        CLOSING,
        CLOSE_UNKNOWN,
        PARTIALLY_CLOSED,
        CLOSING,
        CLOSED,
    ]
    for current, following in zip(recovery_path, recovery_path[1:]):
        assert transition(current, following) == following

    assert transition(ENTRY_PARTIALLY_FILLED, FAILED_UNPROTECTED) == FAILED_UNPROTECTED
    assert transition(FAILED_UNPROTECTED, PROTECTED) == PROTECTED
    assert transition(PLACING_PROTECTION, PROTECTION_PARTIAL) == PROTECTION_PARTIAL