From 48dfc3e76f491714050b5910139a141427f37a26 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 24 Aug 2026 13:38:04 +0000 Subject: [PATCH] fix: harden skipped ledger resolution paths Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- .../persistence/test_ledger_skip_store.py | 74 ++++++++ tests/application/test_execution_ledger.py | 105 ++++++++++- tests/runtime/test_cli_ledger_skips.py | 52 ++++- .../adapters/persistence/ledger_skip_store.py | 177 ++++++++++++------ trex_core/application/execution.py | 21 --- trex_core/application/ledger_outbox.py | 30 ++- trex_core/runtime/cli.py | 86 ++++++++- trex_core/runtime/daemon.py | 2 + 8 files changed, 451 insertions(+), 96 deletions(-) diff --git a/tests/adapters/persistence/test_ledger_skip_store.py b/tests/adapters/persistence/test_ledger_skip_store.py index 4235be0..b2c5f65 100644 --- a/tests/adapters/persistence/test_ledger_skip_store.py +++ b/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 diff --git a/tests/application/test_execution_ledger.py b/tests/application/test_execution_ledger.py index 8c7e164..d993c2c 100644 --- a/tests/application/test_execution_ledger.py +++ b/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 = ( - LedgerWriter(outbox, ledger or InMemoryLedger(), clock=clock) - if clock else LedgerWriter(outbox, ledger or InMemoryLedger()) + writer = LedgerWriter( + outbox, + 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 diff --git a/tests/runtime/test_cli_ledger_skips.py b/tests/runtime/test_cli_ledger_skips.py index b67fc35..b152735 100644 --- a/tests/runtime/test_cli_ledger_skips.py +++ b/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(): - return SimpleNamespace(dsn="postgresql://x/y", rabbitmq=None) +def _cfg(*, live: bool = False): + 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" diff --git a/trex_core/adapters/persistence/ledger_skip_store.py b/trex_core/adapters/persistence/ledger_skip_store.py index 62e95a7..14fa399 100644 --- a/trex_core/adapters/persistence/ledger_skip_store.py +++ b/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: - cur.execute( - "SELECT status FROM ledger_skip WHERE skip_key = %s FOR UPDATE", - (skip_key,), - ) - row = cur.fetchone() - if row is None: - notify = True - cur.execute( - "INSERT INTO ledger_skip " - "(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, created_at, updated_at) " - "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, " - " 'unresolved', %s, %s)", - ( - skip_key, - instance_key, - slot_id, - portfolio_id, - symbol, - order_id, - execution_id, - side, - qty, - price, - executed_at, - order_venue or "unknown", - execution_venue or "unknown", - skip_reason, - now, - now, - ), - ) - else: - notify = row[0] == "resolved" - cur.execute( - "UPDATE ledger_skip SET portfolio_id=%s, symbol=%s, order_id=%s, " - "execution_id=%s, qty=%s, price=%s, executed_at=%s, " - "order_venue=%s, execution_venue=%s, status='unresolved', " - "resolution_reason=NULL, resolved_at=NULL, updated_at=%s " - "WHERE skip_key=%s", - ( - portfolio_id, - symbol, - order_id, - execution_id, - qty, - price, - executed_at, - order_venue or "unknown", - execution_venue or "unknown", - now, - skip_key, - ), - ) - conn.commit() + for _ in range(2): + with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur: + try: + cur.execute( + "INSERT INTO ledger_skip " + "(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, created_at, updated_at) " + "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, " + " 'unresolved', %s, %s)", + ( + skip_key, + instance_key, + slot_id, + portfolio_id, + symbol, + order_id, + execution_id, + side, + qty, + price, + executed_at, + order_venue or "unknown", + execution_venue or "unknown", + skip_reason, + now, + now, + ), + ) + conn.commit() + notify = True + break + except self._psycopg.errors.UniqueViolation: + conn.rollback() + cur.execute( + "SELECT status FROM ledger_skip WHERE skip_key = %s FOR UPDATE", + (skip_key,), + ) + row = cur.fetchone() + if row is None: + continue + notify = row[0] == "resolved" + cur.execute( + "UPDATE ledger_skip SET portfolio_id=%s, symbol=%s, order_id=%s, " + "execution_id=%s, qty=%s, price=%s, executed_at=%s, " + "order_venue=%s, execution_venue=%s, status='unresolved', " + "resolution_reason=NULL, resolved_at=NULL, updated_at=%s " + "WHERE skip_key=%s", + ( + portfolio_id, + symbol, + order_id, + execution_id, + qty, + price, + executed_at, + 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() diff --git a/trex_core/application/execution.py b/trex_core/application/execution.py index 5b427fd..bc8d25c 100644 --- a/trex_core/application/execution.py +++ b/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). diff --git a/trex_core/application/ledger_outbox.py b/trex_core/application/ledger_outbox.py index 3e65490..574fe87 100644 --- a/trex_core/application/ledger_outbox.py +++ b/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.""" diff --git a/trex_core/runtime/cli.py b/trex_core/runtime/cli.py index 37da86b..7b276cc 100644 --- a/trex_core/runtime/cli.py +++ b/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"): diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index a7c88a2..f1fc5b7 100644 --- a/trex_core/runtime/daemon.py +++ b/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,