chore: plan for review blockers follow-up

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

tests/application/test_execution_cancel.py

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

tests/application/test_execution_fill_price_backfill.py

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

tests/application/test_execution_ledger.py

@@ -119,7 +119,11 @@ def _wire(broker, *, ledger=None, clock=None, alarm_emitter=None, ledger_skip_st
     svc = SlotService(repo)
     intents = IntentService(svc, repo, InMemoryIntentStore(),
                             resolve_symbol=lambda pid: "005930")
-    outbox = InMemoryLedgerOutbox(clock=clock) if clock else InMemoryLedgerOutbox()
+    outbox = (
+        InMemoryLedgerOutbox(clock=clock, instance_key="default")
+        if clock
+        else InMemoryLedgerOutbox(instance_key="default")
+    )
     execution = ExecutionService(
         svc,
         repo,
@@ -430,7 +434,7 @@ def test_duplicate_execution_id_results_in_zero_duplicate_in_ledger():
     # And even if the writer somehow re-processes (simulated by a fresh outbox
     # row for the same execution_id, e.g. after a manual replay), the ledger
     # adapter dedup'd it server-side and the outcome is duplicate_completed.
-    replay_outbox = InMemoryLedgerOutbox()
+    replay_outbox = InMemoryLedgerOutbox(instance_key="test-instance")
     replay_writer = LedgerWriter(replay_outbox, ledger)
     replay_outbox.enqueue(first.trade)
     stats = replay_writer.run_once()

tests/application/test_ledger_option_d.py

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

tests/application/test_ledger_outbox.py

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

tests/application/test_ledger_outbox_backoff.py

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

tests/application/test_messaging_event_publisher.py

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

tests/application/test_reconcile_notification.py

@@ -896,7 +896,7 @@ def test_b_sync_does_not_publish_resolved_when_clear_warned_fails():
     from trex_core.application.reconciliation import sync_slot_to_broker
 
     class FailingWarnStore(InMemoryReconcileWarningStore):
-        def clear_warned(self, slot_id):
+        def clear_warned(self, slot_id, *, case_code=None):
             raise RuntimeError("DB unavailable")
 
     ns = InMemoryReconcileNotificationStore()

tests/application/test_reconcile_warning_blocking.py

@@ -340,6 +340,52 @@ def test_intent_service_sell_rejected_when_reconcile_warned():
     assert d.reason == "reconcile_warned"
 
 
+def test_unknown_warning_case_blocks_buy_and_sell_intents():
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    warn_store = InMemoryReconcileWarningStore()
+    slot = _make_holding_slot(repo, svc, qty=10)
+    warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="X")
+
+    intent_svc = IntentService(
+        svc,
+        repo,
+        InMemoryIntentStore(),
+        instance_key="default",
+        warn_store=warn_store,
+    )
+
+    buy = intent_svc.submit_buy(BuyIntent(
+        intent_id="b-unknown-case",
+        slot_id=slot.slot_id,
+        entry_price=100.0,
+        qty=1.0,
+    ))
+    sell = intent_svc.submit_sell(SellIntent(
+        intent_id="s-unknown-case",
+        slot_id=slot.slot_id,
+        reason="strategy_exit",
+    ))
+
+    assert buy.status == "rejected"
+    assert buy.reason == "reconcile_warned"
+    assert sell.status == "rejected"
+    assert sell.reason == "reconcile_warned"
+
+
+def test_slot_to_dict_marks_unknown_warning_case_ineligible():
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    warn_store = InMemoryReconcileWarningStore()
+    slot = _make_holding_slot(repo, svc, qty=10)
+    warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="X")
+
+    row = slot_to_dict(slot, warn_store=warn_store)
+
+    assert row["reconcile_warning_cases"] == ["X"]
+    assert row["trade_eligible"] is False
+
+
 def test_reconcile_warned_buy_emits_alarm_with_allowlisted_fields_after_finalize():
     repo = InMemorySlotRepository()
     svc = SlotService(repo)
@@ -892,6 +938,8 @@ def test_slot_to_dict_reconcile_warned_false_when_no_warn_store():
     row = slot_to_dict(slot, warn_store=None)
     assert "reconcile_warned" in row
     assert row["reconcile_warned"] is False
+    assert row["trade_eligible"] is True
+    assert row["reconcile_warning_cases"] == []
 
 
 def test_slot_to_dict_reconcile_warned_true_when_warned():
@@ -902,6 +950,8 @@ def test_slot_to_dict_reconcile_warned_true_when_warned():
     warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="B")
     row = slot_to_dict(slot, warn_store=warn_store)
     assert row["reconcile_warned"] is True
+    assert row["trade_eligible"] is False
+    assert row["reconcile_warning_cases"] == ["B"]
 
 
 def test_slot_to_dict_paused_and_warned_independent():

tests/application/test_reconcile_warning_store.py

@@ -33,14 +33,27 @@ def test_clear_warned_returns_false():
     assert store.is_warned("slot-1") is False
 
 
-# AC-3 (REQ-102): duplicate set_warned on same slot_id => upsert (single entry).
-def test_set_warned_upsert_same_slot():
+# AC-3 (REQ-123): same slot can carry multiple active warning causes.
+def test_set_warned_preserves_multiple_cases_on_same_slot():
     store = _store()
     store.set_warned("slot-1", portfolio_id="pf-A", case_code="B", reason="first")
     store.set_warned("slot-1", portfolio_id="pf-A", case_code="C", reason="second")
-    # After second set, is_warned still True — single entry (no duplicate row).
     assert store.is_warned("slot-1") is True
     assert store.list_warned() == {"slot-1"}
+    info = store.get_warning_info("slot-1")
+    assert info is not None
+    assert info["case_codes"] == ["B", "C"]
+
+
+def test_clear_warned_single_case_preserves_other_case():
+    store = _store()
+    store.set_warned("slot-1", portfolio_id="pf-A", case_code="B")
+    store.set_warned("slot-1", portfolio_id="pf-A", case_code="C")
+    store.clear_warned("slot-1", case_code="C")
+    info = store.get_warning_info("slot-1")
+    assert info is not None
+    assert info["case_codes"] == ["B"]
+    assert store.is_warned("slot-1") is True
 
 
 # AC-5 (REQ-102): list_warned returns set of warned slot_ids.

tests/application/test_slots_handler_quantity_mismatch.py

@@ -128,6 +128,44 @@ def test_ac1_single_portfolio_c_case_excluded():
 
     assert resp.payload["slots"] == []
     assert resp.payload["excluded_slots"]["reconcile_quantity_mismatch"] == 1
+    assert resp.payload["excluded_slots"]["active_reconcile_warning_cases"] == {"C": 1}
+
+
+def test_holding_slot_with_b_case_is_excluded_for_strategy():
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    warn_store = InMemoryReconcileWarningStore()
+
+    slot = _make_holding_slot(repo, svc, portfolio_id="pf-A", symbol=SYM_A, qty=10)
+    warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="B")
+
+    handler = _make_handler(repo, warn_store)
+    resp = handler(_req(filter="holding"))
+
+    assert resp.payload["slots"] == []
+    assert resp.payload["excluded_slots"]["reconcile_quantity_mismatch"] == 1
+    assert resp.payload["excluded_slots"]["active_reconcile_warning_cases"] == {"B": 1}
+
+
+def test_holding_slot_with_b_and_c_cases_keeps_both_and_reports_both():
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    warn_store = InMemoryReconcileWarningStore()
+
+    slot = _make_holding_slot(repo, svc, portfolio_id="pf-A", symbol=SYM_A, qty=10)
+    slot_id = str(slot.slot_id)
+    warn_store.set_warned(slot_id, portfolio_id="pf-A", case_code="B")
+    warn_store.set_warned(slot_id, portfolio_id="pf-A", case_code="C")
+
+    handler = _make_handler(repo, warn_store)
+    resp = handler(_req(filter="holding"))
+
+    assert resp.payload["slots"] == []
+    assert resp.payload["excluded_slots"]["reconcile_quantity_mismatch"] == 1
+    assert resp.payload["excluded_slots"]["active_reconcile_warning_cases"] == {"B": 1, "C": 1}
+    info = warn_store.get_warning_info(slot_id)
+    assert info is not None
+    assert info["case_codes"] == ["B", "C"]
 
 
 # ---------------------------------------------------------------------------
@@ -286,7 +324,8 @@ def test_ac5_excluded_slots_count_zero_when_no_warnings():
     resp = handler(_req(filter="holding"))
 
     assert "excluded_slots" in resp.payload
-    assert resp.payload["excluded_slots"] == {"reconcile_quantity_mismatch": 0}
+    assert resp.payload["excluded_slots"]["reconcile_quantity_mismatch"] == 0
+    assert resp.payload["excluded_slots"]["active_reconcile_warning_cases"] == {}
 
 
 # ---------------------------------------------------------------------------
@@ -312,8 +351,8 @@ def test_ac6_no_warn_store_no_filtering_no_excluded_slots_key():
 # ---------------------------------------------------------------------------
 
 class BrokenWarnStore(InMemoryReconcileWarningStore):
-    """Simulates a warn_store that fails on list_c_case_warned."""
-    def list_c_case_warned(self):
+    """Simulates a warn_store that fails on warning reads."""
+    def list_warned(self):
         raise RuntimeError("db connection lost")
 
 
@@ -332,6 +371,24 @@ def test_ac7_warn_store_failure_fail_closed():
     assert "excluded_slots" in resp.payload
 
 
+def test_unknown_warning_case_fails_closed():
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    warn_store = InMemoryReconcileWarningStore()
+    slot = _make_holding_slot(repo, svc, portfolio_id="pf-A", symbol=SYM_A, qty=10)
+    warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="X")
+
+    resp = _make_handler(repo, warn_store)(_req())
+
+    assert resp.payload["slots"] == []
+    assert resp.payload["degraded"] is True
+    assert resp.payload["reason"] == "warn_store_unavailable"
+    assert resp.payload["excluded_slots"] == {
+        "reconcile_quantity_mismatch": 0,
+        "active_reconcile_warning_cases": {},
+    }
+
+
 def test_ac23_exclusion_precedes_broker_and_strategy_state_readers():
     repo = InMemorySlotRepository()
     svc = SlotService(repo)
@@ -356,7 +413,7 @@ def test_ac23_exclusion_precedes_broker_and_strategy_state_readers():
     )
     response = handler(_req(filter="holding")).payload
     assert response["slots"] == []
-    assert response["excluded_slots"] == {"reconcile_quantity_mismatch": 1}
+    assert response["excluded_slots"]["reconcile_quantity_mismatch"] == 1
 
 
 # ---------------------------------------------------------------------------

tests/application/test_snapshot_scheduler.py

@@ -1,7 +1,8 @@
 """Snapshot scheduler tests (REQ-030 AC-1; REQ-114 AC-1~9; REQ-115 AC-1~6)."""
 from __future__ import annotations
 
-from datetime import datetime, timezone, timedelta
+from datetime import UTC, datetime, timedelta, timezone
+from types import SimpleNamespace
 from unittest.mock import MagicMock
 
 from trex_core.adapters.broker.paper import PaperBroker
@@ -44,10 +45,42 @@ def test_run_once_failure_sets_degraded_signal():
     store = SnapshotStore()
     sched = SnapshotScheduler(read_port=BadRead(), store=store, broker=PaperBroker())
     assert sched.run_once() is False
-    reason, msg = store.degraded_state()
+    reason, _msg = store.degraded_state()
     assert reason is not None and reason.value == "portfolio_hub_unavailable"
 
 
+def test_run_once_uses_authoritative_core_holdings_broker(monkeypatch):
+    read = StubPortfolioRead(
+        daemon_id="d-1",
+        items=[PortfolioItem(portfolio_id="pf-X", symbol="X", cash_balance=100, asset_quantity=0)],
+    )
+    store = SnapshotStore()
+    projected_broker = object()
+    authoritative_broker = object()
+    captured = {}
+
+    def _capture(**kwargs):
+        captured["broker"] = kwargs["broker"]
+        return {}
+
+    monkeypatch.setattr(
+        "trex_core.application.reconciliation.reconcile_core_holdings",
+        _capture,
+    )
+    sched = SnapshotScheduler(
+        read_port=read,
+        store=store,
+        broker=projected_broker,
+        core_holdings_broker=authoritative_broker,
+        slot_repo=object(),
+        slot_service=SimpleNamespace(instance_key="inst-1"),
+    )
+    sched._reconcile_slots = lambda _items: None
+
+    assert sched.run_once() is True
+    assert captured["broker"] is authoritative_broker
+
+
 def test_swap_preserves_warning_for_carried_over_portfolio():
     store = SnapshotStore()
     store.swap([PortfolioItem(portfolio_id="pf-X", symbol="X",
@@ -99,7 +132,7 @@ def _fixed_clock(kst_hour=8, kst_minute=0):
     """Return a clock callable fixed at a given KST time today."""
     now_kst = datetime.now(tz=_KST).replace(hour=kst_hour, minute=kst_minute,
                                              second=0, microsecond=0)
-    return lambda: now_kst.astimezone(timezone.utc)
+    return lambda: now_kst.astimezone(UTC)
 
 
 # ---------------------------------------------------------------------------
@@ -616,8 +649,8 @@ def test_kis_broker_reinitialize_preserves_base_url_and_flags():
 def test_build_kis_reinit_callback_returns_none_for_paper():
     """REQ-113 AC-2: broker=paper이면 callback=None."""
     from trex_core.adapters.broker.paper import PaperBroker as PB
-    from trex_core.runtime.daemon import _build_kis_reinit_callback
     from trex_core.runtime.config import DaemonConfig
+    from trex_core.runtime.daemon import _build_kis_reinit_callback
 
     # minimal config: paper broker, credential_fetch_enabled=True
     cfg = DaemonConfig(dsn="x", broker="paper", execution_interval_sec=1.0,
@@ -629,8 +662,8 @@ def test_build_kis_reinit_callback_returns_none_for_paper():
 def test_build_kis_reinit_callback_returns_none_when_disabled():
     """REQ-113 AC-1: credential_fetch_enabled=False이면 callback=None."""
     from trex_core.adapters.broker.paper import PaperBroker as PB
-    from trex_core.runtime.daemon import _build_kis_reinit_callback
     from trex_core.runtime.config import DaemonConfig
+    from trex_core.runtime.daemon import _build_kis_reinit_callback
 
     cfg = DaemonConfig(dsn="x", broker="kis", execution_interval_sec=1.0,
                        credential_fetch_enabled=False)
@@ -642,8 +675,8 @@ def test_build_kis_reinit_callback_returns_callable_for_kis():
     """REQ-113 AC-2 / REQ-115 AC-1: broker=kis + fetch enabled → callable 반환 + 실제 reinitialize 호출됨."""
     from trex_core.adapters.broker.kis import KISBroker
     from trex_core.application.portfolio import BrokerCredentialPayload
-    from trex_core.runtime.daemon import _build_kis_reinit_callback
     from trex_core.runtime.config import DaemonConfig
+    from trex_core.runtime.daemon import _build_kis_reinit_callback
 
     cfg = DaemonConfig(dsn="x", broker="kis", execution_interval_sec=1.0,
                        credential_fetch_enabled=True)

tests/application/test_sync.py

@@ -199,6 +199,22 @@ def test_sync_holding_qty_mismatch_updated():
     assert not ws_out.is_warned(str(slot.slot_id))
 
 
+def test_sync_clears_only_b_case_when_slot_has_multiple_warning_causes():
+    slot = _make_slot(SlotState.HOLDING, qty=5.0)
+    ws = InMemoryReconcileWarningStore()
+    slot_id = str(slot.slot_id)
+    ws.set_warned(slot_id, portfolio_id="pf-005930", case_code="B")
+    ws.set_warned(slot_id, portfolio_id="pf-005930", case_code="C")
+
+    result, _repo, ws_out = _setup(slot, {"005930": 10}, warn_store=ws)
+
+    assert result["warning_cleared"] is True
+    assert ws_out.is_warned(slot_id) is True
+    info = ws_out.get_warning_info(slot_id)
+    assert info is not None
+    assert info["case_codes"] == ["C"]
+
+
 # ---------------------------------------------------------------------------
 # AC-4: WAC seed from cost_basis_port when available
 # ---------------------------------------------------------------------------

tests/application/test_unmanaged_position_lifecycle.py

@@ -0,0 +1,307 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from trex_core.adapters.messaging.in_memory import InMemoryTransport
+from trex_core.application.alarm_emitter import AlarmEmitter
+from trex_core.application.broker_port import BalanceItem, BrokerBalanceError
+from trex_core.application.portfolio import InMemoryUnmanagedPositionStore
+from trex_core.application.reconciliation import reconcile_core_holdings
+from trex_core.domain.slot import SlotService
+from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
+
+INSTANCE = "inst-1"
+
+
+class FakeBroker:
+    def __init__(self, balances):
+        self._balances = balances
+
+    def name(self):
+        return "fake"
+
+    def submit_order(self, req): ...
+    def poll_fill(self, *args, **kwargs): ...
+    def cancel_order(self, *args, **kwargs): ...
+
+    def get_balance(self):
+        return [BalanceItem(symbol=symbol, hldg_qty=qty) for symbol, qty in self._balances.items()]
+
+
+class RaisingBroker(FakeBroker):
+    def get_balance(self):
+        raise BrokerBalanceError("transport_failed")
+
+
+class AuthoritativeBroker(FakeBroker):
+    def exposes_full_balance_snapshot(self) -> bool:
+        return True
+
+
+class FilteredBroker(FakeBroker):
+    def exposes_full_balance_snapshot(self) -> bool:
+        return False
+
+
+def _wire():
+    repo = InMemorySlotRepository()
+    return repo, SlotService(repo, instance_key=INSTANCE)
+
+
+def _make_emitter():
+    transport = InMemoryTransport()
+    return AlarmEmitter(transport), transport
+
+
+def test_in_memory_store_tracks_lifecycle_and_operator_audit():
+    store = InMemoryUnmanagedPositionStore()
+    t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
+    t2 = t1 + timedelta(minutes=1)
+    t3 = t2 + timedelta(minutes=1)
+
+    first = store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t1)
+    repeat = store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t2)
+    changed = store.observe(instance_key=INSTANCE, symbol="008110", quantity=4, observed_at=t3)
+    recorded = store.record_operator_action(
+        instance_key=INSTANCE,
+        symbol="008110",
+        action="investigating",
+        evidence_ref="kis-balance:2026-09-07",
+        note="confirmed at broker",
+    )
+    resolved = store.resolve_absent(
+        instance_key=INSTANCE,
+        observed_symbols=set(),
+        observed_at=t3 + timedelta(minutes=1),
+    )
+
+    assert first["transition"] == "detected"
+    assert repeat["transition"] == "unchanged"
+    assert changed["transition"] == "quantity_changed"
+    assert recorded is not None
+    assert recorded["evidence_ref"] == "kis-balance:2026-09-07"
+    assert len(resolved) == 1
+    state = store.get_state(instance_key=INSTANCE, symbol="008110")
+    assert state is not None
+    assert state["quantity"] == 4
+    assert state["first_seen_at"] == t1
+    assert state["last_seen_at"] == t3
+    assert state["quantity_changed_at"] == t3
+    assert state["lifecycle_state"] == "resolved"
+    assert state["resolved_at"] == t3 + timedelta(minutes=1)
+    audit = store.list_audit(instance_key=INSTANCE, symbol="008110")
+    assert [row["event_type"] for row in audit] == [
+        "detected",
+        "quantity_changed",
+        "operator_action",
+        "resolved",
+    ]
+    assert all("recorded_at" in row for row in audit)
+
+
+def test_in_memory_store_isolates_same_symbol_by_instance():
+    store = InMemoryUnmanagedPositionStore()
+
+    store.observe(instance_key="inst-A", symbol="008110", quantity=3)
+    store.observe(instance_key="inst-B", symbol="008110", quantity=7)
+    store.resolve_absent(instance_key="inst-A", observed_symbols=set())
+
+    state_a = store.get_state(instance_key="inst-A", symbol="008110")
+    state_b = store.get_state(instance_key="inst-B", symbol="008110")
+    assert state_a is not None and state_a["lifecycle_state"] == "resolved"
+    assert state_b is not None and state_b["lifecycle_state"] == "active"
+    assert state_b["quantity"] == 7
+
+
+def test_in_memory_store_preserves_first_seen_when_resolved_symbol_reappears():
+    store = InMemoryUnmanagedPositionStore()
+    t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
+    t2 = t1 + timedelta(minutes=1)
+    t3 = t2 + timedelta(minutes=1)
+
+    store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t1)
+    store.record_operator_action(
+        instance_key=INSTANCE,
+        symbol="008110",
+        action="investigating",
+        evidence_ref="kis-balance:2026-09-07",
+        note="before resolve",
+    )
+    store.resolve_absent(instance_key=INSTANCE, observed_symbols=set(), observed_at=t2)
+    reopened = store.observe(instance_key=INSTANCE, symbol="008110", quantity=5, observed_at=t3)
+
+    assert reopened["transition"] == "detected"
+    assert reopened["state"]["first_seen_at"] == t1
+    assert reopened["state"]["last_seen_at"] == t3
+    assert reopened["state"]["quantity_changed_at"] == t3
+    assert reopened["state"]["operator_action"] is None
+    assert reopened["state"]["evidence_ref"] is None
+
+
+def test_in_memory_store_ignores_stale_resolve_after_newer_observation():
+    store = InMemoryUnmanagedPositionStore()
+    t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
+    t2 = t1 + timedelta(minutes=1)
+    t3 = t2 + timedelta(minutes=1)
+
+    store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t1)
+    store.observe(instance_key=INSTANCE, symbol="008110", quantity=5, observed_at=t3)
+    resolved = store.resolve_absent(
+        instance_key=INSTANCE,
+        observed_symbols=set(),
+        observed_at=t2,
+    )
+
+    assert resolved == []
+    state = store.get_state(instance_key=INSTANCE, symbol="008110")
+    assert state is not None
+    assert state["lifecycle_state"] == "active"
+    assert state["quantity"] == 5
+    assert store.list_audit(instance_key=INSTANCE, symbol="008110")[-1]["event_type"] == "quantity_changed"
+
+
+def test_in_memory_store_ignores_stale_observation_after_newer_resolution():
+    store = InMemoryUnmanagedPositionStore()
+    t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
+    t2 = t1 + timedelta(minutes=1)
+    t3 = t2 + timedelta(minutes=1)
+
+    store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t1)
+    store.resolve_absent(instance_key=INSTANCE, observed_symbols=set(), observed_at=t3)
+    stale = store.observe(instance_key=INSTANCE, symbol="008110", quantity=5, observed_at=t2)
+
+    assert stale["transition"] == "stale"
+    state = store.get_state(instance_key=INSTANCE, symbol="008110")
+    assert state is not None
+    assert state["lifecycle_state"] == "resolved"
+    assert state["quantity"] == 3
+    assert [row["event_type"] for row in store.list_audit(instance_key=INSTANCE, symbol="008110")] == [
+        "detected",
+        "resolved",
+    ]
+
+
+def test_reconcile_core_holdings_suppresses_repeated_unmanaged_alarm_spam():
+    repo, svc = _wire()
+    emitter, transport = _make_emitter()
+    store = InMemoryUnmanagedPositionStore()
+
+    for _ in range(3):
+        decisions = reconcile_core_holdings(
+            slot_repo=repo,
+            slot_service=svc,
+            broker=AuthoritativeBroker({"008110": 3}),
+            instance_key=INSTANCE,
+            alarm_emitter=emitter,
+            unmanaged_position_store=store,
+        )
+        assert decisions == {"broker:008110": "unmanaged_position"}
+
+    state = store.get_state(instance_key=INSTANCE, symbol="008110")
+    assert state is not None
+    assert state["quantity"] == 3
+    codes = [payload["code"] for _, payload in transport.alarm_events()]
+    assert codes == ["unmanaged_position_detected"]
+    audit = store.list_audit(instance_key=INSTANCE, symbol="008110")
+    assert [row["event_type"] for row in audit] == ["detected"]
+
+
+def test_reconcile_core_holdings_tracks_change_and_resolution():
+    repo, svc = _wire()
+    emitter, transport = _make_emitter()
+    store = InMemoryUnmanagedPositionStore()
+
+    reconcile_core_holdings(
+        slot_repo=repo,
+        slot_service=svc,
+        broker=FakeBroker({"008110": 3}),
+        instance_key=INSTANCE,
+        alarm_emitter=emitter,
+        unmanaged_position_store=store,
+    )
+    reconcile_core_holdings(
+        slot_repo=repo,
+        slot_service=svc,
+        broker=AuthoritativeBroker({"008110": 5}),
+        instance_key=INSTANCE,
+        alarm_emitter=emitter,
+        unmanaged_position_store=store,
+    )
+    reconcile_core_holdings(
+        slot_repo=repo,
+        slot_service=svc,
+        broker=AuthoritativeBroker({}),
+        instance_key=INSTANCE,
+        alarm_emitter=emitter,
+        unmanaged_position_store=store,
+    )
+
+    state = store.get_state(instance_key=INSTANCE, symbol="008110")
+    assert state is not None
+    assert state["quantity"] == 5
+    assert state["lifecycle_state"] == "resolved"
+    codes = [payload["code"] for _, payload in transport.alarm_events()]
+    assert codes == [
+        "unmanaged_position_detected",
+        "unmanaged_position_changed",
+        "unmanaged_position_resolved",
+    ]
+
+
+def test_broker_failure_does_not_resolve_unmanaged_state():
+    repo, svc = _wire()
+    store = InMemoryUnmanagedPositionStore()
+    store.observe(instance_key=INSTANCE, symbol="008110", quantity=3)
+
+    with pytest.raises(BrokerBalanceError):
+        reconcile_core_holdings(
+            slot_repo=repo,
+            slot_service=svc,
+            broker=RaisingBroker({}),
+            instance_key=INSTANCE,
+            unmanaged_position_store=store,
+        )
+
+    state = store.get_state(instance_key=INSTANCE, symbol="008110")
+    assert state is not None
+    assert state["lifecycle_state"] == "active"
+
+
+def test_missing_authoritative_marker_does_not_resolve_unmanaged_state():
+    repo, svc = _wire()
+    store = InMemoryUnmanagedPositionStore()
+    store.observe(instance_key=INSTANCE, symbol="008110", quantity=3)
+
+    decisions = reconcile_core_holdings(
+        slot_repo=repo,
+        slot_service=svc,
+        broker=FakeBroker({}),
+        instance_key=INSTANCE,
+        unmanaged_position_store=store,
+    )
+
+    assert decisions == {}
+    state = store.get_state(instance_key=INSTANCE, symbol="008110")
+    assert state is not None
+    assert state["lifecycle_state"] == "active"
+
+
+def test_non_authoritative_broker_view_does_not_resolve_unmanaged_state():
+    repo, svc = _wire()
+    store = InMemoryUnmanagedPositionStore()
+    store.observe(instance_key=INSTANCE, symbol="008110", quantity=3)
+
+    decisions = reconcile_core_holdings(
+        slot_repo=repo,
+        slot_service=svc,
+        broker=FilteredBroker({}),
+        instance_key=INSTANCE,
+        unmanaged_position_store=store,
+    )
+
+    assert decisions == {}
+    state = store.get_state(instance_key=INSTANCE, symbol="008110")
+    assert state is not None
+    assert state["lifecycle_state"] == "active"

tests/application/test_verify.py

@@ -313,6 +313,40 @@ def test_verify_c_case_clear_preserves_b_case_warning():
     assert b_info is not None and b_info["case_code"] == "B"
 
 
+def test_verify_c_case_clear_on_slot_with_b_and_c_keeps_b():
+    sym = "005930"
+    slot = _make_slot(status=SlotState.HOLDING, symbol=sym, pid="pf-c")
+
+    repo = InMemorySlotRepository()
+    repo.add_slot(slot)
+
+    ws = InMemoryReconcileWarningStore()
+    slot_id = str(slot.slot_id)
+    ws.set_warned(slot_id, portfolio_id="pf-c", case_code="B")
+    ws.set_warned(slot_id, portfolio_id="pf-c", case_code="C")
+
+    store = SnapshotStore()
+    store.swap([
+        PortfolioItem(portfolio_id="pf-c", symbol=sym, cash_balance=0.0, asset_quantity=10.0),
+    ])
+    broker = FakeBroker({sym: 10})
+
+    result = verify_slot_reconcile(
+        portfolio_id="pf-c",
+        store=store,
+        broker=broker,
+        slot_repo=repo,
+        instance_key=INSTANCE,
+        warn_store=ws,
+    )
+
+    assert result["result"] == "cleared"
+    assert ws.is_warned(slot_id) is True
+    info = ws.get_warning_info(slot_id)
+    assert info is not None
+    assert info["case_codes"] == ["B"]
+
+
 def test_verify_c_case_clear_non_holding_slot():
     """A non-HOLDING slot with C-case warning for the same symbol must also be cleared."""
     sym = "005930"
@@ -360,11 +394,11 @@ class _FailAfterNStore(InMemoryReconcileWarningStore):
         self._calls = 0
         self._fail_after = fail_after
 
-    def clear_warned(self, slot_id: str) -> None:
+    def clear_warned(self, slot_id: str, *, case_code: str | None = None) -> None:
         self._calls += 1
         if self._calls > self._fail_after:
             raise RuntimeError("simulated DB error")
-        super().clear_warned(slot_id)
+        super().clear_warned(slot_id, case_code=case_code)
 
 
 def test_verify_partial_clear_failure_returns_clear_failed():