fix: harden skipped ledger resolution paths

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

tests/adapters/persistence/test_ledger_skip_store.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import threading
 from datetime import UTC, date, datetime
 from uuid import uuid4
 
@@ -87,3 +88,76 @@ def test_ledger_skip_store_filters_and_resolves(pg_clean):
     assert len(resolved) == 1
     assert resolved[0].status == "resolved"
     assert resolved[0].resolution_reason == "manual_ledger_backfill"
+
+
+def test_ledger_skip_store_resolves_null_execution_id_on_backfill(pg_clean):
+    store = PostgreSQLLedgerSkipStore(pg_clean)
+    slot_id = uuid4()
+    record, _ = store.record_skip(
+        instance_key="inst-1",
+        slot_id=slot_id,
+        portfolio_id="PF-1",
+        symbol="005930",
+        order_id="O-3",
+        execution_id=None,
+        side="buy",
+        qty=1.0,
+        price=1010.0,
+        executed_at=datetime(2026, 8, 24, 3, 0, tzinfo=UTC),
+        order_venue="SOR",
+        execution_venue="unknown",
+        skip_reason="missing_execution_id",
+    )
+
+    resolved = store.resolve_for_fill(
+        instance_key="inst-1",
+        slot_id=slot_id,
+        side="buy",
+        order_id="O-3",
+        execution_id="E-3",
+        resolution_reason="manual_ledger_backfill",
+    )
+
+    assert [row.skip_key for row in resolved] == [record.skip_key]
+    assert resolved[0].execution_id == "E-3"
+    assert resolved[0].status == "resolved"
+
+
+def test_ledger_skip_store_retries_unique_violation_during_concurrent_dedupe(pg_clean):
+    slot_id = uuid4()
+    start = threading.Barrier(2)
+    results = []
+
+    def worker():
+        store = PostgreSQLLedgerSkipStore(pg_clean)
+        start.wait()
+        results.append(
+            store.record_skip(
+                instance_key="inst-1",
+                slot_id=slot_id,
+                portfolio_id="PF-1",
+                symbol="005930",
+                order_id="O-4",
+                execution_id=None,
+                side="buy",
+                qty=1.0,
+                price=1111.0,
+                executed_at=datetime(2026, 8, 24, 4, 0, tzinfo=UTC),
+                order_venue="SOR",
+                execution_venue="unknown",
+                skip_reason="missing_execution_id",
+            )
+        )
+
+    t1 = threading.Thread(target=worker)
+    t2 = threading.Thread(target=worker)
+    t1.start()
+    t2.start()
+    t1.join()
+    t2.join()
+
+    assert len(results) == 2
+    assert len({record.skip_key for record, _notify in results}) == 1
+    assert sorted(notify for _record, notify in results) == [False, True]
+    listed = PostgreSQLLedgerSkipStore(pg_clean).list_records(instance_key="inst-1", status="unresolved")
+    assert len(listed) == 1

tests/application/test_execution_ledger.py

@@ -6,7 +6,7 @@ critical path while the ledger write is eventually-consistent.
 """
 from __future__ import annotations
 
-from datetime import UTC
+from datetime import UTC, datetime
 
 from trex_core.adapters.broker.kis import KISBroker
 from trex_core.adapters.broker.paper import PaperBroker
@@ -63,6 +63,55 @@ class _SkipStore:
     def resolve_for_fill(self, **_kwargs):
         return []
 
+    def resolve_for_delivery(
+        self,
+        *,
+        instance_key,
+        portfolio_id,
+        side,
+        order_id,
+        execution_id,
+        resolution_reason,
+    ):
+        resolved = []
+        updated = {}
+        for key, record in self._by_key.items():
+            if (
+                record.instance_key == instance_key
+                and record.portfolio_id == portfolio_id
+                and record.side == side
+                and (
+                    record.execution_id == execution_id
+                    or (record.execution_id is None and record.order_id == order_id)
+                )
+            ):
+                record = LedgerSkipRecord(
+                    skip_key=record.skip_key,
+                    instance_key=record.instance_key,
+                    slot_id=record.slot_id,
+                    portfolio_id=record.portfolio_id,
+                    symbol=record.symbol,
+                    order_id=record.order_id,
+                    execution_id=record.execution_id or execution_id,
+                    side=record.side,
+                    qty=record.qty,
+                    price=record.price,
+                    executed_at=record.executed_at,
+                    order_venue=record.order_venue,
+                    execution_venue=record.execution_venue,
+                    skip_reason=record.skip_reason,
+                    status="resolved",
+                    resolution_reason=resolution_reason,
+                    resolved_at=record.executed_at,
+                    created_at=record.created_at,
+                    updated_at=record.executed_at,
+                )
+                resolved.append(record)
+            updated[key] = record
+        self._by_key = updated
+        self.records = list(self._by_key.values())
+        return resolved
+
 
 
 def _wire(broker, *, ledger=None, clock=None, alarm_emitter=None, ledger_skip_store=None):
@@ -79,9 +128,13 @@ def _wire(broker, *, ledger=None, clock=None, alarm_emitter=None, ledger_skip_st
         alarm_emitter=alarm_emitter,
         ledger_skip_store=ledger_skip_store,
     )
-    writer = (
+    writer = LedgerWriter(
-        LedgerWriter(outbox, ledger or InMemoryLedger(), clock=clock)
+        outbox,
-        if clock else LedgerWriter(outbox, ledger or InMemoryLedger())
+        ledger or InMemoryLedger(),
+        clock=clock,
+        alarm_emitter=alarm_emitter,
+        ledger_skip_store=ledger_skip_store,
+        instance_key="default",
     )
     repo.reconcile_universe("default", {"PF-1": "SYM"})
     return repo, svc, intents, execution, outbox, writer
@@ -294,6 +347,50 @@ def test_missing_execution_id_persists_skip_record() -> None:
     assert record.execution_id is None
 
 
+def test_skip_resolves_only_after_writer_completes_delivery() -> None:
+    transport = InMemoryTransport()
+    skip_store = _SkipStore()
+    ledger = InMemoryLedger()
+    _repo, _, intents, execution, outbox, writer = _wire(
+        PaperBroker(fill_after_polls=0),
+        ledger=ledger,
+        alarm_emitter=AlarmEmitter(transport),
+        ledger_skip_store=skip_store,
+    )
+    sid = _buy(intents, iid="delayed-skip-resolution").slot_id
+    record, _ = skip_store.record_skip(
+        instance_key="default",
+        slot_id=sid,
+        portfolio_id="PF-1",
+        symbol="005930",
+        order_id="PAPER-000001",
+        execution_id=None,
+        side="buy",
+        qty=10.0,
+        price=1000.0,
+        executed_at=datetime(2026, 8, 24, 0, 0, tzinfo=UTC),
+        order_venue="SOR",
+        execution_venue="unknown",
+        skip_reason="missing_execution_id",
+    )
+
+    result = execution.run_once()
+
+    assert result.ledger_enqueued == 1
+    assert outbox.all_items()
+    assert skip_store._by_key[record.skip_key].status == "unresolved"
+    assert transport.alarm_events() == []
+
+    stats = writer.run_once()
+
+    assert stats.completed == 1
+    resolved = skip_store._by_key[record.skip_key]
+    assert resolved.status == "resolved"
+    assert resolved.execution_id is not None
+    assert resolved.resolution_reason == "ledger_recorded"
+    assert [payload["code"] for _, payload in transport.alarm_events()] == ["ledger_skip_resolved"]
+
+
 def test_ledger_skip_emits_warning(caplog):
     """dcy-011 / REQ-018 FR-4 (S-3): ledger skip is logged, not silent."""
     import logging

tests/runtime/test_cli_ledger_skips.py

@@ -9,8 +9,13 @@ from trex_core.application.ledger_skip import LedgerSkipRecord
 from trex_core.runtime import cli
 
 
-def _cfg():
+def _cfg(*, live: bool = False):
-    return SimpleNamespace(dsn="postgresql://x/y", rabbitmq=None)
+    return SimpleNamespace(
+        dsn="postgresql://x/y",
+        rabbitmq=None,
+        broker="kis",
+        kis=SimpleNamespace(allow_live=live),
+    )
 
 
 def _record(*, status="unresolved", resolution_reason=None):
@@ -46,7 +51,7 @@ def test_ledger_skips_parser_accepts_list_and_resolve():
     args = parser.parse_args(["ledger", "skips", "list", "--status", "all"])
     assert args.ledger_cmd == "skips"
     assert args.ledger_skips_cmd == "list"
-    args = parser.parse_args(["ledger", "skips", "resolve", "--skip-key", "abc"])
+    args = parser.parse_args(["ledger", "skips", "resolve", "--skip-key", "abc", "--yes"])
     assert args.ledger_skips_cmd == "resolve"
 
 
@@ -91,8 +96,9 @@ def test_ledger_skips_resolve_marks_record_and_emits_alert(monkeypatch, capsys):
         def __init__(self, *_args, **_kwargs):
             pass
 
-        def resolve_by_key(self, skip_key, *, resolution_reason):
+        def resolve_by_key(self, skip_key, *, instance_key, resolution_reason):
             assert skip_key == "abc"
+            assert instance_key == "inst-1"
             return _record(status="resolved", resolution_reason=resolution_reason)
 
     published = []
@@ -117,6 +123,9 @@ def test_ledger_skips_resolve_marks_record_and_emits_alert(monkeypatch, capsys):
         skip_key="abc",
         reason="operator_reconciled",
         instance="inst-1",
+        yes=True,
+        live=False,
+        audit_dir="audit-dir",
         config=None,
         config_path=None,
         env_file=None,
@@ -124,9 +133,42 @@ def test_ledger_skips_resolve_marks_record_and_emits_alert(monkeypatch, capsys):
         timeout=10.0,
     )
 
-    rc = cli._cmd_ledger_skips_resolve(args, _cfg())
+    rc = cli._cmd_ledger_skips_resolve(args, _cfg(), lambda _: True)
 
     assert rc == 0
     out = json.loads(capsys.readouterr().out)
     assert out["result"] == "resolved"
+    assert out["audit_path"].endswith(".json")
     assert published[0][0] == "ledger_skip_resolved"
+
+
+def test_ledger_skips_resolve_live_requires_live_flag(monkeypatch, tmp_path, capsys):
+    monkeypatch.setattr(
+        "trex_core.adapters.persistence.ledger_skip_store.PostgreSQLLedgerSkipStore",
+        lambda *_args, **_kwargs: object(),
+    )
+    args = SimpleNamespace(
+        cmd="ledger",
+        ledger_cmd="skips",
+        ledger_skips_cmd="resolve",
+        skip_key="abc",
+        reason="operator_reconciled",
+        instance="inst-1",
+        yes=True,
+        live=False,
+        audit_dir=str(tmp_path),
+        config=None,
+        config_path=None,
+        env_file=None,
+        json=True,
+        timeout=10.0,
+    )
+
+    rc = cli._cmd_ledger_skips_resolve(args, _cfg(live=True), lambda _: True)
+
+    assert rc == 2
+    err = capsys.readouterr().err
+    assert "--live" in err
+    audit = json.loads(next(tmp_path.glob("*.json")).read_text(encoding="utf-8"))
+    assert audit["request"]["instance"] == "inst-1"
+    assert audit["result"]["status"] == "rejected"

trex_core/adapters/persistence/ledger_skip_store.py

@@ -40,63 +40,72 @@ class PostgreSQLLedgerSkipStore:
         )
         now = datetime.now(tz=UTC)
         notify = False
-        with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur:
+        for _ in range(2):
-            cur.execute(
+            with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur:
-                "SELECT status FROM ledger_skip WHERE skip_key = %s FOR UPDATE",
+                try:
-                (skip_key,),
+                    cur.execute(
-            )
+                        "INSERT INTO ledger_skip "
-            row = cur.fetchone()
+                        "(skip_key, instance_key, slot_id, portfolio_id, symbol, order_id, "
-            if row is None:
+                        " execution_id, side, qty, price, executed_at, order_venue, "
-                notify = True
+                        " execution_venue, skip_reason, status, created_at, updated_at) "
-                cur.execute(
+                        "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, "
-                    "INSERT INTO ledger_skip "
+                        "        'unresolved', %s, %s)",
-                    "(skip_key, instance_key, slot_id, portfolio_id, symbol, order_id, "
+                        (
-                    " execution_id, side, qty, price, executed_at, order_venue, "
+                            skip_key,
-                    " execution_venue, skip_reason, status, created_at, updated_at) "
+                            instance_key,
-                    "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, "
+                            slot_id,
-                    "        'unresolved', %s, %s)",
+                            portfolio_id,
-                    (
+                            symbol,
-                        skip_key,
+                            order_id,
-                        instance_key,
+                            execution_id,
-                        slot_id,
+                            side,
-                        portfolio_id,
+                            qty,
-                        symbol,
+                            price,
-                        order_id,
+                            executed_at,
-                        execution_id,
+                            order_venue or "unknown",
-                        side,
+                            execution_venue or "unknown",
-                        qty,
+                            skip_reason,
-                        price,
+                            now,
-                        executed_at,
+                            now,
-                        order_venue or "unknown",
+                        ),
-                        execution_venue or "unknown",
+                    )
-                        skip_reason,
+                    conn.commit()
-                        now,
+                    notify = True
-                        now,
+                    break
-                    ),
+                except self._psycopg.errors.UniqueViolation:
-                )
+                    conn.rollback()
-            else:
+                    cur.execute(
-                notify = row[0] == "resolved"
+                        "SELECT status FROM ledger_skip WHERE skip_key = %s FOR UPDATE",
-                cur.execute(
+                        (skip_key,),
-                    "UPDATE ledger_skip SET portfolio_id=%s, symbol=%s, order_id=%s, "
+                    )
-                    "execution_id=%s, qty=%s, price=%s, executed_at=%s, "
+                    row = cur.fetchone()
-                    "order_venue=%s, execution_venue=%s, status='unresolved', "
+                    if row is None:
-                    "resolution_reason=NULL, resolved_at=NULL, updated_at=%s "
+                        continue
-                    "WHERE skip_key=%s",
+                    notify = row[0] == "resolved"
-                    (
+                    cur.execute(
-                        portfolio_id,
+                        "UPDATE ledger_skip SET portfolio_id=%s, symbol=%s, order_id=%s, "
-                        symbol,
+                        "execution_id=%s, qty=%s, price=%s, executed_at=%s, "
-                        order_id,
+                        "order_venue=%s, execution_venue=%s, status='unresolved', "
-                        execution_id,
+                        "resolution_reason=NULL, resolved_at=NULL, updated_at=%s "
-                        qty,
+                        "WHERE skip_key=%s",
-                        price,
+                        (
-                        executed_at,
+                            portfolio_id,
-                        order_venue or "unknown",
+                            symbol,
-                        execution_venue or "unknown",
+                            order_id,
-                        now,
+                            execution_id,
-                        skip_key,
+                            qty,
-                    ),
+                            price,
-                )
+                            executed_at,
-            conn.commit()
+                            order_venue or "unknown",
+                            execution_venue or "unknown",
+                            now,
+                            skip_key,
+                        ),
+                    )
+                    conn.commit()
+                    break
+        else:
+            raise RuntimeError(f"failed to persist ledger_skip row for key {skip_key!r}")
         record = self.get(skip_key)
         if record is None:
             raise RuntimeError(f"ledger_skip row missing after upsert for key {skip_key!r}")
@@ -166,21 +175,24 @@ class PostgreSQLLedgerSkipStore:
             if execution_id:
                 cur.execute(
                     "UPDATE ledger_skip SET status='resolved', resolution_reason=%s, "
+                    "execution_id=COALESCE(execution_id, %s), "
                     "resolved_at=%s, updated_at=%s "
                     "WHERE status='unresolved' AND instance_key=%s AND slot_id=%s "
-                    "AND side=%s AND execution_id=%s "
+                    "AND side=%s AND (execution_id=%s OR (execution_id IS NULL AND order_id=%s)) "
                     "RETURNING skip_key, instance_key, slot_id, portfolio_id, symbol, "
                     "order_id, execution_id, side, qty, price, executed_at, order_venue, "
                     "execution_venue, skip_reason, status, resolution_reason, resolved_at, "
                     "created_at, updated_at",
                     (
                         resolution_reason,
+                        execution_id,
                         now,
                         now,
                         instance_key,
                         slot_id,
                         side,
                         execution_id,
+                        order_id,
                     ),
                 )
             else:
@@ -207,17 +219,60 @@ class PostgreSQLLedgerSkipStore:
             conn.commit()
         return [self._row_to_record(row) for row in rows]
 
-    def resolve_by_key(self, skip_key: str, *, resolution_reason: str) -> LedgerSkipRecord | None:
+    def resolve_for_delivery(
+        self,
+        *,
+        instance_key: str,
+        portfolio_id: str,
+        side: str,
+        order_id: str,
+        execution_id: str,
+        resolution_reason: str,
+    ) -> list[LedgerSkipRecord]:
+        now = datetime.now(tz=UTC)
+        with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur:
+            cur.execute(
+                "UPDATE ledger_skip SET status='resolved', resolution_reason=%s, "
+                "execution_id=COALESCE(execution_id, %s), "
+                "resolved_at=%s, updated_at=%s "
+                "WHERE status='unresolved' AND instance_key=%s AND portfolio_id=%s AND side=%s "
+                "AND (execution_id=%s OR (execution_id IS NULL AND order_id=%s)) "
+                "RETURNING skip_key, instance_key, slot_id, portfolio_id, symbol, order_id, "
+                "execution_id, side, qty, price, executed_at, order_venue, execution_venue, "
+                "skip_reason, status, resolution_reason, resolved_at, created_at, updated_at",
+                (
+                    resolution_reason,
+                    execution_id,
+                    now,
+                    now,
+                    instance_key,
+                    portfolio_id,
+                    side,
+                    execution_id,
+                    order_id,
+                ),
+            )
+            rows = cur.fetchall()
+            conn.commit()
+        return [self._row_to_record(row) for row in rows]
+
+    def resolve_by_key(
+        self,
+        skip_key: str,
+        *,
+        instance_key: str,
+        resolution_reason: str,
+    ) -> LedgerSkipRecord | None:
         now = datetime.now(tz=UTC)
         with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur:
             cur.execute(
                 "UPDATE ledger_skip SET status='resolved', resolution_reason=%s, "
                 "resolved_at=%s, updated_at=%s "
-                "WHERE skip_key=%s AND status='unresolved' "
+                "WHERE skip_key=%s AND instance_key=%s AND status='unresolved' "
                 "RETURNING skip_key, instance_key, slot_id, portfolio_id, symbol, order_id, "
                 "execution_id, side, qty, price, executed_at, order_venue, execution_venue, "
                 "skip_reason, status, resolution_reason, resolved_at, created_at, updated_at",
-                (resolution_reason, now, now, skip_key),
+                (resolution_reason, now, now, skip_key, instance_key),
             )
             row = cur.fetchone()
             conn.commit()

trex_core/application/execution.py

@@ -831,7 +831,6 @@ class ExecutionService:
         )
         self._ledger_outbox.enqueue(trade)
         r.ledger_enqueued += 1
-        self._resolve_ledger_skip(slot, side, broker_order_id, str(fs.execution_id))
 
     def _persist_ledger_skip(
         self,
@@ -868,26 +867,6 @@ class ExecutionService:
         if should_notify:
             publish_ledger_skip_alarm(self._alarm_emitter, record, resolved=False)
 
-    def _resolve_ledger_skip(
-        self,
-        slot,
-        side: str,
-        broker_order_id: str,
-        execution_id: str | None,
-    ) -> None:
-        if self._ledger_skip_store is None:
-            return
-        records = self._ledger_skip_store.resolve_for_fill(
-            instance_key=self._instance_key,
-            slot_id=slot.slot_id,
-            side=side,
-            order_id=str(broker_order_id) if broker_order_id else None,
-            execution_id=execution_id,
-            resolution_reason="ledger_enqueued",
-        )
-        for record in records:
-            publish_ledger_skip_alarm(self._alarm_emitter, record, resolved=True)
-
     def cancel_pending_slot(self, slot_id) -> str:
         """Operator backstop: send cancel_order for a pending slot (dcy-022/REQ-071).
 

trex_core/application/ledger_outbox.py

@@ -25,10 +25,11 @@ from __future__ import annotations
 
 import logging
 import random
+from collections.abc import Callable
 from dataclasses import dataclass, replace
 from datetime import datetime, timedelta, timezone
 from enum import Enum
-from typing import Callable, Optional, Protocol
+from typing import Optional, Protocol
 
 from trex_core.application.alarm_emitter import AlarmEmitter
 from trex_core.application.ledger_port import (
@@ -39,6 +40,7 @@ from trex_core.application.ledger_port import (
     LedgerTrade,
     LedgerWriteResult,
 )
+from trex_core.application.ledger_skip import publish_ledger_skip_alarm
 
 _log = logging.getLogger(__name__)
 
@@ -171,6 +173,8 @@ class LedgerWriter:
         clock: Optional[Callable[[], datetime]] = None,
         rng: Optional[Callable[[], float]] = None,
         alarm_emitter: Optional[AlarmEmitter] = None,
+        ledger_skip_store=None,
+        instance_key: str | None = None,
     ) -> None:
         self._outbox = outbox
         self._ledger = ledger
@@ -185,6 +189,8 @@ class LedgerWriter:
         self._rng = rng or random.random  # jitter source; inject for determinism
         # 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
 
     def enqueue_fill(self, trade: LedgerTrade) -> LedgerWorkItem:
         return self._outbox.enqueue(trade)
@@ -301,10 +307,32 @@ class LedgerWriter:
         # 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._resolve_ledger_skip(item.trade)
             return OutboxOutcome.DUPLICATE_COMPLETED
         self._outbox.mark_outcome(item.correlation_id, OutboxOutcome.COMPLETED)
+        self._resolve_ledger_skip(item.trade)
         return OutboxOutcome.COMPLETED
 
+    def _resolve_ledger_skip(self, trade: LedgerTrade) -> None:
+        if (
+            self._ledger_skip_store is None
+            or self._instance_key is None
+            or not trade.asset_id
+            or not trade.order_id
+            or not trade.execution_id
+        ):
+            return
+        records = self._ledger_skip_store.resolve_for_delivery(
+            instance_key=self._instance_key,
+            portfolio_id=trade.asset_id,
+            side=trade.side,
+            order_id=trade.order_id,
+            execution_id=trade.execution_id,
+            resolution_reason="ledger_recorded",
+        )
+        for record in records:
+            publish_ledger_skip_alarm(self._alarm_emitter, record, resolved=True)
+
     def _writeback(self, trade: LedgerTrade, outcome) -> None:
         """Option D cash/quantity write-back (dcy-018). best-effort; skips when
         no SnapshotStore is wired or the response omits authoritative values."""

trex_core/runtime/cli.py

@@ -60,6 +60,7 @@ from trex_core.runtime.config import (
 
 _DEFAULT_TIMEOUT = 10.0
 _DEFAULT_LEDGER_BACKFILL_AUDIT_DIR = "artifacts/validation/ledger_backfill"
+_DEFAULT_LEDGER_SKIP_RESOLVE_AUDIT_DIR = "artifacts/validation/ledger_skip_resolve"
 _log = logging.getLogger(__name__)
 
 
@@ -406,6 +407,24 @@ def _write_ledger_backfill_audit(args, report: dict) -> str:
     return str(path)
 
 
+def _ledger_skip_resolve_audit_path(args, skip_key: str) -> Path:
+    base = Path(getattr(args, "audit_dir", _DEFAULT_LEDGER_SKIP_RESOLVE_AUDIT_DIR))
+    stamp = datetime.now(tz=UTC).strftime("%Y%m%dT%H%M%SZ")
+    name = (
+        f"{stamp}--{_sanitize_path_component(_resolve_instance(args))}"
+        f"--{_sanitize_path_component(skip_key)}.json"
+    )
+    return base.resolve() / name
+
+
+def _write_ledger_skip_resolve_audit(args, report: dict) -> str:
+    path = _ledger_skip_resolve_audit_path(args, report["request"]["skip_key"])
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(_json.dumps(report, ensure_ascii=False, indent=2, default=str) + "\n",
+                    encoding="utf-8")
+    return str(path)
+
+
 def _lookup_ledger_backfill_evidence(cfg: DaemonConfig, *, trading_date: str, execution_id: str):
     from trex_core.runtime.broker_provider import build_broker_provider
 
@@ -661,6 +680,14 @@ def _build_parser() -> argparse.ArgumentParser:
     )
     p_skips_resolve.add_argument("--skip-key", required=True)
     p_skips_resolve.add_argument("--reason", default="operator_reconciled")
+    p_skips_resolve.add_argument("--yes", action="store_true", help="skip confirmation")
+    p_skips_resolve.add_argument("--live", action="store_true",
+                                 help="confirm a LIVE ledger skip mutation (with --yes)")
+    p_skips_resolve.add_argument(
+        "--audit-dir",
+        default=_DEFAULT_LEDGER_SKIP_RESOLVE_AUDIT_DIR,
+        help="directory for persistent JSON audit reports",
+    )
     _add_common(p_skips_resolve)
 
     return parser
@@ -1057,13 +1084,61 @@ def _cmd_ledger_skips_list(args, cfg) -> int:
     return 0
 
 
-def _cmd_ledger_skips_resolve(args, cfg) -> int:
+def _cmd_ledger_skips_resolve(args, cfg, confirm) -> int:
     from trex_core.adapters.persistence.ledger_skip_store import PostgreSQLLedgerSkipStore
 
+    instance = _resolve_instance(args)
+    broker_name, is_live = _broker_mode(cfg)
+    report = {
+        "request": {
+            "instance": instance,
+            "broker": broker_name,
+            "live_account": is_live,
+            "skip_key": args.skip_key,
+            "resolution_reason": args.reason,
+        },
+        "result": None,
+        "record": None,
+        "audit_path": None,
+    }
+    if is_live and not getattr(args, "live", False):
+        report["result"] = {"status": "rejected", "reason": "live account requires --live (with --yes)"}
+        report["audit_path"] = _write_ledger_skip_resolve_audit(args, report)
+        _fail("live account requires --live (with --yes)", args.json)
+        if args.json:
+            _emit(report, as_json=True)
+        else:
+            _emit_pretty_json(report)
+        return 2
+    if not getattr(args, "yes", False):
+        if args.json:
+            report["result"] = {"status": "rejected", "reason": "--json ledger skips resolve requires --yes"}
+            report["audit_path"] = _write_ledger_skip_resolve_audit(args, report)
+            _fail("--json ledger skips resolve requires --yes", True)
+            _emit(report, as_json=True)
+            return 2
+        prompt = (
+            f"Resolve skipped ledger record on instance={instance} broker={broker_name}"
+            f"{' LIVE' if is_live else ''}: {args.skip_key}?"
+        )
+        if not confirm(prompt):
+            report["result"] = {"status": "aborted", "reason": "operator declined confirmation"}
+            report["audit_path"] = _write_ledger_skip_resolve_audit(args, report)
+            print("aborted")
+            _emit_pretty_json(report)
+            return 1
     store = PostgreSQLLedgerSkipStore(cfg.dsn)
-    record = store.resolve_by_key(args.skip_key, resolution_reason=args.reason)
+    record = store.resolve_by_key(
+        args.skip_key,
+        instance_key=instance,
+        resolution_reason=args.reason,
+    )
     if record is None:
+        report["result"] = {"status": "not_found", "reason": f"unresolved ledger skip not found: {args.skip_key}"}
+        report["audit_path"] = _write_ledger_skip_resolve_audit(args, report)
         _fail(f"unresolved ledger skip not found: {args.skip_key}", args.json)
+        if args.json:
+            _emit(report, as_json=True)
         return 1
     emitter, transport = _default_alarm_emitter(cfg)
     try:
@@ -1071,8 +1146,11 @@ def _cmd_ledger_skips_resolve(args, cfg) -> int:
     finally:
         transport.close()
     payload = {"result": "resolved", "record": _render_ledger_skip_record(record)}
+    report["result"] = {"status": "resolved"}
+    report["record"] = payload["record"]
+    report["audit_path"] = _write_ledger_skip_resolve_audit(args, report)
     if args.json:
-        _emit(payload, as_json=True)
+        _emit({**payload, "audit_path": report["audit_path"]}, as_json=True)
     else:
         print(f"resolved {record.skip_key} reason={record.resolution_reason}")
     return 0
@@ -1338,7 +1416,7 @@ def main(argv: list[str] | None = None, *, rpc=None, confirm=None) -> int:
         and getattr(args, "ledger_cmd", None) == "skips"
         and getattr(args, "ledger_skips_cmd", None) == "resolve"
     ):
-        return _cmd_ledger_skips_resolve(args, cfg)
+        return _cmd_ledger_skips_resolve(args, cfg, confirm)
 
     # direct-DB control commands (pause/resume stay direct-DB, OQ-3).
     if args.cmd in ("pause", "resume", "paused"):

trex_core/runtime/daemon.py

@@ -250,6 +250,8 @@ class Daemon:
             retry_cap_sec=config.ledger_retry_cap_sec,
             max_attempts=config.ledger_max_attempts,
             alarm_emitter=self.alarm_emitter,  # dcy-033 / REQ-100
+            ledger_skip_store=self.ledger_skip_store,
+            instance_key=config.instance_key,
         )
         self.kill_switch = KillSwitchService(
             slot_repo=self.slot_repo,