feat(safety-sell): dcy-017 워치독 소유권 가시화 — snapshot 상태 + 발동 로깅 (BAT-021)

전략 이탈로 슬롯이 내부 청산 루틴(워치독) 소유로 넘어간 상태가 무성이라
운영자가 인지할 수 없던 문제(issue #12)를 가시화한다.

거버넌스: 전부 change-req (신규 REQ 없음, dcy-017 OQ-3; Redis 미채택).
- REQ-034/021: WatchdogStore.last_seen + snapshot RPC 응답에 holding 슬롯
  워치독 상태(expired/last_seen) read-only 동봉(best-effort)
- REQ-032: trex-core slots 가 expired 슬롯을 WATCHDOG-OWNED(전략 이탈) 표시
- REQ-037/038/039: safety-sell 워치독 소유 진입(WARN)/이탈(INFO) + 발동
  (forced_close·stop_loss=WARN, take_profit·market_close=INFO) 로그 +
  in-process 카운터(SafetySellService.counters), 구조화 필드(slot_id/
  portfolio_id/reason/phase/instance_key). per-tick 스킵은 카운터만
- REQ-042: runtime-flows/interface-contract 정합

best-effort 격리(NFR-1): 모든 가시화는 예외 격리 — 워치독 판정·청산 불변.
dcy-016 억제 읽기로 관측이 워치독을 리셋하지 않음(비침습).

검증: 단위 335 passed/34 skipped, 인프라 게이트(PG+RabbitMQ) 369 passed, ruff clean.
신규 test_safety_sell_visibility(6) + snapshot watchdog(1) + CLI(1). BAT-021 rc.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

tests/application/test_messaging_snapshot_consumer.py

@@ -117,3 +117,21 @@ def test_operator_query_suppresses_watchdog_touch():
     )
     handler(req)
     assert wd.updated == []  # dcy-016 R-4: operator query is not a liveness signal
+
+
+def test_snapshot_exposes_watchdog_state_for_holding_slots():
+    """dcy-017 / REQ-034: snapshot response carries holding-slot watchdog state."""
+    from trex_core.application.watchdog import WatchdogStore
+    repo, svc, intents = _wire()
+    sid = _holding_slot(repo, svc)
+    wd = WatchdogStore()  # never updated => expired (strategy absent)
+    handler = make_snapshot_rpc_handler(repo, svc, watchdog_store=wd, watchdog_timeout_sec=120)
+    # operator query so the response does not touch the watchdog before we read it
+    req = RpcRequest(
+        payload={"view": "all", "trading_date": DAY.isoformat(), "actor": "operator"},
+        correlation_id="op",
+    )
+    states = handler(req).payload["watchdog"]
+    assert len(states) == 1
+    assert states[0]["slot_id"] == sid and states[0]["expired"] is True
+    assert states[0]["last_seen"] is None

tests/application/test_safety_sell_visibility.py

@@ -0,0 +1,116 @@
+"""Safety-sell watchdog-ownership + fire visibility tests (dcy-017 / REQ-037, issue #12)."""
+from __future__ import annotations
+
+import logging
+from datetime import date, datetime, timedelta, timezone
+from decimal import Decimal
+
+from trex_core.application.safety_sell import SafetySellService
+from trex_core.application.watchdog import WatchdogStore
+from trex_core.domain.slot import SlotService, SlotState
+from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
+from trex_core.runtime.config import SafetySellConfig
+
+_DAY = date(2026, 6, 7)
+_NOW = datetime(2026, 6, 7, 6, 0, 0, tzinfo=timezone.utc)
+
+
+class _FakeBroker:
+    def __init__(self):
+        self.prices = {}
+
+    def name(self): return "fake"
+    def submit_order(self, req): ...
+    def poll_fill(self, *a, **k): ...
+    def cancel_order(self, *a, **k): ...
+    def get_balance(self): return []
+    def get_current_price(self, symbol): return self.prices.get(symbol)
+
+
+def _cfg(**over):
+    base = dict(watchdog_timeout_sec=120, watchdog_grace_period_sec=0,
+                forced_close_check_interval_sec=60, stop_loss_poll_interval_sec=30,
+                market_close_liquidation_time=None)
+    base.update(over)
+    return SafetySellConfig(**base)
+
+
+def _wire(**cfg_over):
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    svc.preallocate_daily_slots(_DAY, 5)
+    broker = _FakeBroker()
+    watchdog = WatchdogStore()
+    service = SafetySellService(
+        slot_repo=repo, slot_service=svc, broker=broker, watchdog=watchdog,
+        config=_cfg(**cfg_over), instance_key="inst-1",
+    )
+    return repo, svc, broker, watchdog, service
+
+
+def _holding(svc, repo, *, symbol="005930", **fields):
+    idle = repo.find_idle_slot(_DAY)
+    p = svc.transit_slot(idle.slot_id, SlotState.IDLE, SlotState.PENDING_BUY,
+                         reason="buy", actor="t", symbol=symbol, portfolio_id="PF-1", **fields)
+    return svc.transit_slot(p.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
+                            reason="filled", actor="t", avg_price=50000.0, qty=1.0)
+
+
+def test_watchdog_ownership_entry_warns(caplog):
+    repo, svc, broker, watchdog, service = _wire()
+    slot = _holding(svc, repo)  # never polled => watchdog-expired (strategy absent)
+    with caplog.at_level(logging.WARNING):
+        service._update_ownership(_NOW)
+    assert service.counters.get("owned:enter") == 1
+    rec = [r for r in caplog.records if "watchdog ownership entered" in r.message]
+    assert rec and str(slot.slot_id) in rec[0].message and "inst-1" in rec[0].message
+
+
+def test_watchdog_ownership_exit_on_return(caplog):
+    repo, svc, broker, watchdog, service = _wire()
+    slot = _holding(svc, repo)
+    service._update_ownership(_NOW)            # enter (expired)
+    watchdog.update(str(slot.slot_id))          # strategy returns (polls)
+    with caplog.at_level(logging.INFO):
+        service._update_ownership(_NOW)        # exit
+    assert service.counters.get("owned:exit") == 1
+    assert any("watchdog ownership released" in r.message for r in caplog.records)
+
+
+def test_no_ownership_churn_when_stable():
+    repo, svc, broker, watchdog, service = _wire()
+    _holding(svc, repo)
+    service._update_ownership(_NOW)
+    service._update_ownership(_NOW)            # same set => no new enter
+    assert service.counters.get("owned:enter") == 1
+
+
+def test_forced_close_fire_logs_warn(caplog):
+    repo, svc, broker, watchdog, service = _wire()
+    slot = _holding(svc, repo, forced_close_at=_NOW - timedelta(seconds=1))
+    with caplog.at_level(logging.WARNING):
+        fired = service.check_forced_close(_NOW)
+    assert slot.slot_id in fired
+    assert service.counters.get("fire:forced_close_at") == 1
+    rec = [r for r in caplog.records if "safety-sell fired" in r.message
+           and r.levelno == logging.WARNING]
+    assert rec and "forced_close_at" in rec[0].message and "inst-1" in rec[0].message
+
+
+def test_take_profit_fire_logs_info(caplog):
+    repo, svc, broker, watchdog, service = _wire()
+    slot = _holding(svc, repo, take_profit_price=Decimal("60000"))
+    broker.prices["005930"] = Decimal("61000")
+    with caplog.at_level(logging.INFO):
+        fired = service.check_prices(_NOW)
+    assert slot.slot_id in fired
+    assert service.counters.get("fire:take_profit") == 1
+    rec = [r for r in caplog.records if "safety-sell fired" in r.message]
+    assert rec and rec[0].levelno == logging.INFO  # take_profit is INFO (OQ-2)
+
+
+def test_grace_period_suppresses_ownership():
+    repo, svc, broker, watchdog, service = _wire(watchdog_grace_period_sec=9999)
+    _holding(svc, repo)
+    service._update_ownership(_NOW)
+    assert service.counters.get("owned:enter") is None  # grace => no judgement

tests/runtime/test_cli_operator.py

@@ -53,6 +53,20 @@ def test_slots_status_holding_maps_to_held_view():
     assert rpc.calls[0]["payload"]["view"] == "held"
 
 
+def test_slots_marks_watchdog_owned(capsys):
+    """dcy-017: slots flags strategy-absent (watchdog-owned) positions."""
+    resp = {
+        "assets": [{"asset_id": "PF-1", "symbol": "005930", "asset_quantity": 10,
+                    "active_strategy_id": "orb"}],
+        "view": "held", "available_slots": 9, "degraded": False, "reason": None,
+        "watchdog": [{"slot_id": "S-1", "portfolio_id": "PF-1", "symbol": "005930",
+                      "expired": True, "last_seen": None}],
+    }
+    cli.main(["slots", "--status", "held"], rpc=_RpcRecorder(resp))
+    out = capsys.readouterr().out
+    assert "watchdog_owned=1" in out and "WATCHDOG-OWNED" in out
+
+
 def test_instance_shortcut_routes_to_instance_queue():
     rpc = _RpcRecorder(_SNAPSHOT)
     cli.main(["universe", "--instance", "inst-2"], rpc=rpc)