From 304a06f97c1228a2dee15ebcddee85da53507ab8 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Sat, 12 Sep 2026 13:17:22 +0000
Subject: [PATCH 1/6] Initial plan


From 09b1b4f4c3d78c985796d3275e2b04815ee984f9 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Sat, 12 Sep 2026 13:24:58 +0000
Subject: [PATCH 2/6] fix: isolate ledger outbox delivery by instance

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
---
 .../persistence/test_migration_and_drift.py   | 112 +++++++++------
 .../test_postgresql_ledger_outbox.py          | 128 +++++++++++++++---
 tests/application/test_ledger_outbox.py       |  27 ++++
 tests/runtime/test_cli_ledger_backfill.py     |  18 ++-
 trex_core/adapters/persistence/postgresql.py  |  91 +++++++++++--
 .../sql/028_ledger_outbox_instance_claims.sql |  22 +++
 trex_core/application/ledger_outbox.py        | 108 ++++++++++++++-
 trex_core/runtime/cli.py                      |   2 +-
 trex_core/runtime/daemon.py                   |   2 +-
 9 files changed, 429 insertions(+), 81 deletions(-)
 create mode 100644 trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql

diff --git a/tests/adapters/persistence/test_migration_and_drift.py b/tests/adapters/persistence/test_migration_and_drift.py
index 3a35cd3..592cc60 100644
--- a/tests/adapters/persistence/test_migration_and_drift.py
+++ b/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,13 +109,12 @@ 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):
@@ -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_028_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() == []
diff --git a/tests/adapters/persistence/test_postgresql_ledger_outbox.py b/tests/adapters/persistence/test_postgresql_ledger_outbox.py
index ef65a3c..6f51388 100644
--- a/tests/adapters/persistence/test_postgresql_ledger_outbox.py
+++ b/tests/adapters/persistence/test_postgresql_ledger_outbox.py
@@ -4,7 +4,7 @@ Skipped without TREX_POSTGRESQL_TEST_DSN (same gate as other PG suites).
 """
 from __future__ import annotations
 
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 
 from trex_core.adapters.ledger.in_memory import InMemoryLedger
 from trex_core.adapters.persistence.postgresql import PostgreSQLLedgerOutbox
@@ -13,34 +13,40 @@ from trex_core.application.ledger_outbox import (
     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 +56,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 +67,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 +85,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 +100,112 @@ 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_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_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
diff --git a/tests/application/test_ledger_outbox.py b/tests/application/test_ledger_outbox.py
index 5ce7f71..e085f30 100644
--- a/tests/application/test_ledger_outbox.py
+++ b/tests/application/test_ledger_outbox.py
@@ -83,3 +83,30 @@ def test_process_correlation_id_only_drains_requested_row():
     assert ledger.count() == 1
     assert outbox.get(first.correlation_id).status is WorkItemStatus.COMPLETED
     assert outbox.get(second.correlation_id).status is WorkItemStatus.PENDING
+
+
+def test_process_correlation_id_skips_foreign_instance_row():
+    outbox = InMemoryLedgerOutbox(instance_key="inst-a")
+    ledger = InMemoryLedger()
+    writer = LedgerWriter(outbox, ledger, instance_key="inst-b")
+    item = outbox.enqueue(_trade(execution_id="e-foreign"))
+
+    outcome = writer.process_correlation_id(item.correlation_id)
+
+    assert outcome is None
+    assert ledger.count() == 0
+    assert writer.counters["claim:foreign_instance"] == 1
+    assert outbox.get(item.correlation_id).status is WorkItemStatus.PENDING
+
+
+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
diff --git a/tests/runtime/test_cli_ledger_backfill.py b/tests/runtime/test_cli_ledger_backfill.py
index e09282f..17455ff 100644
--- a/tests/runtime/test_cli_ledger_backfill.py
+++ b/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(
diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py
index 6a52b3d..64cabc7 100644
--- a/trex_core/adapters/persistence/postgresql.py
+++ b/trex_core/adapters/persistence/postgresql.py
@@ -13,9 +13,10 @@ domain layer (NFR-1).
 from __future__ import annotations
 
 import json
+from collections.abc import Iterator
 from contextlib import contextmanager
 from datetime import datetime, timedelta, timezone
-from typing import Any, Iterator, Optional
+from typing import Any, Optional
 from uuid import UUID, uuid4
 
 from trex_core.application.ledger_outbox import (
@@ -26,9 +27,9 @@ from trex_core.application.ledger_outbox import (
 from trex_core.application.ledger_port import LedgerTrade
 from trex_core.application.ports import IntentRecord
 from trex_core.application.strategy_state import (
+    _STATE_CAP_BYTES,
     SetResult,
     StrategyStateEntry,
-    _STATE_CAP_BYTES,
 )
 from trex_core.domain.slot import Slot, SlotNotFoundError, SlotState, SlotTransitionRecord
 
@@ -565,16 +566,32 @@ class _PgIntentRecord:
 _OUTBOX_COLUMNS = (
     "correlation_id, asset_id, order_id, execution_id, side, qty, price, "
     "fee, tax, executed_at, status, attempts, last_outcome, last_error, "
-    "created_at, updated_at, next_retry_at, venue, order_venue"
+    "created_at, updated_at, next_retry_at, venue, order_venue, instance_key"
 )
 
 
 class PostgreSQLLedgerOutbox:
     """ledger_outbox-backed LedgerOutbox. correlation_id PK is the dedup gate."""
 
-    def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None:
+    def __init__(
+        self,
+        dsn: str,
+        *,
+        instance_key: str = "default",
+        claim_ttl_sec: float = 60.0,
+        worker_id: str | None = None,
+        clock=None,
+        psycopg_module: Any = None,
+    ) -> None:
         self._psycopg = _require_psycopg(psycopg_module)
         self._dsn = dsn
+        self._instance_key = instance_key
+        self._claim_ttl_sec = claim_ttl_sec
+        self._worker_id = worker_id or f"ledger-writer-{uuid4().hex}"
+        self._clock = clock or (lambda: datetime.now(tz=timezone.utc))
+
+    def _now(self) -> datetime:
+        return self._clock()
 
     def enqueue(self, trade: LedgerTrade) -> LedgerWorkItem:
         with self._psycopg.connect(self._dsn, autocommit=False) as conn:
@@ -582,8 +599,8 @@ class PostgreSQLLedgerOutbox:
                 cur.execute(
                     "INSERT INTO ledger_outbox "
                     "(correlation_id, asset_id, order_id, execution_id, side, "
-                    " qty, price, fee, tax, executed_at, venue, order_venue) "
-                    "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) "
+                    " qty, price, fee, tax, executed_at, venue, order_venue, instance_key) "
+                    "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) "
                     "ON CONFLICT (correlation_id) DO NOTHING",
                     (
                         trade.correlation_id,
@@ -598,6 +615,7 @@ class PostgreSQLLedgerOutbox:
                         trade.executed_at,
                         trade.venue,
                         trade.order_venue,
+                        self._instance_key,
                     ),
                 )
                 conn.commit()
@@ -611,12 +629,53 @@ class PostgreSQLLedgerOutbox:
                 cur.execute(
                     f"SELECT {_OUTBOX_COLUMNS} FROM ledger_outbox "
                     "WHERE status = 'pending' "
+                    "AND instance_key = %s "
                     "AND (next_retry_at IS NULL OR next_retry_at <= now()) "
-                    "ORDER BY created_at LIMIT %s",
-                    (limit,),
+                    "AND (claim_expires_at IS NULL OR claim_expires_at <= now()) "
+                    "ORDER BY created_at, correlation_id LIMIT %s",
+                    (self._instance_key, limit),
                 )
                 return [self._row_to_item(r) for r in cur.fetchall()]
 
+    def claim_pending(
+        self,
+        limit: int = 32,
+        *,
+        correlation_id: str | None = None,
+    ) -> list[LedgerWorkItem]:
+        now = self._now()
+        lease_until = now + timedelta(seconds=self._claim_ttl_sec)
+        correlation_sql = ""
+        params: list[Any] = [self._instance_key]
+        if correlation_id is not None:
+            correlation_sql = "AND correlation_id = %s "
+            params.append(correlation_id)
+        params.extend([limit, self._worker_id, lease_until, now])
+        with self._psycopg.connect(self._dsn, autocommit=False) as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    "WITH candidates AS ("
+                    "SELECT correlation_id FROM ledger_outbox "
+                    "WHERE status = 'pending' "
+                    "AND instance_key = %s "
+                    "AND (next_retry_at IS NULL OR next_retry_at <= now()) "
+                    "AND (claim_expires_at IS NULL OR claim_expires_at <= now()) "
+                    f"{correlation_sql}"
+                    "ORDER BY created_at, correlation_id "
+                    "LIMIT %s "
+                    "FOR UPDATE SKIP LOCKED"
+                    ") "
+                    "UPDATE ledger_outbox o "
+                    "SET claim_owner = %s, claim_expires_at = %s, updated_at = %s "
+                    "FROM candidates "
+                    "WHERE o.correlation_id = candidates.correlation_id "
+                    f"RETURNING {_OUTBOX_COLUMNS}",
+                    params,
+                )
+                rows = [self._row_to_item(r) for r in cur.fetchall()]
+                conn.commit()
+                return rows
+
     def pending_cash_delta(self) -> dict[str, float]:
         # dcy-018 / REQ-029 (OQ-4): estimated cash delta of all pending rows by
         # asset_id. SUM in SQL: sell → +(qty*price - fee - tax), buy → negative.
@@ -627,7 +686,10 @@ class PostgreSQLLedgerOutbox:
                     "SUM(CASE WHEN side='sell' "
                     "         THEN (qty*price - fee - tax) "
                     "         ELSE -(qty*price + fee + tax) END) "
-                    "FROM ledger_outbox WHERE status='pending' GROUP BY asset_id"
+                    "FROM ledger_outbox "
+                    "WHERE status='pending' AND instance_key=%s "
+                    "GROUP BY asset_id",
+                    (self._instance_key,),
                 )
                 return {r[0]: float(r[1]) for r in cur.fetchall() if r[1] is not None}
 
@@ -671,16 +733,20 @@ class PostgreSQLLedgerOutbox:
                 nra = next_retry_at if outcome is OutboxOutcome.RETRYABLE_FAILURE else None
                 cur.execute(
                     "UPDATE ledger_outbox SET status=%s, attempts=attempts+%s, "
-                    "last_outcome=%s, last_error=%s, next_retry_at=%s, updated_at=%s "
-                    "WHERE correlation_id=%s",
+                    "last_outcome=%s, last_error=%s, next_retry_at=%s, updated_at=%s, "
+                    "claim_owner=NULL, claim_expires_at=NULL "
+                    "WHERE correlation_id=%s AND instance_key=%s "
+                    "AND (claim_owner=%s OR claim_owner IS NULL)",
                     (
                         status_value,
                         attempts_delta,
                         outcome.value,
                         error,
                         nra,
-                        datetime.now(tz=timezone.utc),
+                        self._now(),
                         correlation_id,
+                        self._instance_key,
+                        self._worker_id,
                     ),
                 )
                 conn.commit()
@@ -714,6 +780,7 @@ class PostgreSQLLedgerOutbox:
             created_at=row[14],
             updated_at=row[15],
             next_retry_at=row[16],
+            instance_key=row[19],
         )
 
 
diff --git a/trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql b/trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql
new file mode 100644
index 0000000..186c809
--- /dev/null
+++ b/trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql
@@ -0,0 +1,22 @@
+-- GitHub #84: persist outbox owner instance and add claim/lease fields so
+-- shared-DB daemon writers only deliver their own rows and can safely recover
+-- abandoned work after lease expiry. Existing rows remain fail-closed when the
+-- owner instance cannot be inferred.
+ALTER TABLE ledger_outbox ADD COLUMN IF NOT EXISTS instance_key TEXT;
+ALTER TABLE ledger_outbox ADD COLUMN IF NOT EXISTS claim_owner TEXT;
+ALTER TABLE ledger_outbox ADD COLUMN IF NOT EXISTS claim_expires_at TIMESTAMPTZ;
+
+UPDATE ledger_outbox
+SET instance_key='__unowned__'
+WHERE instance_key IS NULL;
+
+ALTER TABLE ledger_outbox ALTER COLUMN instance_key SET DEFAULT '__unowned__';
+ALTER TABLE ledger_outbox ALTER COLUMN instance_key SET NOT NULL;
+
+CREATE INDEX IF NOT EXISTS idx_ledger_outbox_instance_pending
+    ON ledger_outbox (instance_key, status, created_at)
+    WHERE status = 'pending';
+
+CREATE INDEX IF NOT EXISTS idx_ledger_outbox_instance_claim_expires
+    ON ledger_outbox (instance_key, claim_expires_at)
+    WHERE status = 'pending';
diff --git a/trex_core/application/ledger_outbox.py b/trex_core/application/ledger_outbox.py
index 574fe87..79e7b32 100644
--- a/trex_core/application/ledger_outbox.py
+++ b/trex_core/application/ledger_outbox.py
@@ -104,6 +104,7 @@ class LedgerWorkItem:
     next_retry_at: Optional[datetime] = None
     created_at: Optional[datetime] = None
     updated_at: Optional[datetime] = None
+    instance_key: Optional[str] = None
 
 
 class LedgerOutbox(Protocol):
@@ -118,6 +119,15 @@ class LedgerOutbox(Protocol):
         """Return pending work items in FIFO order, up to `limit`."""
         ...
 
+    def claim_pending(
+        self,
+        limit: int = 32,
+        *,
+        correlation_id: str | None = None,
+    ) -> list[LedgerWorkItem]:
+        """Atomically claim pending items for delivery."""
+        ...
+
     def pending_cash_delta(self) -> dict[str, float]:
         """Sum the estimated cash delta of ALL not-yet-settled (pending) rows,
         grouped by asset_id (dcy-018 / REQ-029, OQ-4 estimate overlay).
@@ -191,13 +201,27 @@ class LedgerWriter:
         self._alarm_emitter = alarm_emitter
         self._ledger_skip_store = ledger_skip_store
         self._instance_key = instance_key
+        self.counters: dict[str, int] = {}
+
+    def _count(self, key: str) -> None:
+        self.counters[key] = self.counters.get(key, 0) + 1
 
     def enqueue_fill(self, trade: LedgerTrade) -> LedgerWorkItem:
         return self._outbox.enqueue(trade)
 
     def run_once(self, limit: int = 32) -> WriterStats:
         stats = WriterStats()
-        for item in self._outbox.list_pending(limit):
+        items = self._outbox.claim_pending(limit)
+        claim_result = "claimed" if items else "empty"
+        self._count(f"claim:{claim_result}")
+        _log.info(
+            "ledger outbox claim: instance_key=%s claim_result=%s count=%d limit=%d",
+            self._instance_key,
+            claim_result,
+            len(items),
+            limit,
+        )
+        for item in items:
             outcome = self._process(item)
             if outcome is OutboxOutcome.COMPLETED:
                 stats.completed += 1
@@ -216,6 +240,32 @@ class LedgerWriter:
         item = self._outbox.get(correlation_id)
         if item is None:
             return None
+        if (
+            self._instance_key is not None
+            and item.instance_key is not None
+            and item.instance_key != self._instance_key
+        ):
+            self._count("claim:foreign_instance")
+            _log.info(
+                "ledger outbox claim: instance_key=%s claim_result=foreign_instance "
+                "correlation_id=%s owner_instance_key=%s",
+                self._instance_key,
+                correlation_id,
+                item.instance_key,
+            )
+            return None
+        if item.status is WorkItemStatus.PENDING:
+            claimed = self._outbox.claim_pending(1, correlation_id=correlation_id)
+            if not claimed:
+                self._count("claim:unavailable")
+                _log.info(
+                    "ledger outbox claim: instance_key=%s claim_result=unavailable "
+                    "correlation_id=%s",
+                    self._instance_key,
+                    correlation_id,
+                )
+                return None
+            item = claimed[0]
         return self._process(item)
 
     def _backoff_delay(self, attempts: int) -> float:
@@ -229,6 +279,7 @@ class LedgerWriter:
         existing = self._outbox.get(item.correlation_id)
         if existing is not None and existing.status is WorkItemStatus.COMPLETED:
             self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.DUPLICATE_COMPLETED)
+            self._log_outcome(item, OutboxOutcome.DUPLICATE_COMPLETED)
             return OutboxOutcome.DUPLICATE_COMPLETED
         try:
             outcome = self._ledger.record_trade(item.trade)
@@ -236,6 +287,7 @@ class LedgerWriter:
             self._outbox.mark_outcome(
                 item.correlation_id, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc)
             )
+            self._log_outcome(item, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc))
             # dcy-033 / REQ-100: CRITICAL alarm — 409 conflict, operator resolution needed.
             if self._alarm_emitter is not None:
                 self._alarm_emitter.publish_alarm(
@@ -252,8 +304,10 @@ class LedgerWriter:
             self._outbox.mark_outcome(
                 item.correlation_id, OutboxOutcome.PERMANENT_FAILURE, error=str(exc)
             )
+            self._log_outcome(item, OutboxOutcome.PERMANENT_FAILURE, error=str(exc))
             _log.warning(
-                "ledger delivery_failed (permanent): correlation_id=%s reason=%s",
+                "ledger delivery_failed (permanent): instance_key=%s correlation_id=%s reason=%s",
+                item.instance_key or self._instance_key,
                 item.correlation_id, exc,
             )
             # dcy-033 / REQ-100: CRITICAL alarm — permanent failure, no retry.
@@ -273,10 +327,14 @@ class LedgerWriter:
                 self._outbox.mark_outcome(
                     item.correlation_id, OutboxOutcome.DEAD_LETTERED, error=str(exc)
                 )
+                self._log_outcome(item, OutboxOutcome.DEAD_LETTERED, error=str(exc))
                 _log.warning(
                     "ledger delivery_failed (dead-letter, attempts=%d): "
-                    "correlation_id=%s reason=%s",
-                    new_attempts, item.correlation_id, exc,
+                    "instance_key=%s correlation_id=%s reason=%s",
+                    new_attempts,
+                    item.instance_key or self._instance_key,
+                    item.correlation_id,
+                    exc,
                 )
                 # dcy-033 / REQ-100: CRITICAL alarm — dead-lettered after max retries.
                 if self._alarm_emitter is not None:
@@ -296,6 +354,7 @@ class LedgerWriter:
                 item.correlation_id, OutboxOutcome.RETRYABLE_FAILURE,
                 error=str(exc), next_retry_at=next_retry_at,
             )
+            self._log_outcome(item, OutboxOutcome.RETRYABLE_FAILURE, error=str(exc))
             return OutboxOutcome.RETRYABLE_FAILURE
         # dcy-018 / REQ-019/030 (option D): write back portfolio-hub authoritative
         # cash/quantity to the snapshot store on new+duplicate (OQ-2 b). *_after
@@ -308,11 +367,32 @@ class LedgerWriter:
         if outcome.result is LedgerWriteResult.DUPLICATE_OK:
             self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.DUPLICATE_COMPLETED)
             self._resolve_ledger_skip(item.trade)
+            self._log_outcome(item, OutboxOutcome.DUPLICATE_COMPLETED)
             return OutboxOutcome.DUPLICATE_COMPLETED
         self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.COMPLETED)
         self._resolve_ledger_skip(item.trade)
+        self._log_outcome(item, OutboxOutcome.COMPLETED)
         return OutboxOutcome.COMPLETED
 
+    def _log_outcome(
+        self,
+        item: LedgerWorkItem,
+        outcome: OutboxOutcome,
+        *,
+        error: str | None = None,
+    ) -> None:
+        self._count(f"outcome:{outcome.value}")
+        msg = "ledger outbox outcome: instance_key=%s correlation_id=%s outcome=%s"
+        args: tuple[object, ...] = (
+            item.instance_key or self._instance_key,
+            item.correlation_id,
+            outcome.value,
+        )
+        if error is not None:
+            msg += " error=%s"
+            args = (*args, error)
+        _log.info(msg, *args)
+
     def _resolve_ledger_skip(self, trade: LedgerTrade) -> None:
         if (
             self._ledger_skip_store is None
@@ -358,7 +438,12 @@ class LedgerWriter:
 class InMemoryLedgerOutbox:
     """Process-local outbox. Thread-safe; idempotent on correlation_id."""
 
-    def __init__(self, *, clock: Optional[Callable[[], datetime]] = None) -> None:
+    def __init__(
+        self,
+        *,
+        clock: Optional[Callable[[], datetime]] = None,
+        instance_key: str = "default",
+    ) -> None:
         import threading
 
         self._rows: dict[str, LedgerWorkItem] = {}
@@ -367,6 +452,7 @@ class InMemoryLedgerOutbox:
         # dcy-015: shared clock lets tests align backoff (next_retry_at) with
         # list_pending's "now" deterministically.
         self._clock = clock or (lambda: datetime.now(tz=timezone.utc))
+        self._instance_key = instance_key
 
     def _now(self) -> datetime:
         return self._clock()
@@ -383,6 +469,7 @@ class InMemoryLedgerOutbox:
                 status=WorkItemStatus.PENDING,
                 created_at=self._now(),
                 updated_at=self._now(),
+                instance_key=self._instance_key,
             )
             self._rows[cid] = row
             self._order.append(cid)
@@ -404,6 +491,17 @@ class InMemoryLedgerOutbox:
                         break
             return out
 
+    def claim_pending(
+        self,
+        limit: int = 32,
+        *,
+        correlation_id: str | None = None,
+    ) -> list[LedgerWorkItem]:
+        pending = self.list_pending(limit if correlation_id is None else max(limit, 1))
+        if correlation_id is None:
+            return pending[:limit]
+        return [row for row in pending if row.correlation_id == correlation_id][:1]
+
     def pending_cash_delta(self) -> dict[str, float]:
         with self._lock:
             out: dict[str, float] = {}
diff --git a/trex_core/runtime/cli.py b/trex_core/runtime/cli.py
index 128ef21..182eeea 100644
--- a/trex_core/runtime/cli.py
+++ b/trex_core/runtime/cli.py
@@ -1333,7 +1333,7 @@ def _cmd_ledger_backfill(args, cfg, confirm) -> int:
             _emit_pretty_json(report)
             return 1
 
-    outbox = PostgreSQLLedgerOutbox(cfg.dsn)
+    outbox = PostgreSQLLedgerOutbox(cfg.dsn, instance_key=instance)
     if not cfg.portfolio_hub or not cfg.portfolio_hub.base_url:
         report["result"] = {"status": "rejected", "reason": "portfolio-hub ledger adapter is not configured"}
         report["audit_path"] = _write_ledger_backfill_audit(args, report)
diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py
index d232973..95bcd66 100644
--- a/trex_core/runtime/daemon.py
+++ b/trex_core/runtime/daemon.py
@@ -190,7 +190,7 @@ class Daemon:
             instance_key=config.instance_key,
         )
 
-        self.outbox = PostgreSQLLedgerOutbox(config.dsn)
+        self.outbox = PostgreSQLLedgerOutbox(config.dsn, instance_key=config.instance_key)
         self.ledger_skip_store = PostgreSQLLedgerSkipStore(config.dsn)
         # BAT-008: portfolio-hub read + snapshot store + durable pause + scheduler.
         # (Created before the ledger writer so option-D write-back can target it.)

From 31d6e5c02b50799b01ed5ff950d53c8b3c70354b Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Sat, 12 Sep 2026 13:27:27 +0000
Subject: [PATCH 3/6] test: cover stale ledger outbox lease updates

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
---
 .../test_postgresql_ledger_outbox.py          | 39 +++++++++++++++++++
 tests/application/test_ledger_outbox.py       |  6 ++-
 trex_core/adapters/persistence/postgresql.py  | 37 +++++++++++-------
 .../sql/028_ledger_outbox_instance_claims.sql |  3 +-
 trex_core/application/ledger_outbox.py        | 36 ++++++++++++++---
 5 files changed, 99 insertions(+), 22 deletions(-)

diff --git a/tests/adapters/persistence/test_postgresql_ledger_outbox.py b/tests/adapters/persistence/test_postgresql_ledger_outbox.py
index 6f51388..d553846 100644
--- a/tests/adapters/persistence/test_postgresql_ledger_outbox.py
+++ b/tests/adapters/persistence/test_postgresql_ledger_outbox.py
@@ -169,6 +169,45 @@ def test_claim_expiry_allows_safe_recovery_after_worker_crash(pg_clean):
     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,
+    )
+
+    stored = 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,
+    )
+
+    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):
diff --git a/tests/application/test_ledger_outbox.py b/tests/application/test_ledger_outbox.py
index e085f30..25e747d 100644
--- a/tests/application/test_ledger_outbox.py
+++ b/tests/application/test_ledger_outbox.py
@@ -71,9 +71,9 @@ 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"))
 
@@ -81,6 +81,8 @@ def test_process_correlation_id_only_drains_requested_row():
 
     assert outcome is OutboxOutcome.COMPLETED
     assert ledger.count() == 1
+    assert writer.counters["claim:claimed"] == 1
+    assert writer.counters["outcome:completed"] == 1
     assert outbox.get(first.correlation_id).status is WorkItemStatus.COMPLETED
     assert outbox.get(second.correlation_id).status is WorkItemStatus.PENDING
 
diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py
index 64cabc7..aa5b0f0 100644
--- a/trex_core/adapters/persistence/postgresql.py
+++ b/trex_core/adapters/persistence/postgresql.py
@@ -566,7 +566,8 @@ class _PgIntentRecord:
 _OUTBOX_COLUMNS = (
     "correlation_id, asset_id, order_id, execution_id, side, qty, price, "
     "fee, tax, executed_at, status, attempts, last_outcome, last_error, "
-    "created_at, updated_at, next_retry_at, venue, order_venue, instance_key"
+    "created_at, updated_at, next_retry_at, venue, order_venue, instance_key, "
+    "claim_expires_at"
 )
 
 
@@ -710,6 +711,7 @@ class PostgreSQLLedgerOutbox:
         *,
         error: Optional[str] = None,
         next_retry_at: Optional[datetime] = None,
+        expected_claim_expires_at: datetime | None = None,
     ) -> LedgerWorkItem:
         with self._psycopg.connect(self._dsn, autocommit=False) as conn:
             with conn.cursor() as cur:
@@ -731,23 +733,29 @@ class PostgreSQLLedgerOutbox:
                     attempts_delta = 1
                 # next_retry_at only persists for a retryable backoff; cleared otherwise.
                 nra = next_retry_at if outcome is OutboxOutcome.RETRYABLE_FAILURE else None
+                params: list[Any] = [
+                    status_value,
+                    attempts_delta,
+                    outcome.value,
+                    error,
+                    nra,
+                    self._now(),
+                    correlation_id,
+                    self._instance_key,
+                ]
+                where = "WHERE correlation_id=%s AND instance_key=%s "
+                if expected_claim_expires_at is None:
+                    where += "AND status=%s AND claim_owner IS NULL AND claim_expires_at IS NULL"
+                    params.append(WorkItemStatus.COMPLETED.value)
+                else:
+                    where += "AND claim_owner=%s AND claim_expires_at=%s"
+                    params.extend([self._worker_id, expected_claim_expires_at])
                 cur.execute(
                     "UPDATE ledger_outbox SET status=%s, attempts=attempts+%s, "
                     "last_outcome=%s, last_error=%s, next_retry_at=%s, updated_at=%s, "
                     "claim_owner=NULL, claim_expires_at=NULL "
-                    "WHERE correlation_id=%s AND instance_key=%s "
-                    "AND (claim_owner=%s OR claim_owner IS NULL)",
-                    (
-                        status_value,
-                        attempts_delta,
-                        outcome.value,
-                        error,
-                        nra,
-                        self._now(),
-                        correlation_id,
-                        self._instance_key,
-                        self._worker_id,
-                    ),
+                    f"{where}",
+                    params,
                 )
                 conn.commit()
         row = self.get(correlation_id)
@@ -781,6 +789,7 @@ class PostgreSQLLedgerOutbox:
             updated_at=row[15],
             next_retry_at=row[16],
             instance_key=row[19],
+            claim_expires_at=row[20],
         )
 
 
diff --git a/trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql b/trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql
index 186c809..1c037c7 100644
--- a/trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql
+++ b/trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql
@@ -13,8 +13,9 @@ WHERE instance_key IS NULL;
 ALTER TABLE ledger_outbox ALTER COLUMN instance_key SET DEFAULT '__unowned__';
 ALTER TABLE ledger_outbox ALTER COLUMN instance_key SET NOT NULL;
 
+DROP INDEX IF EXISTS idx_ledger_outbox_instance_pending;
 CREATE INDEX IF NOT EXISTS idx_ledger_outbox_instance_pending
-    ON ledger_outbox (instance_key, status, created_at)
+    ON ledger_outbox (instance_key, status, created_at, correlation_id)
     WHERE status = 'pending';
 
 CREATE INDEX IF NOT EXISTS idx_ledger_outbox_instance_claim_expires
diff --git a/trex_core/application/ledger_outbox.py b/trex_core/application/ledger_outbox.py
index 79e7b32..e0bdd5c 100644
--- a/trex_core/application/ledger_outbox.py
+++ b/trex_core/application/ledger_outbox.py
@@ -105,6 +105,7 @@ class LedgerWorkItem:
     created_at: Optional[datetime] = None
     updated_at: Optional[datetime] = None
     instance_key: Optional[str] = None
+    claim_expires_at: Optional[datetime] = None
 
 
 class LedgerOutbox(Protocol):
@@ -146,6 +147,7 @@ class LedgerOutbox(Protocol):
         *,
         error: Optional[str] = None,
         next_retry_at: Optional[datetime] = None,
+        expected_claim_expires_at: Optional[datetime] = None,
     ) -> LedgerWorkItem:
         """Record the outcome of a writer attempt and advance status.
 
@@ -266,6 +268,14 @@ class LedgerWriter:
                 )
                 return None
             item = claimed[0]
+            self._count("claim:claimed")
+            _log.info(
+                "ledger outbox claim: instance_key=%s claim_result=claimed "
+                "count=%d correlation_id=%s",
+                self._instance_key,
+                len(claimed),
+                correlation_id,
+            )
         return self._process(item)
 
     def _backoff_delay(self, attempts: int) -> float:
@@ -285,7 +295,10 @@ class LedgerWriter:
             outcome = self._ledger.record_trade(item.trade)
         except LedgerConflictError as exc:
             self._outbox.mark_outcome(
-                item.correlation_id, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc)
+                item.correlation_id,
+                OutboxOutcome.CONFLICT_BLOCKED,
+                error=str(exc),
+                expected_claim_expires_at=item.claim_expires_at,
             )
             self._log_outcome(item, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc))
             # dcy-033 / REQ-100: CRITICAL alarm — 409 conflict, operator resolution needed.
@@ -302,7 +315,10 @@ class LedgerWriter:
             # dcy-015 / REQ-017/018 (issue #13): non-retryable client error →
             # terminal delivery_failed immediately (no retry spam). Surface for ops.
             self._outbox.mark_outcome(
-                item.correlation_id, OutboxOutcome.PERMANENT_FAILURE, error=str(exc)
+                item.correlation_id,
+                OutboxOutcome.PERMANENT_FAILURE,
+                error=str(exc),
+                expected_claim_expires_at=item.claim_expires_at,
             )
             self._log_outcome(item, OutboxOutcome.PERMANENT_FAILURE, error=str(exc))
             _log.warning(
@@ -325,7 +341,10 @@ class LedgerWriter:
             new_attempts = item.attempts + 1
             if new_attempts >= self._max_attempts:
                 self._outbox.mark_outcome(
-                    item.correlation_id, OutboxOutcome.DEAD_LETTERED, error=str(exc)
+                    item.correlation_id,
+                    OutboxOutcome.DEAD_LETTERED,
+                    error=str(exc),
+                    expected_claim_expires_at=item.claim_expires_at,
                 )
                 self._log_outcome(item, OutboxOutcome.DEAD_LETTERED, error=str(exc))
                 _log.warning(
@@ -352,7 +371,9 @@ class LedgerWriter:
             )
             self._outbox.mark_outcome(
                 item.correlation_id, OutboxOutcome.RETRYABLE_FAILURE,
-                error=str(exc), next_retry_at=next_retry_at,
+                error=str(exc),
+                next_retry_at=next_retry_at,
+                expected_claim_expires_at=item.claim_expires_at,
             )
             self._log_outcome(item, OutboxOutcome.RETRYABLE_FAILURE, error=str(exc))
             return OutboxOutcome.RETRYABLE_FAILURE
@@ -369,7 +390,11 @@ class LedgerWriter:
             self._resolve_ledger_skip(item.trade)
             self._log_outcome(item, OutboxOutcome.DUPLICATE_COMPLETED)
             return OutboxOutcome.DUPLICATE_COMPLETED
-        self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.COMPLETED)
+        self._outbox.mark_outcome(
+            item.correlation_id,
+            OutboxOutcome.COMPLETED,
+            expected_claim_expires_at=item.claim_expires_at,
+        )
         self._resolve_ledger_skip(item.trade)
         self._log_outcome(item, OutboxOutcome.COMPLETED)
         return OutboxOutcome.COMPLETED
@@ -522,6 +547,7 @@ class InMemoryLedgerOutbox:
         *,
         error: Optional[str] = None,
         next_retry_at: Optional[datetime] = None,
+        expected_claim_expires_at: Optional[datetime] = None,
     ) -> LedgerWorkItem:
         with self._lock:
             row = self._rows[correlation_id]

From 0b5b4660e57c17a0f584973fd850234b171b6bdd Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Sat, 12 Sep 2026 13:29:05 +0000
Subject: [PATCH 4/6] fix: guard ledger outbox outcomes by lease

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
---
 trex_core/adapters/persistence/postgresql.py | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py
index aa5b0f0..5864957 100644
--- a/trex_core/adapters/persistence/postgresql.py
+++ b/trex_core/adapters/persistence/postgresql.py
@@ -745,8 +745,7 @@ class PostgreSQLLedgerOutbox:
                 ]
                 where = "WHERE correlation_id=%s AND instance_key=%s "
                 if expected_claim_expires_at is None:
-                    where += "AND status=%s AND claim_owner IS NULL AND claim_expires_at IS NULL"
-                    params.append(WorkItemStatus.COMPLETED.value)
+                    where += "AND claim_owner IS NULL AND claim_expires_at IS NULL"
                 else:
                     where += "AND claim_owner=%s AND claim_expires_at=%s"
                     params.extend([self._worker_id, expected_claim_expires_at])

From f801c53e0c30edb31250e536e6ac0e8d8d5c861e Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Sat, 12 Sep 2026 13:30:50 +0000
Subject: [PATCH 5/6] test: cover targeted and expired ledger claims

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
---
 .../test_postgresql_ledger_outbox.py          | 23 +++++++++++++++++++
 tests/application/test_ledger_outbox.py       |  6 ++---
 trex_core/adapters/persistence/postgresql.py  |  7 ++++--
 trex_core/application/ledger_outbox.py        | 12 +++++++---
 4 files changed, 40 insertions(+), 8 deletions(-)

diff --git a/tests/adapters/persistence/test_postgresql_ledger_outbox.py b/tests/adapters/persistence/test_postgresql_ledger_outbox.py
index d553846..d9d465d 100644
--- a/tests/adapters/persistence/test_postgresql_ledger_outbox.py
+++ b/tests/adapters/persistence/test_postgresql_ledger_outbox.py
@@ -208,6 +208,29 @@ def test_stale_claim_cannot_overwrite_new_owner_outcome(pg_clean):
     assert stored.last_outcome is OutboxOutcome.COMPLETED
 
 
+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):
diff --git a/tests/application/test_ledger_outbox.py b/tests/application/test_ledger_outbox.py
index 25e747d..651efc6 100644
--- a/tests/application/test_ledger_outbox.py
+++ b/tests/application/test_ledger_outbox.py
@@ -77,14 +77,14 @@ def test_process_correlation_id_only_drains_requested_row():
     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 writer.counters["claim:claimed"] == 1
     assert writer.counters["outcome:completed"] == 1
-    assert outbox.get(first.correlation_id).status is WorkItemStatus.COMPLETED
-    assert outbox.get(second.correlation_id).status is WorkItemStatus.PENDING
+    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():
diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py
index 5864957..3812448 100644
--- a/trex_core/adapters/persistence/postgresql.py
+++ b/trex_core/adapters/persistence/postgresql.py
@@ -747,8 +747,11 @@ class PostgreSQLLedgerOutbox:
                 if expected_claim_expires_at is None:
                     where += "AND claim_owner IS NULL AND claim_expires_at IS NULL"
                 else:
-                    where += "AND claim_owner=%s AND claim_expires_at=%s"
-                    params.extend([self._worker_id, expected_claim_expires_at])
+                    where += (
+                        "AND claim_owner=%s "
+                        "AND (claim_expires_at=%s OR (status='pending' AND claim_expires_at<=%s))"
+                    )
+                    params.extend([self._worker_id, expected_claim_expires_at, self._now()])
                 cur.execute(
                     "UPDATE ledger_outbox SET status=%s, attempts=attempts+%s, "
                     "last_outcome=%s, last_error=%s, next_retry_at=%s, updated_at=%s, "
diff --git a/trex_core/application/ledger_outbox.py b/trex_core/application/ledger_outbox.py
index e0bdd5c..2dc7eac 100644
--- a/trex_core/application/ledger_outbox.py
+++ b/trex_core/application/ledger_outbox.py
@@ -522,10 +522,16 @@ class InMemoryLedgerOutbox:
         *,
         correlation_id: str | None = None,
     ) -> list[LedgerWorkItem]:
-        pending = self.list_pending(limit if correlation_id is None else max(limit, 1))
         if correlation_id is None:
-            return pending[:limit]
-        return [row for row in pending if row.correlation_id == correlation_id][:1]
+            return self.list_pending(limit)
+        now = self._now()
+        with self._lock:
+            row = self._rows.get(correlation_id)
+            if row is None or row.status is not WorkItemStatus.PENDING:
+                return []
+            if row.next_retry_at is not None and row.next_retry_at > now:
+                return []
+            return [row]
 
     def pending_cash_delta(self) -> dict[str, float]:
         with self._lock:

From 762a953c95574c8a2cd5e1a7665bd1a91655321d Mon Sep 17 00:00:00 2001
From: Hermes Agent <noreply@nousresearch.com>
Date: Sun, 13 Sep 2026 09:41:51 +0900
Subject: [PATCH 6/6] fix: fence ledger outbox claims by instance

---
 .proofline/criteria/ac-0029.md                |  21 +++
 .proofline/identities.json                    |   2 +-
 scripts/virtual_execution_final_state.py      |   2 +-
 .../persistence/test_migration_and_drift.py   |   4 +-
 .../test_postgresql_ledger_outbox.py          |  43 ++++-
 tests/application/test_execution_cancel.py    |   2 +-
 .../test_execution_fill_price_backfill.py     |   4 +-
 tests/application/test_execution_ledger.py    |   8 +-
 tests/application/test_ledger_option_d.py     |  16 +-
 tests/application/test_ledger_outbox.py       | 116 ++++++++++++--
 .../application/test_ledger_outbox_backoff.py |   2 +-
 .../test_messaging_event_publisher.py         |   4 +-
 trex_core/adapters/persistence/postgresql.py  |  31 +++-
 ... => 029_ledger_outbox_instance_claims.sql} |   0
 trex_core/application/ledger_outbox.py        | 149 +++++++++++++-----
 15 files changed, 316 insertions(+), 88 deletions(-)
 create mode 100644 .proofline/criteria/ac-0029.md
 rename trex_core/adapters/persistence/sql/{028_ledger_outbox_instance_claims.sql => 029_ledger_outbox_instance_claims.sql} (100%)

diff --git a/.proofline/criteria/ac-0029.md b/.proofline/criteria/ac-0029.md
new file mode 100644
index 0000000..04b4574
--- /dev/null
+++ b/.proofline/criteria/ac-0029.md
@@ -0,0 +1,21 @@
+---
+id: "ac-0029"
+status: active
+---
+
+# Ledger outbox instance ownership and lease fencing
+
+## Criterion
+
+`trex-core`의 durable ledger outbox row는 enqueue 시점의 required non-empty `instance_key`를 immutable owner로 저장해야 하며, daemon·operator backfill·targeted processing을 포함한 모든 delivery 경로는 자기 instance의 row만 claim하고 처리해야 한다. Shared PostgreSQL에서 claim은 pending·retry-due·unclaimed 또는 lease-expired row를 원자적으로 선택해 worker별 bounded lease를 설정해야 하고, 성공·중복·retryable·conflict·permanent·dead-letter outcome은 해당 claim의 fencing 정보가 일치할 때만 반영해야 한다. 다른 instance, legacy `instance_key IS NULL`, 아직 유효한 foreign lease 또는 이미 새 worker가 reclaim한 row는 처리·상태 변경·성공 통계·ledger-skip resolution 없이 fail closed 해야 한다.
+
+Claim owner가 crash하면 lease 만료 후 같은 instance의 다른 worker가 row를 reclaim할 수 있어야 한다. Outcome CAS가 0건이면 claim 상실로 명시적으로 처리하고 성공 outcome으로 보고해서는 안 된다. InMemory와 PostgreSQL outbox는 instance filtering, active lease exclusion, expiry recovery와 stale-claim rejection에서 같은 의미를 가져야 한다. Claim·outcome 관측 정보에는 `instance_key`, sanitized result/reason과 correlation identity만 허용하며 credential, account identifier, token, password, DSN, provider payload 또는 exception 원문을 로그·metric·alert에 포함해서는 안 된다.
+
+## Verification
+
+- 격리 조건: 서로 다른 `instance_key`의 daemon/writer 두 개가 한 PostgreSQL을 공유할 때 각 writer는 자기 row만 claim·전송·집계한다. `list_pending`, `pending_cash_delta`, batch claim과 correlation-targeted claim 모두 foreign row와 `instance_key IS NULL` legacy row를 제외한다. 누락·빈 instance key로 outbox 또는 writer를 생성하면 외부 연결이나 row 생성 전에 거부된다.
+- lease 조건: worker A가 claim한 row는 lease 동안 worker B가 claim하지 못한다. A가 outcome을 기록하면 lease가 해제되고 terminal 또는 retry 상태가 정확히 저장된다. A가 crash한 경우 expiry 전에는 B가 claim하지 못하고 expiry 뒤에는 B가 reclaim할 수 있다. Reclaim 뒤 A의 stale outcome은 typed claim-lost 결과로 거부되며 B의 row 상태, outcome, attempt count, next retry와 lease를 변경하지 않는다.
+- outcome 조건: adapter의 신규 성공과 `DUPLICATE_OK`, retryable, conflict, permanent 및 dead-letter 모든 분기는 claim fencing을 전달한다. CAS 실패 시 completion/retry counter, 성공 로그, snapshot write-back 이후의 ledger-skip resolution을 실행하지 않는다. Adapter duplicate는 row를 `completed/duplicate_completed`로 만들고 lease를 즉시 해제하여 재전송되지 않는다.
+- InMemory parity: 두 번의 동시 claim에서 첫 claim만 성공하고 active lease row는 pending scan에서 제외된다. clock을 이동하면 expiry recovery가 가능하며 이전 claim token으로 outcome을 기록할 수 없다. Instance filter와 targeted processing의 fail-closed 동작이 PostgreSQL과 같다.
+- 보안 조건: exception에 credential·account·token·password·DSN·provider payload를 식별하는 sentinel을 넣어 claim 실패, retry, permanent, dead-letter와 write-back 오류를 실행해도 rendered log, metric key/value와 alert에 sentinel 또는 exception 원문이 나타나지 않는다.
+- migration 및 회귀: migration 번호는 저장소에서 유일하고 기존 row를 자동 귀속시키지 않으며 `instance_key IS NULL`로 격리한다. 빈 DB와 기존 schema 모두에서 migration apply/replay, PostgreSQL claim round-trip, restart/expiry 회수, daemon·CLI wiring을 검증한다. 전체 repository test, Ruff, compileall, ProofLine validation과 `git diff --check`를 실행한다.
diff --git a/.proofline/identities.json b/.proofline/identities.json
index 34f4cf3..e90cfd1 100644
--- a/.proofline/identities.json
+++ b/.proofline/identities.json
@@ -1,5 +1,5 @@
 {
   "schema_version": 1,
   "next_line_number": 10,
-  "next_ac_number": 29
+  "next_ac_number": 30
 }
diff --git a/scripts/virtual_execution_final_state.py b/scripts/virtual_execution_final_state.py
index 80eba0f..f05ea30 100644
--- a/scripts/virtual_execution_final_state.py
+++ b/scripts/virtual_execution_final_state.py
@@ -32,7 +32,7 @@ def main() -> int:
     )
     repo.reconcile_universe(instance, {portfolio_id: symbol})
     broker = PaperBroker(fill_after_polls=0)
-    outbox = InMemoryLedgerOutbox()
+    outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     ledger = InMemoryLedger()
     execution = ExecutionService(
         slots,
diff --git a/tests/adapters/persistence/test_migration_and_drift.py b/tests/adapters/persistence/test_migration_and_drift.py
index 592cc60..71c5c21 100644
--- a/tests/adapters/persistence/test_migration_and_drift.py
+++ b/tests/adapters/persistence/test_migration_and_drift.py
@@ -118,7 +118,7 @@ def test_migration_replay_converts_existing_auto_venue_row_to_sor(pg_clean):
 
 
 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",
@@ -177,7 +177,7 @@ def test_migration_024_separates_legacy_sor_and_is_replay_safe(pg_clean):
             )
 
 
-def test_migration_028_fail_closes_unowned_legacy_outbox_rows(pg_clean):
+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",
diff --git a/tests/adapters/persistence/test_postgresql_ledger_outbox.py b/tests/adapters/persistence/test_postgresql_ledger_outbox.py
index d9d465d..eef6e87 100644
--- a/tests/adapters/persistence/test_postgresql_ledger_outbox.py
+++ b/tests/adapters/persistence/test_postgresql_ledger_outbox.py
@@ -4,12 +4,16 @@ Skipped without TREX_POSTGRESQL_TEST_DSN (same gate as other PG suites).
 """
 from __future__ import annotations
 
+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,
 )
@@ -196,18 +200,27 @@ def test_stale_claim_cannot_overwrite_new_owner_outcome(pg_clean):
         expected_claim_expires_at=claim_b.claim_expires_at,
     )
 
-    stored = 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,
-    )
+    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(
@@ -255,6 +268,22 @@ def test_writer_records_retryable_403_without_cross_instance_delivery(pg_clean):
     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):
diff --git a/tests/application/test_execution_cancel.py b/tests/application/test_execution_cancel.py
index a88fd8c..3fa65f9 100644
--- a/tests/application/test_execution_cancel.py
+++ b/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,
diff --git a/tests/application/test_execution_fill_price_backfill.py b/tests/application/test_execution_fill_price_backfill.py
index 6bb6ac7..2baf2d2 100644
--- a/tests/application/test_execution_fill_price_backfill.py
+++ b/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"})
 
diff --git a/tests/application/test_execution_ledger.py b/tests/application/test_execution_ledger.py
index d993c2c..e1aa009 100644
--- a/tests/application/test_execution_ledger.py
+++ b/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()
diff --git a/tests/application/test_ledger_option_d.py b/tests/application/test_ledger_option_d.py
index 8685773..e0e8769 100644
--- a/tests/application/test_ledger_option_d.py
+++ b/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))
diff --git a/tests/application/test_ledger_outbox.py b/tests/application/test_ledger_outbox.py
index 651efc6..64fc05c 100644
--- a/tests/application/test_ledger_outbox.py
+++ b/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"))
@@ -90,17 +96,29 @@ def test_process_correlation_id_only_drains_requested_row():
 def test_process_correlation_id_skips_foreign_instance_row():
     outbox = InMemoryLedgerOutbox(instance_key="inst-a")
     ledger = InMemoryLedger()
-    writer = LedgerWriter(outbox, ledger, instance_key="inst-b")
     item = outbox.enqueue(_trade(execution_id="e-foreign"))
 
-    outcome = writer.process_correlation_id(item.correlation_id)
-
-    assert outcome is None
+    with pytest.raises(ValueError, match="must match"):
+        LedgerWriter(outbox, ledger, instance_key="inst-b")
     assert ledger.count() == 0
-    assert writer.counters["claim:foreign_instance"] == 1
     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()
@@ -112,3 +130,81 @@ def test_writer_exposes_claim_and_outcome_counters():
     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
diff --git a/tests/application/test_ledger_outbox_backoff.py b/tests/application/test_ledger_outbox_backoff.py
index 5c5cd5d..577c2bd 100644
--- a/tests/application/test_ledger_outbox_backoff.py
+++ b/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
diff --git a/tests/application/test_messaging_event_publisher.py b/tests/application/test_messaging_event_publisher.py
index 03a3bd5..e612a1c 100644
--- a/tests/application/test_messaging_event_publisher.py
+++ b/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"})
diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py
index 3812448..8439926 100644
--- a/trex_core/adapters/persistence/postgresql.py
+++ b/trex_core/adapters/persistence/postgresql.py
@@ -21,6 +21,7 @@ from uuid import UUID, uuid4
 
 from trex_core.application.ledger_outbox import (
     LedgerWorkItem,
+    OutboxClaimLostError,
     OutboxOutcome,
     WorkItemStatus,
 )
@@ -569,6 +570,9 @@ _OUTBOX_COLUMNS = (
     "created_at, updated_at, next_retry_at, venue, order_venue, instance_key, "
     "claim_expires_at"
 )
+_OUTBOX_RETURN_COLUMNS = ", ".join(
+    f"o.{column.strip()}" for column in _OUTBOX_COLUMNS.split(",")
+)
 
 
 class PostgreSQLLedgerOutbox:
@@ -578,12 +582,16 @@ class PostgreSQLLedgerOutbox:
         self,
         dsn: str,
         *,
-        instance_key: str = "default",
+        instance_key: str,
         claim_ttl_sec: float = 60.0,
         worker_id: str | None = None,
         clock=None,
         psycopg_module: Any = None,
     ) -> None:
+        if not isinstance(instance_key, str) or not instance_key.strip():
+            raise ValueError("instance_key must be a non-empty string")
+        if claim_ttl_sec <= 0:
+            raise ValueError("claim_ttl_sec must be positive")
         self._psycopg = _require_psycopg(psycopg_module)
         self._dsn = dsn
         self._instance_key = instance_key
@@ -594,6 +602,10 @@ class PostgreSQLLedgerOutbox:
     def _now(self) -> datetime:
         return self._clock()
 
+    @property
+    def instance_key(self) -> str:
+        return self._instance_key
+
     def enqueue(self, trade: LedgerTrade) -> LedgerWorkItem:
         with self._psycopg.connect(self._dsn, autocommit=False) as conn:
             with conn.cursor() as cur:
@@ -625,16 +637,17 @@ class PostgreSQLLedgerOutbox:
         return existing
 
     def list_pending(self, limit: int = 32) -> list[LedgerWorkItem]:
+        now = self._now()
         with self._psycopg.connect(self._dsn) as conn:
             with conn.cursor() as cur:
                 cur.execute(
                     f"SELECT {_OUTBOX_COLUMNS} FROM ledger_outbox "
                     "WHERE status = 'pending' "
                     "AND instance_key = %s "
-                    "AND (next_retry_at IS NULL OR next_retry_at <= now()) "
-                    "AND (claim_expires_at IS NULL OR claim_expires_at <= now()) "
+                    "AND (next_retry_at IS NULL OR next_retry_at <= %s) "
+                    "AND (claim_expires_at IS NULL OR claim_expires_at <= %s) "
                     "ORDER BY created_at, correlation_id LIMIT %s",
-                    (self._instance_key, limit),
+                    (self._instance_key, now, now, limit),
                 )
                 return [self._row_to_item(r) for r in cur.fetchall()]
 
@@ -647,7 +660,7 @@ class PostgreSQLLedgerOutbox:
         now = self._now()
         lease_until = now + timedelta(seconds=self._claim_ttl_sec)
         correlation_sql = ""
-        params: list[Any] = [self._instance_key]
+        params: list[Any] = [self._instance_key, now, now]
         if correlation_id is not None:
             correlation_sql = "AND correlation_id = %s "
             params.append(correlation_id)
@@ -659,8 +672,8 @@ class PostgreSQLLedgerOutbox:
                     "SELECT correlation_id FROM ledger_outbox "
                     "WHERE status = 'pending' "
                     "AND instance_key = %s "
-                    "AND (next_retry_at IS NULL OR next_retry_at <= now()) "
-                    "AND (claim_expires_at IS NULL OR claim_expires_at <= now()) "
+                    "AND (next_retry_at IS NULL OR next_retry_at <= %s) "
+                    "AND (claim_expires_at IS NULL OR claim_expires_at <= %s) "
                     f"{correlation_sql}"
                     "ORDER BY created_at, correlation_id "
                     "LIMIT %s "
@@ -670,7 +683,7 @@ class PostgreSQLLedgerOutbox:
                     "SET claim_owner = %s, claim_expires_at = %s, updated_at = %s "
                     "FROM candidates "
                     "WHERE o.correlation_id = candidates.correlation_id "
-                    f"RETURNING {_OUTBOX_COLUMNS}",
+                    f"RETURNING {_OUTBOX_RETURN_COLUMNS}",
                     params,
                 )
                 rows = [self._row_to_item(r) for r in cur.fetchall()]
@@ -759,6 +772,8 @@ class PostgreSQLLedgerOutbox:
                     f"{where}",
                     params,
                 )
+                if cur.rowcount != 1:
+                    raise OutboxClaimLostError(correlation_id)
                 conn.commit()
         row = self.get(correlation_id)
         assert row is not None
diff --git a/trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql b/trex_core/adapters/persistence/sql/029_ledger_outbox_instance_claims.sql
similarity index 100%
rename from trex_core/adapters/persistence/sql/028_ledger_outbox_instance_claims.sql
rename to trex_core/adapters/persistence/sql/029_ledger_outbox_instance_claims.sql
diff --git a/trex_core/application/ledger_outbox.py b/trex_core/application/ledger_outbox.py
index 2dc7eac..cf043ed 100644
--- a/trex_core/application/ledger_outbox.py
+++ b/trex_core/application/ledger_outbox.py
@@ -66,6 +66,10 @@ class WorkItemStatus(str, Enum):
     DELIVERY_FAILED = "delivery_failed"    # dcy-015 — permanent or dead-lettered (terminal)
 
 
+class OutboxClaimLostError(RuntimeError):
+    """The caller no longer owns the lease required for an outcome update."""
+
+
 # Outcomes that move a row to the terminal delivery_failed state.
 _DELIVERY_FAILED_OUTCOMES = frozenset(
     {OutboxOutcome.PERMANENT_FAILURE, OutboxOutcome.DEAD_LETTERED}
@@ -111,6 +115,11 @@ class LedgerWorkItem:
 class LedgerOutbox(Protocol):
     """Durable work-item store backing the ledger writer."""
 
+    @property
+    def instance_key(self) -> str:
+        """Immutable owner partition for rows handled by this outbox."""
+        ...
+
     def enqueue(self, trade: LedgerTrade) -> LedgerWorkItem:
         """Enqueue a trade. Idempotent on correlation_id (returns the
         existing row unchanged if it is already present)."""
@@ -163,6 +172,7 @@ class WriterStats:
     retryable: int = 0
     conflicts: int = 0
     delivery_failed: int = 0   # dcy-015: permanent + dead-lettered terminals
+    claim_lost: int = 0
 
 
 class LedgerWriter:
@@ -202,7 +212,12 @@ class LedgerWriter:
         # dcy-033 / REQ-100: best-effort alarm emitter.
         self._alarm_emitter = alarm_emitter
         self._ledger_skip_store = ledger_skip_store
-        self._instance_key = instance_key
+        resolved_instance_key = outbox.instance_key if instance_key is None else instance_key
+        if not isinstance(resolved_instance_key, str) or not resolved_instance_key.strip():
+            raise ValueError("instance_key must be a non-empty string")
+        if resolved_instance_key != outbox.instance_key:
+            raise ValueError("writer instance_key must match outbox instance_key")
+        self._instance_key = resolved_instance_key
         self.counters: dict[str, int] = {}
 
     def _count(self, key: str) -> None:
@@ -224,7 +239,19 @@ class LedgerWriter:
             limit,
         )
         for item in items:
-            outcome = self._process(item)
+            if not self._validate_instance(item):
+                continue
+            try:
+                outcome = self._process(item)
+            except OutboxClaimLostError:
+                self._count("claim:lost")
+                stats.claim_lost += 1
+                _log.warning(
+                    "ledger outbox claim lost: instance_key=%s correlation_id=%s",
+                    self._instance_key,
+                    item.correlation_id,
+                )
+                continue
             if outcome is OutboxOutcome.COMPLETED:
                 stats.completed += 1
             elif outcome is OutboxOutcome.DUPLICATE_COMPLETED:
@@ -242,19 +269,7 @@ class LedgerWriter:
         item = self._outbox.get(correlation_id)
         if item is None:
             return None
-        if (
-            self._instance_key is not None
-            and item.instance_key is not None
-            and item.instance_key != self._instance_key
-        ):
-            self._count("claim:foreign_instance")
-            _log.info(
-                "ledger outbox claim: instance_key=%s claim_result=foreign_instance "
-                "correlation_id=%s owner_instance_key=%s",
-                self._instance_key,
-                correlation_id,
-                item.instance_key,
-            )
+        if not self._validate_instance(item):
             return None
         if item.status is WorkItemStatus.PENDING:
             claimed = self._outbox.claim_pending(1, correlation_id=correlation_id)
@@ -276,7 +291,28 @@ class LedgerWriter:
                 len(claimed),
                 correlation_id,
             )
-        return self._process(item)
+        try:
+            return self._process(item)
+        except OutboxClaimLostError:
+            self._count("claim:lost")
+            _log.warning(
+                "ledger outbox claim lost: instance_key=%s correlation_id=%s",
+                self._instance_key,
+                correlation_id,
+            )
+            return None
+
+    def _validate_instance(self, item: LedgerWorkItem) -> bool:
+        if item.instance_key == self._instance_key:
+            return True
+        self._count("claim:foreign_instance")
+        _log.warning(
+            "ledger outbox claim rejected: instance_key=%s "
+            "claim_result=foreign_or_unowned correlation_id=%s",
+            self._instance_key,
+            item.correlation_id,
+        )
+        return False
 
     def _backoff_delay(self, attempts: int) -> float:
         """Exponential backoff with jitter: min(base*2^(n-1), cap) + up to 25%."""
@@ -288,7 +324,6 @@ class LedgerWriter:
         # outbox row. Skip the adapter call.
         existing = self._outbox.get(item.correlation_id)
         if existing is not None and existing.status is WorkItemStatus.COMPLETED:
-            self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.DUPLICATE_COMPLETED)
             self._log_outcome(item, OutboxOutcome.DUPLICATE_COMPLETED)
             return OutboxOutcome.DUPLICATE_COMPLETED
         try:
@@ -300,7 +335,7 @@ class LedgerWriter:
                 error=str(exc),
                 expected_claim_expires_at=item.claim_expires_at,
             )
-            self._log_outcome(item, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc))
+            self._log_outcome(item, OutboxOutcome.CONFLICT_BLOCKED)
             # dcy-033 / REQ-100: CRITICAL alarm — 409 conflict, operator resolution needed.
             if self._alarm_emitter is not None:
                 self._alarm_emitter.publish_alarm(
@@ -320,11 +355,11 @@ class LedgerWriter:
                 error=str(exc),
                 expected_claim_expires_at=item.claim_expires_at,
             )
-            self._log_outcome(item, OutboxOutcome.PERMANENT_FAILURE, error=str(exc))
+            self._log_outcome(item, OutboxOutcome.PERMANENT_FAILURE)
             _log.warning(
-                "ledger delivery_failed (permanent): instance_key=%s correlation_id=%s reason=%s",
+                "ledger delivery_failed (permanent): instance_key=%s correlation_id=%s",
                 item.instance_key or self._instance_key,
-                item.correlation_id, exc,
+                item.correlation_id,
             )
             # dcy-033 / REQ-100: CRITICAL alarm — permanent failure, no retry.
             if self._alarm_emitter is not None:
@@ -346,14 +381,13 @@ class LedgerWriter:
                     error=str(exc),
                     expected_claim_expires_at=item.claim_expires_at,
                 )
-                self._log_outcome(item, OutboxOutcome.DEAD_LETTERED, error=str(exc))
+                self._log_outcome(item, OutboxOutcome.DEAD_LETTERED)
                 _log.warning(
                     "ledger delivery_failed (dead-letter, attempts=%d): "
-                    "instance_key=%s correlation_id=%s reason=%s",
+                    "instance_key=%s correlation_id=%s",
                     new_attempts,
                     item.instance_key or self._instance_key,
                     item.correlation_id,
-                    exc,
                 )
                 # dcy-033 / REQ-100: CRITICAL alarm — dead-lettered after max retries.
                 if self._alarm_emitter is not None:
@@ -375,7 +409,7 @@ class LedgerWriter:
                 next_retry_at=next_retry_at,
                 expected_claim_expires_at=item.claim_expires_at,
             )
-            self._log_outcome(item, OutboxOutcome.RETRYABLE_FAILURE, error=str(exc))
+            self._log_outcome(item, OutboxOutcome.RETRYABLE_FAILURE)
             return OutboxOutcome.RETRYABLE_FAILURE
         # dcy-018 / REQ-019/030 (option D): write back portfolio-hub authoritative
         # cash/quantity to the snapshot store on new+duplicate (OQ-2 b). *_after
@@ -386,7 +420,11 @@ class LedgerWriter:
         # duplicate_completed; the ledger has exactly one entry for this
         # execution_id either way (REQ-019 AC-1, AC-4).
         if outcome.result is LedgerWriteResult.DUPLICATE_OK:
-            self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.DUPLICATE_COMPLETED)
+            self._outbox.mark_outcome(
+                item.correlation_id,
+                OutboxOutcome.DUPLICATE_COMPLETED,
+                expected_claim_expires_at=item.claim_expires_at,
+            )
             self._resolve_ledger_skip(item.trade)
             self._log_outcome(item, OutboxOutcome.DUPLICATE_COMPLETED)
             return OutboxOutcome.DUPLICATE_COMPLETED
@@ -403,8 +441,6 @@ class LedgerWriter:
         self,
         item: LedgerWorkItem,
         outcome: OutboxOutcome,
-        *,
-        error: str | None = None,
     ) -> None:
         self._count(f"outcome:{outcome.value}")
         msg = "ledger outbox outcome: instance_key=%s correlation_id=%s outcome=%s"
@@ -413,9 +449,6 @@ class LedgerWriter:
             item.correlation_id,
             outcome.value,
         )
-        if error is not None:
-            msg += " error=%s"
-            args = (*args, error)
         _log.info(msg, *args)
 
     def _resolve_ledger_skip(self, trade: LedgerTrade) -> None:
@@ -454,7 +487,7 @@ class LedgerWriter:
         except Exception:  # noqa: BLE001 - write-back must not block the ledger path
             _log.warning(
                 "snapshot cash/position write-back failed: asset_id=%s",
-                trade.asset_id, exc_info=True,
+                trade.asset_id,
             )
 
 
@@ -467,7 +500,8 @@ class InMemoryLedgerOutbox:
         self,
         *,
         clock: Optional[Callable[[], datetime]] = None,
-        instance_key: str = "default",
+        instance_key: str,
+        claim_ttl_sec: float = 60.0,
     ) -> None:
         import threading
 
@@ -477,7 +511,16 @@ class InMemoryLedgerOutbox:
         # dcy-015: shared clock lets tests align backoff (next_retry_at) with
         # list_pending's "now" deterministically.
         self._clock = clock or (lambda: datetime.now(tz=timezone.utc))
+        if not isinstance(instance_key, str) or not instance_key.strip():
+            raise ValueError("instance_key must be a non-empty string")
+        if claim_ttl_sec <= 0:
+            raise ValueError("claim_ttl_sec must be positive")
         self._instance_key = instance_key
+        self._claim_ttl_sec = claim_ttl_sec
+
+    @property
+    def instance_key(self) -> str:
+        return self._instance_key
 
     def _now(self) -> datetime:
         return self._clock()
@@ -508,8 +551,11 @@ class InMemoryLedgerOutbox:
                 row = self._rows[cid]
                 # dcy-015 / REQ-019: skip rows still in backoff so they don't
                 # occupy the FIFO batch (head-of-line protection, NFR-3).
-                if row.status is WorkItemStatus.PENDING and (
-                    row.next_retry_at is None or row.next_retry_at <= now
+                if (
+                    row.status is WorkItemStatus.PENDING
+                    and row.instance_key == self._instance_key
+                    and (row.next_retry_at is None or row.next_retry_at <= now)
+                    and (row.claim_expires_at is None or row.claim_expires_at <= now)
                 ):
                     out.append(row)
                     if len(out) >= limit:
@@ -522,16 +568,27 @@ class InMemoryLedgerOutbox:
         *,
         correlation_id: str | None = None,
     ) -> list[LedgerWorkItem]:
-        if correlation_id is None:
-            return self.list_pending(limit)
         now = self._now()
+        lease_until = now + timedelta(seconds=self._claim_ttl_sec)
         with self._lock:
-            row = self._rows.get(correlation_id)
-            if row is None or row.status is not WorkItemStatus.PENDING:
-                return []
-            if row.next_retry_at is not None and row.next_retry_at > now:
-                return []
-            return [row]
+            claimed: list[LedgerWorkItem] = []
+            candidates = [correlation_id] if correlation_id is not None else self._order
+            for cid in candidates:
+                row = self._rows.get(cid)
+                if row is None or row.status is not WorkItemStatus.PENDING:
+                    continue
+                if row.instance_key != self._instance_key:
+                    continue
+                if row.next_retry_at is not None and row.next_retry_at > now:
+                    continue
+                if row.claim_expires_at is not None and row.claim_expires_at > now:
+                    continue
+                updated = replace(row, claim_expires_at=lease_until, updated_at=now)
+                self._rows[cid] = updated
+                claimed.append(updated)
+                if len(claimed) >= limit:
+                    break
+            return claimed
 
     def pending_cash_delta(self) -> dict[str, float]:
         with self._lock:
@@ -557,6 +614,11 @@ class InMemoryLedgerOutbox:
     ) -> LedgerWorkItem:
         with self._lock:
             row = self._rows[correlation_id]
+            if expected_claim_expires_at is None:
+                if row.claim_expires_at is not None:
+                    raise OutboxClaimLostError(correlation_id)
+            elif row.claim_expires_at != expected_claim_expires_at:
+                raise OutboxClaimLostError(correlation_id)
             if outcome is OutboxOutcome.COMPLETED:
                 new_status = WorkItemStatus.COMPLETED
             elif outcome is OutboxOutcome.CONFLICT_BLOCKED:
@@ -578,6 +640,7 @@ class InMemoryLedgerOutbox:
                     next_retry_at if outcome is OutboxOutcome.RETRYABLE_FAILURE else None
                 ),
                 updated_at=self._now(),
+                claim_expires_at=None,
             )
             self._rows[correlation_id] = updated
             return updated