Merge pull request #90 from fourblocks2020/copilot/bug-ledger-shared-outbox-consumption

Isolate ledger outbox delivery by instance and add lease-based claims

tests/adapters/persistence/test_migration_and_drift.py

@@ -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 (
@@ -33,24 +34,22 @@ _EXPECTED_TABLES = {
 
 
 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 +60,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"}
 
@@ -111,17 +109,16 @@ 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:
+        with 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):
-    outbox = PostgreSQLLedgerOutbox(pg_clean)
+    outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="test-instance")
     normal = LedgerTrade(
         asset_id="PF-NORMAL",
         order_id="O-NORMAL",
@@ -129,7 +126,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 +137,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 +169,39 @@ 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, 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,),
+            )
+
+
+def test_migration_029_fail_closes_unowned_legacy_outbox_rows(pg_clean):
+    outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a")
+    trade = LedgerTrade(
+        asset_id="PF-LEGACY",
+        order_id="O-LEGACY",
+        execution_id="E-LEGACY",
+        side="buy",
+        qty=1,
+        price=1000,
+        executed_at=datetime(2026, 9, 12, tzinfo=UTC),
+    )
+    outbox.enqueue(trade)
+
     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,),
-                )
+            cur.execute("ALTER TABLE ledger_outbox ALTER COLUMN instance_key DROP NOT NULL")
+            cur.execute("ALTER TABLE ledger_outbox ALTER COLUMN instance_key DROP DEFAULT")
+            cur.execute(
+                "UPDATE ledger_outbox SET instance_key=NULL WHERE correlation_id=%s",
+                (trade.correlation_id,),
+            )
+        conn.commit()
+
+    apply_migrations(pg_clean)
+    stored = outbox.get(trade.correlation_id)
+    assert stored is not None
+    assert stored.instance_key == "__unowned__"
+    assert outbox.list_pending() == []

tests/adapters/persistence/test_postgresql_ledger_outbox.py

@@ -4,43 +4,53 @@ Skipped without TREX_POSTGRESQL_TEST_DSN (same gate as other PG suites).
 """
 from __future__ import annotations
 
-from datetime import datetime, timezone
+from concurrent.futures import ThreadPoolExecutor
+from datetime import UTC, datetime
+
+import pytest
 
 from trex_core.adapters.ledger.in_memory import InMemoryLedger
 from trex_core.adapters.persistence.postgresql import PostgreSQLLedgerOutbox
 from trex_core.application.ledger_outbox import (
     LedgerWriter,
+    OutboxClaimLostError,
     OutboxOutcome,
     WorkItemStatus,
 )
-from trex_core.application.ledger_port import LedgerTrade
+from trex_core.application.ledger_port import (
+    LedgerRetryableError,
+    LedgerTrade,
+    LedgerWriteOutcome,
+    LedgerWriteResult,
+)
 
 
 def _trade(execution_id="paper:1:10") -> LedgerTrade:
     return LedgerTrade(
         asset_id="PF-1", order_id="ORD-1", execution_id=execution_id,
         side="buy", qty=10, price=1000.0, fee=1.5, tax=0.0,
-        executed_at=datetime(2026, 6, 4, tzinfo=timezone.utc),
+        executed_at=datetime(2026, 6, 4, tzinfo=UTC),
         venue="NXT", order_venue="SOR",
     )
 
 
 def test_enqueue_persists_and_is_idempotent(pg_clean):
-    outbox = PostgreSQLLedgerOutbox(pg_clean)
+    outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a")
     a = outbox.enqueue(_trade())
     b = outbox.enqueue(_trade())
     assert a.correlation_id == b.correlation_id
     pending = outbox.list_pending()
     assert len(pending) == 1
     assert pending[0].status is WorkItemStatus.PENDING
+    assert pending[0].instance_key == "inst-a"
     assert pending[0].trade.venue == "NXT"
     assert pending[0].trade.order_venue == "SOR"
 
 
 def test_writer_completes_persisted_item(pg_clean):
-    outbox = PostgreSQLLedgerOutbox(pg_clean)
+    outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a")
     ledger = InMemoryLedger()
-    writer = LedgerWriter(outbox, ledger)
+    writer = LedgerWriter(outbox, ledger, instance_key="inst-a")
     outbox.enqueue(_trade(execution_id="pg-1"))
     stats = writer.run_once()
     assert stats.completed == 1 and ledger.count() == 1
@@ -50,9 +60,9 @@ def test_writer_completes_persisted_item(pg_clean):
 
 
 def test_writer_marks_conflict_blocked_on_pg(pg_clean):
-    outbox = PostgreSQLLedgerOutbox(pg_clean)
+    outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a")
     ledger = InMemoryLedger(conflict_executions={"pg-x"})
-    writer = LedgerWriter(outbox, ledger)
+    writer = LedgerWriter(outbox, ledger, instance_key="inst-a")
     outbox.enqueue(_trade(execution_id="pg-x"))
     writer.run_once()
     row = outbox.get("PF-1:ORD-1:pg-x")
@@ -61,10 +71,10 @@ def test_writer_marks_conflict_blocked_on_pg(pg_clean):
 
 
 def test_writer_retries_on_pg(pg_clean):
-    outbox = PostgreSQLLedgerOutbox(pg_clean)
+    outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a")
     ledger = InMemoryLedger(fail_n_then_succeed=1)
     # dcy-015: base 0 => next_retry_at == now, so the row is immediately due again.
-    writer = LedgerWriter(outbox, ledger, retry_base_sec=0.0)
+    writer = LedgerWriter(outbox, ledger, retry_base_sec=0.0, instance_key="inst-a")
     outbox.enqueue(_trade(execution_id="pg-r"))
     writer.run_once()
     row1 = outbox.get("PF-1:ORD-1:pg-r")
@@ -79,9 +89,9 @@ def test_writer_retries_on_pg(pg_clean):
 def test_writer_marks_delivery_failed_permanent_on_pg(pg_clean):
     """dcy-015 / REQ-019: 404/permanent → terminal delivery_failed persisted
     (migration 007 status enum + last_outcome). No retry."""
-    outbox = PostgreSQLLedgerOutbox(pg_clean)
+    outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a")
     ledger = InMemoryLedger(permanent_executions={"pg-perm"})
-    writer = LedgerWriter(outbox, ledger)
+    writer = LedgerWriter(outbox, ledger, instance_key="inst-a")
     outbox.enqueue(_trade(execution_id="pg-perm"))
     s = writer.run_once()
     assert s.delivery_failed == 1
@@ -94,22 +104,199 @@ def test_writer_marks_delivery_failed_permanent_on_pg(pg_clean):
 
 def test_pending_cash_delta_on_pg(pg_clean):
     """dcy-018 / REQ-029: estimated cash delta of pending rows summed by asset (SQL)."""
-    outbox = PostgreSQLLedgerOutbox(pg_clean)
+    outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a")
     outbox.enqueue(LedgerTrade(asset_id="PF-1", order_id="O-1", execution_id="cd-buy",
                                side="buy", qty=10, price=1000.0, fee=1.0, tax=0.0,
-                               executed_at=datetime(2026, 6, 4, tzinfo=timezone.utc)))
+                               executed_at=datetime(2026, 6, 4, tzinfo=UTC)))
     deltas = outbox.pending_cash_delta()
     assert deltas["PF-1"] == -(10 * 1000.0 + 1.0)  # buy => cash down by notional+fee
 
 
 def test_backoff_row_skipped_by_pending_scan_on_pg(pg_clean):
     """dcy-015 / REQ-019: a future next_retry_at excludes the row from list_pending."""
-    outbox = PostgreSQLLedgerOutbox(pg_clean)
+    outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a")
     ledger = InMemoryLedger(fail_n_then_succeed=99)
     # large base => next_retry_at well in the future
-    writer = LedgerWriter(outbox, ledger, retry_base_sec=600.0, max_attempts=99)
+    writer = LedgerWriter(
+        outbox, ledger, retry_base_sec=600.0, max_attempts=99, instance_key="inst-a"
+    )
     outbox.enqueue(_trade(execution_id="pg-bo"))
     writer.run_once()  # -> retryable, next_retry_at ~ now + 600s
     row = outbox.get("PF-1:ORD-1:pg-bo")
     assert row.status is WorkItemStatus.PENDING and row.next_retry_at is not None
     assert outbox.list_pending() == []  # still backing off -> skipped
+
+
+def test_pending_claims_are_isolated_per_instance(pg_clean):
+    ledger = InMemoryLedger()
+    outbox_a = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a")
+    outbox_b = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-b")
+    writer_a = LedgerWriter(outbox_a, ledger, instance_key="inst-a")
+    writer_b = LedgerWriter(outbox_b, ledger, instance_key="inst-b")
+    row_a = outbox_a.enqueue(_trade(execution_id="pg-a"))
+    row_b = outbox_b.enqueue(_trade(execution_id="pg-b"))
+
+    stats_a = writer_a.run_once()
+    assert stats_a.completed == 1 and ledger.count() == 1
+    assert outbox_a.get(row_a.correlation_id).status is WorkItemStatus.COMPLETED
+    assert outbox_b.get(row_b.correlation_id).status is WorkItemStatus.PENDING
+
+    stats_b = writer_b.run_once()
+    assert stats_b.completed == 1 and ledger.count() == 2
+    assert outbox_b.get(row_b.correlation_id).status is WorkItemStatus.COMPLETED
+
+
+def test_claim_expiry_allows_safe_recovery_after_worker_crash(pg_clean):
+    now = {"t": datetime(2026, 6, 4, tzinfo=UTC)}
+    outbox_a = PostgreSQLLedgerOutbox(
+        pg_clean,
+        instance_key="inst-a",
+        claim_ttl_sec=5.0,
+        worker_id="worker-a",
+        clock=lambda: now["t"],
+    )
+    outbox_b = PostgreSQLLedgerOutbox(
+        pg_clean,
+        instance_key="inst-a",
+        claim_ttl_sec=5.0,
+        worker_id="worker-b",
+        clock=lambda: now["t"],
+    )
+    row = outbox_a.enqueue(_trade(execution_id="pg-crash"))
+
+    claimed_a = outbox_a.claim_pending()
+    assert [item.correlation_id for item in claimed_a] == [row.correlation_id]
+    assert outbox_b.claim_pending() == []
+
+    now["t"] = datetime(2026, 6, 4, 0, 0, 6, tzinfo=UTC)
+    claimed_b = outbox_b.claim_pending()
+    assert [item.correlation_id for item in claimed_b] == [row.correlation_id]
+
+
+def test_stale_claim_cannot_overwrite_new_owner_outcome(pg_clean):
+    now = {"t": datetime(2026, 6, 4, tzinfo=UTC)}
+    outbox_a = PostgreSQLLedgerOutbox(
+        pg_clean,
+        instance_key="inst-a",
+        claim_ttl_sec=5.0,
+        worker_id="worker-a",
+        clock=lambda: now["t"],
+    )
+    outbox_b = PostgreSQLLedgerOutbox(
+        pg_clean,
+        instance_key="inst-a",
+        claim_ttl_sec=5.0,
+        worker_id="worker-b",
+        clock=lambda: now["t"],
+    )
+    row = outbox_a.enqueue(_trade(execution_id="pg-stale"))
+
+    [claim_a] = outbox_a.claim_pending()
+    now["t"] = datetime(2026, 6, 4, 0, 0, 6, tzinfo=UTC)
+    [claim_b] = outbox_b.claim_pending()
+    outbox_b.mark_outcome(
+        row.correlation_id,
+        OutboxOutcome.COMPLETED,
+        expected_claim_expires_at=claim_b.claim_expires_at,
+    )
+
+    with pytest.raises(OutboxClaimLostError):
+        outbox_a.mark_outcome(
+            row.correlation_id,
+            OutboxOutcome.RETRYABLE_FAILURE,
+            error="stale owner",
+            next_retry_at=now["t"],
+            expected_claim_expires_at=claim_a.claim_expires_at,
+        )
+
+    stored = outbox_a.get(row.correlation_id)
+    assert stored is not None
+    assert stored.status is WorkItemStatus.COMPLETED
+    assert stored.last_outcome is OutboxOutcome.COMPLETED
+
+
+@pytest.mark.parametrize("instance_key", ["", "   "])
+def test_postgresql_outbox_rejects_empty_instance_before_connect(pg_clean, instance_key):
+    with pytest.raises(ValueError, match="non-empty"):
+        PostgreSQLLedgerOutbox(pg_clean, instance_key=instance_key)
+
+
+def test_original_owner_can_finalize_expired_unreclaimed_claim(pg_clean):
+    now = {"t": datetime(2026, 6, 4, tzinfo=UTC)}
+    outbox = PostgreSQLLedgerOutbox(
+        pg_clean,
+        instance_key="inst-a",
+        claim_ttl_sec=5.0,
+        worker_id="worker-a",
+        clock=lambda: now["t"],
+    )
+    row = outbox.enqueue(_trade(execution_id="pg-expired"))
+
+    [claim] = outbox.claim_pending()
+    now["t"] = datetime(2026, 6, 4, 0, 0, 6, tzinfo=UTC)
+    stored = outbox.mark_outcome(
+        row.correlation_id,
+        OutboxOutcome.COMPLETED,
+        expected_claim_expires_at=claim.claim_expires_at,
+    )
+
+    assert stored.status is WorkItemStatus.COMPLETED
+    assert stored.last_outcome is OutboxOutcome.COMPLETED
+
+
+def test_writer_records_retryable_403_without_cross_instance_delivery(pg_clean):
+    class _ForbiddenLedger:
+        def __init__(self):
+            self.calls: list[str] = []
+
+        def record_trade(self, trade):
+            self.calls.append(trade.execution_id)
+            raise LedgerRetryableError("portfolio-hub 403 Forbidden")
+
+    outbox_a = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a")
+    outbox_b = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-b")
+    writer_b = LedgerWriter(
+        outbox_b, _ForbiddenLedger(), retry_base_sec=0.0, instance_key="inst-b"
+    )
+    row_a = outbox_a.enqueue(_trade(execution_id="pg-403-a"))
+    row_b = outbox_b.enqueue(_trade(execution_id="pg-403-b"))
+
+    stats_b = writer_b.run_once()
+
+    assert stats_b.retryable == 1
+    assert outbox_b.get(row_b.correlation_id).last_outcome is OutboxOutcome.RETRYABLE_FAILURE
+    assert outbox_a.get(row_a.correlation_id).status is WorkItemStatus.PENDING
+
+
+def test_concurrent_workers_claim_each_row_at_most_once(pg_clean):
+    outbox_a = PostgreSQLLedgerOutbox(
+        pg_clean, instance_key="inst-a", worker_id="worker-a"
+    )
+    outbox_b = PostgreSQLLedgerOutbox(
+        pg_clean, instance_key="inst-a", worker_id="worker-b"
+    )
+    row = outbox_a.enqueue(_trade(execution_id="pg-concurrent"))
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        claims = list(pool.map(lambda box: box.claim_pending(), (outbox_a, outbox_b)))
+
+    claimed_ids = [item.correlation_id for batch in claims for item in batch]
+    assert claimed_ids == [row.correlation_id]
+
+
+def test_writer_marks_duplicate_completed_from_adapter_response_on_pg(pg_clean):
+    class _DuplicateLedger:
+        def record_trade(self, _trade):
+            return LedgerWriteOutcome(LedgerWriteResult.DUPLICATE_OK)
+
+    outbox = PostgreSQLLedgerOutbox(pg_clean, instance_key="inst-a")
+    writer = LedgerWriter(outbox, _DuplicateLedger(), instance_key="inst-a")
+    row = outbox.enqueue(_trade(execution_id="pg-dup"))
+
+    outcome = writer.process_correlation_id(row.correlation_id)
+
+    assert outcome is OutboxOutcome.DUPLICATE_COMPLETED
+    stored = outbox.get(row.correlation_id)
+    assert stored is not None
+    assert stored.status is WorkItemStatus.COMPLETED
+    assert stored.last_outcome is OutboxOutcome.DUPLICATE_COMPLETED

tests/application/test_execution_cancel.py

@@ -70,7 +70,7 @@ def _wire(broker, *, stale_sec=None):
     svc = SlotService(repo)
     intents = IntentService(svc, repo, InMemoryIntentStore(),
                             resolve_symbol=lambda pid: "005930")
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     ledger = InMemoryLedger()
     execution = ExecutionService(
         svc, repo, broker, ledger_outbox=outbox,

tests/application/test_execution_fill_price_backfill.py

@@ -120,7 +120,7 @@ def _wire(broker):
         resolve_symbol=lambda pid: "005930",
         instance_key=INST,
     )
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     execution = ExecutionService(svc, repo, broker, ledger_outbox=outbox, instance_key=INST)
     repo.reconcile_universe(INST, {"PF-1": "005930"})
     return repo, svc, intents, execution, outbox
@@ -966,7 +966,7 @@ class TestPaperBrokerRegression:
             resolve_symbol=lambda pid: "005930",
             instance_key=INST,
         )
-        outbox2 = InMemoryLedgerOutbox()
+        outbox2 = InMemoryLedgerOutbox(instance_key="test-instance")
         execution2 = ExecutionService(svc2, repo2, paper, ledger_outbox=outbox2, instance_key=INST)
         repo2.reconcile_universe(INST, {"PF-1": "005930"})
 

tests/application/test_execution_ledger.py

@@ -119,7 +119,11 @@ def _wire(broker, *, ledger=None, clock=None, alarm_emitter=None, ledger_skip_st
     svc = SlotService(repo)
     intents = IntentService(svc, repo, InMemoryIntentStore(),
                             resolve_symbol=lambda pid: "005930")
-    outbox = InMemoryLedgerOutbox(clock=clock) if clock else InMemoryLedgerOutbox()
+    outbox = (
+        InMemoryLedgerOutbox(clock=clock, instance_key="default")
+        if clock
+        else InMemoryLedgerOutbox(instance_key="default")
+    )
     execution = ExecutionService(
         svc,
         repo,
@@ -430,7 +434,7 @@ def test_duplicate_execution_id_results_in_zero_duplicate_in_ledger():
     # 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_outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     replay_writer = LedgerWriter(replay_outbox, ledger)
     replay_outbox.enqueue(first.trade)
     stats = replay_writer.run_once()

tests/application/test_ledger_option_d.py

@@ -39,7 +39,7 @@ def _store_with(pid="PF-1", cash=1_000_000.0, qty=0.0):
 # --- write-back (REQ-019/030) ----------------------------------------------
 def test_writeback_updates_snapshot_on_new():
     store = _store_with()
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     ledger = InMemoryLedger(cash_after=300_000.0, quantity_after=10.0)
     writer = LedgerWriter(outbox, ledger, snapshot_store=store)
     outbox.enqueue(_trade())
@@ -50,7 +50,7 @@ def test_writeback_updates_snapshot_on_new():
 
 def test_writeback_applies_on_duplicate_too():
     store = _store_with(cash=1_000_000.0)
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     # execution already recorded once → second write is duplicate, *_after still authoritative
     ledger = InMemoryLedger(cash_after=300_000.0, quantity_after=10.0)
     ledger.record_trade(_trade(execution_id="dup"))   # pre-record → next is duplicate
@@ -62,7 +62,7 @@ def test_writeback_applies_on_duplicate_too():
 
 def test_writeback_skipped_when_after_missing():
     store = _store_with(cash=1_000_000.0)
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     ledger = InMemoryLedger()  # no cash_after/quantity_after → None
     writer = LedgerWriter(outbox, ledger, snapshot_store=store)
     outbox.enqueue(_trade())
@@ -74,7 +74,7 @@ def test_writeback_best_effort_does_not_block_completion():
     class _BoomStore:
         def update_cash_position(self, *a, **k):
             raise RuntimeError("boom")
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     ledger = InMemoryLedger(cash_after=1.0, quantity_after=1.0)
     writer = LedgerWriter(outbox, ledger, snapshot_store=_BoomStore())
     outbox.enqueue(_trade())
@@ -83,7 +83,7 @@ def test_writeback_best_effort_does_not_block_completion():
 
 
 def test_no_writeback_without_store():
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     ledger = InMemoryLedger(cash_after=1.0, quantity_after=1.0)
     writer = LedgerWriter(outbox, ledger)  # no snapshot_store
     outbox.enqueue(_trade())
@@ -92,7 +92,7 @@ def test_no_writeback_without_store():
 
 # --- estimate overlay (REQ-029 / OQ-4) — via make_slots_rpc_handler ----------
 def test_pending_cash_delta_sums_by_asset():
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     outbox.enqueue(_trade(execution_id="b", side="buy", qty=10, price=1000.0, fee=1.0))
     deltas = outbox.pending_cash_delta()
     assert deltas["PF-1"] == trade_cash_delta(_trade(side="buy", qty=10, price=1000.0, fee=1.0))
@@ -101,7 +101,7 @@ def test_pending_cash_delta_sums_by_asset():
 
 def test_apply_cash_estimate_overlays_pending_delta():
     """Cash-estimated overlay applied through slots handler (REQ-029 OQ-4)."""
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     outbox.enqueue(_trade(execution_id="b", side="buy", qty=10, price=1000.0, fee=1.0))
     store = _store_with(pid="PF-1", cash=1_000_000.0, qty=0.0)
     repo = InMemorySlotRepository()
@@ -117,7 +117,7 @@ def test_apply_cash_estimate_overlays_pending_delta():
 def test_estimate_clears_after_writeback():
     """Once the ledger write completes (no longer pending), the estimate is gone."""
     store = _store_with(cash=1_000_000.0)
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     ledger = InMemoryLedger(cash_after=989_999.0, quantity_after=10.0)
     writer = LedgerWriter(outbox, ledger, snapshot_store=store)
     outbox.enqueue(_trade(side="buy", qty=10, price=1000.0, fee=1.0))

tests/application/test_ledger_outbox.py

@@ -1,16 +1,20 @@
 """LedgerOutbox + LedgerWriter direct unit tests (REQ-019)."""
 from __future__ import annotations
 
+from dataclasses import replace
 from datetime import UTC, datetime, timedelta
 
+import pytest
+
 from trex_core.adapters.ledger.in_memory import InMemoryLedger
 from trex_core.application.ledger_outbox import (
     InMemoryLedgerOutbox,
     LedgerWriter,
+    OutboxClaimLostError,
     OutboxOutcome,
     WorkItemStatus,
 )
-from trex_core.application.ledger_port import LedgerTrade
+from trex_core.application.ledger_port import LedgerPermanentError, LedgerTrade
 
 
 def _trade(execution_id="paper:PAPER-1:10") -> LedgerTrade:
@@ -22,7 +26,7 @@ def _trade(execution_id="paper:PAPER-1:10") -> LedgerTrade:
 
 
 def test_enqueue_is_idempotent_on_correlation_id():
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     a = outbox.enqueue(_trade())
     b = outbox.enqueue(_trade())
     assert a.correlation_id == b.correlation_id
@@ -30,7 +34,7 @@ def test_enqueue_is_idempotent_on_correlation_id():
 
 
 def test_writer_run_once_drains_pending_and_marks_completed():
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     ledger = InMemoryLedger()
     writer = LedgerWriter(outbox, ledger)
     outbox.enqueue(_trade(execution_id="e-1"))
@@ -42,7 +46,7 @@ def test_writer_run_once_drains_pending_and_marks_completed():
 
 
 def test_writer_marks_conflict_blocked_and_does_not_retry():
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     ledger = InMemoryLedger(conflict_executions={"e-x"})
     writer = LedgerWriter(outbox, ledger)
     outbox.enqueue(_trade(execution_id="e-x"))
@@ -58,7 +62,9 @@ def test_writer_retries_until_success():
     # dcy-015: retryable rows now back off; share an injected clock between the
     # outbox (list_pending "now") and the writer (next_retry_at) to advance past it.
     now = {"t": datetime(2026, 6, 4, tzinfo=UTC)}
-    outbox = InMemoryLedgerOutbox(clock=lambda: now["t"])
+    outbox = InMemoryLedgerOutbox(
+        clock=lambda: now["t"], instance_key="test-instance"
+    )
     ledger = InMemoryLedger(fail_n_then_succeed=1)
     writer = LedgerWriter(outbox, ledger, clock=lambda: now["t"])
     outbox.enqueue(_trade(execution_id="e-r"))
@@ -71,15 +77,134 @@ def test_writer_retries_until_success():
 
 
 def test_process_correlation_id_only_drains_requested_row():
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="inst-a")
     ledger = InMemoryLedger()
-    writer = LedgerWriter(outbox, ledger)
+    writer = LedgerWriter(outbox, ledger, instance_key="inst-a")
     first = outbox.enqueue(_trade(execution_id="e-1"))
     second = outbox.enqueue(_trade(execution_id="e-2"))
 
-    outcome = writer.process_correlation_id(first.correlation_id)
+    outcome = writer.process_correlation_id(second.correlation_id)
 
     assert outcome is OutboxOutcome.COMPLETED
     assert ledger.count() == 1
-    assert outbox.get(first.correlation_id).status is WorkItemStatus.COMPLETED
-    assert outbox.get(second.correlation_id).status is WorkItemStatus.PENDING
+    assert writer.counters["claim:claimed"] == 1
+    assert writer.counters["outcome:completed"] == 1
+    assert outbox.get(first.correlation_id).status is WorkItemStatus.PENDING
+    assert outbox.get(second.correlation_id).status is WorkItemStatus.COMPLETED
+
+
+def test_process_correlation_id_skips_foreign_instance_row():
+    outbox = InMemoryLedgerOutbox(instance_key="inst-a")
+    ledger = InMemoryLedger()
+    item = outbox.enqueue(_trade(execution_id="e-foreign"))
+
+    with pytest.raises(ValueError, match="must match"):
+        LedgerWriter(outbox, ledger, instance_key="inst-b")
+    assert ledger.count() == 0
+    assert outbox.get(item.correlation_id).status is WorkItemStatus.PENDING
+
+
+def test_process_correlation_id_rejects_legacy_row_without_instance_owner():
+    class _LegacyOutbox(InMemoryLedgerOutbox):
+        def get(self, correlation_id):
+            item = super().get(correlation_id)
+            return replace(item, instance_key=None) if item is not None else None
+
+    outbox = _LegacyOutbox(instance_key="inst-a")
+    ledger = InMemoryLedger()
+    writer = LedgerWriter(outbox, ledger, instance_key="inst-a")
+    item = outbox.enqueue(_trade())
+
+    assert writer.process_correlation_id(item.correlation_id) is None
+    assert ledger.count() == 0
+
+
+def test_writer_exposes_claim_and_outcome_counters():
+    outbox = InMemoryLedgerOutbox(instance_key="inst-a")
+    ledger = InMemoryLedger()
+    writer = LedgerWriter(outbox, ledger, instance_key="inst-a")
+    outbox.enqueue(_trade(execution_id="e-count"))
+
+    stats = writer.run_once()
+
+    assert stats.completed == 1
+    assert writer.counters["claim:claimed"] == 1
+    assert writer.counters["outcome:completed"] == 1
+
+
+@pytest.mark.parametrize("instance_key", ["", "   "])
+def test_writer_rejects_explicit_empty_instance_key(instance_key):
+    outbox = InMemoryLedgerOutbox(instance_key="inst-a")
+    with pytest.raises(ValueError, match="non-empty"):
+        LedgerWriter(outbox, InMemoryLedger(), instance_key=instance_key)
+
+
+@pytest.mark.parametrize("instance_key", ["", "   "])
+def test_outbox_rejects_empty_instance_key(instance_key):
+    with pytest.raises(ValueError, match="non-empty"):
+        InMemoryLedgerOutbox(instance_key=instance_key)
+
+
+def test_in_memory_claim_is_exclusive_and_stale_lease_is_fenced():
+    now = {"t": datetime(2026, 6, 4, tzinfo=UTC)}
+    outbox = InMemoryLedgerOutbox(
+        instance_key="inst-a",
+        claim_ttl_sec=5.0,
+        clock=lambda: now["t"],
+    )
+    item = outbox.enqueue(_trade(execution_id="e-lease"))
+
+    [claim_a] = outbox.claim_pending()
+    assert outbox.claim_pending() == []
+    assert outbox.list_pending() == []
+
+    now["t"] += timedelta(seconds=6)
+    [claim_b] = outbox.claim_pending()
+    assert claim_b.claim_expires_at != claim_a.claim_expires_at
+    with pytest.raises(OutboxClaimLostError):
+        outbox.mark_outcome(
+            item.correlation_id,
+            OutboxOutcome.COMPLETED,
+            expected_claim_expires_at=claim_a.claim_expires_at,
+        )
+
+    stored = outbox.mark_outcome(
+        item.correlation_id,
+        OutboxOutcome.COMPLETED,
+        expected_claim_expires_at=claim_b.claim_expires_at,
+    )
+    assert stored.status is WorkItemStatus.COMPLETED
+    assert stored.claim_expires_at is None
+
+
+def test_outcome_logs_do_not_render_exception_text(caplog):
+    sentinel = "TOKEN_SENTINEL_ACCOUNT_DSN_PAYLOAD"
+
+    class _FailingLedger:
+        def record_trade(self, _trade):
+            raise LedgerPermanentError(sentinel)
+
+    outbox = InMemoryLedgerOutbox(instance_key="inst-a")
+    writer = LedgerWriter(outbox, _FailingLedger(), instance_key="inst-a")
+    outbox.enqueue(_trade(execution_id="e-sensitive"))
+
+    with caplog.at_level("INFO"):
+        stats = writer.run_once()
+
+    assert stats.delivery_failed == 1
+    assert sentinel not in caplog.text
+
+
+def test_lost_claim_is_counted_without_false_completion():
+    class _ClaimLosingOutbox(InMemoryLedgerOutbox):
+        def mark_outcome(self, correlation_id, outcome, **kwargs):
+            raise OutboxClaimLostError(correlation_id)
+
+    outbox = _ClaimLosingOutbox(instance_key="inst-a")
+    outbox.enqueue(_trade())
+    writer = LedgerWriter(outbox, InMemoryLedger(), instance_key="inst-a")
+
+    stats = writer.run_once()
+
+    assert stats.claim_lost == 1
+    assert stats.completed == 0

tests/application/test_ledger_outbox_backoff.py

@@ -47,7 +47,7 @@ class _ProgrammableLedger:
 
 
 def _wire(behavior, *, clock, **writer_kw):
-    outbox = InMemoryLedgerOutbox(clock=clock)
+    outbox = InMemoryLedgerOutbox(clock=clock, instance_key="test-instance")
     ledger = _ProgrammableLedger(behavior)
     writer = LedgerWriter(outbox, ledger, clock=clock, rng=lambda: 0.0, **writer_kw)
     return outbox, ledger, writer

tests/application/test_messaging_event_publisher.py

@@ -22,7 +22,7 @@ def _wire(transport):
     svc = SlotService(repo)
     intents = IntentService(svc, repo, InMemoryIntentStore(),
                             resolve_symbol=lambda pid: "005930")
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     ledger = InMemoryLedger()
     writer = LedgerWriter(outbox, ledger)
     repo.reconcile_universe("default", {"PF-1": "SYM"})
@@ -203,7 +203,7 @@ def test_sell_fill_dlq_sentinel_absent_when_strategy_id_set():
     intents = IntentService(svc, repo, InMemoryIntentStore(),
                             resolve_symbol=lambda pid: "005930")
     from trex_core.application.ledger_outbox import InMemoryLedgerOutbox
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     execution = ExecutionService(svc, repo, PaperBroker(fill_after_polls=0),
                                  ledger_outbox=outbox, event_emitter=emitter)
     repo.reconcile_universe("default", {"PF-1": "SYM"})

tests/runtime/test_cli_ledger_backfill.py

@@ -187,7 +187,11 @@ def test_ledger_backfill_existing_completed_row_is_idempotent(monkeypatch, tmp_p
     import trex_core.adapters.ledger.portfolio_hub as hub_mod
     import trex_core.adapters.persistence.postgresql as pg_mod
 
-    monkeypatch.setattr(pg_mod, "PostgreSQLLedgerOutbox", lambda dsn: _Outbox())
+    monkeypatch.setattr(
+        pg_mod,
+        "PostgreSQLLedgerOutbox",
+        lambda dsn, *, instance_key: _Outbox(),
+    )
     monkeypatch.setattr(hub_mod, "PortfolioHubLedger", lambda *args, **kwargs: object())
     monkeypatch.setattr(cli, "LedgerWriter", _Writer)
     monkeypatch.setattr(
@@ -246,7 +250,11 @@ def test_ledger_backfill_live_enqueue_and_processes_exact_trade(monkeypatch, tmp
     import trex_core.adapters.ledger.portfolio_hub as hub_mod
     import trex_core.adapters.persistence.postgresql as pg_mod
 
-    monkeypatch.setattr(pg_mod, "PostgreSQLLedgerOutbox", lambda dsn: outbox)
+    monkeypatch.setattr(
+        pg_mod,
+        "PostgreSQLLedgerOutbox",
+        lambda dsn, *, instance_key: outbox,
+    )
     monkeypatch.setattr(hub_mod, "PortfolioHubLedger", lambda *args, **kwargs: object())
     monkeypatch.setattr(cli, "LedgerWriter", _Writer)
     monkeypatch.setattr(
@@ -310,7 +318,11 @@ def test_ledger_backfill_resolves_matching_skip(monkeypatch, tmp_path, capsys):
     import trex_core.adapters.ledger.portfolio_hub as hub_mod
     import trex_core.adapters.persistence.postgresql as pg_mod
 
-    monkeypatch.setattr(pg_mod, "PostgreSQLLedgerOutbox", lambda dsn: _Outbox())
+    monkeypatch.setattr(
+        pg_mod,
+        "PostgreSQLLedgerOutbox",
+        lambda dsn, *, instance_key: _Outbox(),
+    )
     monkeypatch.setattr(hub_mod, "PortfolioHubLedger", lambda *args, **kwargs: object())
     monkeypatch.setattr(cli, "LedgerWriter", _Writer)
     monkeypatch.setattr(