diff --git a/.proofline/criteria/ac-0025.md b/.proofline/criteria/ac-0025.md new file mode 100644 index 0000000..177afd3 --- /dev/null +++ b/.proofline/criteria/ac-0025.md @@ -0,0 +1,19 @@ +--- +id: "ac-0025" +status: active +--- + +# 최초 `reconcile_warned` strategy intent 거부 알림 + +## Criterion + +재조정 경고가 설정된 slot에서 외부 전략 buy/sell intent가 `rejected/reconcile_warned`로 최종 확정되면, `IntentStore.finalize()` 이후에만 `trex.alerts`로 `strategy_intent_rejected`/`warn` 알림을 intent_id 기준 최초 1회 발행해야 한다. 같은 `intent_id` 재요청과 처리 중 중복(`duplicate_in_flight`)은 추가 알림을 만들지 않으며, 같은 slot의 서로 다른 `intent_id`는 각각 독립 발행한다. 알림 payload는 필수 allowlist fields를 항상 포함하고 optional fields는 신뢰 가능한 값이 있을 때만 포함해야 하며, 비허용 필드·민감정보·raw exception body를 alarm/log에 포함해서는 안 된다. emitter 부재/전송 실패/optional context 조회 실패는 저장된 거부 결과와 RPC 응답, broker·slot·warning·ledger·portfolio-hub 동작을 바꾸지 않는 best-effort 경계여야 한다. daemon은 기존 공용 `AlarmEmitter`를 `IntentService`에 주입하고, emitter 없이 `IntentService`를 생성하는 기존 경로는 계속 호환되어야 한다. + +## Verification + +- 성공 조건: warned slot의 새 buy/sell intent 각각에서 finalize 후 `strategy_intent_rejected`/`warn` 알림이 한 건 발행된다. +- dedup 조건: 같은 `intent_id` 재시도(동일 process 및 durable-store 재구성 후)에서 추가 알림이 없고, 같은 slot의 다른 `intent_id`는 각각 한 건 발행한다. +- 예외 경계: `duplicate_in_flight`, admitted, `reconcile_warning_unavailable`, `capacity_full`, validation rejection, 그 밖의 non-`reconcile_warned` rejection에서는 알림이 없어야 한다. +- 순서/실패 조건: finalize 실패 시 알림이 없어야 하며 finalize 완료 전에 publish가 실행되면 FAIL이다. optional context 조회 실패·publisher false·publisher exception·emitter 부재에서도 RPC 응답은 `rejected/reconcile_warned`를 유지해야 한다. +- payload/보안 조건: 필수 fields와 `broker_submitted=false`가 포함되고, optional fields는 확인 가능 시에만 포함된다. buy/sell strategy id는 공통 `strategy_id`로 정규화되며 비허용 필드와 민감정보·raw exception body가 alarm/log에 없어야 한다. +- wiring/회귀 조건: daemon이 공용 `AlarmEmitter`를 `IntentService`에 전달하고, emitter 없는 직접 생성 경로 및 기존 intent dedup/alarm 관련 회귀가 유지되어야 한다. diff --git a/.proofline/identities.json b/.proofline/identities.json index efe7dfe..484c3db 100644 --- a/.proofline/identities.json +++ b/.proofline/identities.json @@ -1,5 +1,5 @@ { "schema_version": 1, - "next_line_number": 8, - "next_ac_number": 25 + "next_line_number": 9, + "next_ac_number": 26 } diff --git a/.proofline/lines/line-0008/dcy-0008.md b/.proofline/lines/line-0008/dcy-0008.md new file mode 100644 index 0000000..3c6d62c --- /dev/null +++ b/.proofline/lines/line-0008/dcy-0008.md @@ -0,0 +1,30 @@ +--- +id: "dcy-0008" +status: confirmed +--- + +# 최초 `reconcile_warned` intent 거부 알림 + +## Problem + +`IntentService`는 재조정 경고(`reconcile_warned`)가 남아 있는 slot의 전략 buy/sell intent를 거부하지만, 현재 경고 알림을 발행하지 않아 운영자가 최초 거부를 즉시 인지하기 어렵다. 기존 intent dedup 기준(`IntentStore.try_insert/get/finalize`)과 차단 정책은 유지되어야 하며, 같은 `intent_id` 재시도에서는 추가 알림이 없어야 한다. + +## Evidence + +- `trex_core/application/intents.py`는 `reconcile_warned` 거부 결과를 `IntentStore.finalize()`로 저장하지만 `trex.alerts` 알림 발행이 없다. +- daemon은 공용 `AlarmEmitter`를 이미 생성해 다른 서비스에 주입하지만 `IntentService`에는 연결하지 않는다. +- issue #74와 승인된 REQ-0008/AC-0025는 finalize 이후 최초 거부 1회 알림, allowlist payload, optional context 조회 실패/발행 실패의 비차단 경계를 요구한다. + +## Scope + +- 대상은 외부 전략 buy/sell intent의 `rejected/reconcile_warned` 결과다. +- `IntentStore.finalize()` 성공 뒤에만 `severity=warn`, `code=strategy_intent_rejected` 알림을 best-effort로 발행한다. +- 필수 fields(`instance_key`, `intent_id`, `intent_type`, `reason`, `broker_submitted=false`)와 확인된 optional fields(`slot_id`, `portfolio_id`, `symbol`, `strategy_id`)만 포함한다. +- 같은 `intent_id`는 기존 durable dedup 결과를 재사용해 추가 발행하지 않고, 다른 `intent_id`는 독립 발행한다. +- 발행기 부재/실패/optional context 조회 실패는 저장 결과와 RPC 응답을 변경하지 않는다. + +## Out of Scope + +- AC-0023/AC-0024의 warning 판정·clear authority 변경 +- broker 주문/취소, slot/warning/ledger/portfolio-hub write side effect 추가 +- 신규 durable outbox, publish retry persistence, restart 후 보장 재발행 diff --git a/.proofline/lines/line-0008/line-0008.md b/.proofline/lines/line-0008/line-0008.md new file mode 100644 index 0000000..286a9f6 --- /dev/null +++ b/.proofline/lines/line-0008/line-0008.md @@ -0,0 +1,3 @@ +--- +id: "line-0008" +--- diff --git a/.proofline/lines/line-0008/req-0008.md b/.proofline/lines/line-0008/req-0008.md new file mode 100644 index 0000000..4cb1edf --- /dev/null +++ b/.proofline/lines/line-0008/req-0008.md @@ -0,0 +1,57 @@ +--- +id: "req-0008" +status: approved +discovery: "dcy-0008" +criteria: + create: + - "ac-0025" + update: [] + retire: [] + satisfy: [] +--- + +# 최초 `reconcile_warned` intent 거부 알림 발행 + +## Objective + +`trex-core`는 재조정 경고가 남아 있는 slot에 대한 외부 전략 buy/sell intent가 `rejected/reconcile_warned`로 확정될 때, 기존 idempotency와 차단 정책을 유지하면서 새 `intent_id`의 최초 거부에 한해 `trex.alerts` 경고를 한 건 발행해야 한다. + +## Scope + +- 경고 발행 조건 + - `IntentStore.try_insert()`가 성공한 최초 처리여야 한다. + - 외부 전략 buy 또는 sell intent여야 한다. + - 최종 결과가 `status="rejected"`, `reason="reconcile_warned"`여야 한다. + - `IntentStore.finalize()`가 성공한 뒤여야 한다. +- 경고 계약 + - `severity=warn` + - `code=strategy_intent_rejected` +- dedup 경계 + - 같은 `intent_id` 재요청은 저장된 기존 결과를 반환하고 경고를 재발행하지 않는다. + - `duplicate_in_flight`와 최초 처리 중 중복 요청은 경고 대상이 아니다. + - 같은 slot이라도 `intent_id`가 다르면 독립 발행한다. + - process-local set, 별도 notification table, `reconcile_notification_state`를 추가하지 않는다. +- payload allowlist + - 필수: `instance_key`, `intent_id`, `intent_type`, `reason="reconcile_warned"`, `broker_submitted=false` + - optional(확인 가능 시에만): `slot_id`, `portfolio_id`, `symbol`, `strategy_id` + - buy `strategy_id`와 sell `sell_strategy_id`는 공통 `strategy_id`로 표현한다. + - `quantity`, `entry_price`, `limit_price`, `venue`, raw request/strategy `extra`는 포함하지 않는다. +- failure isolation + - emitter 부재/실패, optional context 조회 실패는 저장 결과와 RPC 응답을 바꾸지 않는다. + - 이미 확정된 거부를 rollback하거나 admitted로 바꾸지 않는다. +- production wiring + - `IntentService`는 optional `AlarmEmitter`를 받는다. + - daemon은 기존 공용 `AlarmEmitter`를 `IntentService`에 전달한다. + - emitter 없이 생성하는 기존 테스트/사용 경로는 계속 동작한다. + +## Constraints + +- AC-0023·AC-0024의 기존 warning 판정, clear authority, intent dedup semantics를 변경하지 않는다. +- 경고 발행을 위해 broker/slot/warning/ledger/portfolio-hub side effect를 추가하지 않는다. +- 경고/로그에 account number/hash, credential/token/password/secret, DSN/connection string, raw payload, exception 본문을 노출하지 않는다. + +## Non-Goals + +- 다른 거부 사유를 위한 범용 경고 정책 추가 +- safety-sell/kill-switch/watchdog 내부 경로 신규 알림 +- notifier 렌더링/채널, RabbitMQ topology, production 배포/릴리즈 절차 변경 diff --git a/tests/application/test_alarm_emitter.py b/tests/application/test_alarm_emitter.py index 3ef4e73..24507d8 100644 --- a/tests/application/test_alarm_emitter.py +++ b/tests/application/test_alarm_emitter.py @@ -15,6 +15,7 @@ import pytest from trex_core.adapters.messaging.in_memory import InMemoryTransport from trex_core.application.alarm_emitter import AlarmEmitter +from trex_core.application.messaging import TransportUnavailableError # ------------------------------------------------------------------ helpers @@ -154,3 +155,40 @@ def test_alarm_emitter_no_telegram_reference(): source = inspect.getsource(ae_mod) assert "TELEGRAM" not in source, "alarm_emitter must not reference TELEGRAM_*" assert "telegram" not in source.lower(), "alarm_emitter must not reference telegram" + + +def test_publish_alarm_transport_unavailable_log_is_sanitized( + caplog: pytest.LogCaptureFixture, +): + class UnavailableTransport: + def publish_alarm_event(self, routing_key, payload): + raise TransportUnavailableError("secret-token-value") + + def health(self): + from trex_core.application.messaging import TransportHealth + return TransportHealth.DISCONNECTED + + emitter = AlarmEmitter(UnavailableTransport()) # type: ignore[arg-type] + with caplog.at_level(logging.ERROR, logger="trex_core.application.alarm_emitter"): + emitter.publish_alarm("strategy_intent_rejected", "alarm title") + + assert any("transport unavailable" in r.message for r in caplog.records) + assert "secret-token-value" not in caplog.text + + +def test_publish_alarm_generic_exception_log_is_sanitized(caplog: pytest.LogCaptureFixture): + class BrokenTransport: + def publish_alarm_event(self, routing_key, payload): + raise RuntimeError("******") + + def health(self): + from trex_core.application.messaging import TransportHealth + return TransportHealth.CONNECTED + + emitter = AlarmEmitter(BrokenTransport()) # type: ignore[arg-type] + with caplog.at_level(logging.ERROR, logger="trex_core.application.alarm_emitter"): + emitter.publish_alarm("strategy_intent_rejected", "alarm title") + + assert any("alarm publish failed" in r.message for r in caplog.records) + assert "RuntimeError" in caplog.text + assert "******" not in caplog.text diff --git a/tests/application/test_reconcile_warning_blocking.py b/tests/application/test_reconcile_warning_blocking.py index baf1af3..c1c5f8b 100644 --- a/tests/application/test_reconcile_warning_blocking.py +++ b/tests/application/test_reconcile_warning_blocking.py @@ -101,6 +101,24 @@ class DuplicateInFlightIntentStore: raise AssertionError("finalize must not be called for duplicate_in_flight") +class ContextFailingSlotRepository(InMemorySlotRepository): + """Raise only on the N+1th fetch_slot to simulate post-finalize lookup failure.""" + + def __init__(self, *, fail_on_fetch_call_no=None): + super().__init__() + self._fetch_calls = 0 + self._fail_on_fetch_call_no = fail_on_fetch_call_no + + def fetch_slot(self, slot_id): + self._fetch_calls += 1 + if self._fail_on_fetch_call_no == self._fetch_calls: + raise RuntimeError("context lookup failed") + return super().fetch_slot(slot_id) + + def fail_on_next_fetch(self): + self._fail_on_fetch_call_no = self._fetch_calls + 1 + + def _make_holding_slot(repo, svc, instance_key="default", portfolio_id="pf-A", symbol="005930", qty=10): """Helper: creates universe, puts slot in HOLDING state with given qty.""" repo.reconcile_universe(instance_key, {portfolio_id: symbol}) @@ -512,6 +530,60 @@ def test_reconcile_warned_alarm_finalize_failure_and_publish_failure_are_isolate assert d3.reason == "reconcile_warned" +def test_reconcile_warned_buy_context_lookup_failure_does_not_change_response(): + repo = ContextFailingSlotRepository(fail_on_fetch_call_no=2) + svc = SlotService(repo) + warn_store = InMemoryReconcileWarningStore() + intent_store = InMemoryIntentStore() + alarms = RecordingAlarmEmitter() + repo.reconcile_universe("default", {"pf-A": "005930"}) + slot = repo.find_slot_by_portfolio("default", "pf-A") + warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="B") + intent_svc = IntentService( + svc, repo, intent_store, instance_key="default", warn_store=warn_store, alarm_emitter=alarms + ) + + decision = intent_svc.submit_buy( + BuyIntent(intent_id="buy-context-fail", slot_id=slot.slot_id, strategy_id="stg-buy") + ) + assert decision.status == "rejected" + assert decision.reason == "reconcile_warned" + assert len(alarms.calls) == 1 + _, _, _, fields = alarms.calls[0] + assert fields["intent_id"] == "buy-context-fail" + assert fields["strategy_id"] == "stg-buy" + assert "slot_id" not in fields + assert "portfolio_id" not in fields + assert "symbol" not in fields + + +def test_reconcile_warned_sell_context_lookup_failure_does_not_change_response(): + repo = ContextFailingSlotRepository() + svc = SlotService(repo) + warn_store = InMemoryReconcileWarningStore() + intent_store = InMemoryIntentStore() + alarms = RecordingAlarmEmitter() + slot = _make_holding_slot(repo, svc, qty=10, portfolio_id="pf-A", symbol="005930") + repo._fail_on_fetch_call_no = repo._fetch_calls + 2 + warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="C") + intent_svc = IntentService( + svc, repo, intent_store, instance_key="default", warn_store=warn_store, alarm_emitter=alarms + ) + + decision = intent_svc.submit_sell( + SellIntent(intent_id="sell-context-fail", slot_id=slot.slot_id, sell_strategy_id="stg-sell") + ) + assert decision.status == "rejected" + assert decision.reason == "reconcile_warned" + assert len(alarms.calls) == 1 + _, _, _, fields = alarms.calls[0] + assert fields["intent_id"] == "sell-context-fail" + assert fields["strategy_id"] == "stg-sell" + assert "slot_id" not in fields + assert "portfolio_id" not in fields + assert "symbol" not in fields + + # REQ-104 AC-3: no warn, no pause => buy/sell admitted normally def test_intent_service_no_warn_admitted_normally(): repo = InMemorySlotRepository() diff --git a/trex_core/application/alarm_emitter.py b/trex_core/application/alarm_emitter.py index 37b2a7a..679b69f 100644 --- a/trex_core/application/alarm_emitter.py +++ b/trex_core/application/alarm_emitter.py @@ -17,6 +17,7 @@ where no broker connection exists. In that case publish_alarm() logs and skips. from __future__ import annotations import logging +import sys from datetime import datetime, timedelta, timezone from trex_core.adapters.messaging.topology import alarm_routing_key @@ -84,15 +85,17 @@ class AlarmEmitter: try: self._transport.publish_alarm_event(routing_key, payload) return True - except TransportUnavailableError as exc: + except TransportUnavailableError: _log.error( - "alarm publish skipped (transport unavailable): code=%s: %s", - code, exc, + "alarm publish skipped (transport unavailable): code=%s title=%s", + code, title, ) return False except Exception: # noqa: BLE001 — alarm must never crash the caller + exc = sys.exc_info()[1] + exc_name = type(exc).__name__ if exc is not None else "Exception" _log.error( - "alarm publish failed: code=%s title=%s", - code, title, exc_info=True, + "alarm publish failed: code=%s title=%s error_type=%s", + code, title, exc_name, ) return False diff --git a/trex_core/application/intents.py b/trex_core/application/intents.py index 2f7cea4..1f4bf2e 100644 --- a/trex_core/application/intents.py +++ b/trex_core/application/intents.py @@ -183,20 +183,11 @@ class IntentService: intent.intent_id, decision.status, decision.slot_id, decision.reason ) if decision.status == "rejected" and decision.reason == "reconcile_warned": - slot = ( - self._repo.fetch_slot(intent.slot_id) - if intent.slot_id is not None - else ( - self._repo.find_slot_by_portfolio(self._instance_key, intent.portfolio_id) - if intent.portfolio_id - else None - ) - ) self._publish_reconcile_warned_rejection_alarm( intent_id=intent.intent_id, intent_type="buy", strategy_id=intent.strategy_id, - slot=slot, + slot=self._resolve_buy_alarm_slot(intent), ) return decision @@ -221,10 +212,28 @@ class IntentService: intent_id=intent.intent_id, intent_type="sell", strategy_id=intent.sell_strategy_id, - slot=self._repo.fetch_slot(intent.slot_id), + slot=self._resolve_sell_alarm_slot(intent), ) return decision + def _resolve_buy_alarm_slot(self, intent: BuyIntent) -> Any: + try: + if intent.slot_id is not None: + return self._repo.fetch_slot(intent.slot_id) + if intent.portfolio_id: + return self._repo.find_slot_by_portfolio( + self._instance_key, intent.portfolio_id + ) + except Exception: # noqa: BLE001 - optional alarm context lookup + return None + return None + + def _resolve_sell_alarm_slot(self, intent: SellIntent) -> Any: + try: + return self._repo.fetch_slot(intent.slot_id) + except Exception: # noqa: BLE001 - optional alarm context lookup + return None + def _publish_reconcile_warned_rejection_alarm( self, *,