Deliver BAT-002: slot-core daemon + HTTP intent API + PostgreSQL

Implement REQ-005~009 (dcy-002 slice 1) as an onion stack on top of the
BAT-001 pure domain:
- application: IntentService (admission + single-writer transition +
  intent_id idempotency via durable intent_log)
- adapters/persistence: PostgreSQLSlotRepository + PostgreSQLIntentStore +
  trex-core-owned migrations (daily_slot per Appendix A, daily_symbol_entries,
  slot_transition_log, intent_log) + in-memory intent store
- adapters/api: FastAPI /v1 (healthz, intents/buy|sell|{id}, slots, capacity)
- runtime: trex-core-daemon entrypoint with lifespan migrate+preallocate

Verification (against local PostgreSQL 18): 74 passed, ruff clean, domain
import sweep shows zero web/db deps (NFR-1), 20-thread concurrency admits
exactly the idle count (NFR-2), idempotent resubmit causes no double
transition (NFR-3), EXPECTED_SLOT_FIELDS<->information_schema drift test
(realizes dcy-001 OQ-2's deferred DB-side check), and a daemon E2E smoke.

Align all four baseline docs to the new daemon/API/persistence layers.
BAT-002 -> release-candidate; REQ-005~009 -> Implemented. Full async outbox,
broker execution, safety floor, and strategy daemons remain deferred to a
follow-up Discovery.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

tests/adapters/api/test_intent_api.py

@@ -0,0 +1,112 @@
+"""HTTP /v1 API tests via in-memory adapters (REQ-005/007/008).
+
+No PostgreSQL needed: the app is wired with the in-memory slot repo + intent
+store, exercising the full request → admission → transition → response path.
+"""
+from __future__ import annotations
+
+from datetime import date
+
+import pytest
+from fastapi.testclient import TestClient
+
+from trex_core.adapters.api.app import create_app
+from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore
+from trex_core.application.intents import IntentService
+from trex_core.domain.slot import SlotService, SlotState
+from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
+
+DAY = date(2026, 6, 3)
+
+
+@pytest.fixture
+def ctx():
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    intents = IntentService(svc, repo, InMemoryIntentStore())
+    svc.preallocate_daily_slots(DAY, 3)
+    app = create_app(
+        slot_service=svc, slot_repo=repo, intent_service=intents, default_total_slots=3
+    )
+    return repo, svc, intents, TestClient(app)
+
+
+def _buy(client, intent_id, symbol="005930"):
+    return client.post(
+        "/v1/intents/buy",
+        json={"intent_id": intent_id, "trading_date": DAY.isoformat(), "symbol": symbol, "qty": 10},
+    )
+
+
+def test_healthz(ctx):
+    _, _, _, client = ctx
+    r = client.get("/v1/healthz")
+    assert r.status_code == 200 and r.json()["status"] == "ok"
+
+
+def test_buy_intent_admits_and_transitions(ctx):
+    repo, _, _, client = ctx
+    r = _buy(client, "buy-1")
+    assert r.status_code == 200
+    body = r.json()
+    assert body["status"] == "admitted" and body["slot_id"]
+    slot = repo.fetch_slot(__import__("uuid").UUID(body["slot_id"]))
+    assert slot.status is SlotState.PENDING_BUY and slot.symbol == "005930"
+
+
+def test_buy_intent_idempotent_resubmit(ctx):
+    repo, _, _, client = ctx
+    r1 = _buy(client, "buy-dup")
+    r2 = _buy(client, "buy-dup")  # same intent_id
+    assert r1.json() == r2.json()  # identical decision
+    # only ONE slot moved out of idle
+    pending = [s for s in repo.list_slots(DAY) if s.status is SlotState.PENDING_BUY]
+    assert len(pending) == 1
+
+
+def test_buy_intent_rejected_when_no_idle(ctx):
+    _, _, _, client = ctx
+    for i in range(3):
+        assert _buy(client, f"buy-fill-{i}").json()["status"] == "admitted"
+    r = _buy(client, "buy-overflow")
+    assert r.json()["status"] == "rejected" and r.json()["reason"] == "no_idle_slot"
+
+
+def test_sell_intent_requires_holding(ctx):
+    repo, _, _, client = ctx
+    # a freshly bought slot is pending_buy, not holding -> sell rejected
+    slot_id = _buy(client, "buy-for-sell").json()["slot_id"]
+    r = client.post("/v1/intents/sell", json={"intent_id": "sell-1", "slot_id": slot_id})
+    assert r.json()["status"] == "rejected" and r.json()["reason"].startswith("slot_not_holding")
+
+
+def test_sell_intent_admits_from_holding(ctx):
+    repo, svc, _, client = ctx
+    import uuid
+
+    slot_id = uuid.UUID(_buy(client, "buy-h").json()["slot_id"])
+    svc.transit_slot(slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="filled", actor="t")
+    r = client.post("/v1/intents/sell", json={"intent_id": "sell-h", "slot_id": str(slot_id)})
+    assert r.json()["status"] == "admitted"
+    assert repo.fetch_slot(slot_id).status is SlotState.PENDING_SELL
+
+
+def test_get_intent_roundtrip(ctx):
+    _, _, _, client = ctx
+    _buy(client, "buy-q")
+    r = client.get("/v1/intents/buy-q")
+    assert r.status_code == 200 and r.json()["status"] == "admitted"
+    assert client.get("/v1/intents/does-not-exist").status_code == 404
+
+
+def test_list_slots_and_capacity(ctx):
+    _, _, _, client = ctx
+    _buy(client, "buy-c")
+    slots = client.get(
+        "/v1/slots", params={"status": "pending_buy", "trading_date": DAY.isoformat()}
+    ).json()
+    assert len(slots) == 1 and slots[0]["status"] == "pending_buy"
+    cap = client.get(
+        "/v1/capacity", params={"total_slots": 3, "trading_date": DAY.isoformat()}
+    ).json()
+    assert cap["idle_count"] == 2 and cap["pending_count"] == 1 and cap["total_slots"] == 3

tests/adapters/persistence/test_migration_and_drift.py

@@ -0,0 +1,55 @@
+"""Migration + column-contract drift test (REQ-006).
+
+Realizes the DB-side drift check dcy-001 OQ-2 deferred to the persistence
+iteration: EXPECTED_SLOT_FIELDS (code contract) vs daily_slot actual columns
+(information_schema). Skipped without TREX_POSTGRESQL_TEST_DSN.
+"""
+from __future__ import annotations
+
+import psycopg
+
+from trex_core.adapters.persistence.migrate import apply_migrations, migration_files
+from trex_core.domain.slot import EXPECTED_SLOT_FIELDS
+
+_EXPECTED_TABLES = {"daily_slot", "daily_symbol_entries", "slot_transition_log", "intent_log"}
+
+
+def _columns(dsn: str, table: str) -> set[str]:
+    with psycopg.connect(dsn) as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                "SELECT column_name FROM information_schema.columns "
+                "WHERE table_schema = 'public' AND table_name = %s",
+                (table,),
+            )
+            return {r[0] for r in cur.fetchall()}
+
+
+def test_migrations_create_expected_tables(pg_clean):
+    assert migration_files(), "no migration files found"
+    with psycopg.connect(pg_clean) as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                "SELECT table_name FROM information_schema.tables WHERE table_schema='public'"
+            )
+            tables = {r[0] for r in cur.fetchall()}
+    assert _EXPECTED_TABLES <= tables
+
+
+def test_daily_slot_columns_match_expected_contract(pg_clean):
+    """AC: daily_slot columns == EXPECTED_SLOT_FIELDS (dcy-001 Appendix A)."""
+    actual = _columns(pg_clean, "daily_slot")
+    assert actual == set(EXPECTED_SLOT_FIELDS)
+
+
+def test_drift_detected_when_contract_extended(pg_clean):
+    """Negative case: a contract field absent from the table is caught."""
+    actual = _columns(pg_clean, "daily_slot")
+    drifted_contract = set(EXPECTED_SLOT_FIELDS) | {"phantom_column"}
+    assert actual != drifted_contract
+
+
+def test_apply_migrations_is_idempotent(pg_dsn):
+    # applying twice must not error (CREATE ... IF NOT EXISTS)
+    apply_migrations(pg_dsn)
+    apply_migrations(pg_dsn)

tests/adapters/persistence/test_postgresql_repository.py

@@ -0,0 +1,61 @@
+"""PostgreSQL integration tests (REQ-006). Skipped without TREX_POSTGRESQL_TEST_DSN."""
+from __future__ import annotations
+
+from datetime import date
+
+from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore  # noqa: F401
+from trex_core.adapters.persistence.postgresql import (
+    PostgreSQLIntentStore,
+    PostgreSQLSlotRepository,
+)
+from trex_core.application.intents import BuyIntent, IntentService, SellIntent
+from trex_core.domain.slot import SlotService, SlotState
+
+DAY = date(2026, 6, 3)
+
+
+def _wire(dsn):
+    repo = PostgreSQLSlotRepository(dsn)
+    svc = SlotService(repo)
+    intents = IntentService(svc, repo, PostgreSQLIntentStore(dsn))
+    return repo, svc, intents
+
+
+def test_preallocate_and_capacity(pg_clean):
+    repo, svc, _ = _wire(pg_clean)
+    assert svc.preallocate_daily_slots(DAY, 10) == 10
+    assert svc.preallocate_daily_slots(DAY, 10) == 0  # idempotent
+    cap = svc.compute_slot_capacity(DAY, 10)
+    assert cap.idle_count == 10 and cap.scarcity_ratio == 0.0
+
+
+def test_buy_intent_persists_transition(pg_clean):
+    repo, svc, intents = _wire(pg_clean)
+    svc.preallocate_daily_slots(DAY, 3)
+    d = intents.submit_buy(BuyIntent(intent_id="pg-buy-1", trading_date=DAY, symbol="005930", qty=10))
+    assert d.admitted
+    slot = repo.fetch_slot(d.slot_id)
+    assert slot.status is SlotState.PENDING_BUY and slot.symbol == "005930"
+
+
+def test_buy_intent_idempotent_on_pg(pg_clean):
+    repo, svc, intents = _wire(pg_clean)
+    svc.preallocate_daily_slots(DAY, 3)
+    d1 = intents.submit_buy(BuyIntent(intent_id="pg-dup", trading_date=DAY, symbol="005930"))
+    d2 = intents.submit_buy(BuyIntent(intent_id="pg-dup", trading_date=DAY, symbol="005930"))
+    assert d1.slot_id == d2.slot_id and d2.status == "admitted"
+    pending = [s for s in repo.list_slots(DAY) if s.status is SlotState.PENDING_BUY]
+    assert len(pending) == 1  # no double transition
+
+
+def test_full_lifecycle_recovery_bumps_entry_count(pg_clean):
+    repo, svc, intents = _wire(pg_clean)
+    svc.preallocate_daily_slots(DAY, 1)
+    d = intents.submit_buy(BuyIntent(intent_id="pg-life", trading_date=DAY, symbol="005930"))
+    sid = d.slot_id
+    svc.transit_slot(sid, SlotState.PENDING_BUY, SlotState.HOLDING, reason="filled", actor="t")
+    intents.submit_sell(SellIntent(intent_id="pg-sell", slot_id=sid))
+    svc.transit_slot(sid, SlotState.PENDING_SELL, SlotState.CLOSED, reason="filled", actor="t")
+    assert repo.get_symbol_entry_count(DAY, "005930") == 0
+    svc.transit_slot(sid, SlotState.CLOSED, SlotState.IDLE, reason="recovery", actor="t")
+    assert repo.get_symbol_entry_count(DAY, "005930") == 1

tests/application/test_intent_concurrency.py

@@ -0,0 +1,74 @@
+"""Concurrency / single-writer test for IntentService (REQ-007 NFR-2/NFR-3).
+
+Many threads submit buy intents competing for a limited number of idle slots;
+exactly that many must be admitted and no slot may be double-used.
+"""
+from __future__ import annotations
+
+import threading
+from datetime import date
+
+from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore
+from trex_core.application.intents import BuyIntent, IntentService
+from trex_core.domain.slot import SlotService, SlotState
+from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
+
+DAY = date(2026, 6, 3)
+
+
+def test_concurrent_buys_admit_exactly_idle_count():
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    intents = IntentService(svc, repo, InMemoryIntentStore())
+    svc.preallocate_daily_slots(DAY, 5)  # only 5 idle slots
+
+    results: list[str] = []
+    results_lock = threading.Lock()
+
+    def worker(i: int) -> None:
+        d = intents.submit_buy(BuyIntent(intent_id=f"buy-{i}", trading_date=DAY, symbol="005930", qty=1))
+        with results_lock:
+            results.append(d.status)
+
+    threads = [threading.Thread(target=worker, args=(i,)) for i in range(20)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    admitted = [s for s in results if s == "admitted"]
+    rejected = [s for s in results if s == "rejected"]
+    assert len(admitted) == 5  # exactly the idle slots
+    assert len(rejected) == 15
+    # every non-idle slot is pending_buy exactly once (no double-use)
+    pending = [s for s in repo.list_slots(DAY) if s.status is SlotState.PENDING_BUY]
+    idle = [s for s in repo.list_slots(DAY) if s.status is SlotState.IDLE]
+    assert len(pending) == 5 and len(idle) == 0
+
+
+def test_duplicate_intent_id_admits_once_under_concurrency():
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    intents = IntentService(svc, repo, InMemoryIntentStore())
+    svc.preallocate_daily_slots(DAY, 5)
+
+    statuses: list[str] = []
+    lock = threading.Lock()
+
+    def worker() -> None:
+        d = intents.submit_buy(BuyIntent(intent_id="same-id", trading_date=DAY, symbol="005930"))
+        with lock:
+            statuses.append(d.status)
+
+    threads = [threading.Thread(target=worker) for _ in range(10)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    # same intent_id submitted 10x concurrently -> exactly ONE slot transitioned
+    # (the core idempotency guarantee). Racing duplicates observe the winner's
+    # decision, which may still be 'processing' before finalize; never 'rejected'.
+    pending = [s for s in repo.list_slots(DAY) if s.status is SlotState.PENDING_BUY]
+    assert len(pending) == 1
+    assert all(s in ("admitted", "processing") for s in statuses)

tests/conftest.py

@@ -0,0 +1,34 @@
+"""Shared test fixtures.
+
+PostgreSQL integration tests are gated on TREX_POSTGRESQL_TEST_DSN; when it
+is unset they are skipped (unit/API tests still run via in-memory adapters).
+"""
+from __future__ import annotations
+
+import os
+
+import pytest
+
+_PG_TABLES = ("intent_log", "slot_transition_log", "daily_symbol_entries", "daily_slot")
+
+
+@pytest.fixture
+def pg_dsn() -> str:
+    dsn = os.environ.get("TREX_POSTGRESQL_TEST_DSN")
+    if not dsn:
+        pytest.skip("TREX_POSTGRESQL_TEST_DSN not set; skipping PostgreSQL integration test")
+    return dsn
+
+
+@pytest.fixture
+def pg_clean(pg_dsn: str):
+    """Apply migrations and truncate slot-core tables before each PG test."""
+    import psycopg
+
+    from trex_core.adapters.persistence.migrate import apply_migrations
+
+    apply_migrations(pg_dsn)
+    with psycopg.connect(pg_dsn, autocommit=True) as conn:
+        with conn.cursor() as cur:
+            cur.execute(f"TRUNCATE {', '.join(_PG_TABLES)}")
+    return pg_dsn