From c66acf6e47872a33b6e365e222ffb851caa01bd3 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Sun, 20 Sep 2026 20:34:00 +0900 Subject: [PATCH] fix: preserve sell event routing owner --- .proofline/criteria/ac-0038.md | 16 ++ .proofline/criteria/ac-0039.md | 16 ++ .proofline/identities.json | 4 +- .proofline/lines/line-0011/dcy-0011.md | 80 +++++++ .../lines/line-0011/evidence/activity-log.md | 21 ++ .proofline/lines/line-0011/line-0011.md | 4 + .proofline/lines/line-0011/req-0011.md | 43 ++++ docs/interface-contract.md | 5 + docs/runtime-flows.md | 6 +- .../messaging/test_publish_reliability.py | 25 +++ .../messaging/test_rabbitmq_transport.py | 58 +++++ .../persistence/test_postgresql_repository.py | 32 +++ tests/application/test_execution_cancel.py | 62 ++++- .../test_messaging_event_publisher.py | 211 ++++++++++++++++++ tests/domain/slot/test_service_sell.py | 3 +- .../runtime/test_sell_event_owner_routing.py | 60 +++++ trex_core/application/execution.py | 86 +++++-- trex_core/application/intents.py | 14 +- trex_core/domain/slot/service.py | 11 + trex_core/runtime/daemon.py | 41 +++- 20 files changed, 766 insertions(+), 32 deletions(-) create mode 100644 .proofline/criteria/ac-0038.md create mode 100644 .proofline/criteria/ac-0039.md create mode 100644 .proofline/lines/line-0011/dcy-0011.md create mode 100644 .proofline/lines/line-0011/evidence/activity-log.md create mode 100644 .proofline/lines/line-0011/line-0011.md create mode 100644 .proofline/lines/line-0011/req-0011.md create mode 100644 tests/runtime/test_sell_event_owner_routing.py diff --git a/.proofline/criteria/ac-0038.md b/.proofline/criteria/ac-0038.md new file mode 100644 index 0000000..d3ca536 --- /dev/null +++ b/.proofline/criteria/ac-0038.md @@ -0,0 +1,16 @@ +--- +id: "ac-0038" +status: active +--- + +# SELL event owner 보존과 lifecycle 일관성 + +## Criterion + +SELL admission은 non-empty `SellIntent.sell_strategy_id`, slot에 보존된 non-empty `sell_strategy_id`, non-empty `strategy_id` 순서로 event owner를 확정하여 durable `sell_strategy_id`에 저장해야 한다. Full fill과 partial fill 후 external cancel은 terminal 처리 직전의 이 canonical field를 routing hint로 캡처하여, 이후 `INACTIVE` 전이가 ownership field를 지우더라도 같은 owner를 사용해야 한다. Dispatch 전 daemon 재시작·재조회 후에도 owner가 같아야 한다. Zero-fill cancel/reject는 fill event 없이 `HOLDING`으로 rollback하면서 owner를 보존하고, 이후 명시적 새 owner가 없는 SELL admission은 보존 owner를 재사용하며 명시적 새 owner만 이를 교체해야 한다. Whitespace-only 값은 owner로 인정하지 않는다. + +## Verification + +- 성공 조건: BUY owner `A`와 명시적 SELL owner `B`가 다른 주문의 full fill 및 partial fill 후 external cancel event가 모두 `slot_filled_sell.B.`로 발행된다. 별도 SELL owner가 없는 safety/kill-switch sell은 보존된 SELL owner가 있으면 이를, 없으면 `A`를 사용한다. Zero-fill cancel 뒤 owner `B`가 `HOLDING`에 남고, owner를 생략한 재매도가 `B`를 사용하며, admission 뒤 durable row를 새 service/repository instance에서 다시 읽어 처리해도 owner가 같다. +- 실패·경계 조건: fill 뒤 clear된 field에서 owner를 재추측하거나, `sell_strategy_id=B`인데 `strategy_id=A`로 route하거나, zero-fill cancel/reject가 fill event를 만들거나 owner를 지우거나, owner 생략 재매도가 보존된 `sell_strategy_id`를 무시하면 FAIL이다. 공백 문자열은 owner로 인정하지 않는다. +- 검사 방법: `IntentService` admission과 `ExecutionService` 단위·통합 테스트로 owner 우선순위, full/partial/zero-fill, safety sell, cancel→재매도, repository reload/restart simulation을 검증하고 disposable PostgreSQL 경로에서 `sell_strategy_id` 영속 round-trip을 확인한다. diff --git a/.proofline/criteria/ac-0039.md b/.proofline/criteria/ac-0039.md new file mode 100644 index 0000000..9318556 --- /dev/null +++ b/.proofline/criteria/ac-0039.md @@ -0,0 +1,16 @@ +--- +id: "ac-0039" +status: active +--- + +# owner 없는 SELL event의 명시적 unroutable 처리와 전달 분리 + +## Criterion + +SELL terminal 처리에서 `sell_strategy_id`와 `strategy_id`가 모두 없거나 공백이면 정상 `trex.events` strategy event 또는 DLQ synthetic-owner message를 publish하지 않고 reason code `sell_event_owner_missing`으로 처리해야 한다. 해당 execution tick 결과의 dedicated counter를 증가시키고, instance·slot identity·symbol·event type·reason을 포함하되 secret, credential, raw broker payload나 raw exception text가 없는 structured ERROR log와 best-effort alarm을 남겨야 한다. 이 전달 실패와 정상 owner event의 transport publish 실패는 이미 완료된 broker fill, durable slot 전이, ledger outbox enqueue 및 dispatch 정리를 rollback하지 않아야 한다. 정상 owner event의 기존 payload `slot_id`, `event_type`, `sequence_at`은 publish/retry 경계에서 바꾸지 않아야 하며, consumer 구현 또는 새로운 replay·exactly-once 보장은 요구하지 않는다. + +## Verification + +- 성공 조건: owner 없는 full fill 및 partial fill 후 external cancel에서 `trex.events` publish 호출과 `_` strategy routing key가 모두 0회이고 counter·stable reason ERROR·alarm이 각각 관측된다. 동시에 slot/ledger/dispatch 결과는 owner가 있는 동일 체결과 같게 완료된다. owner가 있는 event의 최초 publish가 실패하고 transport가 재시도하는 경우 payload identity와 `sequence_at`이 변하지 않는다. +- 실패·경계 조건: `slot_filled_sell._.` 또는 다른 sentinel owner가 정상 exchange에 발행되거나, owner 미확정 또는 publish exception이 금융 state/outbox를 rollback하거나, 관측 reason이 동적 exception 문자열이면 FAIL이다. alarm transport 실패는 금융 결과를 변경하지 않아야 한다. +- 검사 방법: fake emitter/transport와 `caplog`·alarm recorder를 사용한 application/runtime 테스트로 publish call count, counter, reason, sanitized fields 및 금융 비변경을 검증한다. 기존 messaging retry/DLQ 테스트와 disposable RabbitMQ integration으로 mandatory publish·confirm·동일 payload retry·전략 queue 격리를 확인한다. RabbitMQ fixture가 없으면 integration은 SKIP으로 명시하고 단위 검증을 통과로 대체 주장하지 않는다. diff --git a/.proofline/identities.json b/.proofline/identities.json index b0c0c04..28811f2 100644 --- a/.proofline/identities.json +++ b/.proofline/identities.json @@ -1,5 +1,5 @@ { "schema_version": 1, - "next_line_number": 11, - "next_ac_number": 38 + "next_line_number": 12, + "next_ac_number": 40 } diff --git a/.proofline/lines/line-0011/dcy-0011.md b/.proofline/lines/line-0011/dcy-0011.md new file mode 100644 index 0000000..853df6f --- /dev/null +++ b/.proofline/lines/line-0011/dcy-0011.md @@ -0,0 +1,80 @@ +--- +id: "dcy-0011" +status: confirmed +--- + +# SELL 체결 이벤트 owner 보존과 unroutable 차단 + +## Problem + +SELL 주문이 체결되면 broker fill, slot 상태 전이, ledger outbox 처리는 정상 완료되더라도 전략 event의 routing owner가 비어 `slot_filled_sell._.`로 발행될 수 있다. `trex.events`의 전략 queue는 `*.{strategy_id}.#`로 binding되므로 `_`에 대응하는 consumer가 없고, RabbitMQ mandatory publish는 해당 event를 `NO_ROUTE`로 판단해 DLQ로 보낸다. 금융 처리는 완료됐지만 전략은 체결 전이를 받지 못해 실제 포지션과 전략 상태가 어긋날 수 있다. + +현재 `SellIntent.sell_strategy_id`와 slot의 `sell_strategy_id`는 SELL admission 및 persistence 경계에 존재하지만 `ExecutionService._on_filled()`와 `_on_partial_cancel()`은 event hint로 `slot.strategy_id`만 캡처한다. 이전 수정 `de97bc8`은 `PENDING_SELL → INACTIVE`가 `strategy_id`를 지우기 전에 값을 보존했지만, SELL owner가 `sell_strategy_id`에 있거나 `strategy_id`가 admission 전부터 비어 있는 경우에는 효력이 없다. Daemon emitter도 owner를 찾지 못하면 `_`를 정상 strategy routing segment처럼 사용한다. + +기대 상태는 SELL admission 시 event owner를 한 번 확정해 pending order lifecycle 동안 durable slot state로 보존하고, full fill·partial fill 후 cancel·zero-fill cancel 뒤 재매도·daemon 재시작에서도 같은 owner를 사용하는 것이다. Owner를 확정할 수 없는 event는 금융 처리와 slot transition을 되돌리지 않되 정상 strategy event로 가장하지 않고 stable reason, metric, log와 alarm으로 관측되어야 한다. + +## Evidence + +- `E-001` GitHub Issue [#95](https://github.com/fourblocks2020/trex-core/issues/95)는 2026-09-16~18 운영에서 symbol `003780`·`000390`의 SELL fill 두 건이 `trex.events.strategy._` 계열 key로 발행되어 `NO_ROUTE`/DLQ가 된 반면 broker fill, outbox completion과 portfolio ledger 저장은 성공했다고 기록한다. +- `E-002` 조사 기준 revision은 `main`의 `3f6a967aa14e7feedbff20967c4e870b46189167`이다. Issue #95 body와 comments 전체를 확인했으며 comment, 연결 PR과 open duplicate PR은 없다. `#95`, `SELL routing owner`, `NO_ROUTE strategy event` 검색에서도 중복 구현 PR이 발견되지 않았다. +- `E-003` `trex_core/application/intents.py`의 `SellIntent`는 `sell_strategy_id`를 받고, `_admit_sell()`은 `HOLDING → PENDING_SELL` 전이에 이를 넘긴다. `trex_core/domain/slot/service.py`는 non-`None` 값을 slot에 보존하며 `Slot.sell_strategy_id`는 PostgreSQL `_SLOT_COLUMNS`, update와 row materialization에 포함되어 재시작 후에도 유지된다. +- `E-004` `trex_core/application/execution.py`의 `_on_filled()`와 `_on_partial_cancel()`은 transition 전에 `routing_strategy_id = slot.strategy_id`만 저장한다. Full SELL은 `INACTIVE` 전이에서 `strategy_id`와 `sell_strategy_id`를 모두 지우며 partial SELL은 `HOLDING`으로 돌아가므로, event hint를 owner field에서 먼저 확보하지 않으면 full fill의 post-transition slot만으로 복구할 수 없다. +- `E-005` `trex_core/runtime/daemon.py::_make_event_emitter()`는 hint, `slot.strategy_id`, `_` 순서로 routing segment를 선택한다. 따라서 owner가 없을 때도 `topology.event_routing_key()`와 `publish_event(mandatory=True)`를 호출하고, RabbitMQ adapter는 `UnroutableError`를 DLQ에 적재한다. +- `E-006` commit `de97bc842d48599e4eb26b6e32130798ec97d5c9`은 sell transition 전 `strategy_id` 캡처로 이전 `NO_ROUTE`를 수정했다. 해당 diff와 commit 설명은 `sell_strategy_id`를 owner 후보로 다루지 않아 이번 운영 재현과 일치하는 부분 수정임을 확인했다. +- `E-007` `trex_core/application/execution.py::_rollback()`은 zero-fill reject/cancel에서 `PENDING_SELL → HOLDING`으로 되돌리고 `sell_strategy_id`를 보존한다. 따라서 새 cancel event를 만드는 대신 owner 보존과 다음 SELL admission의 동일 owner 사용을 검증할 수 있다. +- `E-008` 현재 event·RabbitMQ baseline인 `tests/application/test_messaging_event_publisher.py`, `tests/adapters/messaging/test_publish_reliability.py`, `tests/adapters/messaging/test_rabbitmq_transport.py` 실행 결과는 `22 passed, 5 skipped`다. Existing regression은 `strategy_id`가 있는 경로만 검증하고 `sell_strategy_id`만 있는 slot, owner 없음, restart, zero-fill cancel 후 재매도를 검증하지 않는다. +- `E-009` 현재 active ProofLine AC 중 `ac-0013`은 cancel/fill 경합의 slot·ledger exactly-once를, `ac-0015`는 미수락 SELL의 broker-authoritative 상태 수렴을, `ac-0020`은 non-recordable ledger fill 감사를 소유한다. Strategy event owner 확정, `_` 금지와 delivery observability를 소유하는 active AC는 없다. + +## Scope + +- 이 Line은 version-independent SELL event ownership 동작을 소유하는 신규 AC 두 건을 `create`한다. `ac-0013`·`ac-0015`의 slot/ledger lifecycle 의미는 변경하지 않고 회귀 대상으로 검증한다. 기존 active AC의 `update`, `retire`, `satisfy` 또는 release evidence 대상은 없다. + +### 신규 AC-A: SELL event owner 확정과 lifecycle 보존 + +- SELL admission은 non-empty `SellIntent.sell_strategy_id`를 우선하고, 없으면 slot에 이미 지정된 non-empty `sell_strategy_id`, 기존 호환을 위해 non-empty `strategy_id` 순서로 event owner를 확정한다. Whitespace-only 값은 owner가 아니다. +- 확정된 owner는 `PENDING_SELL` slot의 durable `sell_strategy_id`에 저장되어 dispatch, polling, full fill, partial fill/cancel, zero-fill cancel 후 `HOLDING`, 다음 SELL admission과 daemon 재시작에서도 유지된다. 별도 schema migration이나 새 identity registry를 만들지 않고 기존 column을 canonical owner field로 사용한다. +- Full fill과 partial-fill terminal event는 transition 전 `sell_strategy_id`를 캡처해 정확히 그 owner의 routing key를 사용한다. `PENDING_SELL → INACTIVE`가 slot identity field를 지운 뒤에도 routing hint는 유지된다. +- Event owner는 routing destination을 선택하는 opaque ID이며 core가 전략 종류, registration 또는 authorization mapping으로 해석하지 않는다. 다른 instance/strategy queue로 broadcast하거나 fallback하지 않는다. +- Zero-fill external cancel/reject는 기존처럼 event를 새로 만들지 않고 `HOLDING`으로 수렴하되 owner를 보존해야 한다. 같은 slot의 재매도는 명시적 새 owner가 없는 한 보존한 owner를 다시 사용한다. +- 동일 broker execution의 ledger exactly-once와 현재 best-effort event semantics를 유지한다. 재시작·반복 polling은 같은 terminal execution에 대한 event 재발행을 새로 보장하지 않으며, 기존 dispatch clear와 ledger execution identity 경계에서 중복 side effect가 없음을 검증한다. +- Verification은 explicit `sell_strategy_id`, pre-existing slot `sell_strategy_id`, legacy `strategy_id` fallback, full fill, partial-fill cancel, zero-fill cancel→재매도, repository round-trip/restart와 다른 strategy queue 미전달을 포함한다. + +### 신규 AC-B: owner 없는 strategy event의 fail-closed routing과 분리 관측 + +- Strategy event owner를 확정할 수 없으면 `trex.events`에 `_` 또는 다른 synthetic owner를 넣어 publish하지 않는다. 정상 strategy queue와 DLQ에 owner 없는 message를 가장해 보내지 않는다. +- Execution 결과는 stable reason `sell_event_owner_missing`에 대응하는 tick-level counter를 증가시키고 structured ERROR log를 남긴다. Alarm channel이 있으면 같은 reason과 instance/slot/symbol/event type만 포함한 alarm을 best-effort로 발행하며 credential·raw broker payload를 포함하지 않는다. +- Owner 없음 또는 event transport publish 실패는 이미 성공한 broker fill, slot transition, ledger outbox append를 rollback하지 않는다. Financial processing, ledger enqueue, strategy event publish/skip 결과를 서로 독립적으로 검증한다. +- Owner가 있어도 mandatory publish가 unroutable이면 기존 RabbitMQ DLQ와 ERROR 관측 계약을 유지한다. 이번 Line은 topology가 존재하는 정상 owner delivery와 owner 없음의 pre-publish skip을 구분한다. +- Verification은 owner 없음에서 strategy publish call 0건, `_` routing key 0건, stable counter/log/alarm, fill·transition·ledger 성공 유지와 alarm 실패 격리를 포함한다. Disposable RabbitMQ가 사용 가능할 때 올바른 owner는 정확한 queue에 1회 전달되고 다른 strategy queue에는 0회이며, subscriber가 없는 정상 owner는 mandatory publish에 의해 기존 DLQ 경로로 가는 회귀를 검증한다. + +### Source·test·문서 경계 + +- 예상 owner는 `trex_core/application/intents.py`, `trex_core/application/execution.py`, `trex_core/runtime/daemon.py`, 기존 slot persistence parity, messaging/execution tests와 `docs/interface-contract.md`다. +- Public wire field 이름 `sell_strategy_id`, event type, routing key 형식과 RabbitMQ topology는 유지한다. 신규 database column, migration, consumer registry와 cross-repository 변경은 없다. +- Issue #95와 legacy source comment의 `REQ-095`는 번호가 우연히 같을 뿐 별개다. 이 Line의 canonical identity는 `line-0011` 및 이후 allocator가 부여하는 신규 REQ/AC다. + +## Out of Scope + +- 기존 운영 DLQ 97건 또는 과거 SELL event의 자동 replay·삭제·재발행 +- 이미 누락된 전략 상태의 운영 보정, broker 주문 제출·취소 또는 production daemon 재시작 +- Strategy consumer의 idempotency 구현, 새 event ID schema 또는 exactly-once messaging 보장 +- RabbitMQ exchange/queue/binding 형식 변경, `_`를 위한 catch-all strategy queue 생성 +- Core-side strategy registry, producer authentication 또는 `strategy_id ↔ sell_strategy_id` 매핑 +- BUY event owner 의미 변경, ledger/outbox schema 변경 또는 slot lifecycle 재설계 +- Zero-fill cancel/reject에 신규 strategy event type 추가 +- 사용자 승인 없는 production credential 접근, live broker/RabbitMQ mutation 또는 DLQ 조작 + +## Risks and Unknowns + +### Risks + +- 기존 caller가 `sell_strategy_id` 없이 `strategy_id`에만 의존할 수 있다. 명시적 precedence와 legacy fallback을 함께 테스트해 호환성을 유지하되, 둘이 다르면 explicit/current `sell_strategy_id`가 routing authority가 된다. +- Event delivery는 금융 처리 뒤 best-effort이며 broker execution과 원자적이지 않다. 이번 변경은 owner 오류를 제거하고 실패를 분리 관측하지만 exactly-once strategy notification을 새로 보장하지 않는다. +- In-memory transport만으로는 RabbitMQ mandatory routing과 DLQ를 증명할 수 없다. Disposable RabbitMQ test는 실행 가능 환경에서 수행하고 사용할 수 없으면 skip을 성공 증거로 해석하지 않는다. +- Alarm 발행 자체가 실패할 수 있으므로 alarm은 금융 처리와 execution loop를 중단해서는 안 된다. Stable counter와 structured log가 독립 관측 경로로 남아야 한다. + +### Assumptions + +- Issue #95의 `sell_strategy_id` 보존 요구와 기존 `SellIntent`·slot column은 SELL strategy event owner의 canonical source를 정한다. 별도 owner field나 migration은 필요하지 않다. +- 현재 event consumer는 기존 `slot_filled_sell..` wire 형식을 소비하므로 key 형식이 아닌 strategy segment의 값과 missing-owner 차단만 바꾼다. +- Source·test·docs 구현과 PR delivery 요청은 production replay, restart 또는 DLQ mutation 승인을 포함하지 않는다. diff --git a/.proofline/lines/line-0011/evidence/activity-log.md b/.proofline/lines/line-0011/evidence/activity-log.md new file mode 100644 index 0000000..92ea222 --- /dev/null +++ b/.proofline/lines/line-0011/evidence/activity-log.md @@ -0,0 +1,21 @@ +# Activity Log — line-0011 + +## 2026-09-20 + +- 사용자 확인으로 `dcy-0011`을 confirmed 처리했다. +- digest-bound 독립 specification review의 blocker를 수정하고 재검토 PASS를 받았다. +- 사용자 승인으로 `req-0011`을 approved, `ac-0038`·`ac-0039`를 active 처리했다. +- 구현 계획: + 1. SELL admission owner 우선순위와 durable `sell_strategy_id` 보존 + 2. full/partial terminal event의 canonical owner routing + 3. owner 없음의 `sell_event_owner_missing` tick counter·ERROR·alarm과 publish skip + 4. 금융 state·ledger·dispatch 비rollback 회귀 + 5. focused/full/disposable RabbitMQ 및 독립 영향 검토 +- production broker 주문, production lifecycle, production daemon 재기동과 기존 DLQ 조작은 실행하지 않는다. +- SELL admission owner precedence, safety-sell fallback, terminal event pre-transition capture, ownerless fail-closed counter/log/alarm 및 runtime 방어를 구현했다. +- full fill, partial-fill cancel, zero-fill cancel 후 재매도, restart/reprocessing, alarm 실패 격리, 중복 tick, PostgreSQL round-trip, sanitized publish failure log 회귀 테스트를 추가했다. +- RabbitMQ retry의 동일 payload·mandatory flag 단위 검증과 disposable broker integration test를 추가했다. `TREX_RABBITMQ_TEST_URL`이 없어 integration은 명시적으로 SKIP했다. +- focused 검증: `29 passed, 15 skipped`; messaging blocker 검증: `21 passed, 6 skipped`. +- 전체 suite: `1261 passed, 98 skipped in 7.85s`. +- 변경 경로 Ruff, `proofline validate`, `git diff --check`를 통과했다. PostgreSQL과 RabbitMQ 실 backend 검증은 fixture 부재로 skipped 상태다. +- 변경 전 base `3f6a967`에 신규 핵심 regression 3개를 적용한 sabotage 검증은 `3 failed`로 원래 결함을 재현했고 수집 오류 없이 종료했다. diff --git a/.proofline/lines/line-0011/line-0011.md b/.proofline/lines/line-0011/line-0011.md new file mode 100644 index 0000000..f20716f --- /dev/null +++ b/.proofline/lines/line-0011/line-0011.md @@ -0,0 +1,4 @@ +--- +id: "line-0011" +status: discovery +--- diff --git a/.proofline/lines/line-0011/req-0011.md b/.proofline/lines/line-0011/req-0011.md new file mode 100644 index 0000000..593e14e --- /dev/null +++ b/.proofline/lines/line-0011/req-0011.md @@ -0,0 +1,43 @@ +--- +id: "req-0011" +status: approved +discovery: "dcy-0011" +criteria: + create: + - "ac-0038" + - "ac-0039" + update: [] + retire: [] + satisfy: [] +--- + +# SELL 체결 이벤트 owner 보존과 unroutable 차단 + +## Objective + +SELL 주문 admission에서 확정된 전략 owner를 체결 event routing까지 보존하여 해당 전략 queue로만 전달한다. owner를 결정할 수 없는 SELL 체결은 정상 전략 event로 위장해 `_` routing key로 발행하지 않고 명시적 전달 실패로 관측하되, 이미 성공한 broker 체결, slot 전이, ledger outbox 처리는 되돌리지 않는다. + +## Scope + +- `HOLDING → PENDING_SELL` admission에서 non-empty `SellIntent.sell_strategy_id`, slot에 보존된 non-empty `sell_strategy_id`, non-empty `strategy_id` 순서로 SELL event owner를 확정하고 기존 durable `sell_strategy_id`에 저장한다. Whitespace-only 값은 owner로 인정하지 않는다. +- 확정된 owner는 기존 durable slot field를 통해 dispatch 전후와 daemon 재시작 후에도 보존하고, full fill 및 partial fill 후 external cancel의 `slot_filled_sell` routing hint로 사용한다. +- zero-fill cancel 또는 reject는 fill event를 만들지 않고 기존 rollback 계약을 유지하며 확정 owner를 `HOLDING`에 보존한다. 이후 명시적 새 owner가 없는 SELL admission은 이 owner를 다시 사용하고, 명시적 새 owner가 있으면 그것으로 교체한다. +- owner 미확정 SELL 체결은 `trex.events` 정상 전략 topic 또는 DLQ에 synthetic owner message를 publish하지 않는다. stable reason code `sell_event_owner_missing`, tick-level counter, ERROR log, best-effort alarm으로 구분해 관측한다. +- 정상 owner가 있는 event의 publish 실패도 금융 처리를 rollback하지 않는다. 기존 payload의 `slot_id`, `event_type`, `sequence_at`을 변경하지 않으며, 재시작·반복 polling에서 같은 terminal execution의 event 재발행을 새로 보장하지 않는다. +- `docs/interface-contract.md`와 `docs/runtime-flows.md`의 RabbitMQ event routing 및 slot event 발행 구간에 owner 선택, pre-publish skip, 전달 실패 의미를 구현과 함께 동기화한다. + +## Constraints + +- `sell_strategy_id`와 `strategy_id`의 의미를 합치거나 legacy mapping을 추가하지 않는다. +- 새 persistence column이나 strategy-event outbox를 추가하지 않는다. restart 보장은 terminal 처리 전에 durable `PENDING_SELL` row를 다시 읽는 범위이며, terminal commit 뒤 실패한 event의 재생 저장소를 이번 Line에서 도입하지 않는다. +- publisher confirm과 1회 reconnect/retry, RabbitMQ mandatory publish 및 기존 transport DLQ 계약은 유지한다. +- alarm payload와 log는 secret, credential, connection string 또는 raw exception text를 포함하지 않는다. +- production broker 주문, production lifecycle, production daemon 재기동은 검증에서 실행하지 않는다. + +## Non-Goals + +- 이미 발생한 과거 DLQ message의 복구 또는 재발행 +- 전략 consumer 구현 변경 및 전역 exactly-once 보장 +- RabbitMQ topology, queue TTL/max-length, exchange 이름 변경 +- BUY event owner 계약 변경 +- durable strategy-event outbox 또는 운영 replay command 추가 diff --git a/docs/interface-contract.md b/docs/interface-contract.md index efefbb5..1749117 100644 --- a/docs/interface-contract.md +++ b/docs/interface-contract.md @@ -68,6 +68,11 @@ | update_thresholds | `trex.intent.req` | `action="update_thresholds"`, `reply_to` | reply-to 큐로 `{status:updated|ignored, slot_id, reason?}` (REQ-040) | | 이벤트 pub/sub | `trex.events` (topic) | routing key `{event_type}.{strategy_id}.{symbol}` | 전략 큐가 자기 `strategy_id` 패턴으로 바인딩 | +- `slot_filled_sell`의 `strategy_id`는 SELL admission에서 확정해 `sell_strategy_id`에 보존한다. 우선순위는 요청의 non-empty `sell_strategy_id` → 기존 보존 `sell_strategy_id` → legacy/safety 경로의 non-empty `strategy_id`다. +- SELL terminal transition이 owner 필드를 clear하기 전에 보존 owner를 routing hint로 캡처한다. full fill, partial-fill cancel, 외부 cancel 후 재매도, 재기동 후 reprocessing에도 같은 owner를 사용한다. +- 위 source가 모두 없으면 `slot_filled_sell._.{symbol}`을 발행하지 않는다. 금융 상태 전이와 ledger/outbox는 유지하고, 해당 tick의 `sell_event_owner_missing` counter·ERROR log·best-effort alarm을 남긴다. ownerless event를 일반 unroutable DLQ 메시지로 합성하지 않는다. +- event payload의 기존 `slot_id`, `event_type`, `sequence_at` identity는 producer가 보존한다. delivery는 at-least-once이며 consumer의 멱등 처리나 exactly-once는 이 인터페이스의 보장이 아니다. + ### Message contracts #### `trex.slots.req` 요청 payload (REQ-073 AC-1, BAT-035; REQ-116 BAT-055) diff --git a/docs/runtime-flows.md b/docs/runtime-flows.md index 9d61d4c..7e7032b 100644 --- a/docs/runtime-flows.md +++ b/docs/runtime-flows.md @@ -65,9 +65,11 @@ ## Flow: `슬롯 이벤트 발행/구독` (RabbitMQ `trex.events`, BAT-007, BAT-047) -1. 체결: execution 루프가 `pending_buy→holding`(또는 `pending_sell→closed`)을 commit + ledger outbox enqueue (REQ-018/019). +1. 체결: execution 루프가 `pending_buy→holding`(또는 `pending_sell→inactive`)을 commit + ledger outbox enqueue (REQ-018/019). 2. 같은 스레드에서 `event_publisher.publish_event(rk, payload)` 호출 (REQ-091 재시도 정책 포함, BAT-047). - routing key: `{event_type}.{strategy_id}.{symbol}` (예: `slot_filled_buy.A.005930`). + - SELL owner는 admission 시 요청의 non-empty `sell_strategy_id` → 기존 보존 `sell_strategy_id` → legacy/safety `strategy_id` 순으로 확정해 pending slot에 저장한다. terminal transition 전에 이를 캡처하므로 full fill, partial-fill cancel, zero-fill cancel 후 ownerless 재매도, 외부 cancel, 재기동/reprocessing에서 mutable post-transition field를 추측하지 않는다. + - SELL owner가 끝내 없으면 정상 `trex.events` publish와 `_` fallback, unroutable DLQ 합성을 모두 skip한다. 대신 `ExecutionResult.sell_event_owner_missing` tick counter + `reason=sell_event_owner_missing` ERROR log + best-effort critical alarm을 남긴다. 이미 commit한 broker fill, slot transition, ledger/outbox는 rollback하지 않는다. - payload: 현재 상태 + `sequence_at`(슬롯 `updated_at` iso) 운반 → 수신자는 last-wins 적용 가능 (순서 보증 불요). - publisher confirms 활성. `is_open=True`를 맹신하지 않음 — 발행 실패 시(half-open 포함) 강제 재연결(`close`+`_connect(timeout=2s)`) 후 1회 재시도. 재시도도 실패하면 `TransportUnavailableError` 전파. - unroutable(구독자 없음): `UnroutableError` → `_publish_dlq`로 `trex.events.dlq`(durable)에 보존. DLQ 유입 시 ERROR 로그 발행(REQ-093). @@ -258,7 +260,7 @@ - **슬리피지 보호 지정가 + KRX 호가단위 스냅(dcy-022/REQ-070, dcy-030/REQ-083)**: 슬롯에 `slippage_pct` 가 있고 `limit_price` 도 있으면, raw 실효가 = `limit_price×(1±pct/100)` 를 KRX 호가단위(REQ-083 `domain/pricing.py`)로 **스냅**해 제출한다. 방향 = **상한 준수**: 매수 = `floor_to_tick(raw)`(slippage 상한 미초과), 매도 = `ceil_to_tick(raw)`(매도 하한 미초과). band<1tick 엣지에서 미체결을 막기 위해 매수 = `max(floor_to_tick(raw), ceil_to_tick(limit_price))`, 매도 = `min(ceil_to_tick(raw), floor_to_tick(limit_price))` 로 entry 유효 호가를 보장한다. `limit_price` 없으면(시장가) 스냅 무시. 미지정 슬롯은 기존 동작(회귀 없음). 주식 전용(ETF/ETN·2,000원 미만 비주식 미지원, dcy-030 OQ-1). 이전 구현은 `int(ceil/floor)` 무스냅이라 호가단위 위반으로 KIS 거부(issue #28) — CHG-20260616-01 에서 스냅·방향(체결 보장→상한 준수) 정정. 2. poll: `list_dispatched_pending(instance_key)`(broker_order_id NOT NULL) → `BrokerPort.poll_fill` → 체결 시 `pending_buy→holding`(avg_price/qty) / `pending_sell→inactive`(점유 초기화·자산 정체성 보존), 거부 시 롤백. 전이 후 `clear_dispatch`. - **체결가 백필 가드(dcy-032/REQ-095/096, BAT-048)**: `_resolve_fill_price(fs, slot)` 헬퍼가 체결가를 결정. (1) `fs.avg_price`가 양수이면 그대로 사용(happy path — WARN 없음). (2) `fs.avg_price=None` 또는 `≤0`이고 `slot.limit_price>0`이면 지정가로 백필 + `reason=avg_price_backfill` **WARN 1건** 발행(슬롯 평단·원장 price에 백필값 사용, 0 저장 방지). (3) 지정가마저 None/≤0이면 체결 처리를 **skip**(슬롯 전이·원장 기록 미수행) + `reason=fill_price_unresolvable` 강 WARN 발행 — 슬롯이 PENDING 상태로 남아 stale_pending 경보가 운영자 개입을 유도. 이 가드는 `_on_filled`·`_on_partial_cancel`·`_record_ledger` 세 경로에 공통 적용된다(`reconcile()` 경로도 동일 헬퍼를 탄다). - - 외부 취소 감지(dcy-013/REQ-012/013, issue #9): 계좌에서 직접 취소된 주문을 KIS `poll_fill` 이 `cncl_yn="Y"` 로 파싱한다. **체결 0 취소** → `FillStatus(state="cancelled")` → 기존 롤백 경로(`pending_sell→holding`, `clear_dispatch`)로 자동 복구(영구 갇힘 제거). **부분 체결 후 취소**(`is_partial_cancel`) → 체결분 원장 1회 기록 + 슬롯 qty 차감(매도: 잔여 held, cost basis 유지 / 매수: 체결분만) + `holding` 복귀(OQ-1 A안, 과매도·이중기록 방지). + - 외부 취소 감지(dcy-013/REQ-012/013, issue #9): 계좌에서 직접 취소된 주문을 KIS `poll_fill` 이 `cncl_yn="Y"` 로 파싱한다. **체결 0 취소** → `FillStatus(state="cancelled")` → 기존 롤백 경로(`pending_sell→holding`, `clear_dispatch`)로 자동 복구하며 보존 `sell_strategy_id`는 유지된다. owner 없는 재매도는 이 owner를 재사용한다. **부분 체결 후 취소**(`is_partial_cancel`) → 체결분 원장 1회 기록 + 슬롯 qty 차감(매도: 잔여 held, cost basis 유지 / 매수: 체결분만) + `holding` 복귀(OQ-1 A안, 과매도·이중기록 방지); SELL event routing도 보존 owner를 사용한다. - pending stale 워치독(dcy-013/REQ-014 FR-3, Alert-only): `accepted`/`partially_filled`(취소 아님)로 다음 tick 대기 중인 슬롯이 `TREX_PENDING_STALE_THRESHOLD`(초, 0=비활성)를 초과해 머물면 WARN 로그 + `ExecutionResult.stale_pending` 카운터로 가시화한다. **상태는 변경하지 않는다**(auto-remediation 후속) — 취소 감지 누락(R-1) 등 미지 원인의 무한 갇힘이 silent 로 남지 않게 하는 안전망. - **timeout 자동 취소(dcy-022/REQ-069, dcy-041)**: `accepted`/`partially_filled` 상태 슬롯에 `timeout_sec`가 있고 경과 시간이 초과하면 외부 호출 전에 `cancel_state=unknown`을 예약 저장한 뒤 취소를 한 번 보낸다. 즉시 결과는 `requested|rejected|unknown`으로 교체하지만 **슬롯 상태는 변경하지 않는다**. 같은 주문은 자동 재취소하지 않으며 다음 poll의 `cancelled|filled`만 최종 상태를 확정한다. 3. reconcile(재기동): 서빙 전 `list_dispatched_pending` 재폴링으로 크래시 후 복구(중복 주문 0). 미디스패치(broker_order_id NULL) 슬롯은 다음 dispatch 가 1회만 송신. diff --git a/tests/adapters/messaging/test_publish_reliability.py b/tests/adapters/messaging/test_publish_reliability.py index 294198d..315a650 100644 --- a/tests/adapters/messaging/test_publish_reliability.py +++ b/tests/adapters/messaging/test_publish_reliability.py @@ -101,6 +101,31 @@ class TestPublishRetryOnHalfOpen: assert reconnect_called["yes"], "_force_reconnect must be called on first failure" + def test_sell_retry_reuses_identical_payload_and_mandatory_flag(self): + t = _make_transport() + _inject_connected(t) + payload = { + "slot_id": "slot-95", + "event_type": "slot_filled_sell", + "sequence_at": 11, + } + routing_key = "slot_filled_sell.SELL-B.005930" + + with ( + patch.object( + t, + "_do_basic_publish", + side_effect=[OSError("first attempt"), None], + ) as publish, + patch.object(t, "_force_reconnect"), + ): + t.publish_event(routing_key, payload) + + assert publish.call_count == 2 + assert publish.call_args_list[0] == publish.call_args_list[1] + assert publish.call_args_list[0].args[0] == routing_key + assert publish.call_args_list[0].kwargs["mandatory"] is True + # --------------------------------------------------------------------------- # REQ-091 AC-2: retry also fails → TransportUnavailableError raised diff --git a/tests/adapters/messaging/test_rabbitmq_transport.py b/tests/adapters/messaging/test_rabbitmq_transport.py index 906a423..69fa693 100644 --- a/tests/adapters/messaging/test_rabbitmq_transport.py +++ b/tests/adapters/messaging/test_rabbitmq_transport.py @@ -5,6 +5,7 @@ queue/routing keys to avoid colliding with other tests or runs. """ from __future__ import annotations +import json import os import threading import time @@ -149,6 +150,63 @@ def test_event_publish_and_per_strategy_consume(mq_url): _drain_queue(mq_url, queue_b) +def test_sell_event_retry_preserves_payload_and_strategy_route(mq_url, monkeypatch): + """AC-0039: live retry keeps identity and uses mandatory confirmed publish.""" + suffix = uuid.uuid4().hex[:8] + owner = f"SELL-{suffix}" + other_owner = f"OTHER-{suffix}" + queue = topology.strategy_event_queue_name(owner) + other_queue = topology.strategy_event_queue_name(other_owner) + routing_key = topology.event_routing_key("slot_filled_sell", owner, "005930") + payload = { + "slot_id": f"slot-{suffix}", + "event_type": "slot_filled_sell", + "sequence_at": 7, + } + publisher = RabbitMQTransport(mq_url, timeout=5.0) + attempts: list[tuple[str, bytes, bool]] = [] + try: + publisher.declare_topology() + publisher.declare_strategy_queue( + queue, topology.strategy_event_binding_pattern(owner), + ) + publisher.declare_strategy_queue( + other_queue, topology.strategy_event_binding_pattern(other_owner), + ) + original_publish = publisher._do_basic_publish + + def fail_first_then_publish(rk: str, body: bytes, *, mandatory: bool) -> None: + attempts.append((rk, body, mandatory)) + if len(attempts) == 1: + raise OSError("forced first-attempt failure") + original_publish(rk, body, mandatory=mandatory) + + monkeypatch.setattr(publisher, "_do_basic_publish", fail_first_then_publish) + publisher.publish_event(routing_key, payload) + + assert len(attempts) == 2 + assert attempts[0] == attempts[1] + assert attempts[0][0] == routing_key + assert attempts[0][2] is True + assert json.loads(attempts[0][1]) == payload + + publisher._ensure_connected() + channel = publisher._channel + assert channel is not None + method, _props, body = channel.basic_get(queue, auto_ack=True) + assert method is not None + assert body is not None + assert json.loads(body) == payload + other_method, _other_props, _other_body = channel.basic_get( + other_queue, auto_ack=True, + ) + assert other_method is None + finally: + publisher.close() + _drain_queue(mq_url, queue) + _drain_queue(mq_url, other_queue) + + def test_call_rpc_timeout(mq_url): queue = f"trex.test.rpc.timeout.{uuid.uuid4().hex[:8]}" client = RabbitMQTransport(mq_url, timeout=5.0) diff --git a/tests/adapters/persistence/test_postgresql_repository.py b/tests/adapters/persistence/test_postgresql_repository.py index 01d13d0..863b57d 100644 --- a/tests/adapters/persistence/test_postgresql_repository.py +++ b/tests/adapters/persistence/test_postgresql_repository.py @@ -63,6 +63,38 @@ def test_buy_intent_persists_transition(pg_clean): assert slot.status is SlotState.PENDING_BUY and slot.symbol == "005930" +def test_sell_admission_owner_survives_postgresql_repository_restart(pg_clean): + repo, svc, intents = _wire(pg_clean) + repo.reconcile_universe(INST, _SYMS) + decision = intents.submit_buy(BuyIntent( + intent_id="pg-owner-buy", + qty=10, + strategy_id="BUY-A", + portfolio_id="PF-1", + )) + assert decision.slot_id is not None + svc.transit_slot( + decision.slot_id, + SlotState.PENDING_BUY, + SlotState.HOLDING, + reason="filled", + actor="test", + qty=10.0, + avg_price=1000.0, + ) + sell = intents.submit_sell(SellIntent( + intent_id="pg-owner-sell", + slot_id=decision.slot_id, + sell_strategy_id="SELL-B", + )) + assert sell.admitted + + reloaded = PostgreSQLSlotRepository(pg_clean).fetch_slot(decision.slot_id) + assert reloaded is not None + assert reloaded.status is SlotState.PENDING_SELL + assert reloaded.sell_strategy_id == "SELL-B" + + def test_buy_intent_idempotent_on_pg(pg_clean): repo, svc, intents = _wire(pg_clean) repo.reconcile_universe(INST, _SYMS) diff --git a/tests/application/test_execution_cancel.py b/tests/application/test_execution_cancel.py index 3fa65f9..508f8b9 100644 --- a/tests/application/test_execution_cancel.py +++ b/tests/application/test_execution_cancel.py @@ -65,7 +65,7 @@ class ProgrammableBroker: return None -def _wire(broker, *, stale_sec=None): +def _wire(broker, *, stale_sec=None, event_emitter=None): repo = InMemorySlotRepository() svc = SlotService(repo) intents = IntentService(svc, repo, InMemoryIntentStore(), @@ -74,7 +74,7 @@ def _wire(broker, *, stale_sec=None): ledger = InMemoryLedger() execution = ExecutionService( svc, repo, broker, ledger_outbox=outbox, - stale_pending_threshold_sec=stale_sec, + stale_pending_threshold_sec=stale_sec, event_emitter=event_emitter, ) writer = LedgerWriter(outbox, ledger) repo.reconcile_universe("default", {"PF-1": "SYM"}) @@ -104,6 +104,64 @@ def test_full_external_cancel_returns_slot_to_holding(): assert r.rolled_back == 1 +def test_zero_fill_cancel_preserves_owner_for_ownerless_resell(): + broker = ProgrammableBroker() + repo, _svc, intents, execution, _ledger, _writer = _wire(broker) + sid = _buy_to_holding(intents, execution) + assert sid is not None + + intents.submit_sell(SellIntent( + intent_id="s-owner-1", slot_id=sid, sell_strategy_id="SELL-B", + )) + broker.responder = lambda side: FillStatus(state="cancelled") + execution.run_once() + + holding = repo.fetch_slot(sid) + assert holding is not None + assert holding.status is SlotState.HOLDING + assert holding.sell_strategy_id == "SELL-B" + + intents.submit_sell(SellIntent(intent_id="s-owner-2", slot_id=sid)) + pending = repo.fetch_slot(sid) + assert pending is not None + assert pending.status is SlotState.PENDING_SELL + assert pending.sell_strategy_id == "SELL-B" + + + +def test_partial_sell_cancel_routes_with_sell_owner_and_preserves_it(): + emitted: list[tuple[str, str | None]] = [] + + def emitter(event_type, _slot, owner=None): + emitted.append((event_type, owner)) + + broker = ProgrammableBroker() + repo, _svc, intents, execution, _ledger, _writer = _wire( + broker, event_emitter=emitter, + ) + sid = _buy_to_holding(intents, execution) + assert sid is not None + intents.submit_sell(SellIntent( + intent_id="s-partial-owner", slot_id=sid, sell_strategy_id="SELL-B", + )) + broker.balance["005930"] = 6 + broker.responder = lambda side: FillStatus( + state="partially_filled", + filled_qty=4, + avg_price=1100.0, + execution_id="exec-partial-owner", + cancelled=True, + ) + + execution.run_once() + + holding = repo.fetch_slot(sid) + assert holding is not None + assert holding.status is SlotState.HOLDING + assert holding.sell_strategy_id == "SELL-B" + assert ("slot_filled_sell", "SELL-B") in emitted + + def test_unresolved_poll_observability_is_sanitized_and_deduplicated(caplog): broker = ProgrammableBroker() repo, _svc, intents, execution, _ledger, _writer = _wire(broker) diff --git a/tests/application/test_messaging_event_publisher.py b/tests/application/test_messaging_event_publisher.py index e612a1c..7254cf9 100644 --- a/tests/application/test_messaging_event_publisher.py +++ b/tests/application/test_messaging_event_publisher.py @@ -1,6 +1,7 @@ """Slot transition event emission tests (REQ-021 + REQ-023 + REQ-018 hook).""" from __future__ import annotations +import logging import threading import time from trex_core.adapters.broker.paper import PaperBroker @@ -218,3 +219,213 @@ def test_sell_fill_dlq_sentinel_absent_when_strategy_id_set(): assert bad_keys == [], ( f"DLQ-sentinel routing keys found (strategy_id=None leak): {bad_keys}" ) + + +def test_sell_fill_prefers_admission_sell_strategy_owner(): + emitted: list[tuple[str, str | None]] = [] + + def emitter(event_type: str, _slot, owner: str | None = None) -> None: + emitted.append((event_type, owner)) + + repo = InMemorySlotRepository() + svc = SlotService(repo) + intents = IntentService( + svc, + repo, + InMemoryIntentStore(), + resolve_symbol=lambda _pid: "005930", + ) + outbox = InMemoryLedgerOutbox(instance_key="test-instance") + execution = ExecutionService( + svc, + repo, + PaperBroker(fill_after_polls=0), + ledger_outbox=outbox, + event_emitter=emitter, + ) + repo.reconcile_universe("default", {"PF-1": "SYM"}) + + decision = intents.submit_buy(BuyIntent( + intent_id="b-owner", qty=10, strategy_id="BUY-A", portfolio_id="PF-1", + )) + assert decision.slot_id is not None + execution.run_once() + intents.submit_sell(SellIntent( + intent_id="s-owner", slot_id=decision.slot_id, sell_strategy_id="SELL-B", + )) + pending = repo.fetch_slot(decision.slot_id) + assert pending is not None + assert pending.sell_strategy_id == "SELL-B" + + execution.run_once() + + assert ("slot_filled_sell", "SELL-B") in emitted + assert ("slot_filled_sell", "BUY-A") not in emitted + execution.run_once() + assert emitted.count(("slot_filled_sell", "SELL-B")) == 1 + + +def test_ownerless_sell_admission_persists_legacy_buy_owner(): + repo, _svc, intents, execution, _writer = _wire(InMemoryTransport()) + decision = intents.submit_buy(BuyIntent( + intent_id="b-legacy-owner", qty=10, strategy_id="BUY-A", portfolio_id="PF-1", + )) + assert decision.slot_id is not None + execution.run_once() + + intents.submit_sell(SellIntent( + intent_id="s-legacy-owner", + slot_id=decision.slot_id, + sell_strategy_id=" ", + )) + + pending = repo.fetch_slot(decision.slot_id) + assert pending is not None + assert pending.sell_strategy_id == "BUY-A" + + +def test_sell_owner_survives_execution_service_restart(): + emitted: list[tuple[str, str | None]] = [] + + def emitter(event_type: str, _slot, owner: str | None = None) -> None: + emitted.append((event_type, owner)) + + repo = InMemorySlotRepository() + svc = SlotService(repo) + intents = IntentService( + svc, repo, InMemoryIntentStore(), resolve_symbol=lambda _pid: "005930", + ) + outbox = InMemoryLedgerOutbox(instance_key="default") + repo.reconcile_universe("default", {"PF-1": "SYM"}) + broker = PaperBroker(fill_after_polls=0) + first_execution = ExecutionService( + svc, repo, broker, ledger_outbox=outbox, + ) + decision = intents.submit_buy(BuyIntent( + intent_id="b-restart-owner", qty=10, strategy_id="BUY-A", portfolio_id="PF-1", + )) + assert decision.slot_id is not None + first_execution.run_once() + intents.submit_sell(SellIntent( + intent_id="s-restart-owner", + slot_id=decision.slot_id, + sell_strategy_id="SELL-B", + )) + + restarted_execution = ExecutionService( + svc, + repo, + broker, + ledger_outbox=outbox, + event_emitter=emitter, + ) + restarted_execution.run_once() + + assert ("slot_filled_sell", "SELL-B") in emitted + + +class _AlarmRecorder: + def __init__(self) -> None: + self.calls: list[tuple[str, dict]] = [] + + def publish_alarm(self, code, _title, *, fields=None, severity="critical"): + self.calls.append((code, {**(fields or {}), "severity": severity})) + return True + + +def test_ownerless_sell_fill_is_skipped_without_financial_rollback(caplog): + emitted: list[tuple[str, str | None]] = [] + alarm = _AlarmRecorder() + + def emitter(event_type: str, _slot, owner: str | None = None) -> None: + emitted.append((event_type, owner)) + + repo = InMemorySlotRepository() + svc = SlotService(repo) + intents = IntentService( + svc, + repo, + InMemoryIntentStore(), + resolve_symbol=lambda _pid: "005930", + ) + outbox = InMemoryLedgerOutbox(instance_key="test-instance") + execution = ExecutionService( + svc, + repo, + PaperBroker(fill_after_polls=0), + ledger_outbox=outbox, + event_emitter=emitter, + alarm_emitter=alarm, # type: ignore[arg-type] + instance_key="default", + ) + repo.reconcile_universe("default", {"PF-1": "SYM"}) + + decision = intents.submit_buy(BuyIntent( + intent_id="b-ownerless", qty=10, portfolio_id="PF-1", + )) + assert decision.slot_id is not None + execution.run_once() + intents.submit_sell(SellIntent(intent_id="s-ownerless", slot_id=decision.slot_id)) + + with caplog.at_level(logging.ERROR, logger="trex_core.application.execution"): + result = execution.run_once() + + sell_calls = [call for call in emitted if call[0] == "slot_filled_sell"] + assert sell_calls == [] + assert result.sell_event_owner_missing == 1 + assert result.filled == 1 + assert result.ledger_enqueued == 1 + final_slot = repo.fetch_slot(decision.slot_id) + assert final_slot is not None + assert final_slot.status is SlotState.INACTIVE + assert any("reason=sell_event_owner_missing" in rec.message for rec in caplog.records) + assert alarm.calls == [( + "sell_event_owner_missing", + { + "instance_key": "default", + "slot_id": str(decision.slot_id), + "symbol": "005930", + "event_type": "slot_filled_sell", + "reason": "sell_event_owner_missing", + "severity": "critical", + }, + )] + + +def test_ownerless_sell_alarm_failure_does_not_rollback_fill(): + class RaisingAlarm: + def publish_alarm(self, *_args, **_kwargs): + raise RuntimeError("alarm unavailable") + + repo = InMemorySlotRepository() + svc = SlotService(repo) + intents = IntentService( + svc, repo, InMemoryIntentStore(), resolve_symbol=lambda _pid: "005930", + ) + outbox = InMemoryLedgerOutbox(instance_key="default") + execution = ExecutionService( + svc, + repo, + PaperBroker(fill_after_polls=0), + ledger_outbox=outbox, + alarm_emitter=RaisingAlarm(), # type: ignore[arg-type] + instance_key="default", + ) + repo.reconcile_universe("default", {"PF-1": "SYM"}) + decision = intents.submit_buy(BuyIntent( + intent_id="b-ownerless-alarm", qty=10, portfolio_id="PF-1", + )) + assert decision.slot_id is not None + execution.run_once() + intents.submit_sell(SellIntent( + intent_id="s-ownerless-alarm", slot_id=decision.slot_id, + )) + + result = execution.run_once() + + final_slot = repo.fetch_slot(decision.slot_id) + assert final_slot is not None + assert final_slot.status is SlotState.INACTIVE + assert result.filled == 1 + assert result.ledger_enqueued == 1 + assert result.sell_event_owner_missing == 1 diff --git a/tests/domain/slot/test_service_sell.py b/tests/domain/slot/test_service_sell.py index 0490309..e8822a6 100644 --- a/tests/domain/slot/test_service_sell.py +++ b/tests/domain/slot/test_service_sell.py @@ -34,7 +34,7 @@ def _holding_slot(service, repo): slot = _inactive_slot(repo) pending = service.transit_slot( slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, - reason="buy", actor="test", symbol="005930", + reason="buy", actor="test", symbol="005930", strategy_id="BUY-A", ) return service.transit_slot( pending.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, @@ -49,6 +49,7 @@ class TestSell: service.sell(holding.slot_id, actor="safety-sell", reason="stop_loss") updated = repo.fetch_slot(holding.slot_id) assert updated.status is SlotState.PENDING_SELL + assert updated.sell_strategy_id == "BUY-A" def test_sell_records_transition_log(self, service, repo): """sell() appends a transition log entry.""" diff --git a/tests/runtime/test_sell_event_owner_routing.py b/tests/runtime/test_sell_event_owner_routing.py new file mode 100644 index 0000000..31948ba --- /dev/null +++ b/tests/runtime/test_sell_event_owner_routing.py @@ -0,0 +1,60 @@ +"""Runtime SELL event owner routing defenses (REQ-0011 / AC-0039).""" +from __future__ import annotations + +from typing import Any, cast +from uuid import uuid4 + +from trex_core.domain.slot import Slot, SlotState +from trex_core.runtime.daemon import Daemon + + +class _Publisher: + def __init__(self) -> None: + self.calls: list[tuple[str, dict]] = [] + self.failure: Exception | None = None + + def publish_event(self, routing_key: str, payload: dict) -> None: + if self.failure is not None: + raise self.failure + self.calls.append((routing_key, payload)) + + +def _daemon_with_publisher() -> tuple[Daemon, _Publisher]: + daemon = object.__new__(Daemon) + publisher = _Publisher() + daemon.event_publisher = cast(Any, publisher) + return daemon, publisher + + +def test_runtime_emitter_blocks_ownerless_sell_without_synthetic_owner(caplog): + daemon, publisher = _daemon_with_publisher() + slot = Slot(slot_id=uuid4(), status=SlotState.INACTIVE, symbol="005930") + + daemon._make_event_emitter()("slot_filled_sell", slot, None) + + assert publisher.calls == [] + assert "reason=sell_event_owner_missing" in caplog.text + + +def test_runtime_emitter_routes_sell_with_explicit_owner(): + daemon, publisher = _daemon_with_publisher() + slot = Slot(slot_id=uuid4(), status=SlotState.INACTIVE, symbol="005930") + + daemon._make_event_emitter()("slot_filled_sell", slot, "SELL-B") + + assert len(publisher.calls) == 1 + assert publisher.calls[0][0] == "slot_filled_sell.SELL-B.005930" + + +def test_runtime_publish_failure_log_excludes_raw_exception(caplog): + daemon, publisher = _daemon_with_publisher() + publisher.failure = RuntimeError("SENTINEL_SECRET_CONNECTION_TEXT") + slot = Slot(slot_id=uuid4(), status=SlotState.INACTIVE, symbol="005930") + + daemon._make_event_emitter()("slot_filled_sell", slot, "SELL-B") + + assert publisher.calls == [] + assert "reason=event_publish_failed" in caplog.text + assert "exc_type=RuntimeError" in caplog.text + assert "SENTINEL_SECRET_CONNECTION_TEXT" not in caplog.text + assert not any(record.exc_info for record in caplog.records) diff --git a/trex_core/application/execution.py b/trex_core/application/execution.py index 777037b..850af47 100644 --- a/trex_core/application/execution.py +++ b/trex_core/application/execution.py @@ -95,6 +95,7 @@ class ExecutionResult: stale_pending: int = 0 # dcy-013 / REQ-014 FR-3: pending orders over threshold sell_skipped_no_balance: int = 0 # dcy-020 / REQ-062: broker flat -> sell skipped timeout_cancelled: int = 0 # dcy-022 / REQ-069: cancel_order sent on timeout + sell_event_owner_missing: int = 0 # req-0011/ac-0039: pre-publish skip count class ExecutionService: @@ -555,13 +556,13 @@ class ExecutionService: def _on_filled(self, slot, fs, side: str, r: ExecutionResult) -> None: broker_order_id = slot.broker_order_id # captured BEFORE clear_dispatch - # Capture strategy_id BEFORE the transition clears it. - # PENDING_SELL -> INACTIVE clears strategy_id on the slot row - # (_apply_transition_fields REQ-057), so we must capture it here to - # supply as a routing hint for the slot_filled_sell event — otherwise the - # routing key uses "_" and the message is NO_ROUTEd to the DLQ instead - # of reaching trex.events.strategy.{strategy_id} (M2 REL-009 feedback). - routing_strategy_id = slot.strategy_id + # req-0011/ac-0038: SELL admission persists the canonical owner in + # sell_strategy_id. Capture before INACTIVE clears identity fields. + routing_strategy_id = ( + self._sell_event_owner(slot) + if slot.status is SlotState.PENDING_SELL + else self._non_empty_strategy_id(slot.strategy_id) + ) # REQ-095: resolve fill price with limit_price backfill guard. fill_price, _was_backfilled = self._resolve_fill_price(fs, slot) if fill_price is None: @@ -633,7 +634,7 @@ class ExecutionService: # record subject). Best-effort: a publish failure must not roll back. # Pass routing_strategy_id so sell fills route correctly even after # strategy_id is cleared from the slot row (slot_filled_sell NO_ROUTE fix). - self._emit_event(event_type, updated, routing_strategy_id) + self._emit_event(event_type, updated, routing_strategy_id, r) self._repo.clear_dispatch(slot.slot_id) # done with this order def _on_partial_cancel(self, slot, fs, side: str, r: ExecutionResult) -> None: @@ -646,9 +647,11 @@ class ExecutionService: portion that actually filled. """ broker_order_id = slot.broker_order_id # captured BEFORE clear_dispatch - # Capture strategy_id BEFORE the transition (same reason as _on_filled: - # PENDING_SELL -> INACTIVE clears strategy_id, causing NO_ROUTE). - routing_strategy_id = slot.strategy_id + routing_strategy_id = ( + self._sell_event_owner(slot) + if slot.status is SlotState.PENDING_SELL + else self._non_empty_strategy_id(slot.strategy_id) + ) # REQ-095: resolve fill price with limit_price backfill guard. fill_price, _was_backfilled = self._resolve_fill_price(fs, slot) if fill_price is None: @@ -700,7 +703,7 @@ class ExecutionService: except SlotStateMismatchError: return # slot moved concurrently; next tick reconciles self._record_ledger(slot, fs, side, broker_order_id, r, resolved_price=fill_price) - self._emit_event(event_type, updated, routing_strategy_id) + self._emit_event(event_type, updated, routing_strategy_id, r) self._repo.clear_dispatch(slot.slot_id) def _check_stale_pending(self, slot, now: datetime, r: ExecutionResult) -> None: @@ -860,7 +863,64 @@ class ExecutionService: }, ) - def _emit_event(self, event_type: str, slot, routing_strategy_id: str | None = None) -> None: + @staticmethod + def _non_empty_strategy_id(value: object) -> str | None: + return value if isinstance(value, str) and value.strip() else None + + @classmethod + def _sell_event_owner(cls, slot) -> str | None: + return ( + cls._non_empty_strategy_id(getattr(slot, "sell_strategy_id", None)) + or cls._non_empty_strategy_id(getattr(slot, "strategy_id", None)) + ) + + def _emit_event( + self, + event_type: str, + slot, + routing_strategy_id: str | None = None, + result: ExecutionResult | None = None, + ) -> None: + if ( + event_type == "slot_filled_sell" + and self._non_empty_strategy_id(routing_strategy_id) is None + ): + if result is not None: + result.sell_event_owner_missing += 1 + reason = "sell_event_owner_missing" + slot_id = getattr(slot, "slot_id", None) or getattr(slot, "id", None) + symbol = getattr(slot, "symbol", None) + _log.error( + "sell event skipped reason=%s instance_key=%s slot_id=%s " + "symbol=%s event_type=%s", + reason, + self._instance_key, + slot_id, + symbol, + event_type, + ) + if self._alarm_emitter is not None: + try: + self._alarm_emitter.publish_alarm( + reason, + "SELL event owner missing; strategy publish skipped", + fields={ + "instance_key": self._instance_key, + "slot_id": str(slot_id) if slot_id is not None else None, + "symbol": symbol, + "event_type": event_type, + "reason": reason, + }, + ) + except Exception: # noqa: BLE001 - alarm is best-effort + _log.error( + "sell owner alarm failed reason=%s instance_key=%s " + "slot_id=%s", + reason, + self._instance_key, + slot_id, + ) + return if self._event_emitter is None: return try: diff --git a/trex_core/application/intents.py b/trex_core/application/intents.py index 7297d06..b30407f 100644 --- a/trex_core/application/intents.py +++ b/trex_core/application/intents.py @@ -469,6 +469,18 @@ class IntentService: return IntentDecision( intent.intent_id, "rejected", None, f"slot_not_holding:{slot.status.value}" ) + # req-0011 / ac-0038: decide the SELL event owner at admission and + # persist it in the existing canonical slot field. An explicit SELL + # owner wins, then a preserved owner from a prior zero-fill cancel, + # then the legacy holding owner used by autonomous safety sells. + def non_empty(value: Optional[str]) -> Optional[str]: + return value if isinstance(value, str) and value.strip() else None + + sell_event_owner = ( + non_empty(intent.sell_strategy_id) + or non_empty(slot.sell_strategy_id) + or non_empty(slot.strategy_id) + ) # dcy-022 / REQ-068: resolve effective timeout. effective_timeout = ( intent.timeout_sec @@ -482,7 +494,7 @@ class IntentService: SlotState.PENDING_SELL, reason=intent.reason, actor="intent_api", - sell_strategy_id=intent.sell_strategy_id, + sell_strategy_id=sell_event_owner, # dcy-012 / REQ-007: optional sell limit (None => market sell). limit_price=intent.limit_price, # dcy-021 / REQ-062: partial-sell order qty (None => sell all); diff --git a/trex_core/domain/slot/service.py b/trex_core/domain/slot/service.py index 29c6789..43b7304 100644 --- a/trex_core/domain/slot/service.py +++ b/trex_core/domain/slot/service.py @@ -327,6 +327,16 @@ class SlotService: raise SlotTransitionError( f"sell() requires holding state, got: {slot.status.value}" ) + # req-0011 / ac-0038: autonomous sells have no SellIntent, so persist + # the already-owned SELL identity or inherit the holding owner. + sell_event_owner = next( + ( + value + for value in (slot.sell_strategy_id, slot.strategy_id) + if isinstance(value, str) and value.strip() + ), + None, + ) self.transit_slot( slot_id, SlotState.HOLDING, @@ -334,6 +344,7 @@ class SlotService: reason=reason, actor=actor, limit_price=limit_price, + sell_strategy_id=sell_event_owner, ) # ------------------------------------------------------------ diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index a6f01fe..46605db 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -386,28 +386,47 @@ class Daemon: def _make_event_emitter(self): def emit(event_type: str, slot, routing_strategy_id: str | None = None) -> None: payload = slot_transition_event_payload(slot, event_type=event_type) - # routing_strategy_id is supplied when the transition clears - # slot.strategy_id (e.g. PENDING_SELL -> INACTIVE for slot_filled_sell). - # Without this hint the routing key would use "_" and the message would - # be NO_ROUTEd to the DLQ instead of reaching the strategy queue - # (M2 REL-009 feedback / slot_filled NO_ROUTE fix). + # routing_strategy_id is captured before transitions that clear owner + # fields (for SELL, the admission-owned sell_strategy_id). Missing + # SELL ownership is blocked below instead of synthesizing "_". effective_strategy_id = ( - routing_strategy_id or getattr(slot, "strategy_id", None) or "_" + routing_strategy_id or getattr(slot, "strategy_id", None) ) + # req-0011/ac-0039 defense in depth: application execution normally + # skips ownerless SELL events before this hook. Never synthesize the + # legacy "_" owner if a direct caller violates that boundary. + if ( + event_type == "slot_filled_sell" + and not ( + isinstance(effective_strategy_id, str) + and effective_strategy_id.strip() + ) + ): + _log.error( + "sell event publish blocked reason=sell_event_owner_missing " + "slot_id=%s symbol=%s event_type=%s", + getattr(slot, "slot_id", None) or getattr(slot, "id", None), + getattr(slot, "symbol", None), + event_type, + ) + return + effective_strategy_id = effective_strategy_id or "_" rk = topology.event_routing_key( event_type, effective_strategy_id, slot.symbol or "_", ) try: self.event_publisher.publish_event(rk, payload) - except Exception: # noqa: BLE001 - publish failure must not roll back - # REQ-093: escalate to ERROR so operators can observe via log - # alert. Include routing key + slot identity for correlation. + except Exception as exc: # noqa: BLE001 - publish failure must not roll back + # req-0011/ac-0039: transport failures are observable without + # rendering raw exception text, traceback, broker payload, or + # connection details into logs. _log.error( - "event publish failed (rk=%s slot_id=%s strategy=%s)", + "event publish failed reason=event_publish_failed " + "rk=%s slot_id=%s strategy=%s exc_type=%s", rk, getattr(slot, "slot_id", None) or getattr(slot, "id", None), getattr(slot, "strategy_id", None), - exc_info=True, + type(exc).__name__, ) return emit