trex/trex-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix unmanaged reconciliation review blockers

Browse Source 
Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-09-13 02:21:37 +00:00
committed by GitHub
co-authored by genichin
parent f9284d1de8
commit 591db7adb7
26 changed files with 484 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/adapters/persistence/test_migration_and_drift.py
+57 -45
View File
@@ -6,9 +6,10 @@ iteration: EXPECTED_SLOT_FIELDS (code contract) vs daily_slot actual columns
"""
from __future__ import annotations
from datetime import UTC, datetime
import psycopg
import pytest
from datetime import datetime, timezone
from trex_core.adapters.persistence.migrate import apply_migrations, migration_files
from trex_core.adapters.persistence.postgresql import (
@@ -29,28 +30,28 @@ _EXPECTED_TABLES = {
"ledger_skip", # GitHub #68 skipped-ledger durable audit
"reconcile_notification_state", # req-0007 / ac-0024
"slot_strategy_state", # req-0009 / ac-0027
"unmanaged_position_state", # req-0010 / ac-0029
"unmanaged_position_audit", # req-0010 / ac-0029
}
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()}
with psycopg.connect(dsn) as conn, 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()}
with psycopg.connect(pg_clean) as conn, 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
@@ -61,20 +62,19 @@ def test_daily_slot_columns_match_expected_contract(pg_clean):
def test_strategy_state_schema_matches_instance_map_contract(pg_clean):
with psycopg.connect(pg_clean) as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT column_name, data_type, udt_name "
"FROM information_schema.columns "
"WHERE table_schema = 'public' AND table_name = 'slot_strategy_state'"
)
columns = {row[0]: (row[1], row[2]) for row in cur.fetchall()}
cur.execute(
"SELECT a.attname FROM pg_index i "
"JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey) "
"WHERE i.indrelid = 'slot_strategy_state'::regclass AND i.indisprimary"
)
primary_key = {row[0] for row in cur.fetchall()}
with psycopg.connect(pg_clean) as conn, conn.cursor() as cur:
cur.execute(
"SELECT column_name, data_type, udt_name "
"FROM information_schema.columns "
"WHERE table_schema = 'public' AND table_name = 'slot_strategy_state'"
)
columns = {row[0]: (row[1], row[2]) for row in cur.fetchall()}
cur.execute(
"SELECT a.attname FROM pg_index i "
"JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey) "
"WHERE i.indrelid = 'slot_strategy_state'::regclass AND i.indisprimary"
)
primary_key = {row[0] for row in cur.fetchall()}
assert columns == {"slot_id": ("uuid", "uuid"), "data": ("jsonb", "jsonb")}
assert primary_key == {"slot_id"}
@@ -92,6 +92,14 @@ def test_apply_migrations_is_idempotent(pg_dsn):
apply_migrations(pg_dsn)
def test_migration_sequence_includes_canonical_unmanaged_position_number():
names = [path.name for path in migration_files()]
assert "030_create_unmanaged_position_state.sql" in names
assert "028_create_unmanaged_position_state.sql" not in names
prefixes = [name.split("_", 1)[0] for name in names]
assert len(prefixes) == len(set(prefixes))
def test_migration_replay_converts_existing_auto_venue_row_to_sor(pg_clean):
repo = PostgreSQLSlotRepository(pg_clean)
repo.reconcile_universe("migration-auto", {"pf-auto": "005930"})
@@ -111,13 +119,15 @@ def test_migration_replay_converts_existing_auto_venue_row_to_sor(pg_clean):
apply_migrations(pg_clean)
assert repo.fetch_slot(slot.slot_id).venue == "SOR"
with psycopg.connect(pg_clean) as conn:
with conn.cursor() as cur:
with pytest.raises(psycopg.errors.CheckViolation):
cur.execute(
"UPDATE daily_slot SET venue = 'AUTO' WHERE slot_id = %s",
(slot.slot_id,),
)
with (
psycopg.connect(pg_clean) as conn,
conn.cursor() as cur,
pytest.raises(psycopg.errors.CheckViolation),
):
cur.execute(
"UPDATE daily_slot SET venue = 'AUTO' WHERE slot_id = %s",
(slot.slot_id,),
)
def test_migration_024_separates_legacy_sor_and_is_replay_safe(pg_clean):
@@ -129,7 +139,7 @@ def test_migration_024_separates_legacy_sor_and_is_replay_safe(pg_clean):
side="buy",
qty=1,
price=1000,
executed_at=datetime(2026, 8, 19, tzinfo=timezone.utc),
executed_at=datetime(2026, 8, 19, tzinfo=UTC),
venue="NXT",
order_venue="KRX",
)
@@ -140,7 +150,7 @@ def test_migration_024_separates_legacy_sor_and_is_replay_safe(pg_clean):
side="buy",
qty=1,
price=1000,
executed_at=datetime(2026, 8, 19, tzinfo=timezone.utc),
executed_at=datetime(2026, 8, 19, tzinfo=UTC),
venue="KRX",
order_venue="SOR",
)
@@ -172,10 +182,12 @@ def test_migration_024_separates_legacy_sor_and_is_replay_safe(pg_clean):
assert legacy_row.trade.venue is None
assert legacy_row.trade.order_venue == "SOR"
with psycopg.connect(pg_clean) as conn:
with conn.cursor() as cur:
with pytest.raises(psycopg.errors.CheckViolation):
cur.execute(
"UPDATE ledger_outbox SET venue='SOR' WHERE correlation_id=%s",
(normal.correlation_id,),
)
with (
psycopg.connect(pg_clean) as conn,
conn.cursor() as cur,
pytest.raises(psycopg.errors.CheckViolation),
):
cur.execute(
"UPDATE ledger_outbox SET venue='SOR' WHERE correlation_id=%s",
(normal.correlation_id,),
)
tests/adapters/portfolio/test_unmanaged_position_store.py
+11 -1
View File
@@ -15,7 +15,7 @@ def pg_unmanaged_store(pg_clean):
return PostgreSQLUnmanagedPositionStore(pg_clean)
class TestMigration028:
class TestMigration030:
def test_tables_exist_after_migration(self, pg_clean):
import psycopg
@@ -111,6 +111,14 @@ class TestCRUDRoundTrip:
quantity=3,
observed_at=t1,
)
pg_unmanaged_store.record_operator_action(
instance_key="inst-A",
symbol="008110",
action="investigating",
evidence_ref="kis:2026-09-07",
actor="cli:test",
note="before resolve",
)
pg_unmanaged_store.resolve_absent(
instance_key="inst-A",
observed_symbols=set(),
@@ -126,3 +134,5 @@ class TestCRUDRoundTrip:
assert reopened["transition"] == "detected"
assert reopened["state"]["first_seen_at"] == t1
assert reopened["state"]["last_seen_at"] == t3
assert reopened["state"]["operator_action"] is None
assert reopened["state"]["evidence_ref"] is None
tests/application/test_snapshot_scheduler.py
+39 -6
View File
@@ -1,7 +1,8 @@
"""Snapshot scheduler tests (REQ-030 AC-1; REQ-114 AC-1~9; REQ-115 AC-1~6)."""
from __future__ import annotations
from datetime import datetime, timezone, timedelta
from datetime import UTC, datetime, timedelta, timezone
from types import SimpleNamespace
from unittest.mock import MagicMock
from trex_core.adapters.broker.paper import PaperBroker
@@ -44,10 +45,42 @@ def test_run_once_failure_sets_degraded_signal():
store = SnapshotStore()
sched = SnapshotScheduler(read_port=BadRead(), store=store, broker=PaperBroker())
assert sched.run_once() is False
reason, msg = store.degraded_state()
reason, _msg = store.degraded_state()
assert reason is not None and reason.value == "portfolio_hub_unavailable"
def test_run_once_uses_authoritative_core_holdings_broker(monkeypatch):
read = StubPortfolioRead(
daemon_id="d-1",
items=[PortfolioItem(portfolio_id="pf-X", symbol="X", cash_balance=100, asset_quantity=0)],
)
store = SnapshotStore()
projected_broker = object()
authoritative_broker = object()
captured = {}
def _capture(**kwargs):
captured["broker"] = kwargs["broker"]
return {}
monkeypatch.setattr(
"trex_core.application.reconciliation.reconcile_core_holdings",
_capture,
)
sched = SnapshotScheduler(
read_port=read,
store=store,
broker=projected_broker,
core_holdings_broker=authoritative_broker,
slot_repo=object(),
slot_service=SimpleNamespace(instance_key="inst-1"),
)
sched._reconcile_slots = lambda _items: None
assert sched.run_once() is True
assert captured["broker"] is authoritative_broker
def test_swap_preserves_warning_for_carried_over_portfolio():
store = SnapshotStore()
store.swap([PortfolioItem(portfolio_id="pf-X", symbol="X",
@@ -99,7 +132,7 @@ def _fixed_clock(kst_hour=8, kst_minute=0):
"""Return a clock callable fixed at a given KST time today."""
now_kst = datetime.now(tz=_KST).replace(hour=kst_hour, minute=kst_minute,
second=0, microsecond=0)
return lambda: now_kst.astimezone(timezone.utc)
return lambda: now_kst.astimezone(UTC)
# ---------------------------------------------------------------------------
@@ -616,8 +649,8 @@ def test_kis_broker_reinitialize_preserves_base_url_and_flags():
def test_build_kis_reinit_callback_returns_none_for_paper():
"""REQ-113 AC-2: broker=paper이면 callback=None."""
from trex_core.adapters.broker.paper import PaperBroker as PB
from trex_core.runtime.daemon import _build_kis_reinit_callback
from trex_core.runtime.config import DaemonConfig
from trex_core.runtime.daemon import _build_kis_reinit_callback
# minimal config: paper broker, credential_fetch_enabled=True
cfg = DaemonConfig(dsn="x", broker="paper", execution_interval_sec=1.0,
@@ -629,8 +662,8 @@ def test_build_kis_reinit_callback_returns_none_for_paper():
def test_build_kis_reinit_callback_returns_none_when_disabled():
"""REQ-113 AC-1: credential_fetch_enabled=False이면 callback=None."""
from trex_core.adapters.broker.paper import PaperBroker as PB
from trex_core.runtime.daemon import _build_kis_reinit_callback
from trex_core.runtime.config import DaemonConfig
from trex_core.runtime.daemon import _build_kis_reinit_callback
cfg = DaemonConfig(dsn="x", broker="kis", execution_interval_sec=1.0,
credential_fetch_enabled=False)
@@ -642,8 +675,8 @@ def test_build_kis_reinit_callback_returns_callable_for_kis():
"""REQ-113 AC-2 / REQ-115 AC-1: broker=kis + fetch enabled → callable 반환 + 실제 reinitialize 호출됨."""
from trex_core.adapters.broker.kis import KISBroker
from trex_core.application.portfolio import BrokerCredentialPayload
from trex_core.runtime.daemon import _build_kis_reinit_callback
from trex_core.runtime.config import DaemonConfig
from trex_core.runtime.daemon import _build_kis_reinit_callback
cfg = DaemonConfig(dsn="x", broker="kis", execution_interval_sec=1.0,
credential_fetch_enabled=True)
tests/application/test_unmanaged_position_lifecycle.py
+36 -3
View File
@@ -35,6 +35,11 @@ class RaisingBroker(FakeBroker):
raise BrokerBalanceError("transport_failed")
class AuthoritativeBroker(FakeBroker):
def exposes_full_balance_snapshot(self) -> bool:
return True
class FilteredBroker(FakeBroker):
def exposes_full_balance_snapshot(self) -> bool:
return False
@@ -116,6 +121,13 @@ def test_in_memory_store_preserves_first_seen_when_resolved_symbol_reappears():
t3 = t2 + timedelta(minutes=1)
store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t1)
store.record_operator_action(
instance_key=INSTANCE,
symbol="008110",
action="investigating",
evidence_ref="kis-balance:2026-09-07",
note="before resolve",
)
store.resolve_absent(instance_key=INSTANCE, observed_symbols=set(), observed_at=t2)
reopened = store.observe(instance_key=INSTANCE, symbol="008110", quantity=5, observed_at=t3)
@@ -123,6 +135,8 @@ def test_in_memory_store_preserves_first_seen_when_resolved_symbol_reappears():
assert reopened["state"]["first_seen_at"] == t1
assert reopened["state"]["last_seen_at"] == t3
assert reopened["state"]["quantity_changed_at"] == t3
assert reopened["state"]["operator_action"] is None
assert reopened["state"]["evidence_ref"] is None
def test_reconcile_core_holdings_suppresses_repeated_unmanaged_alarm_spam():
@@ -134,7 +148,7 @@ def test_reconcile_core_holdings_suppresses_repeated_unmanaged_alarm_spam():
decisions = reconcile_core_holdings(
slot_repo=repo,
slot_service=svc,
broker=FakeBroker({"008110": 3}),
broker=AuthoritativeBroker({"008110": 3}),
instance_key=INSTANCE,
alarm_emitter=emitter,
unmanaged_position_store=store,
@@ -166,7 +180,7 @@ def test_reconcile_core_holdings_tracks_change_and_resolution():
reconcile_core_holdings(
slot_repo=repo,
slot_service=svc,
broker=FakeBroker({"008110": 5}),
broker=AuthoritativeBroker({"008110": 5}),
instance_key=INSTANCE,
alarm_emitter=emitter,
unmanaged_position_store=store,
@@ -174,7 +188,7 @@ def test_reconcile_core_holdings_tracks_change_and_resolution():
reconcile_core_holdings(
slot_repo=repo,
slot_service=svc,
broker=FakeBroker({}),
broker=AuthoritativeBroker({}),
instance_key=INSTANCE,
alarm_emitter=emitter,
unmanaged_position_store=store,
@@ -211,6 +225,25 @@ def test_broker_failure_does_not_resolve_unmanaged_state():
assert state["lifecycle_state"] == "active"
def test_missing_authoritative_marker_does_not_resolve_unmanaged_state():
repo, svc = _wire()
store = InMemoryUnmanagedPositionStore()
store.observe(instance_key=INSTANCE, symbol="008110", quantity=3)
decisions = reconcile_core_holdings(
slot_repo=repo,
slot_service=svc,
broker=FakeBroker({}),
instance_key=INSTANCE,
unmanaged_position_store=store,
)
assert decisions == {}
state = store.get_state(instance_key=INSTANCE, symbol="008110")
assert state is not None
assert state["lifecycle_state"] == "active"
def test_non_authoritative_broker_view_does_not_resolve_unmanaged_state():
repo, svc = _wire()
store = InMemoryUnmanagedPositionStore()
tests/runtime/test_broker_provider.py
+21
View File
@@ -23,6 +23,15 @@ class _Broker:
def reinitialize(self, credential): self.credentials.append(credential)
class _ProjectedBroker(_Broker):
def __init__(self, authoritative):
super().__init__()
self.authoritative = authoritative
def authoritative_balance_reader(self):
return self.authoritative
class _Read:
def __init__(self, ctx=None, error=None): self.ctx, self.error, self.calls = ctx, error, 0
def fetch_broker_credential(self):
@@ -72,6 +81,18 @@ def test_enabled_paper_never_fetches_kis_credentials(monkeypatch):
assert read.calls == 0
def test_reconciliation_broker_uses_authoritative_balance_reader(monkeypatch):
authoritative = _Broker()
projected = _ProjectedBroker(authoritative)
monkeypatch.setattr("trex_core.runtime.broker_provider._make_env_broker", lambda cfg: projected)
provider = ConfigBrokerProvider(_cfg(broker="paper"), read_port=_Read(_ctx()))
broker, source = provider.get_broker_for_reconciliation()
assert broker is authoritative
assert source.mode == "env"
def test_break_glass_is_distinct_and_skips_fetch(monkeypatch):
broker = _Broker()
monkeypatch.setenv("TREX_KIS_BROKER_BREAK_GLASS", "true")
tests/runtime/test_cli_operator.py
+84 -4
View File
@@ -523,12 +523,15 @@ def test_verify_no_db_no_live_flag_needed(capsys, monkeypatch):
# unmanaged parser/tests
# ---------------------------------------------------------------------------
def test_unmanaged_parser_accepts_list_and_record():
def test_unmanaged_parser_accepts_list_audit_and_record():
parser = cli._build_parser()
args = parser.parse_args(["unmanaged", "list", "--status", "all", "--symbol", "008110"])
assert args.cmd == "unmanaged"
assert args.unmanaged_cmd == "list"
assert args.status == "all"
args = parser.parse_args(["unmanaged", "audit", "008110"])
assert args.unmanaged_cmd == "audit"
assert args.symbol == "008110"
args = parser.parse_args(
[
"unmanaged",
@@ -617,7 +620,6 @@ def test_unmanaged_record_requires_existing_state(monkeypatch, capsys):
action="approved_exception",
evidence_ref="kis-balance:2026-09-11",
note="confirmed",
actor="operator",
instance="inst-1",
config=None,
config_path=None,
@@ -634,6 +636,8 @@ def test_unmanaged_record_requires_existing_state(monkeypatch, capsys):
def test_unmanaged_record_persists_operator_action(monkeypatch, capsys):
monkeypatch.setattr(cli, "_operator_actor", lambda: "cli:test")
class _Store:
def __init__(self, *_args, **_kwargs):
pass
@@ -671,7 +675,6 @@ def test_unmanaged_record_persists_operator_action(monkeypatch, capsys):
action="managed_inclusion",
evidence_ref="kis-balance:2026-09-11",
note="manual inclusion planned",
actor="operator",
instance="inst-1",
config=None,
config_path=None,
@@ -687,9 +690,12 @@ def test_unmanaged_record_persists_operator_action(monkeypatch, capsys):
assert out["result"] == "recorded"
assert out["state"]["operator_action"] == "managed_inclusion"
assert out["state"]["evidence_ref"] == "kis-balance:2026-09-11"
assert out["state"]["operator_actor"] == "cli:test"
def test_unmanaged_record_text_output(monkeypatch, capsys):
monkeypatch.setattr(cli, "_operator_actor", lambda: "cli:test")
class _Store:
def __init__(self, *_args, **_kwargs):
pass
@@ -724,7 +730,6 @@ def test_unmanaged_record_text_output(monkeypatch, capsys):
action="approved_exception",
evidence_ref="kis-balance:2026-09-11",
note="confirmed",
actor="operator",
instance="inst-1",
config=None,
config_path=None,
@@ -742,6 +747,8 @@ def test_unmanaged_record_text_output(monkeypatch, capsys):
def test_unmanaged_record_main_dispatch(monkeypatch, capsys):
monkeypatch.setattr(cli, "_operator_actor", lambda: "cli:test")
class _Store:
def __init__(self, *_args, **_kwargs):
pass
@@ -788,6 +795,79 @@ def test_unmanaged_record_main_dispatch(monkeypatch, capsys):
assert out["state"]["symbol"] == "008110"
def test_unmanaged_audit_outputs_lifecycle_and_operator_history(monkeypatch, capsys):
class _Store:
def __init__(self, *_args, **_kwargs):
pass
def get_state(self, **kwargs):
assert kwargs == {"instance_key": "inst-1", "symbol": "008110"}
return {
"instance_key": "inst-1",
"symbol": "008110",
"quantity": 3,
"lifecycle_state": "active",
"first_seen_at": "2026-09-07T00:00:00+00:00",
"last_seen_at": "2026-09-11T00:00:00+00:00",
"quantity_changed_at": "2026-09-07T00:00:00+00:00",
"resolved_at": None,
"operator_action": "investigating",
"evidence_ref": "kis-balance:2026-09-11",
"operator_note": "confirmed",
"operator_actor": "cli:test",
"operator_updated_at": "2026-09-11T00:00:00+00:00",
"created_at": "2026-09-07T00:00:00+00:00",
"updated_at": "2026-09-11T00:00:00+00:00",
}
def list_audit(self, **kwargs):
assert kwargs == {"instance_key": "inst-1", "symbol": "008110"}
return [
{
"event_type": "detected",
"recorded_at": "2026-09-07T00:00:00+00:00",
"quantity": 3,
"action": None,
"actor": "reconcile",
"evidence_ref": None,
"note": "",
},
{
"event_type": "operator_action",
"recorded_at": "2026-09-11T00:00:00+00:00",
"quantity": 3,
"action": "investigating",
"actor": "cli:test",
"evidence_ref": "kis-balance:2026-09-11",
"note": "confirmed",
},
]
monkeypatch.setattr(
"trex_core.adapters.portfolio.unmanaged_position_store.PostgreSQLUnmanagedPositionStore",
_Store,
)
args = SimpleNamespace(
cmd="unmanaged",
unmanaged_cmd="audit",
symbol="008110",
instance="inst-1",
config=None,
config_path=None,
env_file=None,
json=True,
timeout=10.0,
)
rc = cli._cmd_unmanaged_audit(args, SimpleNamespace(dsn="postgresql://x/y"))
assert rc == 0
out = json.loads(capsys.readouterr().out)
assert out["symbol"] == "008110"
assert out["state"]["operator_actor"] == "cli:test"
assert [row["event_type"] for row in out["audit"]] == ["detected", "operator_action"]
# ---------------------------------------------------------------------------
# REL-010 regression: _cmd_verify broker import must not raise ImportError
# ---------------------------------------------------------------------------
tests/runtime/test_daemon_resolve_symbol.py
+41
View File
@@ -73,3 +73,44 @@ def test_daemon_wires_shared_alarm_emitter_into_intent_service():
found = True
break
assert found, "Daemon must pass self.alarm_emitter into IntentService(alarm_emitter=...)"
def test_daemon_wires_reconciliation_broker_into_snapshot_scheduler():
src = (
Path(__file__).resolve().parents[2]
/ "trex_core"
/ "runtime"
/ "daemon.py"
).read_text(encoding="utf-8")
tree = ast.parse(src)
found = False
for node in ast.walk(tree):
if not isinstance(node, ast.Assign):
continue
if not any(
isinstance(t, ast.Attribute) and t.attr == "snapshot_scheduler"
for t in node.targets
):
continue
call = node.value
if not (
isinstance(call, ast.Call)
and isinstance(call.func, ast.Name)
and call.func.id == "SnapshotScheduler"
):
continue
for kw in call.keywords:
if kw.arg != "core_holdings_broker":
continue
if (
isinstance(kw.value, ast.Attribute)
and kw.value.attr == "reconciliation_broker"
and isinstance(kw.value.value, ast.Name)
and kw.value.value.id == "self"
):
found = True
break
assert found, (
"Daemon must pass self.reconciliation_broker into "
"SnapshotScheduler(core_holdings_broker=...)"
)