trex-core/tests/application/test_execution_ledger.py

"""Execution loop ledger-hook tests (REQ-018, REQ-019, dcy-004).

Exercises the full buy/sell lifecycle with the in-memory paper broker and
in-memory outbox + ledger writer, ensuring the slot transition is on the
critical path while the ledger write is eventually-consistent.
"""
from __future__ import annotations

from trex_core.adapters.broker.paper import PaperBroker
from trex_core.adapters.ledger.in_memory import InMemoryLedger
from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore
from trex_core.application.execution import ExecutionService
from trex_core.application.intents import BuyIntent, IntentService, SellIntent
from trex_core.application.ledger_outbox import (
    InMemoryLedgerOutbox,
    LedgerWriter,
    OutboxOutcome,
    WorkItemStatus,
)
from trex_core.domain.slot import SlotService, SlotState
from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository



def _wire(broker, *, ledger=None, clock=None):
    repo = InMemorySlotRepository()
    svc = SlotService(repo)
    intents = IntentService(svc, repo, InMemoryIntentStore(),
                            resolve_symbol=lambda pid: "005930")
    outbox = InMemoryLedgerOutbox(clock=clock) if clock else InMemoryLedgerOutbox()
    execution = ExecutionService(svc, repo, broker, ledger_outbox=outbox)
    writer = (
        LedgerWriter(outbox, ledger or InMemoryLedger(), clock=clock)
        if clock else LedgerWriter(outbox, ledger or InMemoryLedger())
    )
    repo.reconcile_universe("default", {"PF-1": "SYM"})
    return repo, svc, intents, execution, outbox, writer


def _buy(intents, *, iid="b1", portfolio_id="PF-1"):
    return intents.submit_buy(BuyIntent(
        intent_id=iid, qty=10, entry_price=1000,
        portfolio_id=portfolio_id,
    ))


def test_buy_fill_enqueues_one_ledger_item_and_writer_records_it():
    ledger = InMemoryLedger()
    repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0),
                                                          ledger=ledger)
    sid = _buy(intents).slot_id
    r = execution.run_once()
    assert r.filled == 1 and r.ledger_enqueued == 1
    assert repo.fetch_slot(sid).status is SlotState.HOLDING
    stats = writer.run_once()
    assert stats.completed == 1
    assert ledger.count() == 1
    t = ledger.trades()[0]
    assert (t.asset_id, t.side, t.qty, t.price) == ("PF-1", "buy", 10, 1000.0)
    assert t.execution_id.startswith("paper:PAPER-")
    assert t.fee > 0.0 and t.tax == 0.0


def test_sell_fill_enqueues_sell_ledger_item():
    ledger = InMemoryLedger()
    repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0),
                                                          ledger=ledger)
    sid = _buy(intents).slot_id
    execution.run_once()        # -> holding + buy ledger item
    writer.run_once()           # drain buy
    intents.submit_sell(SellIntent(intent_id="s1", slot_id=sid))
    r = execution.run_once()    # dispatch sell + fill -> closed + sell ledger item
    assert r.ledger_enqueued == 1
    assert repo.fetch_slot(sid).status is SlotState.INACTIVE
    writer.run_once()
    sides = [t.side for t in ledger.trades()]
    assert sides == ["buy", "sell"]


def test_portfolio_id_absent_skips_ledger_but_transitions_slot():
    ledger = InMemoryLedger()
    repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0),
                                                          ledger=ledger)
    # dcy-011: buy admission now requires portfolio_id, so create the
    # pending_buy slot directly (bypass admission) to exercise the
    # execution-loop ledger skip + visibility (REQ-018 FR-4).
    from trex_core.domain.slot.models import Slot
    from uuid import uuid4
    anon = Slot(slot_id=uuid4(), status=SlotState.INACTIVE)
    repo.add_slot(anon)
    svc.transit_slot(anon.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY,
                     reason="buy", actor="test", symbol="005930", portfolio_id=None)
    sid = anon.slot_id
    r = execution.run_once()
    assert r.filled == 1 and r.ledger_enqueued == 0 and r.ledger_skipped == 1
    assert repo.fetch_slot(sid).status is SlotState.HOLDING
    assert outbox.all_items() == []
    assert ledger.count() == 0


def test_ledger_skip_emits_warning(caplog):
    """dcy-011 / REQ-018 FR-4 (S-3): ledger skip is logged, not silent."""
    import logging
    ledger = InMemoryLedger()
    repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0),
                                                          ledger=ledger)
    from trex_core.domain.slot.models import Slot
    from uuid import uuid4
    anon = Slot(slot_id=uuid4(), status=SlotState.INACTIVE)
    repo.add_slot(anon)
    svc.transit_slot(anon.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY,
                     reason="buy", actor="test", symbol="005930", portfolio_id=None)
    with caplog.at_level(logging.WARNING, logger="trex_core.application.execution"):
        execution.run_once()
    msgs = [r.getMessage() for r in caplog.records]
    assert any("ledger skip" in m and "missing_portfolio_id" in m for m in msgs)


def test_duplicate_execution_id_results_in_zero_duplicate_in_ledger():
    """REQ-019 AC-1: same execution_id processed twice = 1 ledger entry."""
    ledger = InMemoryLedger()
    repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0),
                                                          ledger=ledger)
    _buy(intents)
    execution.run_once()
    writer.run_once()
    assert ledger.count() == 1
    # Same correlation_id arriving again (re-enqueue) is dedup'd at the outbox.
    first = outbox.all_items()[0]
    outbox.enqueue(first.trade)
    assert len(outbox.all_items()) == 1
    # And even if the writer somehow re-processes (simulated by a fresh outbox
    # row for the same execution_id, e.g. after a manual replay), the ledger
    # adapter dedup'd it server-side and the outcome is duplicate_completed.
    replay_outbox = InMemoryLedgerOutbox()
    replay_writer = LedgerWriter(replay_outbox, ledger)
    replay_outbox.enqueue(first.trade)
    stats = replay_writer.run_once()
    assert stats.duplicates == 1 and stats.completed == 0
    assert ledger.count() == 1    # no duplicate in the ledger


def test_retryable_failure_then_eventually_completes():
    """REQ-019 AC-2: retry until completed; slot transition unaffected.

    dcy-015: retryable rows back off, so advance a shared clock past each
    next_retry_at between writer passes.
    """
    from datetime import datetime, timedelta, timezone
    now = {"t": datetime(2026, 6, 4, tzinfo=timezone.utc)}
    ledger = InMemoryLedger(fail_n_then_succeed=2)
    repo, svc, intents, execution, outbox, writer = _wire(
        PaperBroker(fill_after_polls=0), ledger=ledger, clock=lambda: now["t"],
    )
    sid = _buy(intents).slot_id
    execution.run_once()
    assert repo.fetch_slot(sid).status is SlotState.HOLDING  # slot already transitioned
    stats1 = writer.run_once()
    assert stats1.retryable == 1 and stats1.completed == 0 and ledger.count() == 0
    now["t"] += timedelta(seconds=600)  # past backoff
    stats2 = writer.run_once()
    assert stats2.retryable == 1 and ledger.count() == 0
    now["t"] += timedelta(seconds=600)  # past backoff again
    stats3 = writer.run_once()
    assert stats3.completed == 1 and ledger.count() == 1
    # outbox row terminal-completed
    [item] = outbox.all_items()
    assert item.status is WorkItemStatus.COMPLETED
    assert item.last_outcome is OutboxOutcome.COMPLETED


def test_conflict_blocks_without_auto_retry():
    """REQ-019 AC-3: 409 conflict halts auto-retry."""
    # The execution_id paper synthesizes is paper:{oid}:{qty}; first dispatched
    # order id is PAPER-000001 / qty=10 => execution_id = paper:PAPER-000001:10.
    conflict_id = "paper:PAPER-000001:10"
    ledger = InMemoryLedger(conflict_executions={conflict_id})
    repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0),
                                                          ledger=ledger)
    _buy(intents)
    execution.run_once()
    stats = writer.run_once()
    assert stats.conflicts == 1
    [item] = outbox.all_items()
    assert item.status is WorkItemStatus.CONFLICT_BLOCKED
    # second writer tick must NOT call the adapter again
    stats2 = writer.run_once()
    assert stats2.completed == 0 and stats2.retryable == 0
    # ledger is still empty (conflict blocked; operator resolves)
    assert ledger.count() == 0


def test_duplicate_completed_when_reprocessing_terminal_item():
    """REQ-019 AC-4: a completed item reports duplicate_completed if re-listed."""
    ledger = InMemoryLedger()
    repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0),
                                                          ledger=ledger)
    _buy(intents)
    execution.run_once()
    writer.run_once()
    # Force a duplicate enqueue of the same trade — outbox keeps single row,
    # ledger keeps single entry.
    [item] = outbox.all_items()
    outbox.enqueue(item.trade)
    assert len(outbox.all_items()) == 1
    # Writer sees no PENDING rows now; stats are all zero (terminal).
    stats = writer.run_once()
    assert stats.completed == 0 and stats.duplicates == 0 and stats.retryable == 0
    assert ledger.count() == 1


def test_slot_transition_proceeds_when_ledger_is_down():
    """REQ-019 AC-5: portfolio-hub outage does not block the fill transition."""
    ledger = InMemoryLedger(fail_n_then_succeed=100)
    repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0),
                                                          ledger=ledger)
    sid = _buy(intents).slot_id
    execution.run_once()
    assert repo.fetch_slot(sid).status is SlotState.HOLDING
    writer.run_once()  # all retries fail
    assert ledger.count() == 0
    [item] = outbox.all_items()
    assert item.status is WorkItemStatus.PENDING  # still waiting for retry