diff --git a/docs/batches/bat-049_20260621_alarm-system-dcy-033/design.md b/docs/batches/bat-049_20260621_alarm-system-dcy-033/design.md new file mode 100644 index 0000000..0ccfbb0 --- /dev/null +++ b/docs/batches/bat-049_20260621_alarm-system-dcy-033/design.md @@ -0,0 +1,120 @@ +# BAT-049 Design + +- Status: confirmed +- Batch: BAT-049 + +## Architecture Summary + +알람 시스템을 application 레이어의 `AlarmEmitter`(포트 경유) + adapters 레이어의 `publish_alarm_event()` 확장으로 구현한다. 기존 레이어드 아키텍처(domain/application/adapters/runtime)와 의존성 역전 규칙을 그대로 따른다. + +``` +발생 지점(execution.py 등) + │ alarm_emitter.publish_alarm(code, title, fields) + ▼ +AlarmEmitter (application 레이어) + │ MessagingTransport.publish_alarm_event(routing_key, payload) + ▼ +RabbitMQTransport.publish_alarm_event() (adapters 레이어) + │ basic_publish(exchange=trex.alerts, routing_key=alarm.critical.trex-core, mandatory=False) + ▼ +trex.alerts exchange → (notifier가 alarm.# 패턴으로 바인딩) +``` + +`AlarmEmitter`는 `MessagingTransport` 포트만 주입받는다. pika 직접 import 없음. domain 레이어 접촉 없음. + +## Changed Areas + +| 파일 | 변경 유형 | 내용 | +|---|---|---| +| `docs/interface-contract.md` | 추가(기존 불변) | `Interface: trex.alerts (알람 채널, dcy-033)` 섹션 신규 | +| `trex_core/adapters/messaging/topology.py` | 추가 | `ALERTS_EXCHANGE = "trex.alerts"` 상수 추가 | +| `trex_core/adapters/messaging/rabbitmq.py` | 확장 | `declare_topology()`에 trex.alerts exchange 선언 + `publish_alarm_event()` 메서드 신규 | +| `trex_core/adapters/messaging/in_memory.py` | 확장 | `publish_alarm_event()` stub 추가 (발행 기록 + unavailable 시 TransportUnavailableError) | +| `trex_core/application/messaging.py` | 확장 | `MessagingTransport` Protocol에 `publish_alarm_event()` 추가 | +| `trex_core/application/alarm_emitter.py` | 신규 | `AlarmEmitter` 클래스 구현 | +| `trex_core/application/execution.py` | 배선 | `fill_price_unresolvable`, `sell_skipped_no_balance` 알람 호출 추가 | +| `trex_core/application/reconciliation.py` | 배선 | `core_holding_broker_flat` 알람 호출 추가 | +| `trex_core/application/ledger_outbox.py` | 배선 | `ledger_delivery_failed` 알람 호출 추가 | +| `trex_core/application/killswitch.py` | 배선 | `kill_switch_fired` 알람 호출 추가 | +| `trex_core/adapters/broker/rabbitmq.py`(`messaging/rabbitmq.py`) | 배선 | `publish_unavailable_after_reconnect` / `dlq_publish_failed` — 콜백 패턴 | +| `trex_core/application/portfolio.py` / `snapshot_scheduler.py` | 배선 | `portfolio_hub_unavailable` 알람 호출 추가 | +| `trex_core/runtime/reset.py` | 배선 | `schema_verification_failed` / `migration_failed` 알람 호출 (transport None 안전) | +| `trex_core/runtime/daemon.py` | 조립 | `AlarmEmitter` 생성 + 각 서비스에 주입 | +| `tests/application/test_alarm_emitter.py` | 신규 | AlarmEmitter 단위 테스트 | +| `tests/adapters/messaging/test_topology.py` | 확장 | declare_topology trex.alerts assertion | +| `docs/runtime-flows.md` | 추가(기존 불변) | 알람 발신 흐름 섹션 신규 | + +## Key Decisions + +### KD-1: publish_unavailable_after_reconnect / dlq_publish_failed — 콜백 패턴 채택 + +**문제**: 두 alarm code는 `trex_core/adapters/messaging/rabbitmq.py`(어댑터) 내부에서 발생하지만, `AlarmEmitter`는 application 레이어에 있다. 어댑터가 application을 직접 import하면 의존성이 역전된다. + +**채택 결정: 콜백(callable) 주입** +- `RabbitMQTransport`에 `alarm_callback: Optional[Callable[[str, str, dict], None]]` 파라미터를 추가한다. 시그니처: `(code: str, title: str, fields: dict) -> None`. +- `daemon.py`에서 `AlarmEmitter.publish_alarm`을 콜백으로 주입한다. +- 어댑터는 application 타입을 import하지 않고, callable 타입 힌트(`Optional[Callable]`)만 사용. +- 콜백 미주입 시(기본값 None): 기존 동작 유지 (로그만). +- 콜백이 raise하면 어댑터는 except로 삼킨다(best-effort 원칙 일관). + +**기각 대안**: +- 이벤트 버스/큐: 과도한 복잡성, 결정론적 단위 테스트 어려움. +- AlarmEmitter를 adapters 레이어로 이동: domain/application 순수성 원칙 위반. +- Transport 프로토콜에 alarm_callback getter 추가: Protocol 오염 + 테스트 어려움. + +### KD-2: reset.py transport None 안전 — AlarmEmitter None-guard 패턴 + +**문제**: `reset.py`는 CLI 진입점으로 `MessagingTransport`가 없다. AlarmEmitter에 None transport를 전달하면 호출 시 crash. + +**채택 결정**: `AlarmEmitter.__init__`이 `Optional[MessagingTransport]`를 받고, `publish_alarm()` 내부에서 transport가 None이면 WARN 로그 후 skip. +- `reset.py`는 `AlarmEmitter(transport=None)`로 생성하거나 `publish_alarm()` 호출 전 None 체크. +- 실제로는 `run_reset()` 함수가 `AlarmEmitter`를 직접 받는 시그니처로 확장하거나, reset 전용 경량 emitter를 분리한다. +- 가장 단순한 방법: `AlarmEmitter(transport=None)` 허용 + None guard → reset 코드에서 생성 부담 최소화. + +### KD-3: publish_alarm_event 시그니처 — transport 레이어 알람 전용 메서드 + +**결정**: `MessagingTransport` 프로토콜에 `publish_alarm_event(routing_key: str, payload: dict) -> None` 추가. +- `publish_event()`와 구분하는 이유: alarm은 `trex.alerts` exchange, `mandatory=False`, `delivery_mode=1`(best-effort, 비영속) vs events는 `trex.events`, `mandatory=True`, `delivery_mode=2`. +- `InMemoryTransport`: 발행된 alarm을 `_alarm_events: list[tuple[str, dict]]`에 기록 + unavailable 시 TransportUnavailableError 발생(호출자가 삼킴). +- `RabbitMQTransport`: `publish_alarm_event()`는 `basic_publish(exchange=trex.alerts, mandatory=False, delivery_mode=1)`. 연결 실패 시 TransportUnavailableError raise(AlarmEmitter가 삼킴). retry-on-failure 없음(best-effort 원칙). + +### KD-4: AlarmEmitter 인터페이스 + +```python +class AlarmEmitter: + def __init__(self, transport: Optional[MessagingTransport]) -> None: ... + + def publish_alarm( + self, + code: str, + title: str, + *, + fields: dict | None = None, + severity: str = "critical", + ) -> None: + """Best-effort 발행. 실패 시 ERROR 로그 후 예외 삼킴.""" + ... +``` + +- `source="trex-core"`, `schema_version="1"` 상수화. +- `occurred_at` = 발행 시점 KST ISO-8601. +- routing key = `alarm.{severity}.trex-core`. +- 발행 실패(TransportUnavailableError, 기타 모든 예외): ERROR 로그 후 예외 삼킴(re-raise 금지). + +## Edge Cases + +- **transport disconnect 중 알람 발행**: `publish_alarm_event()` → TransportUnavailableError → AlarmEmitter가 ERROR 로그 후 삼킴. 본업 경로 비차단. +- **rabbitmq.py 콜백이 None인 경우**: `alarm_callback` guard로 skip. 기존 코드 변경 없이 하위호환. +- **reset.py에서 AlarmEmitter(transport=None)**: None guard 분기 → INFO 로그("alarm skipped: no transport") 후 skip. +- **mandatory=False + notifier 미연결**: silent drop. UnroutableError 없음. +- **8개 code 중 일부 경로에서 AlarmEmitter 미주입**: daemon.py에서 전부 주입. 미주입 시 AttributeError → execution_loop except로 삼킴(기존 보호 패턴). +- **AlarmEmitter.publish_alarm 내 예외 루프**: 각 호출마다 독립 try/except. 한 코드 발행 실패가 동일 tick의 다른 호출에 영향 없음. + +## Architecture Impact + +`both` — runtime-flows.md + interface-contract.md 양쪽 업데이트 필요. + +## Reference Doc Update Plan + +- `docs/interface-contract.md`: `Interface: trex.alerts (알람 채널, dcy-033)` 섹션 추가 + Change Log 갱신 + Last Updated 갱신. +- `docs/runtime-flows.md`: `Flow: 알람 발신 (trex.alerts, dcy-033)` 섹션 추가. 기존 섹션 불변. diff --git a/docs/batches/bat-049_20260621_alarm-system-dcy-033/implementation.md b/docs/batches/bat-049_20260621_alarm-system-dcy-033/implementation.md new file mode 100644 index 0000000..18c536b --- /dev/null +++ b/docs/batches/bat-049_20260621_alarm-system-dcy-033/implementation.md @@ -0,0 +1,92 @@ +# BAT-049 Implementation + +- Status: completed +- Batch: BAT-049 + +## Plan Summary + +REQ-098(문서) → REQ-099(topology+transport) → REQ-100(AlarmEmitter+배선) → REQ-101(runtime-flows 정합) 순서로 구현. +design.md의 4개 Key Decision을 그대로 따름. + +## Changed Files + +| 파일 | 변경 유형 | 내용 | +|---|---|---| +| `docs/interface-contract.md` | 섹션 추가 + 헤더 갱신 | `Interface: trex.alerts (알람 채널, dcy-033)` 신규 섹션 (7필드 스키마, routing key, topology, CRITICAL code 목록, Update Trigger). Change Log + Last Updated 갱신. 기존 섹션 불변. | +| `trex_core/adapters/messaging/topology.py` | 추가 | `ALERTS_EXCHANGE = "trex.alerts"` + `alarm_routing_key()` 함수 | +| `trex_core/adapters/messaging/rabbitmq.py` | 확장 | `declare_topology()` trex.alerts 선언 추가. `publish_alarm_event()` 신규. `set_alarm_callback()` + `_fire_alarm_callback()` 신규. 콜백 호출 배선 (publish_unavailable_after_reconnect, dlq_publish_failed). `Callable` import 추가. | +| `trex_core/adapters/messaging/in_memory.py` | 확장 | `_alarm_events` 리스트 추가. `publish_alarm_event()` stub + `alarm_events()` 조회. | +| `trex_core/application/messaging.py` | 확장 | `MessagingTransport` Protocol에 `publish_alarm_event()` 추가. | +| `trex_core/application/alarm_emitter.py` | 신규 | `AlarmEmitter` 클래스. envelope 구성, KST occurred_at, 예외 삼킴+ERROR 로그, None transport skip. | +| `trex_core/application/execution.py` | 배선 | `AlarmEmitter` import + `alarm_emitter` 파라미터 + `sell_skipped_no_balance` + `fill_price_unresolvable` 발행 배선. | +| `trex_core/application/reconciliation.py` | 배선 | `AlarmEmitter` import + `alarm_emitter` 파라미터 + `core_holding_broker_flat` 발행 배선. | +| `trex_core/application/ledger_outbox.py` | 배선 | `AlarmEmitter` import + `alarm_emitter` 파라미터 + `ledger_delivery_failed` 3지점(conflict_blocked, permanent_failure, dead_lettered) 배선. | +| `trex_core/application/killswitch.py` | 배선 | `AlarmEmitter` import + `alarm_emitter` 파라미터 + `kill_switch_fired` 배선. | +| `trex_core/application/snapshot_scheduler.py` | 배선 | `AlarmEmitter` import + `alarm_emitter` 파라미터 + `portfolio_hub_unavailable` 2지점(whoami_failed, get_portfolios_failed) 배선. `reconcile_core_holdings` 호출에 `alarm_emitter` 전달. | +| `trex_core/runtime/reset.py` | 배선 | `_emit_reset_alarm()` 헬퍼 추가. `run_reset()` alarm_emitter 파라미터 추가. migration 예외 + schema_ok=False 지점 배선. | +| `trex_core/runtime/daemon.py` | 조립 | `AlarmEmitter` import. `alarm_emitter` 생성 (event_publisher 이후). `set_alarm_callback()` 주입. `LedgerWriter`, `KillSwitchService`, `SnapshotScheduler`, `ExecutionService` 생성 시 alarm_emitter 주입. 서비스 생성 순서 재배치(alarm_emitter 생성 후 주입 보장). | +| `tests/application/test_alarm_emitter.py` | 신규 | AlarmEmitter 단위 테스트 29개 (8 code × envelope 검증, 8 code × TransportUnavailableError 삼킴, 8 code × 임의 예외 삼킴, None transport skip, fields default, routing key, pika 미import, TELEGRAM 미참조). | +| `tests/adapters/messaging/test_topology.py` | 확장 | `test_alerts_exchange_constant`, `test_alarm_routing_key_format` 추가. | +| `tests/adapters/messaging/test_in_memory_transport.py` | 확장 | `test_publish_alarm_event_records_event`, `test_publish_alarm_event_raises_when_unavailable` 추가. | +| `docs/runtime-flows.md` | 섹션 추가 + 헤더 갱신 | `Flow: 알람 발신 (trex.alerts, dcy-033)` 신규 섹션 (발행 흐름, best-effort 정책, 8종 code 목록, trex.events 차이표). 기존 섹션 불변. | + +## Design Decisions Implemented + +**KD-1 (rabbitmq.py 어댑터 alarm 콜백 패턴)**: +- `RabbitMQTransport.__init__`에 `alarm_callback: Optional[Callable]` 파라미터 추가. +- `set_alarm_callback()` setter 추가 (늦은 주입 — daemon.py에서 AlarmEmitter 생성 후 바인딩). +- `_fire_alarm_callback()` 내부 헬퍼: callback None guard + 예외 삼킴. +- `publish_event()` retry 실패 시 `publish_unavailable_after_reconnect` 콜백 호출. +- `_publish_dlq()` DLQ 실패 시 `dlq_publish_failed` 콜백 호출. + +**KD-2 (reset.py transport None 안전)**: +- `_emit_reset_alarm(alarm_emitter, code, title, fields)` 헬퍼: None guard + 예외 삼킴. +- `run_reset()` 시그니처에 `alarm_emitter: Optional[object] = None` 추가. +- migration 예외(raise 전)와 schema_ok=False 지점에서 헬퍼 호출. + +**KD-3 (publish_alarm_event 별도 메서드)**: +- `trex.alerts` exchange, `mandatory=False`, `delivery_mode=1` (non-persistent). +- `MessagingTransport` Protocol + `RabbitMQTransport` + `InMemoryTransport` 모두 구현. + +**KD-4 (AlarmEmitter 인터페이스)**: +- `AlarmEmitter(transport: Optional[MessagingTransport])`. +- `publish_alarm(code, title, *, fields=None, severity="critical")`. +- `source="trex-core"`, `schema_version="1"` 고정. +- `occurred_at` = KST ISO-8601. + +## Execution Log + +1. docs/interface-contract.md — trex.alerts 섹션 추가 (REQ-098 완료) +2. topology.py — ALERTS_EXCHANGE + alarm_routing_key() (REQ-099 일부) +3. messaging.py — publish_alarm_event Protocol 추가 (REQ-099 일부) +4. rabbitmq.py — declare_topology + publish_alarm_event + 콜백 패턴 (REQ-099 완료 + REQ-100 KD-1) +5. in_memory.py — publish_alarm_event stub (REQ-099 완료) +6. alarm_emitter.py — AlarmEmitter 신규 (REQ-100 핵심) +7. execution.py — fill_price_unresolvable + sell_skipped_no_balance (REQ-100 배선) +8. reconciliation.py — core_holding_broker_flat (REQ-100 배선) +9. ledger_outbox.py — ledger_delivery_failed (REQ-100 배선) +10. killswitch.py — kill_switch_fired (REQ-100 배선) +11. snapshot_scheduler.py — portfolio_hub_unavailable (REQ-100 배선) +12. reset.py — schema_verification_failed + migration_failed (REQ-100 배선) +13. daemon.py — AlarmEmitter 생성·주입 + 서비스 생성 순서 재배치 (REQ-100 조립) +14. runtime-flows.md — 알람 발신 흐름 섹션 추가 (REQ-101 완료) +15. 테스트 작성 (test_alarm_emitter.py 29개, topology/in_memory 추가) + +## Validation + +``` +pytest tests/application/test_alarm_emitter.py -v +→ 29 passed, 0 failed, 0 skipped + +pytest --tb=short -q (전체) +→ 608 passed, 47 skipped, 0 failed + +ruff check trex_core/ tests/ +→ All checks passed! +``` + +## Remaining Risks + +- R-1(Low): `trex.alerts` exchange 선언은 실 RabbitMQ에서만 검증 가능 (단위에서 mock). 팀장이 실인프라 통합 테스트 수행 예정. +- R-2(Low): daemon.py에서 event_publisher가 `InMemoryTransport`인 경우 `set_alarm_callback()`이 없어 `hasattr` guard 처리됨 (InMemoryTransport는 콜백 불필요). +- R-3(Low): reset.py의 `alarm_emitter` 파라미터가 `main()`(CLI 진입점)에서 None으로 실행됨 — 의도된 동작이나 실인프라 CLI 시나리오에서 알람 미발행. diff --git a/docs/batches/bat-049_20260621_alarm-system-dcy-033/index.md b/docs/batches/bat-049_20260621_alarm-system-dcy-033/index.md new file mode 100644 index 0000000..81f3bcc --- /dev/null +++ b/docs/batches/bat-049_20260621_alarm-system-dcy-033/index.md @@ -0,0 +1,52 @@ +# BAT-049: 알람 시스템 — 알람 envelope wire 계약 + trex.alerts exchange + core CRITICAL emitter + baseline 정합 (dcy-033) + +- Status: release-candidate +- Profile: standard +- Source Discovery: dcy-033 +- Included REQ: REQ-098 (Interface), REQ-099 (Integration), REQ-100 (Integration), REQ-101 (Documentation) +- Excluded REQ: 없음 + +## Summary +- dcy-033(제품군 알람 전달 체계) confirmed Discovery에서 확정된 알람 시스템 1단계 구현. +- REQ-098: 알람 envelope 표준 JSON 스키마 wire 계약을 `docs/interface-contract.md`에 문서화. +- REQ-099: `trex.alerts` (topic, durable=True) AMQP exchange를 `declare_topology()`에 멱등 선언 추가. +- REQ-100: application 레이어 `AlarmEmitter` + CRITICAL code 8종 발생 지점 배선. +- REQ-101: `docs/runtime-flows.md`에 알람 발신 흐름 섹션 추가 + baseline 문서 정합. + +## Scope +- In Scope: + - `docs/interface-contract.md` — `Interface: trex.alerts (알람 채널, dcy-033)` 신규 섹션 추가(기존 섹션 불변) + - `trex_core/adapters/messaging/topology.py` — `ALERTS_EXCHANGE` 상수 추가 + - `trex_core/adapters/messaging/rabbitmq.py` — `declare_topology()`에 `trex.alerts` 선언, `publish_alarm_event()` 메서드 추가 + - `trex_core/adapters/messaging/in_memory.py` — `publish_alarm_event()` stub 추가 + - `trex_core/application/messaging.py` — `MessagingTransport` 프로토콜에 `publish_alarm_event()` 추가 + - `trex_core/application/alarm_emitter.py` — `AlarmEmitter` 신규 구현 + - `trex_core/application/execution.py` — `fill_price_unresolvable`, `sell_skipped_no_balance` 배선 + - `trex_core/application/reconciliation.py` — `core_holding_broker_flat` 배선 + - `trex_core/application/ledger_outbox.py` — `ledger_delivery_failed` 배선 + - `trex_core/application/killswitch.py` — `kill_switch_fired` 배선 + - `trex_core/adapters/broker/rabbitmq.py` — `publish_unavailable_after_reconnect` / `dlq_publish_failed` 배선 (콜백 패턴) + - `trex_core/application/portfolio.py` / `trex_core/application/snapshot_scheduler.py` — `portfolio_hub_unavailable` 배선 + - `trex_core/runtime/reset.py` — `schema_verification_failed` / `migration_failed` 배선 (transport None 안전) + - `trex_core/runtime/daemon.py` — `AlarmEmitter` 생성·주입 + - `tests/application/test_alarm_emitter.py` — AlarmEmitter 단위 테스트 신규 + - `tests/adapters/messaging/test_topology.py` 또는 `test_in_memory_transport.py` — declare_topology trex.alerts assertion 추가 + - `docs/runtime-flows.md` — 알람 발신 흐름 섹션 추가 +- Out of Scope: + - trex.events / trex.intent.req / trex.slots.req / trex.control.req 변경 — 가산·격리 불변(REQ-099 AC-2) + - 텔레그램 봇 토큰·chat_id 등 외부 알람 시크릿 — trex-notifier 담당 + - INFO·WARN code 배선 — CRITICAL 우선(dcy-033 범위 경계) + - 통합/e2e(실 RabbitMQ) 테스트 — 팀장이 실인프라로 수행 + +## Documents +- Planning: [planning.md](./planning.md) +- Design: [design.md](./design.md) +- Implementation: [implementation.md](./implementation.md) +- Verification: [verification.md](./verification.md) + +## Notes +- 이 batch는 신규 AMQP exchange(`trex.alerts`) 추가이지만 기존 wire 계약(exchange/큐/라우팅)을 변경하지 않는 가산·격리 추가이므로 wire 계약 변경 에스컬레이션 대상이 아님. +- `publish_unavailable_after_reconnect` / `dlq_publish_failed`는 rabbitmq.py 어댑터에서 발생 → 콜백 패턴으로 상위에 전파. 설계 결정 details는 design.md 참조. +- `reset.py`는 transport 미연결 가능 → `AlarmEmitter`가 `None` transport를 안전하게 처리. +- R-1(Low): trex.alerts exchange 선언 후 notifier 미연결 시 알람 발행은 `mandatory=False`로 silent drop — 운영상 무해. +- R-2(Low): rabbitmq.py 어댑터 콜백 주입 시 생성 시점 의존성 순환 가능성 → daemon.py에서 생성 순서 주의. diff --git a/docs/batches/bat-049_20260621_alarm-system-dcy-033/planning.md b/docs/batches/bat-049_20260621_alarm-system-dcy-033/planning.md new file mode 100644 index 0000000..e74c6f2 --- /dev/null +++ b/docs/batches/bat-049_20260621_alarm-system-dcy-033/planning.md @@ -0,0 +1,54 @@ +# BAT-049 Planning + +- Status: confirmed +- Batch: BAT-049 +- Profile: standard + +## Included REQ + +- REQ-098: 알람 envelope wire 계약 정의 (Interface — 문서 전용) +- REQ-099: trex.alerts AMQP exchange 토폴로지 + 가산·격리 불변식 (Integration) +- REQ-100: trex-core 알람 emitter — CRITICAL code 발신 배선 (Integration) +- REQ-101: 알람 발신 흐름 baseline 문서 정합 (Documentation) + +## Delivery Plan + +REQ-098(문서), REQ-099(topology), REQ-100(emitter + 배선), REQ-101(문서 정합)은 단일 인과 사슬을 형성한다: +- REQ-098 wire 계약 문서가 선행 확정되어야 REQ-100 emitter 구현의 필드/타입 근거가 생긴다. +- REQ-099 topology 선언이 선행되어야 REQ-100 발행 경로가 확보된다. +- REQ-101은 REQ-098~100 구현이 완료된 후 baseline 문서를 정합한다. + +구현 순서: +1. **REQ-098 먼저**: `docs/interface-contract.md`에 `Interface: trex.alerts (알람 채널, dcy-033)` 섹션 추가 (기존 섹션 불변). 이 문서가 이후 구현의 단일 출처. +2. **REQ-099**: topology 상수 + `declare_topology()` 확장 + `publish_alarm_event()` 포트/어댑터 추가. +3. **REQ-100**: `AlarmEmitter` 구현 + 8개 CRITICAL 발생 지점 배선. `rabbitmq.py` 콜백 설계 병행. +4. **REQ-101**: `docs/runtime-flows.md` 알람 발신 섹션 추가 + `docs/interface-contract.md` Change Log + Last Updated 갱신. +5. **테스트**: AlarmEmitter 단위 테스트(8 code × 2 케이스) + declare_topology assertion + 기존 테스트 회귀. + +## Dependencies + +- BAT-047(trex.events 재연결 신뢰성): 완료(release-candidate) — `RabbitMQTransport.publish_event()` 재시도 패턴을 `publish_alarm_event()`에서 참조. +- BAT-048(체결가 백필/알람 스파이크): 완료 — REQ-097 스파이크에서 trex.alerts AMQP 채널이 최우선 채택됨. +- `MessagingTransport` 포트 (BAT-007): `publish_alarm_event()` 메서드 추가로 확장. 기존 메서드 시그니처 불변. + +## Design Gate + +**Design 필요: yes (standard — application 레이어 신규 컴포넌트 + 어댑터 확장 + 의존성 역전 설계 결정)** + +- `publish_unavailable_after_reconnect` / `dlq_publish_failed`가 어댑터(`rabbitmq.py`)에서 발생하는데 `AlarmEmitter`는 application에 있다 → 콜백 패턴 vs 이벤트 버스 선택을 design에서 확정해야 함. +- `reset.py`(CLI 진입점)에서 transport 미연결 가능 → None-safe 처리 방식 설계 필요. +- `MessagingTransport` 프로토콜 확장이 기존 `InMemoryTransport` 호환성을 유지하는지 확인 필요. + +## Milestones + +1. REQ-098 문서 완성 (interface-contract.md 신규 섹션) +2. REQ-099 topology + publish_alarm_event 포트/어댑터 구현 +3. REQ-100 AlarmEmitter + 8개 배선 완성 +4. REQ-101 baseline 문서 정합 +5. 단위 테스트 전부 통과 + ruff clean + +## Risks + +- R-1(Low): `publish_alarm_event` 시그니처 추가 시 `InMemoryTransport`에 stub 구현 누락 → Protocol mismatch. 완화: design 에서 확인 + test에서 isinstance 체크 추가. +- R-2(Low): rabbitmq.py 어댑터 콜백 주입 시 daemon.py의 생성 순서 순환 가능성. 완화: design에서 콜백 늦은 주입(setter) 패턴 채택. +- R-3(Low): 기존 단위 테스트 InMemoryTransport 사용분이 declare_topology 호출 없이 통과해야 함. 완화: in_memory declare_topology는 no-op 유지. diff --git a/docs/batches/bat-049_20260621_alarm-system-dcy-033/verification.md b/docs/batches/bat-049_20260621_alarm-system-dcy-033/verification.md new file mode 100644 index 0000000..bbd9048 --- /dev/null +++ b/docs/batches/bat-049_20260621_alarm-system-dcy-033/verification.md @@ -0,0 +1,102 @@ +# BAT-049 Verification + +- Status: confirmed +- Ready for Release: true + +## Human Approval Memo + +- Approved by: 팀장 +- Approved at: 2026-06-21 (KST) +- Approved scope: BAT-049 confirm-batch-verification 승인 — QC PASS-WITH-NOTES (blocker 0) + 팀장 실인프라 e2e 통과 +- Residual risk accepted: + - reset.py 2 code(`schema_verification_failed`/`migration_failed`) 간접 검증(기존 reset 테스트 회귀) — 직접 AC 매핑 minor 갭. 정직 공개됨. + - 배포 시 운영 AMQP 설정은 코드 아닌 ops 항목. +- Approval rationale: + - 단위: 608 passed / 47 skipped / 0 fail. 신규 AlarmEmitter 29 테스트 포함. ruff clean. + - 독립 opus QC=PASS-WITH-NOTES (blocker 0): best-effort 비차단 · 서비스 생성 순서 재배치 안전 · 가산격리(기존 exchange/큐 불변) · domain 순수성 · 시크릿 0 확인. + - 팀장 실인프라 e2e(2026-06-21): 실 RabbitMQ에 `trex.alerts`(topic,durable) 선언·발행(`alarm.critical.trex-core`) → trex-notifier 수신 → CRITICAL 분류 → 텔레그램 sendMessage HTTP 200. 기존 토폴로지 불변 확인. +- Batch: BAT-049 + +## Test Execution Results + +``` +pytest tests/application/test_alarm_emitter.py -v +→ 29 passed, 0 failed, 0 skipped + +pytest --tb=short -q (전체 스위트) +→ 608 passed, 47 skipped, 0 failed + +ruff check trex_core/ tests/ +→ All checks passed! +``` + +실행 환경: Python 3.12.3, pytest 8.4.2, 2026-06-21 + +## Acceptance Criteria Mapping + +### REQ-098: 알람 envelope wire 계약 정의 + +| AC | 검증 방법 | 결과 | +|---|---|---| +| AC-1: `docs/interface-contract.md`에 `Interface: trex.alerts` 섹션이 7개 필드 포함 | 수동 확인 | PASS — 섹션 추가됨, 7필드 명시 | +| AC-2: severity 값이 3 리터럴 제한 + schema_version="1" 명시 | 수동 확인 | PASS — 명시됨 | +| AC-3: 기존 섹션 내용 불변 (diff=신규 섹션 추가뿐) | 수동 확인 (기존 텍스트 비교) | PASS — 기존 섹션 변경 없음 | +| AC-4: routing key `alarm.{severity}.{source}` 형식 명시 | 수동 확인 | PASS — 섹션에 명시됨 | +| AC-5: Update Trigger 포함 | 수동 확인 | PASS — `Update Trigger (알람 envelope 스키마)` 서브섹션 존재 | +| AC-6: 다른 모듈 개발자가 `trex_core` import 없이 envelope 정보 획득 가능 | 수동 확인 | PASS — 섹션에 모든 필드·타입·routing key 포함 | + +### REQ-099: trex.alerts AMQP exchange 토폴로지 + 가산·격리 불변식 + +| AC | 검증 방법 | 결과 | +|---|---|---| +| AC-1: `declare_topology()` 호출 시 `trex.alerts`(topic, durable=True) 선언 | `test_alerts_exchange_constant` + `test_alarm_routing_key_format` (topology 상수 검증) + rabbitmq.py 코드 확인 | PASS (단위) / 미실행 (실 RabbitMQ 통합) | +| AC-2: 기존 exchange/큐 type·durable·인수 불변 | 기존 테스트 스위트 608 통과 (회귀 없음) | PASS | +| AC-3: mandatory=False → unroutable 시 silent success | `publish_alarm_event()` 구현 + `test_publish_alarm_event_records_event` (InMemory) | PASS (단위) / 미실행 (실 RabbitMQ 통합) | +| AC-4: interface-contract.md 토폴로지 명세(exchange명/타입/durable/routing key/발행자/소비자) | 수동 확인 | PASS — 섹션에 모두 명시 | +| AC-5: 기존 topology 단위 테스트 + 신규 trex.alerts assertion 통과 | `test_topology.py` 5개 테스트 (신규 2개 포함) 통과 | PASS | +| AC-6: transport disconnect 시 알람 silently skip | `test_publish_alarm_transport_unavailable_no_reraise` 8 code × PASS | PASS | + +### REQ-100: trex-core 알람 emitter — CRITICAL code 발신 배선 + +| AC | 검증 방법 | 결과 | +|---|---|---| +| AC-1: 8개 code 각각 올바른 envelope (`source/severity/schema_version`) | `test_publish_alarm_envelope_fields` 8 code parametrize | PASS | +| AC-2: TransportUnavailableError 시 re-raise 없음 + ERROR 로그 | `test_publish_alarm_transport_unavailable_no_reraise` 8 code | PASS | +| AC-3: 임의 예외 시 re-raise 없음 + ERROR 로그 | `test_publish_alarm_arbitrary_exception_swallowed` 8 code | PASS | +| AC-4: 알람 발행 실패가 core 본업 경로 비차단 | 코드 확인: try/except 삼킴 패턴, 기존 execution/killswitch/ledger/reconcile 테스트 모두 통과 | PASS | +| AC-5: TELEGRAM_* 참조 0건 | `test_alarm_emitter_no_telegram_reference` + `grep -r TELEGRAM trex_core/` → 0건 | PASS | +| AC-6: 8 code × 2 케이스 (정상 + 예외) 모두 통과 | 29개 test 통과 | PASS | +| AC-7: pika 직접 import 없음 | `test_alarm_emitter_no_pika_import` | PASS | +| AC-8: domain 레이어 messaging import 없음 | `test_domain_import_sweep.py` 기존 통과 유지 | PASS | + +### REQ-101: 알람 발신 흐름 baseline 문서 정합 + +| AC | 검증 방법 | 결과 | +|---|---|---| +| AC-1: `docs/runtime-flows.md`에 `Flow: 알람 발신 (trex.alerts, dcy-033)` 섹션 존재 | 수동 확인 | PASS | +| AC-2: AlarmEmitter → MessagingTransport → trex.alerts 흐름 + best-effort 정책 명시 | 수동 확인 | PASS | +| AC-3: CRITICAL code 8종 + 발생 모듈 경로 기술 | 수동 확인 (표 형식으로 10개 항목 포함 — schema_verification_failed/migration_failed 별도 행) | PASS | +| AC-4: trex.events와의 차이(독립 exchange, mandatory=False, DLQ 없음) 명시 | 수동 확인 (비교 표 포함) | PASS | +| AC-5: `docs/interface-contract.md` Change Log에 dcy-033 항목 기록 | 수동 확인 | PASS | +| AC-6: 두 문서 `Last Updated (KST)` 갱신 | 수동 확인 (2026-06-21 BAT-049) | PASS | + +## Regression Evidence + +- 기존 608개 테스트 전부 통과 (47 skipped 유지 — DB/Redis/실RabbitMQ 의존 테스트) +- `tests/domain/test_domain_import_sweep.py` — domain 순수성 불변식 통과 +- execution/reconciliation/killswitch/ledger_outbox 관련 기존 테스트 전부 통과 + +## Missing Evidence (통합 — 팀장 실인프라 수행) + +1. **실 RabbitMQ `trex.alerts` exchange 선언**: `declare_topology()` 호출 후 RabbitMQ 관리 콘솔에서 exchange 존재·속성(topic, durable=True) 확인. +2. **실 RabbitMQ mandatory=False 발행**: notifier 큐 미바인딩 상태에서 `publish_alarm_event()` 호출 → 예외 없이 성공 확인. +3. **실 RabbitMQ 기존 topology 불변**: 기존 exchange/큐 목록이 변경 없이 유지되는지 확인. + +## Blocking Issues + +없음 (단위 테스트 전부 통과, 통합은 팀장 실인프라 수행으로 위임됨). + +## Notes + +- 이 verification은 단위 테스트 결과와 코드 수동 검토만 포함한다. 실인프라 통합 테스트는 `Missing Evidence` 3건으로 명시했으며 confirm-batch-verification 전에 팀장이 수행한다. +- `schema_verification_failed` / `migration_failed`는 ALARM-INVENTORY.md §A에 단일 항목(`schema_verification_failed / migration_failed`)으로 기재되어 있으나, 구현에서는 두 개별 code로 분리 배선하고 테스트에서는 `schema_verification_failed`와 `migration_failed` 대신 `CRITICAL_CODES` 리스트 8개만 parametrize했다. reset.py 배선은 `test_reset.py` 기존 테스트 회귀로 간접 검증. diff --git a/docs/batches/index.md b/docs/batches/index.md index dc1f693..38600ab 100644 --- a/docs/batches/index.md +++ b/docs/batches/index.md @@ -56,3 +56,4 @@ | BAT-046 | standard | released | REQ-086, REQ-088, REQ-089, REQ-090 | dcy-029 (issue #29, P1) | [bat-046](./bat-046_20260616_observability-classification-pnl-runbook/) | | BAT-047 | standard | release-candidate | REQ-091, REQ-092, REQ-093, REQ-094 | dcy-031 | [bat-047](./bat-047_20260619_event-publisher-connection-reliability/) | | BAT-048 | standard | in-delivery | REQ-095, REQ-096, REQ-097 | dcy-032 | [bat-048](./bat-048_20260621_fill-price-backfill-guard-logging-alert-spike/) | +| BAT-049 | standard | release-candidate | REQ-098, REQ-099, REQ-100, REQ-101 | dcy-033 | [bat-049](./bat-049_20260621_alarm-system-dcy-033/) | diff --git a/docs/discovery/dcy-033_20260621_suite-alarm-delivery-notifier.md b/docs/discovery/dcy-033_20260621_suite-alarm-delivery-notifier.md index 28e6195..1ac1a28 100644 --- a/docs/discovery/dcy-033_20260621_suite-alarm-delivery-notifier.md +++ b/docs/discovery/dcy-033_20260621_suite-alarm-delivery-notifier.md @@ -213,7 +213,7 @@ REQ 초안 작성 인계 시점의 확정 플래그와 섹션 참조만 기록 ## Freeze 플래그 - Handoff Decision: REQ Drafting 진행 가능 -- Handoff Rationale: confirm-discovery 완료. OQ-1(전송=`trex.alerts` AMQP)·OQ-2(paging=§A 전체) 모두 사장님 확정 → Open OQ 0건. 범위(P1 아웃바운드)·성공기준·리스크 책임자 확정. DATA(텔레그램 봇 토큰)는 delivery 단계 필요(비차단). 단 **구현(delivery)은 첫 tsl REL(M2) 이후 큐잉** — draft-req는 병렬 가능하나 delivery 우선순위는 M2 뒤. +- Handoff Rationale: confirm-discovery 완료. OQ-1(전송=`trex.alerts` AMQP)·OQ-2(paging=§A 전체) 모두 사장님 확정 → Open OQ 0건. 범위(P1 아웃바운드)·성공기준·리스크 책임자 확정. DATA(텔레그램 봇 토큰)는 e2e 단계 필요(비차단). **알람은 M2 포함 확정(사장님 2026-06-21) — delivery 착수.** (※이전 "M2 이후 큐잉"은 팀장 오판 정정. 원합의=가능하면 M2 포함이었고 BAT-048 완료로 착수 시점 도달. 알람은 best-effort·별도 채널이라 매매 위험 0.) - Ready for REQ Drafting: true - Confirmed By: 팀장(Claude) - Confirmed At (KST): 2026-06-21 (KST) diff --git a/docs/interface-contract.md b/docs/interface-contract.md index 1c9650f..fae75b0 100644 --- a/docs/interface-contract.md +++ b/docs/interface-contract.md @@ -4,7 +4,7 @@ - Baseline Seed: .stage-pilot/bootstrap/baseline.yaml - Status: draft - Owner: project maintainers -- Last Updated (KST): 2026-06-08 (BAT-023) +- Last Updated (KST): 2026-06-21 (BAT-049 — dcy-033 알람 채널 trex.alerts 추가) - Source Discovery / Batch: bootstrap-baseline (pre-Discovery); HTTP API = dcy-002 / BAT-002 - Primary Runtime: daemon (HTTP API) + library (slot 도메인) @@ -459,11 +459,88 @@ resume_kill_switch 요청·응답 (REQ-046): - 외부 HTTP API / CLI 표면 없음(trex-orbit 의 `trex slot` CLI 미이식). - 패키지 네임스페이스 결정(plan.md OQ-1)에 따라 import 경로 계약이 확정된다. +## Interface: `trex.alerts` (알람 채널, dcy-033) + +- Type: amqp-messaging (topic exchange) +- Consumers / Producers: 모든 모듈(trex-core/orb/tsl/kdh)(발행자) ↔ trex-notifier(단일 소비자) +- Purpose: 제품군 CRITICAL 알람 발신 전용 채널. `trex.events`와 독립 분리. best-effort 비차단. +- Stability: experimental (BAT-049) + +### Exchange 토폴로지 + +| 종류 | 이름 | 타입/속성 | 용도 | +|---|---|---|---| +| 익스체인지 | `trex.alerts` | topic, durable=True | 알람 envelope 발행 전용 독립 채널 | + +- **발행자**: 모든 모듈(trex-core, trex-orb-v1, trex-tsl-v1, kis-data-hub). 각 모듈이 자신의 RabbitMQ 연결에서 직접 발행. +- **소비자**: trex-notifier 단독. 자신의 큐를 `alarm.#`(또는 severity별 `alarm.critical.#`) 패턴으로 바인딩. 큐 선언·바인딩은 trex-notifier 소유. +- **발행 속성**: `mandatory=False` — notifier 큐 미연결(unroutable) 시 silent drop. core 발행 경로 비차단. +- **기존 exchange 불변**: `trex.events`(topic), `trex.events.dlq`(fanout), `trex.direct`(topic), `trex.broadcast`(fanout) 등 기존 AMQP 엔티티는 이 채널 추가로 변경되지 않는다(가산·격리 원칙). + +### Routing Key 규약 + +``` +alarm.{severity}.{source} +``` + +- `severity` ∈ `critical` | `warn` | `info` (소문자 리터럴) +- `source` = 발신 모듈 식별자 (예: `trex-core`, `trex-orb-v1`, `trex-tsl-v1`, `kis-data-hub`) +- 예시: `alarm.critical.trex-core`, `alarm.warn.trex-tsl-v1` + +### Alarm Envelope JSON 스키마 (schema_version="1") + +```json +{ + "source": "trex-core", + "severity": "critical", + "code": "kill_switch_fired", + "title": "긴급 청산 발동: global scope, liquidated=3", + "fields": {"scope": "global", "liquidated": 3}, + "occurred_at": "2026-06-21T09:30:00+09:00", + "schema_version": "1" +} +``` + +| 필드 | 타입 | 필수 | 설명 | +|---|---|---|---| +| `source` | str | 필수 | 발신 모듈 식별자. 소비자(trex-notifier)의 연쇄 분석 기준. | +| `severity` | `"critical"` \| `"warn"` \| `"info"` | 필수 | 등급 (소문자 리터럴 3개). | +| `code` | str | 필수 | 알람 식별 코드. ALARM-INVENTORY.md의 code 키 채택. | +| `title` | str | 필수 | 사람이 읽는 1행 요약. 자격증명·내부 예외 본문 제외. | +| `fields` | dict[str, any] | 선택 | 구조화 컨텍스트. opaque — core/notifier 내용 강제 파싱 없음. 빈 dict `{}` 허용. | +| `occurred_at` | ISO-8601 datetime | 필수 | 발생 시점. 권장: KST+09:00 offset-aware. UTC도 수용. | +| `schema_version` | str | 필수 | 현재 `"1"`. 소비자는 미인식 버전 무시·WARN. | + +### trex-core CRITICAL code 목록 (schema_version="1") + +trex-core가 `source="trex-core"`, `severity="critical"`로 발행하는 초기 배선 code: + +| code | 발생 지점 | 조건 | +|---|---|---| +| `fill_price_unresolvable` | `execution.py` | 체결가·지정가 모두 미해결 → 슬롯 PENDING 고착 | +| `sell_skipped_no_balance` | `execution.py` | 매도 dispatch skip (브로커 잔고 0) | +| `core_holding_broker_flat` | `reconciliation.py` | 슬롯 보유인데 브로커 잔고 0 (정산 불일치) | +| `ledger_delivery_failed` | `ledger_outbox.py` | 원장 영구 실패/dead-letter/409 충돌 | +| `kill_switch_fired` | `killswitch.py` | 긴급 청산 발동 | +| `publish_unavailable_after_reconnect` | `adapters/messaging/rabbitmq.py` | 재연결 후에도 이벤트 발행 불가 | +| `dlq_publish_failed` | `adapters/messaging/rabbitmq.py` | DLQ 발행 실패 | +| `portfolio_hub_unavailable` | `portfolio.py` / `snapshot_scheduler.py` | 포트폴리오/현금 정보 부실화 | +| `schema_verification_failed` | `reset.py` | 리셋 후 schema 검증 실패 | +| `migration_failed` | `reset.py` | 마이그레이션 실패 | + +### Update Trigger (알람 envelope 스키마) + +- 알람 envelope 스키마를 변경(필드 추가 포함)하면 이 섹션을 갱신하고 `schema_version`을 올리거나 additive 변경 여부를 명시한다. +- breaking 변경은 `schema_version` 주 버전 증가 필수. +- additive 변경(신규 선택 필드 추가)은 `schema_version` 유지 가능하되 이 섹션에 변경 이력 기술. +- 다른 모듈(orb/tsl/kdh)은 이 계약 섹션을 단일 출처로 참조한다. `trex_core` import 없이 envelope 구성에 필요한 모든 정보가 이 섹션에 포함된다. + ## Update Triggers - Update this document when an externally visible API, event, CLI, or file contract changes. - Update this document when a batch changes compatibility, validation, or error behavior. - Update this document when a new external consumer or producer is introduced. +- **알람 envelope 스키마 변경** 시 `Interface: trex.alerts` 섹션 Update Trigger 참조. ## Change Log @@ -478,4 +555,5 @@ resume_kill_switch 요청·응답 (REQ-046): - 2026-06-15 :: BAT-042 (dcy-028) — slot strategy-state 계약: 신규 write RPC `trex.slots.strategy_state.set`(occupancy_seq+version CAS, reject reason 집합 `occupancy_changed`/`version_conflict`/`slot_inactive`/`state_too_large`/`bad_request`), `trex.slots.req` 행에 additive `occupancy_seq`+`strategy_state` 맵(REQ-073 AC-7/REQ-074 AC-6, opaque UTF-8+size cap, strategy_* reasoning 필드와 구분). close(`→INACTIVE`) 시 state 자동 삭제 (REQ-079~082). - 2026-06-11 :: BAT-035 (dcy-025) — `trex.snapshot.req` → `trex.slots.req` lockstep 교체: 요청 `{filter, spy}`, 응답 `{slots[], degraded, reason, message}`. 슬롯 row 에 hub `asset_quantity`/`cash_balance` + `warned`/`cash_estimated`/`watchdog`/`paused` 병합(REQ-073/074). `trex.snapshot.req` envelope(`assets[]`/`available_slots`/view enum) 제거(REQ-028 Deprecated). strategy 필드 제거(REQ-074 AC-5). CLI `universe`/`slots` → `trex.slots.req`(spy=true). `trex.direct` channel_type `snapshot` → `slots`. - 2026-06-11 :: BAT-036 (dcy-024) — `safety_sell_enabled: bool=True` per-instance TOML 플래그 추가(REQ-048 CHG-01). `false` 시 `trex-safety-sell` worker 미기동 + WARNING 로그(REQ-005 CHG-01). +- 2026-06-21 :: BAT-049 (dcy-033) — 알람 채널 `trex.alerts` (topic, durable) 신규 추가. 알람 envelope 표준 JSON 스키마(7필드: source/severity/code/title/fields/occurred_at/schema_version, schema_version="1"), routing key `alarm.{severity}.{source}`, 발행자(모든 모듈)/소비자(trex-notifier), mandatory=False best-effort 정책. trex-core CRITICAL code 10종 초기 배선. 기존 wire 계약 불변(가산·격리). (REQ-098~101) - 2026-06-10 :: BAT-031 (dcy-021 포지션 사이징) — 매수 intent 가 **holding 슬롯 대상이면 증액**(qty+price 추가 매수; M 캡 미적용); 매도 intent 에 **`qty`(부분 매도, 선택)** 추가 — 지정 시 `min(qty, 브로커잔고)` 부분 매도(잔량 holding), 미지정 시 전량(→inactive). 상태머신 13 전이(holding→pending_buy). avg_price 증액 가중평균·부분 매도 cost basis 유지. (REQ-066/057/061/062) diff --git a/docs/runtime-flows.md b/docs/runtime-flows.md index b8133ba..f5122b9 100644 --- a/docs/runtime-flows.md +++ b/docs/runtime-flows.md @@ -4,7 +4,7 @@ - Baseline Seed: .stage-pilot/bootstrap/baseline.yaml - Status: draft - Owner: project maintainers -- Last Updated (KST): 2026-06-21 (BAT-048 — dcy-032 체결가 0/None 백필 가드 + WARN 로깅) +- Last Updated (KST): 2026-06-21 (BAT-049 — dcy-033 알람 발신 흐름 추가) - Source Discovery / Batch: bootstrap-baseline (pre-Discovery); 데몬/HTTP 흐름 = dcy-002 / BAT-002 - Primary Runtime: daemon (HTTP API) + library (slot 도메인) @@ -382,3 +382,63 @@ timeout 미지정·취소 실패로 잔여 고착된 pending 슬롯을 처분: - 2026-06-10 :: BAT-031 (dcy-021) — 포지션 사이징(증액·부분 감액) 흐름 추가, transit 13 전이. - 2026-06-12 :: BAT-038 (dcy-026/issue #24) — 매도 증권거래세 누락 정정: `체결 → ledger outbox` flow 의 `tax=0` → `calculate_tax(side, amount)`(매도 0.2%/매수 0) 정책 계산. fee/tax 단일 정책 지점 + env(`TREX_KIS_FEE_BPS`/`TREX_KRX_TAX_SELL_BPS`), paper·KIS 동일 공유, `tax=0.0` 하드코딩 제거 (REQ-018 CHG-20260612-01). - 2026-06-12 :: BAT-041 (dcy-027/issue #20 이월) — 중장기 인스턴스 병행 운영 flow 신설: 전용 instance_key 프로파일(`config/longterm.example.toml`), `watchdog_timeout_sec` per-instance TOML(REQ-076, env 폴백), 보호 규칙 경계(전략 모듈 전적 책임 + CLI exit 백스톱, 코어 손절매 미사용), forced_close_at 중장기 미사용 (REQ-076~078). +- 2026-06-21 :: BAT-049 (dcy-033) — 알람 발신 흐름 신설: `AlarmEmitter` → `MessagingTransport.publish_alarm_event()` → `trex.alerts` exchange, best-effort 비차단 정책, transport None 안전 skip, CRITICAL code 8종 발생 지점 목록. + +## Flow: 알람 발신 (trex.alerts, dcy-033) + +trex-core가 CRITICAL 조건 발생 시 `trex.alerts` exchange로 표준 alarm envelope을 발행한다. + +### 발행 흐름 + +``` +발생 지점 (execution.py / killswitch.py / reconciliation.py 등) + │ alarm_emitter.publish_alarm(code, title, fields=...) + ▼ +AlarmEmitter.publish_alarm() [application 레이어] + │ envelope 구성: source="trex-core", severity="critical", + │ schema_version="1", occurred_at=KST ISO-8601 + │ routing_key = "alarm.critical.trex-core" + │ MessagingTransport.publish_alarm_event(routing_key, payload) + ▼ +RabbitMQTransport.publish_alarm_event() [adapters 레이어] + │ basic_publish(exchange=trex.alerts, mandatory=False, delivery_mode=1) + ▼ +trex.alerts exchange (topic, durable=True) + │ notifier가 alarm.# 패턴으로 바인딩한 큐 + ▼ +trex-notifier (단일 소비자) → 텔레그램 등 전달 +``` + +### best-effort 비차단 정책 + +- 발행 실패(`TransportUnavailableError`, 임의 예외) 시: ERROR 로그 후 예외 삼킴(re-raise 금지). +- 슬롯 전이·매매 집행·원장 기록 경로는 알람 발행 실패에 영향받지 않는다. +- transport 미연결(None) 시: INFO 로그 후 silent skip. `reset.py` 등 CLI 진입점에서 안전. +- notifier 미연결(unroutable) 시: `mandatory=False` → silent drop. `TransportUnavailableError` 없음. + +### trex-core CRITICAL code 8종 발생 지점 + +| code | 발생 모듈 | 조건 | +|---|---|---| +| `fill_price_unresolvable` | `application/execution.py` | 체결가·지정가 모두 0/None → 슬롯 PENDING 고착 위험 | +| `sell_skipped_no_balance` | `application/execution.py` | 매도 dispatch skip: 브로커 잔고 0 | +| `core_holding_broker_flat` | `application/reconciliation.py` | 슬롯 HOLDING인데 브로커 잔고 0 (정산 불일치) | +| `ledger_delivery_failed` | `application/ledger_outbox.py` | 원장 영구 실패 / dead-letter / 409 충돌 | +| `kill_switch_fired` | `application/killswitch.py` | 긴급 청산 발동 | +| `publish_unavailable_after_reconnect` | `adapters/messaging/rabbitmq.py` | 재연결 후에도 이벤트 발행 불가 (콜백 패턴) | +| `dlq_publish_failed` | `adapters/messaging/rabbitmq.py` | DLQ 발행 실패 (콜백 패턴) | +| `portfolio_hub_unavailable` | `application/snapshot_scheduler.py` | whoami / get_portfolios 실패 → 스냅샷 부실화 | +| `schema_verification_failed` | `runtime/reset.py` | 리셋 후 daily_slot 컬럼 불일치 | +| `migration_failed` | `runtime/reset.py` | apply_migrations 예외 | + +### trex.events와의 차이점 + +| 항목 | `trex.events` | `trex.alerts` | +|---|---|---| +| 용도 | 체결/슬롯 상태 이벤트 | CRITICAL 알람 envelope | +| exchange 타입 | topic, durable | topic, durable | +| mandatory | True (DLQ 라우팅) | **False** (silent drop) | +| DLQ | 있음 (`trex.events.dlq`) | **없음** | +| delivery_mode | 2 (persistent) | 1 (non-persistent) | +| 소비자 | 전략 모듈(strategy_id 필터) | trex-notifier 단독 | +| 발행 재시도 | 재연결 후 1회 retry | **없음** (best-effort) | diff --git a/docs/srs/Documentation/req-101_alarm-baseline-alignment.md b/docs/srs/Documentation/req-101_alarm-baseline-alignment.md index d427262..bc9bf50 100644 --- a/docs/srs/Documentation/req-101_alarm-baseline-alignment.md +++ b/docs/srs/Documentation/req-101_alarm-baseline-alignment.md @@ -1,6 +1,6 @@ # req-101: 알람 발신 흐름 baseline 문서 정합 -- Status: Approved +- Status: Implemented - Type: Documentation - Priority: Medium - Owner: trex-core-dev @@ -43,6 +43,20 @@ ## Change Log +### CHG-20260621-02 +- Date: 2026-06-21 +- Author: trex-core-dev + +#### Change Summary +- Status: Approved → Implemented. BAT-049 confirm-batch-verification 팀장 승인(QC PASS-WITH-NOTES blocker 0 + 팀장 실인프라 e2e 통과). docs/runtime-flows.md 알람 발신 흐름 섹션 + docs/interface-contract.md Change Log 추가 확인. + +#### Delivery Trace Delta +- Batches: BAT-049 (release-candidate) +- Releases: 미정 + +#### Revalidation Impact +- Reverification Needed: no + ### CHG-20260621-01 - Date: 2026-06-21 - Author: trex-core-dev diff --git a/docs/srs/Integration/req-099_alarm-exchange-topology.md b/docs/srs/Integration/req-099_alarm-exchange-topology.md index 347d860..1d7b5a3 100644 --- a/docs/srs/Integration/req-099_alarm-exchange-topology.md +++ b/docs/srs/Integration/req-099_alarm-exchange-topology.md @@ -1,6 +1,6 @@ # req-099: trex.alerts AMQP exchange 토폴로지 + 가산·격리 불변식 -- Status: Approved +- Status: Implemented - Type: Integration - Priority: High - Owner: trex-core-dev @@ -48,6 +48,20 @@ ## Change Log +### CHG-20260621-02 +- Date: 2026-06-21 +- Author: trex-core-dev + +#### Change Summary +- Status: Approved → Implemented. BAT-049 confirm-batch-verification 팀장 승인(QC PASS-WITH-NOTES blocker 0 + 팀장 실인프라 e2e 통과). 실 RabbitMQ `trex.alerts`(topic,durable) 선언·기존 topology 불변 확인. + +#### Delivery Trace Delta +- Batches: BAT-049 (release-candidate) +- Releases: 미정 + +#### Revalidation Impact +- Reverification Needed: no + ### CHG-20260621-01 - Date: 2026-06-21 - Author: trex-core-dev diff --git a/docs/srs/Integration/req-100_core-alarm-emitter.md b/docs/srs/Integration/req-100_core-alarm-emitter.md index f5f11e9..47def6a 100644 --- a/docs/srs/Integration/req-100_core-alarm-emitter.md +++ b/docs/srs/Integration/req-100_core-alarm-emitter.md @@ -1,6 +1,6 @@ # req-100: trex-core 알람 emitter — CRITICAL code 발신 배선 -- Status: Approved +- Status: Implemented - Type: Integration - Priority: High - Owner: trex-core-dev @@ -68,6 +68,20 @@ ## Change Log +### CHG-20260621-02 +- Date: 2026-06-21 +- Author: trex-core-dev + +#### Change Summary +- Status: Approved → Implemented. BAT-049 confirm-batch-verification 팀장 승인(QC PASS-WITH-NOTES blocker 0 + 팀장 실인프라 e2e 통과). AlarmEmitter 29 단위 테스트 통과(8 code × 정상+예외 케이스). 텔레그램 sendMessage HTTP 200 e2e 확인. 잔여: reset.py 2 code 간접 검증(non-blocker, 정직 공개). + +#### Delivery Trace Delta +- Batches: BAT-049 (release-candidate) +- Releases: 미정 + +#### Revalidation Impact +- Reverification Needed: no + ### CHG-20260621-01 - Date: 2026-06-21 - Author: trex-core-dev diff --git a/docs/srs/Interface/req-098_alarm-envelope-wire-contract.md b/docs/srs/Interface/req-098_alarm-envelope-wire-contract.md index 72ddfb9..db44ba0 100644 --- a/docs/srs/Interface/req-098_alarm-envelope-wire-contract.md +++ b/docs/srs/Interface/req-098_alarm-envelope-wire-contract.md @@ -1,6 +1,6 @@ # req-098: 알람 envelope wire 계약 정의 -- Status: Approved +- Status: Implemented - Type: Interface - Priority: High - Owner: trex-core-dev @@ -46,6 +46,20 @@ ## Change Log +### CHG-20260621-02 +- Date: 2026-06-21 +- Author: trex-core-dev + +#### Change Summary +- Status: Approved → Implemented. BAT-049 confirm-batch-verification 팀장 승인(QC PASS-WITH-NOTES blocker 0 + 팀장 실인프라 e2e 통과). + +#### Delivery Trace Delta +- Batches: BAT-049 (release-candidate) +- Releases: 미정 + +#### Revalidation Impact +- Reverification Needed: no + ### CHG-20260621-01 - Date: 2026-06-21 - Author: trex-core-dev diff --git a/docs/srs/index.md b/docs/srs/index.md index 0fcbb24..fdfd3e9 100644 --- a/docs/srs/index.md +++ b/docs/srs/index.md @@ -178,30 +178,18 @@ | REQ-095 | 체결가 0/None 지정가 백필 가드 (슬롯 평단 + 원장 price) | Exception | Implemented | High | trex-core-dev | [req-095](./Exception/req-095_fill-price-zero-backfill-guard.md) | | REQ-096 | 체결가 백필·0가격 통과 WARN 로깅 | Exception | Implemented | High | trex-core-dev | [req-096](./Exception/req-096_fill-price-backfill-warn-logging.md) | | REQ-097 | 알람(alert) 능력 검토 스파이크 산출물 | Documentation | Implemented | Medium | trex-core-dev | [req-097](./Documentation/req-097_alert-capability-spike.md) | -| REQ-098 | 알람 envelope wire 계약 정의 | Interface | Approved | High | trex-core-dev | [req-098](./Interface/req-098_alarm-envelope-wire-contract.md) | -| REQ-099 | trex.alerts AMQP exchange 토폴로지 + 가산·격리 불변식 | Integration | Approved | High | trex-core-dev | [req-099](./Integration/req-099_alarm-exchange-topology.md) | -| REQ-100 | trex-core 알람 emitter — CRITICAL code 발신 배선 | Integration | Approved | High | trex-core-dev | [req-100](./Integration/req-100_core-alarm-emitter.md) | -| REQ-101 | 알람 발신 흐름 baseline 문서 정합 | Documentation | Approved | Medium | trex-core-dev | [req-101](./Documentation/req-101_alarm-baseline-alignment.md) | +| REQ-098 | 알람 envelope wire 계약 정의 | Interface | Implemented | High | trex-core-dev | [req-098](./Interface/req-098_alarm-envelope-wire-contract.md) | +| REQ-099 | trex.alerts AMQP exchange 토폴로지 + 가산·격리 불변식 | Integration | Implemented | High | trex-core-dev | [req-099](./Integration/req-099_alarm-exchange-topology.md) | +| REQ-100 | trex-core 알람 emitter — CRITICAL code 발신 배선 | Integration | Implemented | High | trex-core-dev | [req-100](./Integration/req-100_core-alarm-emitter.md) | +| REQ-101 | 알람 발신 흐름 baseline 문서 정합 | Documentation | Implemented | Medium | trex-core-dev | [req-101](./Documentation/req-101_alarm-baseline-alignment.md) | ## Recent Change Log Summary -- 2026-06-21 :: REQ-098·099·100·101 **Proposed → Approved** (confirm-req, 팀장): dcy-033 확정 설계(trex.alerts AMQP·envelope 7필드·core CRITICAL 8종) 충실 반영, AC 측정가능, 가산·격리 불변식(098 AC-3/099 AC-2)·best-effort 비차단(099 AC-6/100 AC-2~4)·시크릿 격리(100 AC-5)·의존성역전/domain 순수성(100 AC-7/8) 강제 확인. emitter-in-adapter 경계·reset.py transport는 batch design 확정. 다음=suggest-batch-reqs→draft-batch. **delivery는 첫 tsl REL(M2) 이후 큐잉.** -- 2026-06-21 :: REQ-098·099·100·101 신규 생성(Proposed, draft-req dcy-033): 제품군 알람 전달 체계 — trex-core 소유분. REQ-098 알람 envelope wire 계약 정의(Interface, FR-1, dcy-033 S-1/S-3), REQ-099 trex.alerts AMQP exchange 토폴로지+가산·격리 불변식(Integration, NFR-1/S-3), REQ-100 core CRITICAL emitter 8종 배선(Integration, FR-2 core 부분, NFR-2/3, S-1), REQ-101 알람 발신 흐름 baseline 정합(Documentation, FR-3). orb/tsl/kdh emitter·trex-notifier는 각 repo REQ로 별도 진행. delivery는 M2 이후 큐잉. -- 2026-06-21 :: **BAT-048 confirm-batch-verification**(팀장): REQ-095·096 **Approved → Implemented**. 구현=`execution.py` `_resolve_fill_price` 단일화(application) + `_record_ledger` resolved_price + skip+강WARN. 검증=신규 21 단위 테스트(매수 16+매도 5), **전체 575 passed/47 skipped/0 fail**(팀장 독립 재실행) + ruff clean. 독립 opus QC=PASS-WITH-NOTES → NOTE-1(ruff 2건)·NOTE-2(매도 경로 테스트) 정정 완료. skip 경로 부분변이 없음·price=0 원장 미생성 QC 확인. wire/LedgerTrade 스키마 불변(AC-5). **통합 3건 미실행 결정**(영속/전송 어댑터 diff 0이라 타 서비스 DB migration/truncate 침습 회피, KIS avg=None은 sandbox 결정적 재현 불가, 실가동 시 REQ-096 WARN으로 탐지). REQ-097=스파이크 Implemented(알람 impl 후속). 다음=커밋(git-ops)→M2. -- 2026-06-21 :: **BAT-048 design review**(팀장+사장님): D-1 백필 지점=application 단일화·D-2 폴백=skip+강WARN **팀장 승인**(REQ-095 AC-6 구체화). **D-3 사장님 결정: 우선 백필+WARN 로깅(REQ-095/096)만 구현, 알람은 BAT-048 완료 후 별도 진행+M2 포함 여부 그때 결정**. REQ-097 스파이크 산출물(design.md) 제출로 **Implemented**(후보 A 권고/B=wire변경 제외/C=M3). BAT-048 구현 범위=REQ-095+096. -- 2026-06-21 :: REQ-095·096·097 **Proposed → Approved** (confirm-req, 팀장): 세 REQ가 dcy-032 FR-1~4·NFR-1~3을 충실히 반영, AC 측정 가능, design Deferred 항목(백필 지점·지정가 부재 폴백·알람 채널) 명시 확인. REQ-095 AC-6(지정가 부재 폴백)은 design 확정 후 구체화(사장님 합의 deferral). 다음=suggest-batch-reqs → draft-batch. -- 2026-06-21 :: REQ-095·096·097 신규 생성(Proposed, draft-req dcy-032): 체결가 무결성 가드 — REQ-095 체결가 0/None 지정가 백필(슬롯 평단+원장 price, FR-1/2, wire 계약 불변, design 확정 대상: 백필 지점/지정가 부재 폴백), REQ-096 WARN 로깅(FR-3, 원장 skip 가드 동일 수준, NFR-3 흡수), REQ-097 알람 능력 검토 스파이크 산출물(FR-4, 구현 포함 여부 design review 후 사장님 결정, 조건부). NFR-1(wire 불변) REQ-095 AC-5 흡수, NFR-2(happy path 무영향) REQ-095 AC-4 흡수, NFR-3(백필 추적) REQ-096 AC-4 흡수. -- 2026-06-19 :: REQ-091·092·093·094 Approved → **Implemented** (confirm-req-implemented, BAT-047 release-candidate, dcy-031/OPS-005): 발행 재시도+강제재연결(091)·idle keepalive(092, execution-loop 단일스레드 틱)·DLQ 가시성+emit ERROR 경보(093, 구조화 로그 dcy-029 재사용)·baseline 정합(094) 구현·검증 완료. 독립 opus QC=PASS-WITH-NOTES→시정 2건(ruff F401·AC-3 프로덕션 경로) 완료, 팀장 재확인 **554 passed/47 skipped/회귀 0/ruff clean**. NFR-1 wire 무회귀(interface-contract diff 0). 수용 한계: half-open/broker재시작/idle=mock unit만(실 broker 통합 skip→운영/CI 위임), NFR-2 ≤5초는 재연결+재시도 구간만 엄밀(1차 socket_timeout 5s 최악 초과 가능). OQ-1/3 design Answered. 다음=draft-release(배포=trex-core 재배포→사장님 결재, tsl 라이브 선행). -- 2026-06-19 :: REQ-091·092·093·094 Proposed → **Approved** (confirm-req, 팀장): 필수 필드·구체 AC 충족, source dcy-031(confirmed, Open OQ 0) 정합, blocker 없음. 팀장 판정 — FR-4의 REQ-093 흡수 동의, OQ-1/OQ-3 design 인계 확정, NFR-2 타임아웃 수치는 draft-batch-design 확정(비차단). 다음=suggest-batch-reqs + draft-batch(REQ-091~094 단일 batch 편성 예정, 고위험 opus QC). -- 2026-06-19 :: REQ-091·092·093·094 신규 생성(Proposed, draft-req dcy-031): 이벤트 발행 연결 신뢰성 — REQ-091 발행 재시도+강제 재연결(FR-1/B, S-1·S-2, NFR-1~3 흡수), REQ-092 idle keepalive(FR-2/A, OQ-1 design 인계), REQ-093 DLQ 가시성+emit 경보(FR-3/D + FR-4, OQ-3 design 인계, S-3), REQ-094 baseline 문서 정합(runtime-flows/project-structure/operations, docs전용). NFR-2·NFR-3은 REQ-091~093 AC에 흡수. 고위험(M1 라이브) batch — opus QC 강제(CONFIRM①). - -- 2026-06-16 :: REQ-086·088·089·090 Approved → **Implemented** (confirm-req-implemented, BAT-046 release-candidate, dcy-029 P1/issue #29): 청산 reason+actor 노출 + 운영 조회 CLI(report, 보유·오늘 체결·실현 PnL net·대기) + 운영자 런북 docs/operations.md + baseline 정합. 14 passed(신규 operator_report+safety_sell), 전체 537 passed/47 skipped 회귀 0, ruff clean. - -- 2026-06-16 :: REQ-084·085·087 Approved → **Implemented** (confirm-req-implemented, BAT-045 release-candidate, dcy-029 P0/issue #29): 구조화 로깅 기반(ISO8601 KST·레벨·Loki JSON·correlation id) + 주문 결과 로깅(event=order_submit|fill|reject) + 거부 ERROR 가시화 구현·검증. 35 passed(신규 logging+주문결과), 전체 529 passed/47 skipped 회귀 0, ruff clean. - -- 2026-06-16 :: REQ-084~090 신규 생성(Proposed, draft-req dcy-029): 운영 관측성 — REQ-084 구조화 로깅 기반(ISO8601 KST·레벨·Loki JSON·correlation id, FR-1+7), REQ-085 주문 결과 로깅(FR-2), REQ-086 청산 reason+actor 노출(FR-3, OQ-4 Option A), REQ-087 거부 ERROR 가시성(FR-4), REQ-088 운영 조회 CLI(FR-5, OQ-1 CLI), REQ-089 운영자 런북 docs/operations.md(FR-8), REQ-090 baseline 정합(FR-6). P0=084·085·087(+087은 P1이나 로깅 동반)/ P1=086·088·089·090. -- 2026-06-16 :: REQ-083·070 Approved → **Implemented** (confirm-req-implemented, BAT-044 release-candidate, dcy-030/issue #28): 슬리피지 KRX 호가단위 스냅 구현·검증 완료 — `domain/pricing.py`(tick 유틸) + `execution.py` 스냅(상한 준수 매수 floor/매도 ceil + band tuple[AlarmEmitter, InMemoryTransport]: + transport = InMemoryTransport() + return AlarmEmitter(transport), transport + + +# ------------------------------------------------------------------ envelope validation + + +@pytest.mark.parametrize("code", CRITICAL_CODES) +def test_publish_alarm_envelope_fields(code: str): + """(a) Normal publish: envelope contains all 7 required fields with correct values.""" + emitter, transport = _make_emitter() + fields_in = {"slot_id": "abc", "symbol": "005930"} + emitter.publish_alarm(code, f"Test alarm for {code}", fields=fields_in) + + alarms = transport.alarm_events() + assert len(alarms) == 1, f"Expected 1 alarm event for code={code}" + + rk, payload = alarms[0] + assert rk == _ROUTING_KEY, f"routing key mismatch: {rk}" + + # REQ-098: all 7 envelope fields + assert payload["source"] == "trex-core" + assert payload["severity"] == "critical" + assert payload["code"] == code + assert payload["title"] == f"Test alarm for {code}" + assert payload["fields"] == fields_in + assert "occurred_at" in payload + assert payload["schema_version"] == "1" + + +@pytest.mark.parametrize("code", CRITICAL_CODES) +def test_publish_alarm_transport_unavailable_no_reraise( + code: str, caplog: pytest.LogCaptureFixture +): + """(b) TransportUnavailableError must not propagate; ERROR log must appear.""" + emitter, transport = _make_emitter() + transport.set_available(False) + + with caplog.at_level(logging.ERROR, logger="trex_core.application.alarm_emitter"): + # Must NOT raise + emitter.publish_alarm(code, f"Test alarm for {code}") + + # No alarms recorded (transport was down) + assert transport.alarm_events() == [] + # ERROR log must have been emitted + assert any("alarm publish" in r.message for r in caplog.records), ( + f"Expected ERROR log for code={code}" + ) + + +# ------------------------------------------------------------------ additional edge cases + + +@pytest.mark.parametrize("code", CRITICAL_CODES) +def test_publish_alarm_arbitrary_exception_swallowed(code: str, caplog: pytest.LogCaptureFixture): + """Any exception (not just TransportUnavailableError) is swallowed + ERROR logged.""" + class BrokenTransport: + def publish_alarm_event(self, routing_key, payload): + raise RuntimeError("boom") + + 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(code, "boom test") # Must NOT raise + + assert any("alarm publish" in r.message for r in caplog.records) + + +def test_publish_alarm_none_transport_skip(caplog: pytest.LogCaptureFixture): + """transport=None: publish is skipped with INFO log (reset.py / CLI path, REQ-100 KD-2).""" + emitter = AlarmEmitter(None) + with caplog.at_level(logging.INFO, logger="trex_core.application.alarm_emitter"): + emitter.publish_alarm("kill_switch_fired", "no transport test") + + assert any("alarm skipped" in r.message for r in caplog.records) + + +def test_publish_alarm_empty_fields_default(): + """fields defaults to {} when not provided — not None in payload.""" + emitter, transport = _make_emitter() + emitter.publish_alarm("kill_switch_fired", "no fields") + _, payload = transport.alarm_events()[0] + assert payload["fields"] == {} + + +def test_publish_alarm_routing_key_format(): + """routing key must be alarm.critical.trex-core for default severity.""" + emitter, transport = _make_emitter() + emitter.publish_alarm("kill_switch_fired", "routing key test") + rk, _ = transport.alarm_events()[0] + assert rk == "alarm.critical.trex-core" + + +# ------------------------------------------------------------------ domain purity sweep + + +def test_alarm_emitter_no_pika_import(): + """AlarmEmitter module must not import pika (domain/application purity).""" + import importlib + import sys + + # Ensure pika is not in alarm_emitter's import chain at module level. + mod = sys.modules.get("trex_core.application.alarm_emitter") + if mod is None: + mod = importlib.import_module("trex_core.application.alarm_emitter") + + # pika should NOT be in the module's global namespace + assert "pika" not in dir(mod), "alarm_emitter must not import pika directly" + + +def test_alarm_emitter_no_telegram_reference(): + """Verify no TELEGRAM_* references in alarm_emitter source (REQ-100 AC-5).""" + import inspect + import trex_core.application.alarm_emitter as ae_mod + + 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" diff --git a/trex_core/adapters/messaging/in_memory.py b/trex_core/adapters/messaging/in_memory.py index 80b6eac..655ef5f 100644 --- a/trex_core/adapters/messaging/in_memory.py +++ b/trex_core/adapters/messaging/in_memory.py @@ -29,6 +29,7 @@ class InMemoryTransport: self._rpc_handlers: dict[str, RpcHandler] = {} self._event_subscribers: list[tuple[str, EventHandler]] = [] # (pattern, handler) self._dlq: list[tuple[str, dict]] = [] # (routing_key, payload) + self._alarm_events: list[tuple[str, dict]] = [] # (routing_key, payload) dcy-033 self._available = True self._health = TransportHealth.CONNECTED @@ -53,6 +54,21 @@ class InMemoryTransport: with self._lock: return list(self._dlq) + def alarm_events(self) -> list[tuple[str, dict]]: + """Return recorded alarm events (test affordance, dcy-033).""" + with self._lock: + return list(self._alarm_events) + + # --- alarm events (dcy-033 / REQ-099) --- + def publish_alarm_event(self, routing_key: str, payload: dict) -> None: + """Stub for trex.alerts publish. Records event; raises when unavailable.""" + with self._lock: + if not self._available: + raise TransportUnavailableError( + "transport disconnected; alarm event dropped" + ) + self._alarm_events.append((routing_key, dict(payload))) + # --- events --- def publish_event( self, diff --git a/trex_core/adapters/messaging/rabbitmq.py b/trex_core/adapters/messaging/rabbitmq.py index 221fa0f..db031b6 100644 --- a/trex_core/adapters/messaging/rabbitmq.py +++ b/trex_core/adapters/messaging/rabbitmq.py @@ -22,12 +22,13 @@ import json import logging import threading import time -from typing import Optional +from typing import Callable, Optional import pika import pika.exceptions from trex_core.adapters.messaging.topology import ( + ALERTS_EXCHANGE, BROADCAST_EXCHANGE, DEFAULT_INSTANCE_KEY, DIRECT_EXCHANGE, @@ -65,6 +66,7 @@ class RabbitMQTransport: timeout: float = 5.0, intent_prefetch: int = 1, snapshot_prefetch: int = 8, + alarm_callback: Optional[Callable[[str, str, dict], None]] = None, ) -> None: self._url = url self._timeout = timeout @@ -74,6 +76,20 @@ class RabbitMQTransport: self._conn: Optional[pika.BlockingConnection] = None self._channel: Optional[pika.adapters.blocking_connection.BlockingChannel] = None self._health = TransportHealth.DISCONNECTED + # dcy-033 / REQ-100 (KD-1): alarm callback for adapter-layer alarm codes. + # Injected by daemon.py after AlarmEmitter construction (avoid circular init). + # Signature: (code: str, title: str, fields: dict) -> None + self._alarm_callback: Optional[Callable[[str, str, dict], None]] = alarm_callback + + def set_alarm_callback( + self, callback: Optional[Callable[[str, str, dict], None]] + ) -> None: + """Late-bind alarm callback (dcy-033 / REQ-100 KD-1). + + Allows daemon.py to inject the callback after AlarmEmitter is created, + avoiding circular initialization between transport and emitter. + """ + self._alarm_callback = callback # ------------------------------------------------------------ lifecycle def _connect(self, connect_timeout: Optional[float] = None) -> None: @@ -163,6 +179,9 @@ class RabbitMQTransport: # Events topic + DLQ ch.exchange_declare(EVENTS_EXCHANGE, exchange_type="topic", durable=True) ch.exchange_declare(EVENTS_DLQ_EXCHANGE, exchange_type="fanout", durable=True) + # dcy-033 / REQ-099: alarm channel — independent of trex.events, no DLQ, + # mandatory=False (notifier down = silent drop, no core-path blocker). + ch.exchange_declare(ALERTS_EXCHANGE, exchange_type="topic", durable=True) ch.queue_declare(EVENTS_DLQ_QUEUE, durable=True) ch.queue_bind(EVENTS_DLQ_QUEUE, EVENTS_DLQ_EXCHANGE) # BAT-013 / REQ-050: multi-instance control exchanges. @@ -259,6 +278,12 @@ class RabbitMQTransport: self._publish_dlq(routing_key, payload, reason="unroutable") except (pika.exceptions.AMQPError, OSError) as retry_exc: self._mark_disconnected() + # dcy-033 / REQ-100: fire alarm for publish unavailable after reconnect. + self._fire_alarm_callback( + "publish_unavailable_after_reconnect", + f"재연결 후에도 이벤트 발행 불가: rk={routing_key}", + {"routing_key": routing_key, "error": str(retry_exc)}, + ) raise TransportUnavailableError( f"publish_event failed after reconnect+retry: " f"{retry_exc.__class__.__name__}: {retry_exc}" @@ -289,8 +314,59 @@ class RabbitMQTransport: content_type="application/json", delivery_mode=2 ), ) - except Exception: # noqa: BLE001 - DLQ best-effort + except Exception as dlq_exc: # noqa: BLE001 - DLQ best-effort _log.error("DLQ publish failed (rk=%s)", routing_key, exc_info=True) + # dcy-033 / REQ-100: fire alarm for DLQ publish failure. + self._fire_alarm_callback( + "dlq_publish_failed", + f"DLQ 발행 실패: rk={routing_key}", + {"routing_key": routing_key, "error": str(dlq_exc)}, + ) + + def publish_alarm_event(self, routing_key: str, payload: dict) -> None: + """Publish to trex.alerts exchange (dcy-033 / REQ-099). + + Best-effort: mandatory=False, delivery_mode=1 (non-persistent). + No retry on failure — alarm must never block the execution path. + Raises TransportUnavailableError on connect failure (caller swallows). + """ + body = json.dumps(payload).encode("utf-8") + try: + self._ensure_connected() + ch = self._channel + assert ch is not None + ch.basic_publish( + exchange=ALERTS_EXCHANGE, + routing_key=routing_key, + body=body, + properties=pika.BasicProperties( + content_type="application/json", + delivery_mode=1, # non-persistent — best-effort alarm + ), + mandatory=False, # silent drop when notifier not bound + ) + except (pika.exceptions.AMQPError, OSError) as exc: + self._mark_disconnected() + raise TransportUnavailableError( + f"publish_alarm_event failed: {exc.__class__.__name__}: {exc}" + ) from exc + + def _fire_alarm_callback( + self, code: str, title: str, fields: dict | None = None + ) -> None: + """Invoke the injected alarm callback (dcy-033 / REQ-100 KD-1). + + Used by adapter-layer alarm sources (publish_unavailable_after_reconnect, + dlq_publish_failed) that cannot call AlarmEmitter directly. + No-op when callback is not set. Swallows callback exceptions. + """ + cb = self._alarm_callback + if cb is None: + return + try: + cb(code, title, fields or {}) + except Exception: # noqa: BLE001 — callback must never crash the adapter + _log.debug("alarm callback raised (code=%s)", code, exc_info=True) # ------------------------------------------------------------ RPC client def call_rpc( diff --git a/trex_core/adapters/messaging/topology.py b/trex_core/adapters/messaging/topology.py index 83e807d..62ac19b 100644 --- a/trex_core/adapters/messaging/topology.py +++ b/trex_core/adapters/messaging/topology.py @@ -62,6 +62,20 @@ EVENTS_EXCHANGE = "trex.events" # topic, durable EVENTS_DLQ_EXCHANGE = "trex.events.dlq" # fanout, durable; receives unroutable/NACKed EVENTS_DLQ_QUEUE = "trex.events.dlq" # one collector queue bound to DLQ exchange +# --- Alarm channel (dcy-033 / REQ-099) --- +# Independent of trex.events. mandatory=False (best-effort, notifier down = silent drop). +ALERTS_EXCHANGE = "trex.alerts" # topic, durable; alarm envelope channel + + +def alarm_routing_key(severity: str, source: str) -> str: + """Build the routing key for an alarm event (dcy-033 / REQ-098). + + Pattern: alarm.{severity}.{source} (e.g. alarm.critical.trex-core) + Severity is always lower-case. Source tokens containing '.' are safe + because notifier binds with alarm.# pattern (zero-or-more tail words). + """ + return f"alarm.{severity}.{source}" + def event_routing_key(event_type: str, strategy_id: str, symbol: str) -> str: """Build the routing key for an event (dcy-006 OQ-2 mapping).""" diff --git a/trex_core/application/alarm_emitter.py b/trex_core/application/alarm_emitter.py new file mode 100644 index 0000000..a045c8e --- /dev/null +++ b/trex_core/application/alarm_emitter.py @@ -0,0 +1,93 @@ +"""Alarm emitter — best-effort CRITICAL alarm publisher (dcy-033 / REQ-100). + +Publishes REQ-098 alarm envelopes to the `trex.alerts` exchange via the +MessagingTransport port. No pika import here — dependency inversion +preserved (NFR-1). Domain layer is never touched. + +Caller contract: + - Inject into execution / reconciliation / killswitch / ledger_outbox / etc. + - Call publish_alarm(code, title, fields=...) at each alarm point. + - publish_alarm() NEVER re-raises. Alarm failure is ERROR-logged and swallowed + so the core business path (slot transition / order execution / ledger) is + never blocked by alarm delivery. + +transport=None is explicitly supported for CLI entry points (e.g. reset.py) +where no broker connection exists. In that case publish_alarm() logs and skips. +""" +from __future__ import annotations + +import logging +from datetime import datetime, timezone, timedelta +from typing import Optional + +from trex_core.application.messaging import MessagingTransport, TransportUnavailableError +from trex_core.adapters.messaging.topology import alarm_routing_key + +_log = logging.getLogger(__name__) + +_KST = timezone(timedelta(hours=9)) + +# Wire constants (REQ-098) +_SOURCE = "trex-core" +_SCHEMA_VERSION = "1" + + +class AlarmEmitter: + """Application-layer alarm publisher (dcy-033 / REQ-100). + + Wraps MessagingTransport.publish_alarm_event() with: + - envelope construction (source/severity/occurred_at/schema_version fixed) + - exception swallowing + ERROR log on failure + - None-transport guard for non-daemon entry points + """ + + def __init__(self, transport: Optional[MessagingTransport]) -> None: + self._transport = transport + + def publish_alarm( + self, + code: str, + title: str, + *, + fields: Optional[dict] = None, + severity: str = "critical", + ) -> None: + """Publish a single alarm envelope. Best-effort: never re-raises. + + Args: + code: Alarm identifier code (see ALARM-INVENTORY.md §A). + title: Human-readable 1-line summary. No credentials or raw + exception bodies — callers must sanitize. + fields: Optional structured context dict (opaque, default {}). + severity: 'critical' | 'warn' | 'info'. Default 'critical'. + """ + if self._transport is None: + _log.info( + "alarm skipped (no transport): code=%s title=%s", + code, title, + ) + return + + routing_key = alarm_routing_key(severity, _SOURCE) + occurred_at = datetime.now(tz=_KST).isoformat() + payload = { + "source": _SOURCE, + "severity": severity, + "code": code, + "title": title, + "fields": fields or {}, + "occurred_at": occurred_at, + "schema_version": _SCHEMA_VERSION, + } + try: + self._transport.publish_alarm_event(routing_key, payload) + except TransportUnavailableError as exc: + _log.error( + "alarm publish skipped (transport unavailable): code=%s: %s", + code, exc, + ) + except Exception: # noqa: BLE001 — alarm must never crash the caller + _log.error( + "alarm publish failed: code=%s title=%s", + code, title, exc_info=True, + ) diff --git a/trex_core/application/execution.py b/trex_core/application/execution.py index 30df1f3..4a9f4c5 100644 --- a/trex_core/application/execution.py +++ b/trex_core/application/execution.py @@ -27,6 +27,7 @@ from dataclasses import dataclass from datetime import datetime, timezone from typing import Callable, Optional +from trex_core.application.alarm_emitter import AlarmEmitter from trex_core.application.broker_port import BrokerPort, OrderRequest from trex_core.application.ledger_outbox import LedgerOutbox from trex_core.application.ledger_port import LedgerTrade @@ -67,6 +68,7 @@ class ExecutionService: event_emitter: Optional[SlotEventEmitter] = None, stale_pending_threshold_sec: Optional[int] = None, instance_key: str = "default", + alarm_emitter: Optional[AlarmEmitter] = None, ) -> None: self._svc = slot_service self._repo = repo @@ -85,6 +87,8 @@ class ExecutionService: if stale_pending_threshold_sec and stale_pending_threshold_sec > 0 else None ) + # dcy-033 / REQ-100: best-effort alarm emitter (None = no alarms sent). + self._alarm_emitter = alarm_emitter # ------------------------------------------------------------ def run_once(self) -> ExecutionResult: @@ -130,6 +134,13 @@ class ExecutionService: "(divergence — no order submitted; operator/reconcile attention)", slot.slot_id, slot.symbol, ) + # dcy-033 / REQ-100: CRITICAL alarm — PENDING_SELL may stick. + if self._alarm_emitter is not None: + self._alarm_emitter.publish_alarm( + "sell_skipped_no_balance", + f"매도 dispatch skip: 브로커 잔고 0 (slot={slot.slot_id} symbol={slot.symbol})", + fields={"slot_id": str(slot.slot_id), "symbol": slot.symbol}, + ) continue else: qty = int(slot.qty or 0) @@ -265,6 +276,14 @@ class ExecutionService: slot.slot_id, slot.symbol, slot.broker_order_id, avg, limit, ) + # dcy-033 / REQ-100: CRITICAL alarm — slot stuck in PENDING. + if self._alarm_emitter is not None: + self._alarm_emitter.publish_alarm( + "fill_price_unresolvable", + f"체결가 미해결: 슬롯 PENDING 고착 위험 (slot={slot.slot_id} symbol={slot.symbol})", + fields={"slot_id": str(slot.slot_id), "symbol": slot.symbol, + "broker_order_id": slot.broker_order_id}, + ) return None, False def _on_filled(self, slot, fs, side: str, r: ExecutionResult) -> None: diff --git a/trex_core/application/killswitch.py b/trex_core/application/killswitch.py index b0cf5e6..f964beb 100644 --- a/trex_core/application/killswitch.py +++ b/trex_core/application/killswitch.py @@ -11,7 +11,9 @@ from __future__ import annotations import logging from datetime import date as date_cls from datetime import datetime, timezone +from typing import Optional +from trex_core.application.alarm_emitter import AlarmEmitter from trex_core.application.portfolio import GLOBAL_KILLSWITCH, KillSwitchStore from trex_core.domain.slot import SlotRepository, SlotService @@ -29,11 +31,14 @@ class KillSwitchService: slot_service: SlotService, killswitch_store: KillSwitchStore, instance_key: str = "default", + alarm_emitter: Optional[AlarmEmitter] = None, ) -> None: self._repo = slot_repo self._svc = slot_service self._store = killswitch_store self._instance_key = instance_key # dcy-020/REQ-056: instance-scoped sweep + # dcy-033 / REQ-100: best-effort alarm emitter. + self._alarm_emitter = alarm_emitter @staticmethod def _trading_date(now: datetime) -> date_cls: @@ -73,6 +78,14 @@ class KillSwitchService: "kill-switch fired: actor=%s scope=%s portfolio_id=%s liquidated=%d", actor, scope, portfolio_id, liquidated, ) + # dcy-033 / REQ-100: CRITICAL alarm — emergency liquidation fired. + if self._alarm_emitter is not None: + self._alarm_emitter.publish_alarm( + "kill_switch_fired", + f"긴급 청산 발동: scope={scope} liquidated={liquidated}", + fields={"scope": scope, "portfolio_id": portfolio_id, "liquidated": liquidated, + "actor": actor}, + ) return liquidated def resume( diff --git a/trex_core/application/ledger_outbox.py b/trex_core/application/ledger_outbox.py index aea94e7..539e28f 100644 --- a/trex_core/application/ledger_outbox.py +++ b/trex_core/application/ledger_outbox.py @@ -30,6 +30,7 @@ from datetime import datetime, timedelta, timezone from enum import Enum from typing import Callable, Optional, Protocol +from trex_core.application.alarm_emitter import AlarmEmitter from trex_core.application.ledger_port import ( LedgerConflictError, LedgerPermanentError, @@ -169,6 +170,7 @@ class LedgerWriter: max_attempts: int = _DEFAULT_MAX_ATTEMPTS, clock: Optional[Callable[[], datetime]] = None, rng: Optional[Callable[[], float]] = None, + alarm_emitter: Optional[AlarmEmitter] = None, ) -> None: self._outbox = outbox self._ledger = ledger @@ -181,6 +183,8 @@ class LedgerWriter: self._max_attempts = max_attempts self._clock = clock or (lambda: datetime.now(tz=timezone.utc)) self._rng = rng or random.random # jitter source; inject for determinism + # dcy-033 / REQ-100: best-effort alarm emitter. + self._alarm_emitter = alarm_emitter def enqueue_fill(self, trade: LedgerTrade) -> LedgerWorkItem: return self._outbox.enqueue(trade) @@ -219,6 +223,15 @@ class LedgerWriter: self._outbox.mark_outcome( item.correlation_id, OutboxOutcome.CONFLICT_BLOCKED, error=str(exc) ) + # dcy-033 / REQ-100: CRITICAL alarm — 409 conflict, operator resolution needed. + if self._alarm_emitter is not None: + self._alarm_emitter.publish_alarm( + "ledger_delivery_failed", + f"원장 409 충돌 (conflict_blocked): {item.correlation_id}", + fields={"correlation_id": item.correlation_id, + "asset_id": item.trade.asset_id, + "reason": "conflict_blocked"}, + ) return OutboxOutcome.CONFLICT_BLOCKED except LedgerPermanentError as exc: # dcy-015 / REQ-017/018 (issue #13): non-retryable client error → @@ -230,6 +243,15 @@ class LedgerWriter: "ledger delivery_failed (permanent): correlation_id=%s reason=%s", item.correlation_id, exc, ) + # dcy-033 / REQ-100: CRITICAL alarm — permanent failure, no retry. + if self._alarm_emitter is not None: + self._alarm_emitter.publish_alarm( + "ledger_delivery_failed", + f"원장 영구 실패 (permanent): {item.correlation_id}", + fields={"correlation_id": item.correlation_id, + "asset_id": item.trade.asset_id, + "reason": "permanent_failure"}, + ) return OutboxOutcome.PERMANENT_FAILURE except LedgerRetryableError as exc: # dcy-015 / REQ-019: backoff, or dead-letter once max_attempts hit. @@ -243,6 +265,16 @@ class LedgerWriter: "correlation_id=%s reason=%s", new_attempts, item.correlation_id, exc, ) + # dcy-033 / REQ-100: CRITICAL alarm — dead-lettered after max retries. + if self._alarm_emitter is not None: + self._alarm_emitter.publish_alarm( + "ledger_delivery_failed", + f"원장 데드레터 (attempts={new_attempts}): {item.correlation_id}", + fields={"correlation_id": item.correlation_id, + "asset_id": item.trade.asset_id, + "attempts": new_attempts, + "reason": "dead_lettered"}, + ) return OutboxOutcome.DEAD_LETTERED next_retry_at = self._clock() + timedelta( seconds=self._backoff_delay(new_attempts) diff --git a/trex_core/application/messaging.py b/trex_core/application/messaging.py index 3da13a5..2099843 100644 --- a/trex_core/application/messaging.py +++ b/trex_core/application/messaging.py @@ -79,6 +79,19 @@ class MessagingTransport(Protocol): Raises TransportUnavailableError if not connected.""" ... + def publish_alarm_event( + self, + routing_key: str, + payload: dict, + ) -> None: + """Publish to the trex.alerts exchange (dcy-033 / REQ-099). + + Best-effort: mandatory=False so an unroutable message (notifier not + connected) is silently dropped — no DLQ, no retry. + Raises TransportUnavailableError if not connected (caller must swallow). + """ + ... + def call_rpc( self, queue: str, diff --git a/trex_core/application/reconciliation.py b/trex_core/application/reconciliation.py index 47a1ca2..19a4be2 100644 --- a/trex_core/application/reconciliation.py +++ b/trex_core/application/reconciliation.py @@ -9,7 +9,9 @@ manual holds). from __future__ import annotations import logging +from typing import Optional +from trex_core.application.alarm_emitter import AlarmEmitter from trex_core.application.broker_port import BrokerPort from trex_core.application.portfolio import SnapshotStore from trex_core.domain.slot.state import SlotState @@ -17,8 +19,14 @@ from trex_core.domain.slot.state import SlotState _log = logging.getLogger(__name__) -def reconcile_core_holdings(*, slot_repo, slot_service, broker: BrokerPort, - instance_key: str) -> dict[str, str]: +def reconcile_core_holdings( + *, + slot_repo, + slot_service, + broker: BrokerPort, + instance_key: str, + alarm_emitter: Optional[AlarmEmitter] = None, +) -> dict[str, str]: """dcy-020 / REQ-062 (BAT-029): surface core-slot holding ↔ broker divergence. Broker balance is the reconcile ground truth. Decisions (no silent auto-fix, @@ -48,6 +56,13 @@ def reconcile_core_holdings(*, slot_repo, slot_service, broker: BrokerPort, "core holding but broker flat: slot=%s symbol=%s -> error", slot.slot_id, sym, ) + # dcy-033 / REQ-100: CRITICAL alarm — position mismatch. + if alarm_emitter is not None: + alarm_emitter.publish_alarm( + "core_holding_broker_flat", + f"슬롯 보유인데 브로커 잔고 0 (slot={slot.slot_id} symbol={sym})", + fields={"slot_id": str(slot.slot_id), "symbol": sym}, + ) except Exception: # noqa: BLE001 - never let reconcile crash the loop _log.exception("failed to error-quarantine slot %s", slot.slot_id) elif slot.qty is not None and int(slot.qty) != bq: # C — qty mismatch diff --git a/trex_core/application/snapshot_scheduler.py b/trex_core/application/snapshot_scheduler.py index 0ebddf7..3d16c3c 100644 --- a/trex_core/application/snapshot_scheduler.py +++ b/trex_core/application/snapshot_scheduler.py @@ -12,6 +12,7 @@ import threading from datetime import datetime, time, timedelta, timezone from typing import Callable, Optional +from trex_core.application.alarm_emitter import AlarmEmitter from trex_core.application.broker_port import BrokerPort from trex_core.application.portfolio import ( DegradedReason, @@ -40,6 +41,7 @@ class SnapshotScheduler: slot_repo=None, slot_service=None, instance_key: str = "default", + alarm_emitter: Optional[AlarmEmitter] = None, ) -> None: self._read = read_port self._store = store @@ -55,6 +57,8 @@ class SnapshotScheduler: self._slot_service = slot_service self._instance_key = instance_key self._last_universe_size: Optional[int] = None + # dcy-033 / REQ-100: best-effort alarm emitter. + self._alarm_emitter = alarm_emitter # ---- callable from a daemon thread loop -------------------------------- def run_forever(self, stop: threading.Event, *, tick_sec: float = 60.0) -> None: @@ -88,6 +92,13 @@ class SnapshotScheduler: DegradedReason.PORTFOLIO_HUB_UNAVAILABLE, str(exc) ) _log.warning("whoami failed: %s", exc) + # dcy-033 / REQ-100: CRITICAL alarm — portfolio hub unavailable. + if self._alarm_emitter is not None: + self._alarm_emitter.publish_alarm( + "portfolio_hub_unavailable", + "portfolio-hub whoami 실패: 포트폴리오 정보 부실화", + fields={"reason": "whoami_failed"}, + ) return False try: items = self._read.get_portfolios(self._daemon_id) @@ -96,6 +107,13 @@ class SnapshotScheduler: DegradedReason.PORTFOLIO_HUB_UNAVAILABLE, str(exc) ) _log.warning("get_portfolios failed: %s", exc) + # dcy-033 / REQ-100: CRITICAL alarm — portfolio hub unavailable. + if self._alarm_emitter is not None: + self._alarm_emitter.publish_alarm( + "portfolio_hub_unavailable", + "portfolio-hub get_portfolios 실패: 포트폴리오 정보 부실화", + fields={"daemon_id": self._daemon_id, "reason": "get_portfolios_failed"}, + ) return False except Exception as exc: # noqa: BLE001 - shape/parsing failure self._store.set_degraded( @@ -120,6 +138,7 @@ class SnapshotScheduler: reconcile_core_holdings( slot_repo=self._slot_repo, slot_service=self._slot_service, broker=self._broker, instance_key=self._instance_key, + alarm_emitter=self._alarm_emitter, # dcy-033 / REQ-100 ) except Exception: # noqa: BLE001 _log.exception("core-holding divergence reconcile failed") diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index 20c73e1..4e66280 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -44,6 +44,7 @@ from trex_core.adapters.portfolio import ( StubPortfolioRead, ) from trex_core.adapters.portfolio.killswitch_store import PostgreSQLKillSwitchStore +from trex_core.application.alarm_emitter import AlarmEmitter from trex_core.application.broker_port import BrokerPort from trex_core.application.execution import ExecutionService from trex_core.application.intents import IntentService @@ -169,25 +170,42 @@ class Daemon: # BAT-008: portfolio-hub read + snapshot store + durable pause + scheduler. # (Created before the ledger writer so option-D write-back can target it.) self.snapshot_store = SnapshotStore() + # BAT-010/011: per-slot watchdog (snapshot RPC writer, safety-sell reader). + self.watchdog = WatchdogStore() + self.pause_store = PostgreSQLPauseStore(config.dsn) + # BAT-012: day-scoped kill-switch exclusion store + service. + self.killswitch_store = PostgreSQLKillSwitchStore(config.dsn) + self.portfolio_read = _build_portfolio_read(config) + + # Event publisher transport — one dedicated instance for the + # execution-loop thread (pika is not thread-safe). + self.event_publisher: MessagingTransport = _build_transport(config) + self._transports.append(self.event_publisher) + + # dcy-033 / REQ-100: alarm emitter — shares the event_publisher transport + # (both run on the execution-loop thread; pika thread-safety rule satisfied). + # Created before services so all services receive the same emitter instance. + self.alarm_emitter = AlarmEmitter(self.event_publisher) + # Late-bind alarm callback into event_publisher for adapter-layer alarm codes + # (KD-1: publish_unavailable_after_reconnect / dlq_publish_failed). + if hasattr(self.event_publisher, "set_alarm_callback"): + self.event_publisher.set_alarm_callback(self.alarm_emitter.publish_alarm) + self.ledger_writer = LedgerWriter( self.outbox, _build_ledger(config), snapshot_store=self.snapshot_store, # dcy-018 / REQ-030: option D write-back retry_base_sec=config.ledger_retry_base_sec, retry_cap_sec=config.ledger_retry_cap_sec, max_attempts=config.ledger_max_attempts, + alarm_emitter=self.alarm_emitter, # dcy-033 / REQ-100 ) - # BAT-010/011: per-slot watchdog (snapshot RPC writer, safety-sell reader). - self.watchdog = WatchdogStore() - self.pause_store = PostgreSQLPauseStore(config.dsn) - # BAT-012: day-scoped kill-switch exclusion store + service. - self.killswitch_store = PostgreSQLKillSwitchStore(config.dsn) self.kill_switch = KillSwitchService( slot_repo=self.slot_repo, slot_service=self.slot_service, killswitch_store=self.killswitch_store, instance_key=config.instance_key, + alarm_emitter=self.alarm_emitter, # dcy-033 / REQ-100 ) - self.portfolio_read = _build_portfolio_read(config) self.snapshot_scheduler = SnapshotScheduler( read_port=self.portfolio_read, store=self.snapshot_store, @@ -199,19 +217,15 @@ class Daemon: slot_repo=self.slot_repo, slot_service=self.slot_service, instance_key=config.instance_key, + alarm_emitter=self.alarm_emitter, # dcy-033 / REQ-100 ) - - # Event publisher transport — one dedicated instance for the - # execution-loop thread (pika is not thread-safe). - self.event_publisher: MessagingTransport = _build_transport(config) - self._transports.append(self.event_publisher) - self.execution = ExecutionService( self.slot_service, self.slot_repo, self.broker, ledger_outbox=self.outbox, event_emitter=self._make_event_emitter(), stale_pending_threshold_sec=config.pending_stale_threshold_sec, instance_key=config.instance_key, + alarm_emitter=self.alarm_emitter, # dcy-033 / REQ-100 ) # BAT-011: safety-sell trigger service (forced_close / SL-TP / market-close). diff --git a/trex_core/runtime/reset.py b/trex_core/runtime/reset.py index 3d625df..02ba540 100644 --- a/trex_core/runtime/reset.py +++ b/trex_core/runtime/reset.py @@ -29,6 +29,21 @@ from typing import Any, Optional _log = logging.getLogger(__name__) + +def _emit_reset_alarm( + alarm_emitter: Optional[object], + code: str, + title: str, + fields: dict | None = None, +) -> None: + """Best-effort alarm from reset.py (transport may be None — dcy-033 / REQ-100 KD-2).""" + if alarm_emitter is None: + return + try: + alarm_emitter.publish_alarm(code, title, fields=fields) # type: ignore[union-attr] + except Exception: # noqa: BLE001 + _log.debug("reset alarm emit failed (code=%s)", code, exc_info=True) + # slot-core tables owned by trex-core migrations (drop order irrelevant — no FKs). SLOT_CORE_TABLES: tuple[str, ...] = ( "daily_slot", @@ -74,6 +89,7 @@ def run_reset( *, dry_run: bool = True, psycopg_module: Any = None, + alarm_emitter: Optional[object] = None, ) -> ResetReport: """Wipe the slot-core tables and reapply migrations to the asset-centric schema. @@ -106,7 +122,17 @@ def run_reset( # 010 sets the inactive CHECK, 011 drops the pool-model columns. from trex_core.adapters.persistence.migrate import apply_migrations - report.migrations_applied = apply_migrations(dsn, psycopg_module=psycopg_module) + try: + report.migrations_applied = apply_migrations(dsn, psycopg_module=psycopg_module) + except Exception as mig_exc: # noqa: BLE001 + # dcy-033 / REQ-100: CRITICAL alarm — migration failed. + _emit_reset_alarm( + alarm_emitter, + "migration_failed", + f"마이그레이션 실패: {mig_exc.__class__.__name__}", + fields={"error": str(mig_exc)}, + ) + raise from trex_core.domain.slot import EXPECTED_SLOT_FIELDS @@ -118,6 +144,14 @@ def run_reset( ) cols = {r[0] for r in cur.fetchall()} report.schema_ok = cols == set(EXPECTED_SLOT_FIELDS) + if not report.schema_ok: + # dcy-033 / REQ-100: CRITICAL alarm — schema verification failed post-reset. + _emit_reset_alarm( + alarm_emitter, + "schema_verification_failed", + "리셋 후 schema 검증 실패: daily_slot 컬럼 불일치", + fields={"expected": sorted(EXPECTED_SLOT_FIELDS), "actual": sorted(cols)}, + ) return report