From 359ea098ff717e7f04cf4206dcbc64ad17b47755 Mon Sep 17 00:00:00 2001
From: Hermes Agent <noreply@nousresearch.com>
Date: Wed, 19 Aug 2026 12:53:56 +0900
Subject: [PATCH] fix: enforce actual venue persistence contract

---
 docs/data-model.md                            |  3 +-
 .../persistence/test_migration_and_drift.py   | 68 ++++++++++++++++++-
 .../sql/024_add_ledger_order_venue.sql        | 29 ++++++--
 trex_core/application/broker_port.py          |  4 +-
 trex_core/application/ledger_port.py          |  5 +-
 5 files changed, 99 insertions(+), 10 deletions(-)

diff --git a/docs/data-model.md b/docs/data-model.md
index 5d01846..ba12c5d 100644
--- a/docs/data-model.md
+++ b/docs/data-model.md
@@ -153,7 +153,8 @@
 - side :: TEXT CHECK ('buy','sell')
 - qty / price / fee / tax :: NUMERIC(18,4)
 - executed_at :: TIMESTAMPTZ?
-- venue :: TEXT NOT NULL DEFAULT 'KRX' :: 실제 제출 거래소. 기존 행은 KRX로 보존한다.
+- venue :: TEXT? CHECK (NULL 또는 'KRX'/'NXT') :: broker-authoritative 실제 체결 거래소. 과거 SOR row처럼 실제 거래소를 복원할 수 없으면 NULL이며 임의 추론하지 않는다.
+- order_venue :: TEXT NOT NULL DEFAULT 'KRX' CHECK ('KRX'/'NXT'/'SOR') :: 원주문/라우팅 venue. 취소·재시작 감사에 사용한다.
 - status :: TEXT CHECK ('pending','completed','conflict_blocked','delivery_failed') — `delivery_failed` 추가(dcy-015, 마이그레이션 007)
 - attempts :: INTEGER :: 재시도 누적
 - last_outcome :: TEXT? CHECK ('completed','duplicate_completed','retryable_failure','conflict_blocked','permanent_failure','dead_lettered')
diff --git a/tests/adapters/persistence/test_migration_and_drift.py b/tests/adapters/persistence/test_migration_and_drift.py
index 1e25857..1d9fe08 100644
--- a/tests/adapters/persistence/test_migration_and_drift.py
+++ b/tests/adapters/persistence/test_migration_and_drift.py
@@ -8,9 +8,14 @@ from __future__ import annotations
 
 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 PostgreSQLSlotRepository
+from trex_core.adapters.persistence.postgresql import (
+    PostgreSQLLedgerOutbox,
+    PostgreSQLSlotRepository,
+)
+from trex_core.application.ledger_port import LedgerTrade
 from trex_core.domain.slot import EXPECTED_SLOT_FIELDS
 
 _EXPECTED_TABLES = {
@@ -91,3 +96,64 @@ def test_migration_replay_converts_existing_auto_venue_row_to_sor(pg_clean):
                     "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):
+    outbox = PostgreSQLLedgerOutbox(pg_clean)
+    normal = LedgerTrade(
+        asset_id="PF-NORMAL",
+        order_id="O-NORMAL",
+        execution_id="E-NORMAL",
+        side="buy",
+        qty=1,
+        price=1000,
+        executed_at=datetime(2026, 8, 19, tzinfo=timezone.utc),
+        venue="NXT",
+        order_venue="KRX",
+    )
+    legacy = LedgerTrade(
+        asset_id="PF-LEGACY",
+        order_id="O-LEGACY",
+        execution_id="E-LEGACY",
+        side="buy",
+        qty=1,
+        price=1000,
+        executed_at=datetime(2026, 8, 19, tzinfo=timezone.utc),
+        venue="KRX",
+        order_venue="SOR",
+    )
+    outbox.enqueue(normal)
+    outbox.enqueue(legacy)
+
+    # Recreate a pre-024 SOR row: one venue column carried routing provenance,
+    # and no separate order_venue was available.
+    with psycopg.connect(pg_clean) as conn:
+        with conn.cursor() as cur:
+            cur.execute("ALTER TABLE ledger_outbox DROP CONSTRAINT ledger_outbox_venue_check")
+            cur.execute("ALTER TABLE ledger_outbox ALTER COLUMN order_venue DROP NOT NULL")
+            cur.execute(
+                "UPDATE ledger_outbox SET venue='SOR', order_venue=NULL "
+                "WHERE correlation_id=%s",
+                (legacy.correlation_id,),
+            )
+        conn.commit()
+
+    apply_migrations(pg_clean)
+    apply_migrations(pg_clean)
+
+    normal_row = outbox.get(normal.correlation_id)
+    legacy_row = outbox.get(legacy.correlation_id)
+    assert normal_row is not None
+    assert normal_row.trade.venue == "NXT"
+    assert normal_row.trade.order_venue == "KRX"
+    assert legacy_row is not None
+    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,),
+                )
diff --git a/trex_core/adapters/persistence/sql/024_add_ledger_order_venue.sql b/trex_core/adapters/persistence/sql/024_add_ledger_order_venue.sql
index a093d69..c2cec8a 100644
--- a/trex_core/adapters/persistence/sql/024_add_ledger_order_venue.sql
+++ b/trex_core/adapters/persistence/sql/024_add_ledger_order_venue.sql
@@ -1,6 +1,18 @@
--- GitHub #61: keep the original submission venue separate from actual execution venue.
+-- GitHub #61: keep original submission venue separate from actual execution venue.
+-- The nullable add + NULL-only backfill makes replay safe: rows written after
+-- this migration already have order_venue and are never reinterpreted.
 ALTER TABLE ledger_outbox
-    ADD COLUMN IF NOT EXISTS order_venue TEXT NOT NULL DEFAULT 'KRX';
+    ADD COLUMN IF NOT EXISTS order_venue TEXT;
+
+UPDATE ledger_outbox
+SET order_venue = venue
+WHERE order_venue IS NULL
+  AND venue IN ('KRX', 'NXT', 'SOR');
+
+ALTER TABLE ledger_outbox
+    ALTER COLUMN order_venue SET DEFAULT 'KRX';
+ALTER TABLE ledger_outbox
+    ALTER COLUMN order_venue SET NOT NULL;
 
 ALTER TABLE ledger_outbox
     DROP CONSTRAINT IF EXISTS ledger_outbox_order_venue_check;
@@ -8,5 +20,14 @@ ALTER TABLE ledger_outbox
     ADD CONSTRAINT ledger_outbox_order_venue_check
     CHECK (order_venue IN ('KRX', 'NXT', 'SOR'));
 
--- Historical rows used venue for both concepts. Preserve that provenance.
-UPDATE ledger_outbox SET order_venue = venue WHERE order_venue = 'KRX' AND venue <> 'KRX';
+-- Historical SOR in `venue` recorded routing provenance, not a proven KRX/NXT
+-- execution exchange. Preserve SOR in order_venue and mark actual as unknown.
+ALTER TABLE ledger_outbox
+    ALTER COLUMN venue DROP NOT NULL;
+UPDATE ledger_outbox SET venue = NULL WHERE venue = 'SOR';
+
+ALTER TABLE ledger_outbox
+    DROP CONSTRAINT IF EXISTS ledger_outbox_venue_check;
+ALTER TABLE ledger_outbox
+    ADD CONSTRAINT ledger_outbox_venue_check
+    CHECK (venue IS NULL OR venue IN ('KRX', 'NXT'));
diff --git a/trex_core/application/broker_port.py b/trex_core/application/broker_port.py
index b4a404b..b56842a 100644
--- a/trex_core/application/broker_port.py
+++ b/trex_core/application/broker_port.py
@@ -71,8 +71,8 @@ class OrderAck:
     accepted: bool
     broker_order_id: Optional[str] = None
     reason: Optional[str] = None
-    # Actual venue used for submission. Execution persists this value with the
-    # broker order id so restart/cancel never has to infer the original venue.
+    # Original submission/routing venue. Execution persists this with the
+    # broker order id so restart/cancel never infer it from wall-clock time.
     venue: Optional[str] = None
     raw: dict[str, Any] = field(default_factory=dict)
     submitted_at: Optional[datetime] = None
diff --git a/trex_core/application/ledger_port.py b/trex_core/application/ledger_port.py
index 8cd2924..8d92236 100644
--- a/trex_core/application/ledger_port.py
+++ b/trex_core/application/ledger_port.py
@@ -57,8 +57,9 @@ class LedgerTrade:
     fee: float = 0.0
     tax: float = 0.0
     executed_at: Optional[datetime] = None
-    # Broker-authoritative execution exchange (KRX/NXT).
-    venue: str = "KRX"
+    # Broker-authoritative execution exchange (KRX/NXT). Historical rows whose
+    # SOR execution exchange cannot be reconstructed are None.
+    venue: Optional[str] = None
     # Original submission/routing venue (KRX/NXT/SOR), retained for audit.
     order_venue: str = "KRX"