Merge remote-tracking branch 'origin/main' into copilot/feat-reconcile-unmanaged-position-durable-lifecycl

# Conflicts:
#	.proofline/criteria/ac-0029.md
#	.proofline/identities.json
#	tests/adapters/persistence/test_migration_and_drift.py

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>

trex_core/adapters/persistence/postgresql.py

@@ -13,22 +13,24 @@ 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 (
     LedgerWorkItem,
+    OutboxClaimLostError,
     OutboxOutcome,
     WorkItemStatus,
 )
 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 +567,44 @@ 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, "
+    "claim_expires_at"
+)
+_OUTBOX_RETURN_COLUMNS = ", ".join(
+    f"o.{column.strip()}" for column in _OUTBOX_COLUMNS.split(",")
 )
 
 
 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,
+        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
+        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()
+
+    @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:
@@ -582,8 +612,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 +628,7 @@ class PostgreSQLLedgerOutbox:
                         trade.executed_at,
                         trade.venue,
                         trade.order_venue,
+                        self._instance_key,
                     ),
                 )
                 conn.commit()
@@ -606,17 +637,59 @@ 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 (next_retry_at IS NULL OR next_retry_at <= now()) "
-                    "ORDER BY created_at LIMIT %s",
-                    (limit,),
+                    "AND instance_key = %s "
+                    "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, now, now, 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, now, now]
+        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 <= %s) "
+                    "AND (claim_expires_at IS NULL OR claim_expires_at <= %s) "
+                    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_RETURN_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 +700,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}
 
@@ -648,6 +724,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:
@@ -669,20 +746,34 @@ 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 claim_owner IS NULL AND claim_expires_at IS NULL"
+                else:
+                    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 "
-                    "WHERE correlation_id=%s",
-                    (
-                        status_value,
-                        attempts_delta,
-                        outcome.value,
-                        error,
-                        nra,
-                        datetime.now(tz=timezone.utc),
-                        correlation_id,
-                    ),
+                    "last_outcome=%s, last_error=%s, next_retry_at=%s, updated_at=%s, "
+                    "claim_owner=NULL, claim_expires_at=NULL "
+                    f"{where}",
+                    params,
                 )
+                if cur.rowcount != 1:
+                    raise OutboxClaimLostError(correlation_id)
                 conn.commit()
         row = self.get(correlation_id)
         assert row is not None
@@ -714,6 +805,8 @@ class PostgreSQLLedgerOutbox:
             created_at=row[14],
             updated_at=row[15],
             next_retry_at=row[16],
+            instance_key=row[19],
+            claim_expires_at=row[20],
         )
 
 

trex_core/adapters/persistence/sql/016_create_slot_reconcile_warning.sql

@@ -1,6 +1,6 @@
 -- BAT-050 / REQ-102: reconcile-warning durable store (dcy-034).
--- PK: slot_id (TEXT, UUID string). pause_store 동형 설계.
--- ON CONFLICT DO UPDATE: 동일 slot_id 재삽입 시 최신 값으로 갱신.
+-- Initial schema; migration 028 upgrades PK to (slot_id, case_code)
+-- so B/C causes can coexist for the same slot.
 CREATE TABLE IF NOT EXISTS slot_reconcile_warning (
     slot_id      TEXT        PRIMARY KEY,
     portfolio_id TEXT        NOT NULL,

trex_core/adapters/persistence/sql/028_preserve_reconcile_warning_cases.sql

@@ -0,0 +1,30 @@
+-- REQ-123 / issue #83:
+-- Preserve simultaneous warning causes by keying warnings per (slot_id, case_code).
+-- NOTE: historical overwritten causes cannot be reconstructed from the prior
+-- slot_id-only schema; this migration enables forward preservation only.
+UPDATE slot_reconcile_warning
+SET case_code = 'B'
+WHERE case_code IS NULL OR case_code = '';
+
+ALTER TABLE slot_reconcile_warning
+    ALTER COLUMN case_code SET NOT NULL;
+
+DELETE FROM slot_reconcile_warning w
+USING (
+    SELECT ctid, ROW_NUMBER() OVER (
+        PARTITION BY slot_id, case_code
+        ORDER BY updated_at DESC, created_at DESC, ctid DESC
+    ) AS rn
+    FROM slot_reconcile_warning
+) ranked
+WHERE w.ctid = ranked.ctid
+  AND ranked.rn > 1;
+
+ALTER TABLE slot_reconcile_warning
+    DROP CONSTRAINT IF EXISTS slot_reconcile_warning_pkey;
+
+ALTER TABLE slot_reconcile_warning
+    ADD CONSTRAINT slot_reconcile_warning_pkey PRIMARY KEY (slot_id, case_code);
+
+CREATE INDEX IF NOT EXISTS idx_slot_reconcile_warning_slot_id
+    ON slot_reconcile_warning (slot_id);

trex_core/adapters/persistence/sql/029_ledger_outbox_instance_claims.sql

@@ -0,0 +1,23 @@
+-- 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;
+
+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, correlation_id)
+    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';

trex_core/adapters/portfolio/reconcile_warning_store.py

@@ -1,8 +1,4 @@
-"""PostgreSQL-backed durable reconcile-warning store (REQ-102, BAT-050).
-
-pause_store.py와 동형 설계. PK: slot_id. case_code: 'B' | 'C'.
-Survives daemon restart (dcy-034 NFR-1).
-"""
+"""PostgreSQL-backed durable reconcile-warning store (REQ-102, BAT-050)."""
 from __future__ import annotations
 
 from datetime import datetime, timezone
@@ -12,7 +8,7 @@ from trex_core.adapters.persistence.postgresql import _require_psycopg
 
 
 class PostgreSQLReconcileWarningStore:
-    """`slot_reconcile_warning` table CRUD. slot_id PK is the gate (REQ-102)."""
+    """`slot_reconcile_warning` table CRUD keyed by (slot_id, case_code)."""
 
     def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None:
         self._psycopg = _require_psycopg(psycopg_module)
@@ -34,9 +30,8 @@ class PostgreSQLReconcileWarningStore:
                     "INSERT INTO slot_reconcile_warning "
                     "(slot_id, portfolio_id, case_code, actor, reason, created_at, updated_at) "
                     "VALUES (%s, %s, %s, %s, %s, %s, %s) "
-                    "ON CONFLICT (slot_id) DO UPDATE "
+                    "ON CONFLICT (slot_id, case_code) DO UPDATE "
                     "SET portfolio_id = EXCLUDED.portfolio_id, "
-                    "    case_code    = EXCLUDED.case_code, "
                     "    actor        = EXCLUDED.actor, "
                     "    reason       = EXCLUDED.reason, "
                     "    updated_at   = EXCLUDED.updated_at",
@@ -44,13 +39,19 @@ class PostgreSQLReconcileWarningStore:
                 )
                 conn.commit()
 
-    def clear_warned(self, slot_id: str) -> None:
+    def clear_warned(self, slot_id: str, *, case_code: str | None = None) -> None:
         with self._psycopg.connect(self._dsn, autocommit=False) as conn:
             with conn.cursor() as cur:
-                cur.execute(
-                    "DELETE FROM slot_reconcile_warning WHERE slot_id = %s",
-                    (slot_id,),
-                )
+                if case_code is None:
+                    cur.execute(
+                        "DELETE FROM slot_reconcile_warning WHERE slot_id = %s",
+                        (slot_id,),
+                    )
+                else:
+                    cur.execute(
+                        "DELETE FROM slot_reconcile_warning WHERE slot_id = %s AND case_code = %s",
+                        (slot_id, case_code),
+                    )
                 conn.commit()
 
     def is_warned(self, slot_id: str) -> bool:
@@ -65,7 +66,7 @@ class PostgreSQLReconcileWarningStore:
     def list_warned(self) -> set[str]:
         with self._psycopg.connect(self._dsn) as conn:
             with conn.cursor() as cur:
-                cur.execute("SELECT slot_id FROM slot_reconcile_warning")
+                cur.execute("SELECT DISTINCT slot_id FROM slot_reconcile_warning")
                 return {row[0] for row in cur.fetchall()}
 
     def list_c_case_warned(self) -> set[str]:
@@ -73,26 +74,30 @@ class PostgreSQLReconcileWarningStore:
         with self._psycopg.connect(self._dsn) as conn:
             with conn.cursor() as cur:
                 cur.execute(
-                    "SELECT slot_id FROM slot_reconcile_warning WHERE case_code = %s",
+                    "SELECT DISTINCT slot_id FROM slot_reconcile_warning WHERE case_code = %s",
                     ("C",),
                 )
                 return {row[0] for row in cur.fetchall()}
 
     def get_warning_info(self, slot_id: str) -> dict | None:
-        """Return {"portfolio_id", "case_code", "actor", "reason"} or None (REQ-105/106)."""
+        """Return warning info with full active case set, or None."""
         with self._psycopg.connect(self._dsn) as conn:
             with conn.cursor() as cur:
                 cur.execute(
                     "SELECT portfolio_id, case_code, actor, reason "
-                    "FROM slot_reconcile_warning WHERE slot_id = %s LIMIT 1",
+                    "FROM slot_reconcile_warning WHERE slot_id = %s "
+                    "ORDER BY CASE WHEN case_code = 'C' THEN 0 ELSE 1 END, updated_at DESC",
                     (slot_id,),
                 )
-                row = cur.fetchone()
-                if row is None:
+                rows = cur.fetchall()
+                if not rows:
                     return None
+                case_codes = sorted({str(row[1]) for row in rows if row[1]})
+                preferred = rows[0]
                 return {
-                    "portfolio_id": row[0],
-                    "case_code": row[1],
-                    "actor": row[2],
-                    "reason": row[3],
+                    "portfolio_id": preferred[0],
+                    "case_code": preferred[1],
+                    "case_codes": case_codes,
+                    "actor": preferred[2],
+                    "reason": preferred[3],
                 }

trex_core/application/intents.py

@@ -29,6 +29,7 @@ from typing import Any, Callable, Optional
 from uuid import UUID
 
 from trex_core.application.alarm_emitter import AlarmEmitter
+from trex_core.application.portfolio import BLOCKING_RECONCILE_CASES
 from trex_core.application.ports import IntentStore
 from trex_core.domain.slot import (
     SlotAdmissionRejectedError,
@@ -270,6 +271,29 @@ class IntentService:
         except Exception:  # noqa: BLE001 - alarm path must not affect intent decision
             pass
 
+    def _is_reconcile_trade_blocked(self, slot_id: str) -> bool:
+        if self._warn_store is None:
+            return False
+        if not self._warn_store.is_warned(slot_id):
+            return False
+        info = self._warn_store.get_warning_info(slot_id)
+        if info is None:
+            return True
+        raw_codes = info.get("case_codes")
+        if isinstance(raw_codes, list):
+            case_codes = {str(code) for code in raw_codes if code}
+        else:
+            case_code = str(info.get("case_code", ""))
+            case_codes = {case_code} if case_code else set()
+        if not case_codes:
+            return True
+        # The warning row is authoritative. Unknown cases are not evidence
+        # that trading is safe: they may come from corrupt data or a newer
+        # producer that this core version cannot interpret.
+        if not case_codes.issubset(BLOCKING_RECONCILE_CASES):
+            return True
+        return bool(case_codes & BLOCKING_RECONCILE_CASES)
+
     # ------------------------------------------------------------
     def _dedup(self, intent_id: str, intent_type: str, payload: dict) -> Optional[IntentDecision]:
         """Return prior decision if intent_id already seen; else None (and reserve it)."""
@@ -339,7 +363,7 @@ class IntentService:
             # those paths call SlotService.sell() directly.
             if self._warn_store is not None:
                 try:
-                    if self._warn_store.is_warned(str(slot.slot_id)):
+                    if self._is_reconcile_trade_blocked(str(slot.slot_id)):
                         return IntentDecision(
                             intent.intent_id, "rejected", None, "reconcile_warned"
                         )
@@ -432,7 +456,7 @@ class IntentService:
         # SlotService.sell() / transit_slot() directly without IntentService (OQ-B).
         if self._warn_store is not None:
             try:
-                if self._warn_store.is_warned(str(slot.slot_id)):
+                if self._is_reconcile_trade_blocked(str(slot.slot_id)):
                     return IntentDecision(
                         intent.intent_id, "rejected", None, "reconcile_warned"
                     )

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}
@@ -104,11 +108,18 @@ class LedgerWorkItem:
     next_retry_at: Optional[datetime] = None
     created_at: Optional[datetime] = None
     updated_at: Optional[datetime] = None
+    instance_key: Optional[str] = None
+    claim_expires_at: Optional[datetime] = None
 
 
 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)."""
@@ -118,6 +129,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).
@@ -136,6 +156,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.
 
@@ -151,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:
@@ -190,15 +212,46 @@ 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:
+        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):
-            outcome = self._process(item)
+        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:
+            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:
@@ -216,7 +269,50 @@ class LedgerWriter:
         item = self._outbox.get(correlation_id)
         if item is None:
             return None
-        return self._process(item)
+        if not self._validate_instance(item):
+            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]
+            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,
+            )
+        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%."""
@@ -228,14 +324,18 @@ 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:
             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)
             # dcy-033 / REQ-100: CRITICAL alarm — 409 conflict, operator resolution needed.
             if self._alarm_emitter is not None:
                 self._alarm_emitter.publish_alarm(
@@ -250,11 +350,16 @@ 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)
             _log.warning(
-                "ledger delivery_failed (permanent): correlation_id=%s reason=%s",
-                item.correlation_id, exc,
+                "ledger delivery_failed (permanent): instance_key=%s correlation_id=%s",
+                item.instance_key or self._instance_key,
+                item.correlation_id,
             )
             # dcy-033 / REQ-100: CRITICAL alarm — permanent failure, no retry.
             if self._alarm_emitter is not None:
@@ -271,12 +376,18 @@ 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)
                 _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",
+                    new_attempts,
+                    item.instance_key or self._instance_key,
+                    item.correlation_id,
                 )
                 # dcy-033 / REQ-100: CRITICAL alarm — dead-lettered after max retries.
                 if self._alarm_emitter is not None:
@@ -294,8 +405,11 @@ 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)
             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
@@ -306,13 +420,37 @@ 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
-        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
 
+    def _log_outcome(
+        self,
+        item: LedgerWorkItem,
+        outcome: OutboxOutcome,
+    ) -> 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,
+        )
+        _log.info(msg, *args)
+
     def _resolve_ledger_skip(self, trade: LedgerTrade) -> None:
         if (
             self._ledger_skip_store is None
@@ -349,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,
             )
 
 
@@ -358,7 +496,13 @@ 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,
+        claim_ttl_sec: float = 60.0,
+    ) -> None:
         import threading
 
         self._rows: dict[str, LedgerWorkItem] = {}
@@ -367,6 +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()
@@ -383,6 +537,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)
@@ -396,14 +551,45 @@ 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:
                         break
             return out
 
+    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)
+        with self._lock:
+            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:
             out: dict[str, float] = {}
@@ -424,9 +610,15 @@ 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]
+            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:
@@ -448,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

trex_core/application/messaging_handlers.py

@@ -17,6 +17,7 @@ from trex_core.application.account_snapshot import AccountSnapshotUnavailable
 from trex_core.application.broker_port import BalanceItem
 from trex_core.application.messaging import RpcRequest, RpcResponse
 from trex_core.application.portfolio import (
+    BLOCKING_RECONCILE_CASES,
     PauseStore,
     ReconcileWarningStore,
     SnapshotStore,
@@ -263,28 +264,27 @@ def make_slots_rpc_handler(
             for s in holding_slots:
                 watchdog_store.update(str(s.slot_id))
 
-        # Build merged slot rows (REQ-074)
-        # BAT-050 / REQ-104: warn_store passed to slot_to_dict for reconcile_warned field.
-        slot_rows = [
-            slot_to_dict(
-                s,
-                snapshot_store=snapshot_store,
-                watchdog_store=watchdog_store,
-                pause_store=pause_store,
-                warn_store=warn_store,
-                watchdog_timeout_sec=watchdog_timeout_sec,
-            )
-            for s in slots
-        ]
-
-        # AC-0023 exclusion must precede all optional data readers so excluded
-        # slots cannot expose state or make an otherwise valid response fail.
-        excluded_count = 0
+        warning_cases_by_slot: dict[str, set[str]] = {}
         if warn_store is not None:
             try:
-                c_warned_ids = warn_store.list_c_case_warned()
+                warned_slot_ids = warn_store.list_warned()
+                for sid in warned_slot_ids:
+                    info = warn_store.get_warning_info(sid)
+                    if info is None:
+                        raise RuntimeError(f"missing warning info for slot_id={sid}")
+                    raw_codes = info.get("case_codes")
+                    if isinstance(raw_codes, list):
+                        codes = {str(code) for code in raw_codes if code}
+                    else:
+                        case_code = str(info.get("case_code", ""))
+                        codes = {case_code} if case_code else set()
+                    if not codes:
+                        raise RuntimeError(f"empty warning cases for slot_id={sid}")
+                    if not codes.issubset(BLOCKING_RECONCILE_CASES):
+                        raise RuntimeError(f"unsupported warning cases for slot_id={sid}")
+                    warning_cases_by_slot[str(sid)] = codes
             except Exception:  # noqa: BLE001
-                _log.exception("warn_store.list_c_case_warned() failed — fail-closed")
+                _log.exception("warn_store warning query failed — fail-closed")
                 return RpcResponse(payload={
                     "slots": [],
                     "available_slots": None,
@@ -294,20 +294,48 @@ def make_slots_rpc_handler(
                         "reconcile warning store unavailable; "
                         "returning no slots (fail-closed)"
                     ),
-                    "excluded_slots": {"reconcile_quantity_mismatch": 0},
+                    "excluded_slots": {
+                        "reconcile_quantity_mismatch": 0,
+                        "active_reconcile_warning_cases": {},
+                    },
                 })
-            if c_warned_ids:
-                included = [
-                    (slot, row)
-                    for slot, row in zip(slots, slot_rows, strict=True)
-                    if not (
-                        slot.status.value == "holding"
-                        and str(slot.slot_id) in c_warned_ids
-                    )
-                ]
-                excluded_count = len(slots) - len(included)
+
+        # Build merged slot rows (REQ-074)
+        slot_rows = [
+            slot_to_dict(
+                s,
+                snapshot_store=snapshot_store,
+                watchdog_store=watchdog_store,
+                pause_store=pause_store,
+                warning_cases_by_slot=warning_cases_by_slot,
+                watchdog_timeout_sec=watchdog_timeout_sec,
+            )
+            for s in slots
+        ]
+
+        # Exclusion must precede all optional data readers so blocked slots cannot
+        # expose state or make an otherwise valid response fail.
+        excluded_count = 0
+        excluded_case_counts: dict[str, int] = {}
+        if warn_store is not None:
+            included: list[tuple[object, dict]] = []
+            for slot, row in zip(slots, slot_rows, strict=True):
+                blocked_holding = (
+                    slot.status.value == "holding"
+                    and bool(set(row.get("reconcile_warning_cases", [])) & BLOCKING_RECONCILE_CASES)
+                )
+                if blocked_holding:
+                    excluded_count += 1
+                    for code in row.get("reconcile_warning_cases", []):
+                        excluded_case_counts[code] = excluded_case_counts.get(code, 0) + 1
+                    continue
+                included.append((slot, row))
+            if included:
                 slots = [slot for slot, _ in included]
                 slot_rows = [row for _, row in included]
+            else:
+                slots = []
+                slot_rows = []
 
         broker_qty_by_symbol: dict[str, int] = {}
         broker_qty_available = False
@@ -410,7 +438,10 @@ def make_slots_rpc_handler(
             "message": message,
         }
         if warn_store is not None:
-            base_payload["excluded_slots"] = {"reconcile_quantity_mismatch": excluded_count}
+            base_payload["excluded_slots"] = {
+                "reconcile_quantity_mismatch": excluded_count,
+                "active_reconcile_warning_cases": excluded_case_counts,
+            }
         return RpcResponse(payload=base_payload)
 
     return handler

trex_core/application/portfolio.py

@@ -21,6 +21,8 @@ from datetime import UTC, datetime
 from enum import Enum
 from typing import Protocol
 
+BLOCKING_RECONCILE_CASES = frozenset({"B", "C"})
+
 # ---------------------------------------------------------------- portfolio-hub
 
 @dataclass(frozen=True)
@@ -359,7 +361,7 @@ class InMemoryKillSwitchStore:
 
 
 class ReconcileWarningStore(Protocol):
-    """Durable reconcile-warning overlay — slot_id PK (REQ-102, dcy-034)."""
+    """Durable reconcile-warning overlay keyed by (slot_id, case_code)."""
 
     def set_warned(
         self,
@@ -372,7 +374,7 @@ class ReconcileWarningStore(Protocol):
     ) -> None:
         ...
 
-    def clear_warned(self, slot_id: str) -> None:
+    def clear_warned(self, slot_id: str, *, case_code: str | None = None) -> None:
         ...
 
     def is_warned(self, slot_id: str) -> bool:
@@ -382,10 +384,14 @@ class ReconcileWarningStore(Protocol):
         ...
 
     def get_warning_info(self, slot_id: str) -> dict | None:
-        """Return {"portfolio_id", "case_code", "actor", "reason"} or None.
+        """Return warning info dict or None.
 
-        Used by verify to determine whether the warning is case_code='C'
-        (verify's responsibility) vs 'B' (sync's responsibility, REQ-106 OQ-C).
+        Returned dict includes:
+          {"portfolio_id", "case_code", "case_codes", "actor", "reason"}
+        where case_code is a backward-compatible preferred case and case_codes is
+        the full active case set for the slot.
+
+        Used by verify to determine whether C-case is active (REQ-106 OQ-C).
         Returns None when the slot has no active warning.
         """
         ...
@@ -403,8 +409,8 @@ class InMemoryReconcileWarningStore:
 
     def __init__(self) -> None:
         self._lock = threading.Lock()
-        # slot_id -> {"portfolio_id", "case_code", "actor", "reason"}
-        self._warned: dict[str, dict] = {}
+        # slot_id -> case_code -> {"portfolio_id", "actor", "reason"}
+        self._warned: dict[str, dict[str, dict]] = {}
 
     def set_warned(
         self,
@@ -416,16 +422,24 @@ class InMemoryReconcileWarningStore:
         reason: str = "",
     ) -> None:
         with self._lock:
-            self._warned[slot_id] = {
+            per_case = self._warned.setdefault(slot_id, {})
+            per_case[case_code] = {
                 "portfolio_id": portfolio_id,
-                "case_code": case_code,
                 "actor": actor,
                 "reason": reason,
             }
 
-    def clear_warned(self, slot_id: str) -> None:
+    def clear_warned(self, slot_id: str, *, case_code: str | None = None) -> None:
         with self._lock:
-            self._warned.pop(slot_id, None)
+            if case_code is None:
+                self._warned.pop(slot_id, None)
+                return
+            per_case = self._warned.get(slot_id)
+            if not per_case:
+                return
+            per_case.pop(case_code, None)
+            if not per_case:
+                self._warned.pop(slot_id, None)
 
     def is_warned(self, slot_id: str) -> bool:
         with self._lock:
@@ -437,12 +451,23 @@ class InMemoryReconcileWarningStore:
 
     def get_warning_info(self, slot_id: str) -> dict | None:
         with self._lock:
-            info = self._warned.get(slot_id)
-            return dict(info) if info is not None else None
+            per_case = self._warned.get(slot_id)
+            if not per_case:
+                return None
+            case_codes = sorted(per_case.keys())
+            preferred_case = "C" if "C" in per_case else case_codes[0]
+            preferred = per_case[preferred_case]
+            return {
+                "portfolio_id": preferred.get("portfolio_id", ""),
+                "case_code": preferred_case,
+                "case_codes": case_codes,
+                "actor": preferred.get("actor", ""),
+                "reason": preferred.get("reason", ""),
+            }
 
     def list_c_case_warned(self) -> set[str]:
         with self._lock:
-            return {sid for sid, info in self._warned.items() if info.get("case_code") == "C"}
+            return {sid for sid, per_case in self._warned.items() if "C" in per_case}
 
 
 class ReconcileNotificationStore(Protocol):
@@ -794,6 +819,7 @@ def slot_to_dict(
     watchdog_store=None,
     pause_store: PauseStore | None = None,
     warn_store: ReconcileWarningStore | None = None,
+    warning_cases_by_slot: dict[str, set[str]] | None = None,
     watchdog_timeout_sec: int = 120,
 ) -> dict:
     """Merged slot row for trex.slots.req (REQ-073/074).
@@ -868,10 +894,25 @@ def slot_to_dict(
     # BAT-050 / REQ-104: reconcile_warned overlay (ReconcileWarningStore) — additive field.
     # Bulk list_warned() called once per RPC (caller pre-fetches; see make_slots_rpc_handler).
     # warn_store=None → reconcile_warned=False (backward compatible, NFR-4).
-    reconcile_warned = False
-    if warn_store is not None:
-        reconcile_warned = str(s.slot_id) in warn_store.list_warned()
+    warning_cases: set[str] = set()
+    if warning_cases_by_slot is not None:
+        warning_cases = set(warning_cases_by_slot.get(str(s.slot_id), set()))
+    elif warn_store is not None:
+        info = warn_store.get_warning_info(str(s.slot_id))
+        if info is not None:
+            raw_cases = info.get("case_codes")
+            if isinstance(raw_cases, list):
+                warning_cases = {str(c) for c in raw_cases if c}
+            else:
+                case_code = str(info.get("case_code", ""))
+                if case_code:
+                    warning_cases = {case_code}
+    reconcile_warned = bool(warning_cases)
     d["reconcile_warned"] = reconcile_warned
+    d["reconcile_warning_cases"] = sorted(warning_cases)
+    known_blocking = bool(warning_cases & BLOCKING_RECONCILE_CASES)
+    unknown_cases = bool(warning_cases - BLOCKING_RECONCILE_CASES)
+    d["trade_eligible"] = not (known_blocking or unknown_cases)
 
     return d
 

trex_core/application/reconciliation.py

@@ -742,7 +742,7 @@ def sync_slot_to_broker(
     Broker qty == 0 for slot   → skipped with WARN (slot-side A-case guard).
     Broker-only (no slot)      → NOT handled here; caller filters (FR-3).
 
-    After update, if slot.qty == broker_qty: clear_warned(slot_id).
+    After update, if slot.qty == broker_qty: clear_warned(slot_id, case_code="B").
     """
     from uuid import UUID
 
@@ -983,14 +983,15 @@ def sync_slot_to_broker(
     after_avg = float(refreshed.avg_price) if (refreshed and refreshed.avg_price is not None) else None
     after = {"qty": after_qty, "status": after_status, "avg_price": after_avg}
 
-    # Clear warning when qty now matches broker
+    # Clear B-case warning when qty now matches broker
     warning_cleared = False
     if after_qty is not None and int(after_qty) == broker_qty:
         try:
-            warn_store.clear_warned(slot_id)
+            warn_store.clear_warned(slot_id, case_code="B")
             warning_cleared = True
             _log.info(
-                "sync: reconcile-warning cleared slot=%s symbol=%s broker_qty=%d actor=sync",
+                "sync: reconcile-warning cleared slot=%s symbol=%s case_code=B "
+                "broker_qty=%d actor=sync",
                 slot_id, sym, broker_qty,
             )
         except Exception:  # noqa: BLE001
@@ -1075,7 +1076,7 @@ def verify_slot_reconcile(
             "reason": "no active reconcile-warning (already cleared or never set)",
         }
 
-    # Check case_code — verify is C-only (REQ-106 OQ-C).
+    # Check case set — verify is C-only (REQ-106 OQ-C).
     info = warn_store.get_warning_info(slot_id_str)
     if info is None:
         return {
@@ -1087,8 +1088,13 @@ def verify_slot_reconcile(
             "reason": "warning exists but get_warning_info returned None (unexpected)",
         }
 
-    case_code = info.get("case_code", "")
-    if case_code != "C":
+    case_codes_raw = info.get("case_codes")
+    if isinstance(case_codes_raw, list):
+        case_codes = {str(code) for code in case_codes_raw if code}
+    else:
+        case_code = str(info.get("case_code", ""))
+        case_codes = {case_code} if case_code else set()
+    if "C" not in case_codes:
         return {
             "portfolio_id": portfolio_id,
             "slot_id": slot_id_str,
@@ -1096,7 +1102,7 @@ def verify_slot_reconcile(
             "hub_qty": None,
             "broker_qty": None,
             "reason": (
-                f"warning case_code='{case_code}' is not 'C' — "
+                f"warning case_codes={sorted(case_codes)} do not include 'C' — "
                 "use 'trex-core sync' to resolve B-case (slot↔broker qty mismatch)"
             ),
         }
@@ -1127,9 +1133,17 @@ def verify_slot_reconcile(
                     continue
                 sid_str = str(s.slot_id)
                 warn_info = warn_store.get_warning_info(sid_str)
-                if warn_info is None or warn_info.get("case_code") != "C":
+                if warn_info is None:
                     continue
-                warn_store.clear_warned(sid_str)
+                warned_codes_raw = warn_info.get("case_codes")
+                if isinstance(warned_codes_raw, list):
+                    warned_codes = {str(code) for code in warned_codes_raw if code}
+                else:
+                    case_code = str(warn_info.get("case_code", ""))
+                    warned_codes = {case_code} if case_code else set()
+                if "C" not in warned_codes:
+                    continue
+                warn_store.clear_warned(sid_str, case_code="C")
                 store.set_warning(s.portfolio_id, False)
                 cleared_slots.append(sid_str)
             _log.info(

trex_core/runtime/cli.py

@@ -961,7 +961,15 @@ def _cmd_verify(args, cfg) -> int:
         c_slots: list[str] = []
         for sid in warn_store.list_warned():
             info = warn_store.get_warning_info(sid)
-            if info and info.get("case_code") == "C":
+            if not info:
+                continue
+            raw_codes = info.get("case_codes")
+            if isinstance(raw_codes, list):
+                case_codes = {str(code) for code in raw_codes if code}
+            else:
+                case_code = str(info.get("case_code", ""))
+                case_codes = {case_code} if case_code else set()
+            if "C" in case_codes:
                 c_slots.append(info.get("portfolio_id", ""))
         target_pids = [p for p in c_slots if p]
 
@@ -1487,7 +1495,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)

trex_core/runtime/daemon.py

@@ -192,7 +192,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.)
@@ -200,7 +200,7 @@ class Daemon:
         # BAT-010/011: per-slot watchdog (snapshot RPC writer, safety-sell reader).
         self.watchdog = WatchdogStore()
         self.pause_store = PostgreSQLPauseStore(config.dsn)
-        # BAT-050 / REQ-102: durable reconcile-warning store (slot_id PK).
+        # BAT-050 / REQ-102: durable reconcile-warning store ((slot_id, case_code) PK).
         self.warn_store = PostgreSQLReconcileWarningStore(config.dsn)
         # req-0007 / ac-0024: durable reconcile notification deduplication store.
         self.notification_store = PostgreSQLReconcileNotificationStore(config.dsn)