diff --git a/SESSION.md b/SESSION.md index 4b57d24..6bbd095 100644 --- a/SESSION.md +++ b/SESSION.md @@ -2,7 +2,7 @@ ## Current Goal -Automation execution-layer P19 is closed locally. P19 adds structured channel routing and channel health observability so alert dispatch can distinguish enabled workbench/webhook paths from disabled WeCom/Email adapters with credential-missing reasons. The default next action is P20 `Operator Digest And Review Queue`. +Automation execution-layer P20 is closed locally. P20 adds an operator digest/review queue that aggregates recent alerts, deliveries, visual reviews, suppression, retries, mismatches, and missing feedback into a compact operator-facing review surface. The default next action is P21 `Runtime Incident Journal`. ## Current Branch @@ -14,8 +14,8 @@ Automation execution-layer P19 is closed locally. P19 adds structured channel ro ## Automation Status -- state: automation_execution_layer_p19_multi_channel_alert_routing_closed -- current_phase: P19 Multi-Channel Alert Routing +- state: automation_execution_layer_p20_operator_digest_review_queue_closed +- current_phase: P20 Operator Digest And Review Queue - latest_p7_selector_choice: continue_periodic_analysis_decision_monitoring_with_maturity_rollup - latest_p14_slice: webhook_transport_retry - latest_p15_pxx_tasklist_status: ready @@ -42,7 +42,12 @@ Automation execution-layer P19 is closed locally. P19 adds structured channel ro - latest_channel_health_status: visible - latest_disabled_adapter_status: visible_credential_missing - latest_channel_routing_smoke_status: no_pending_alerts_safe_noop -- next_execution_layer_phase: P20 Operator Digest And Review Queue +- latest_p20_slice: operator_digest_review_queue +- latest_operator_digest_status: implemented +- latest_operator_digest_route_status: ok +- latest_operator_digest_workbench_status: visible +- latest_operator_digest_smoke_status: safe_noop_or_completed +- next_execution_layer_phase: P21 Runtime Incident Journal - latest_visual_review_script_status: idle - latest_runtime_visual_watch_task_status: ok - latest_visual_capture_cleanup_task_status: ok @@ -118,6 +123,9 @@ Automation execution-layer P19 is closed locally. P19 adds structured channel ro - Added structured channel route/capability helpers with workbench, webhook, WeCom, and email channel descriptions. - Added channel health summary and disabled adapter metadata for WeCom/Email credential-missing states. - Extended alert dispatcher payloads/results, delivery summary, runtime `alert_dispatch` details, dispatcher script output, and workbench delivery cards with per-channel observability. +- Added `OperatorDigestService` to aggregate recent alerts, deliveries, visual reviews, quality flags, and missing feedback into an operator review queue. +- Added `scripts/run_operator_digest.py` and `GET /trading/alerts/operator-digest`. +- Extended the workbench `Alert Quality` panel with operator digest status, review queue counts, and top review item types. ## Current Working State @@ -141,6 +149,8 @@ Automation execution-layer P19 is closed locally. P19 adds structured channel ro - P18 visibility is implemented: dispatcher payloads/results, delivery summary, runtime `alert_dispatch` details, script output, visual review API, and workbench panels expose media evidence status. - P19 multi-channel alert routing is implemented: routing now returns `channel_route`, `channel_health`, channel capability metadata, and explicit disabled-adapter states for WeCom/Email when credentials/adapters are unavailable. - P19 visibility is implemented: dispatcher result exposes `channel_summary` / `channel_health` / `channel_disabled`, delivery summary exposes `channel_status_counts`, runtime `alert_dispatch` details expose channel health, and the workbench `Alert Quality` panel shows channel counts. +- P20 operator digest/review queue is implemented: recent high-value alerts, suppressed deliveries, failed/retry deliveries, visual mismatches, and pending feedback items are aggregated into a bounded review queue. +- P20 visibility is implemented: `scripts/run_operator_digest.py`, `/trading/alerts/operator-digest`, focused tests, and the workbench `Alert Quality` panel expose digest status, counts, warnings, and top queue items. - Latest `scripts/run_visual_review.py` smoke found no pending reviews: attempted=0, reviewed=0, status=idle. - Latest capture smoke returned `status=ignored` when no watch run was supplied, which is the expected safe no-op. - Latest cleanup smoke deleted one expired temporary artifact through `scripts/run_visual_capture_cleanup.py`. @@ -150,7 +160,7 @@ Automation execution-layer P19 is closed locally. P19 adds structured channel ro - Latest `scripts/run_trading_runtime.py` smoke returned blockers=none and includes a healthy `visual_watch` runtime task. - Latest local runtime smoke confirmed `runtime_status.tasks.alert_dispatch.details` now exposes retry and silent-window counters with blockers=none. -## P19 Blockers +## P20 Blockers - none @@ -170,6 +180,7 @@ Automation execution-layer P19 is closed locally. P19 adds structured channel ro - P17 visual trigger routing remains advisory evidence routing only and must not change execution permission, live readiness, trust bucket, risk governor, or operator confirmation. - P18 media evidence remains delivery metadata/fallback routing only and must not change execution permission, live readiness, trust bucket, risk governor, or operator confirmation. - P19 channel routing remains notification routing/observability only and must not change execution permission, live readiness, trust bucket, risk governor, or operator confirmation. +- P20 operator digest remains read-only review aggregation only and must not change execution permission, live readiness, trust bucket, risk governor, or operator confirmation. - A final rerun of `scripts/run_alert_dispatcher.py` and `scripts/run_trading_runtime.py` hit local PostgreSQL connection timeout because nothing was listening on `127.0.0.1:5432`; the P14 code path is still covered by full unit tests and earlier successful smoke during implementation. - P15 dispatcher smoke succeeded with `suppressed_noise_policy`, `workbench_only`, and `policy_external_allowed` counters present. P15 runtime smoke reached `alert_dispatch` successfully, but the overall runtime returned external environment blockers for market stream disconnected and OKX read transport connection refused. - P16 direct browser smoke succeeded locally with `browser_capture_status=captured`. Script-level capture without a watch run still returns safe no-op `watch_run_missing`. @@ -182,11 +193,13 @@ Automation execution-layer P19 is closed locally. P19 adds structured channel ro - P19 dispatcher smoke returned attempted=0 and exposed `channel_health.enabled_channels=["workbench"]` plus disabled `webhook`, `wecom`, and `email` credential-missing states. - P19 quality-summary API smoke returned 200 and exposed `delivery_summary.channel_status_counts`. - Latest P19 runtime smoke returned blockers=none and exposes `alert_dispatch.details.channel_health` and `channel_summary`. +- P20 focused tests passed for digest service, script JSON/text output, API route, and dashboard route inventory; full `tests/unit` passed with 685 tests. +- P20 smoke returned `status=completed` through `scripts/run_operator_digest.py` with review_queue=10, API route smoke returned 200 with review_queue=3, and runtime smoke returned blockers=none. - Real-live order submission remains intentionally disabled because the OKX order adapter is still not a real-live submitter. ## Exact Next Step -P19 is locally closed. Default next action is P20 `Operator Digest And Review Queue`: aggregate recent alerts, visual reviews, feedback, suppression, retries, media, and channel health into an operator-facing digest/review queue. +P20 is locally closed. Default next action is P21 `Runtime Incident Journal`: persist runtime incidents and recovery suggestions as an operator-visible incident log without changing trading permissions. ## Validation Commands Used Recently @@ -207,4 +220,6 @@ D:\ai-exchange\.venv\Scripts\python.exe scripts\run_visual_review.py D:\ai-exchange\.venv\Scripts\python.exe -m unittest tests.unit.test_visual_media_evidence_service tests.unit.test_alert_feedback_service tests.unit.test_alert_quality_summary_service tests.unit.test_trading_runtime_service tests.unit.test_api_and_main D:\ai-exchange\.venv\Scripts\python.exe scripts\run_alert_dispatcher.py D:\ai-exchange\.venv\Scripts\python.exe -m unittest tests.unit.test_channel_router_service tests.unit.test_alert_feedback_service tests.unit.test_alert_quality_summary_service tests.unit.test_trading_runtime_service +D:\ai-exchange\.venv\Scripts\python.exe -m unittest tests.unit.test_operator_digest_service tests.unit.test_run_operator_digest tests.unit.test_api_server.ApiServerRoutingTests.test_routes_trading_alert_operator_digest tests.unit.test_api_static_server.ApiStaticServerTests.test_dashboard_uses_canonical_trading_routes +D:\ai-exchange\.venv\Scripts\python.exe scripts\run_operator_digest.py ``` diff --git a/TASKS.md b/TASKS.md index 36ca65b..a3ffe82 100644 --- a/TASKS.md +++ b/TASKS.md @@ -13,7 +13,7 @@ ## In Progress -- [x] P19 Multi-Channel Alert Routing local closure complete +- [x] P20 Operator Digest And Review Queue local closure complete ## Next Up @@ -30,14 +30,15 @@ - [x] P17 local closure complete - [x] P18 local closure complete - [x] P19 local closure complete -- [ ] Start P20 Operator Digest And Review Queue +- [x] P20 local closure complete +- [ ] Start P21 Runtime Incident Journal - [ ] Continue periodic analysis-decision monitoring with maturity rollup in parallel - [ ] Treat browser screenshot capture and additional external media adapters as future enhancements ## Automation Status -- [x] state: automation_execution_layer_p19_multi_channel_alert_routing_closed -- [x] current_phase: P19 Multi-Channel Alert Routing +- [x] state: automation_execution_layer_p20_operator_digest_review_queue_closed +- [x] current_phase: P20 Operator Digest And Review Queue - [x] latest_p7_selector_choice: continue_periodic_analysis_decision_monitoring_with_maturity_rollup - [x] latest_p14_slice: webhook_transport_retry - [x] latest_p15_pxx_tasklist_status: ready @@ -64,7 +65,12 @@ - [x] latest_channel_health_status: visible - [x] latest_disabled_adapter_status: visible_credential_missing - [x] latest_channel_routing_smoke_status: no_pending_alerts_safe_noop -- [x] next_execution_layer_phase: P20 Operator Digest And Review Queue +- [x] latest_p20_slice: operator_digest_review_queue +- [x] latest_operator_digest_status: implemented +- [x] latest_operator_digest_route_status: ok +- [x] latest_operator_digest_workbench_status: visible +- [x] latest_operator_digest_smoke_status: safe_noop_or_completed +- [x] next_execution_layer_phase: P21 Runtime Incident Journal - [x] latest_visual_review_script_status: idle - [x] latest_runtime_visual_watch_task_status: ok - [x] latest_visual_capture_cleanup_task_status: ok @@ -155,6 +161,11 @@ - [x] Added disabled adapter states for WeCom and email when credentials or adapter support are unavailable - [x] Added per-channel summary to dispatcher results, delivery summary, runtime `alert_dispatch`, dispatcher script output, and workbench Alert Quality - [x] Added focused P19 channel router, disabled channel, summary, and runtime tests +- [x] Added `OperatorDigestService` for recent alert, delivery, visual review, quality flag, and feedback-gap aggregation +- [x] Added `scripts/run_operator_digest.py` +- [x] Added `GET /trading/alerts/operator-digest` +- [x] Added operator digest/review queue visibility to the workbench `Alert Quality` panel +- [x] Added focused P20 digest service, script, API, and dashboard route tests - [x] Added runtime `visual_watch` task - [x] Added focused service, script, API, and runtime tests - [x] Smoked the visual review script with no pending reviews @@ -177,6 +188,8 @@ - [x] Smoked P19 dispatcher safe no-op with channel health visible - [x] Smoked P19 quality-summary API with channel status counts visible - [x] Smoked P19 trading runtime with blockers=none and alert dispatch channel health visible +- [x] Smoked P20 operator digest script/API completed state with review queue fields visible +- [x] Completed P20 full validation: 685 unit tests, 68 script static checks, dashboard JS syntax, operator digest script/API smoke, and runtime smoke blockers=none ## Blocked @@ -232,7 +245,7 @@ - [x] `P17` Visual Trigger ROI Router - [x] `P18` Visual Evidence Media Dispatch - [x] `P19` Multi-Channel Alert Routing -- [ ] `P20` Operator Digest And Review Queue +- [x] `P20` Operator Digest And Review Queue - [ ] `P21` Runtime Incident Journal - [ ] `P22` Maintenance And Retention Rollup - [ ] `P23` End-To-End Unattended Harness diff --git a/docs/AI_ICT_盯盘告警开发任务清单_v1.md b/docs/AI_ICT_盯盘告警开发任务清单_v1.md index 9cd8db9..39bd660 100644 --- a/docs/AI_ICT_盯盘告警开发任务清单_v1.md +++ b/docs/AI_ICT_盯盘告警开发任务清单_v1.md @@ -409,6 +409,7 @@ candidate -> decision alert -> dispatcher -> 外部消息 -> delivery record - 2026-05-12 P17 已完成 `visual trigger ROI router` 本地闭环：视觉触发策略现在区分 `triggered` / `skipped`，普通 observe/monitoring 状态不触发视觉复核，actionable、blocked、market degraded、broker/reconciliation degraded 与 council watch item 会输出可审计 trigger type / reason codes，并在 runtime、alert quality summary、脚本和 workbench 中暴露 trigger/skip/ROI 计数。 - 2026-05-12 P18 已完成 `visual evidence media dispatch` 本地闭环：visual review 现在会生成稳定 `media_evidence` block，外部文本通道不支持文件时降级为 `media_linked`，workbench 可消费本地截图路径时标记 `media_attached`，无视觉证据时标记 `media_unavailable`；dispatcher、delivery summary、runtime、脚本和 workbench 均可见媒体证据状态。 - 2026-05-12 P19 已完成 `multi-channel alert routing` 本地闭环：channel router 现在输出 workbench/webhook/WeCom/Email capability 与 health summary，WeCom/Email 在凭据缺失或 adapter 未实现时会以 disabled adapter 形式返回 `credential_missing` / `adapter_not_implemented`，dispatcher、delivery summary、runtime、脚本和 workbench 均可见 per-channel 状态。 +- 2026-05-12 P20 已完成 `operator digest / review queue` 本地闭环：operator digest 会把高价值告警、被抑制 delivery、失败/重试 delivery、视觉 mismatch 与待反馈项聚合成 bounded review queue，并通过脚本、API 和 workbench `Alert Quality` 面板暴露。 ### 验收 - 非重大错误不打断主循环 diff --git a/docs/AI_ICT_盯盘告警技术设计稿_v1.md b/docs/AI_ICT_盯盘告警技术设计稿_v1.md index 6cf1649..859ad3f 100644 --- a/docs/AI_ICT_盯盘告警技术设计稿_v1.md +++ b/docs/AI_ICT_盯盘告警技术设计稿_v1.md @@ -109,6 +109,11 @@ 2026-05-12 P19 已落地：当前 channel router 保持现有 workbench/webhook 行为，同时输出结构化 `channel_route`、channel capability 与 channel health summary。WeCom / Email 已预留为 disabled adapter；凭据缺失时暴露 `credential_missing`，凭据存在但 adapter 未实现时暴露 `adapter_not_implemented`，不会把未配置通道伪装为成功发送。dispatcher、delivery summary、runtime 与 workbench 均可查看 per-channel 状态。 +#### `src/services/alerts/operator_digest_service.py` +职责：把 recent alerts、delivery history、visual reviews、alert/visual feedback 和 alert quality summary 聚合成 operator digest。 + +2026-05-12 P20 已落地：digest 输出 high-value alerts、suppressed deliveries、failed/retry deliveries、visual mismatches、pending feedback items、quality flags 与 bounded review queue。该服务只做复盘聚合与 operator 可观测性，不改变 execution allowed、real live readiness、risk governor、trust bucket 或 operator confirmation protocol。 + #### `src/services/alerts/delivery_policy_service.py` 职责： diff --git a/docs/AI_ICT_自动化执行层_P15-PXX闭环任务清单_v1.md b/docs/AI_ICT_自动化执行层_P15-PXX闭环任务清单_v1.md index 934402c..65a2d8b 100644 --- a/docs/AI_ICT_自动化执行层_P15-PXX闭环任务清单_v1.md +++ b/docs/AI_ICT_自动化执行层_P15-PXX闭环任务清单_v1.md @@ -217,6 +217,10 @@ - digest 能列出需要 operator 复核的 top items。 - 不改变交易权限或自动下单边界。 +- 2026-05-12 P20 已完成 `Operator Digest And Review Queue` 本地闭环：新增 `OperatorDigestService` 聚合 recent alerts、delivery history、visual reviews、quality flags 与 feedback gaps，输出 `high_value_alerts`、`suppressed_alerts`、`failed_retry_items`、`visual_mismatch_items`、`pending_feedback_items` 与 bounded `review_queue`。 +- 2026-05-12 P20 已新增 `scripts/run_operator_digest.py` 与 `GET /trading/alerts/operator-digest`，并将 digest status/counts/top queue items 接入 workbench `Alert Quality` 面板。 +- 2026-05-12 P20 本地闭环已验证：focused P20 tests 通过，`tests/unit` 全量 685 个测试通过，script static checks 68 个脚本通过，`node --check src/web/app.js` 通过；digest 在无数据时返回 safe idle/no-op，有数据时可列出 operator top review items，脚本/API/runtime smoke 均可见 review queue/counts 且 runtime blockers=none；该层只读聚合，不改变交易权限或自动下单边界。 + --- ## 11. P21 Runtime Incident Journal diff --git a/scripts/run_operator_digest.py b/scripts/run_operator_digest.py new file mode 100644 index 0000000..f96edcd --- /dev/null +++ b/scripts/run_operator_digest.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +from pathlib import Path +import json +import os +import sys + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from src.services.alerts import OperatorDigestService + + +def main(env=None) -> int: + env = os.environ if env is None else env + instrument_id = int(env.get("OPERATOR_DIGEST_INSTRUMENT_ID", "1")) + alert_limit = int(env.get("OPERATOR_DIGEST_LIMIT", "50")) + visual_limit = int(env.get("OPERATOR_DIGEST_VISUAL_LIMIT", "50")) + top_limit = int(env.get("OPERATOR_DIGEST_TOP_LIMIT", "10")) + output_json = _get_bool(env, "OPERATOR_DIGEST_OUTPUT_JSON", False) + payload = OperatorDigestService().build_digest( + instrument_id=instrument_id, + alert_limit=alert_limit, + visual_review_limit=visual_limit, + top_limit=top_limit, + ) + + if output_json: + print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str)) + else: + counts = dict(payload.get("summary_counts") or {}) + print(f"Operator digest status: {payload.get('status', 'unknown')}") + print(payload.get("operator_summary") or "Operator digest unavailable.") + print( + "Alerts / suppressed / failed-retry / visual mismatches / pending feedback: " + f"{counts.get('recent_alert_count', 0)} / " + f"{counts.get('suppressed_delivery_count', 0)} / " + f"{counts.get('failed_or_retry_delivery_count', 0)} / " + f"{counts.get('visual_mismatch_count', 0)} / " + f"{counts.get('pending_feedback_count', 0)}" + ) + print(f"Review queue items: {len(payload.get('review_queue') or [])}") + for item in payload.get("review_queue") or []: + print( + f"- {item.get('priority', 'unknown')} {item.get('queue_type', 'unknown')} " + f"alert={item.get('alert_id', 'n/a')} review={item.get('visual_review_id', 'n/a')} " + f"reason={item.get('reason', 'n/a')}" + ) + if payload.get("warnings"): + print(f"Warnings: {', '.join(payload.get('warnings') or [])}") + return 0 + + +def _get_bool(env, key: str, default: bool = False) -> bool: + value = env.get(key) + if value is None or value == "": + return default + return str(value).strip().lower() in {"1", "true", "yes", "on"} + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/api/__init__.py b/src/api/__init__.py index 6a0b806..36b4ec3 100644 --- a/src/api/__init__.py +++ b/src/api/__init__.py @@ -48,7 +48,13 @@ from src.api.trading_runtime import ( ) from src.api.trading_journal import get_recent_trading_journal, get_trading_journal_summary from src.api.trading_plans import get_recent_trading_plans -from src.api.alerts import dispatch_pending_alerts, get_alert_deliveries, get_alert_quality_summary, get_alert_silent_window_status +from src.api.alerts import ( + dispatch_pending_alerts, + get_alert_deliveries, + get_alert_quality_summary, + get_alert_silent_window_status, + get_operator_digest, +) from src.api.feedback import ( create_alert_feedback, create_visual_review_feedback, @@ -124,6 +130,7 @@ __all__ = [ "dispatch_pending_alerts", "get_alert_deliveries", "get_alert_quality_summary", + "get_operator_digest", "get_alert_silent_window_status", "create_alert_feedback", "create_visual_review_feedback", diff --git a/src/api/alerts.py b/src/api/alerts.py index 9973144..339a222 100644 --- a/src/api/alerts.py +++ b/src/api/alerts.py @@ -1,13 +1,16 @@ from src.repositories.alert_delivery_repository import AlertDeliveryRepository -from src.services.alerts import AlertDispatcherService, AlertQualitySummaryService, AlertSilentWindowService - +from src.services.alerts import ( + AlertDispatcherService, + AlertQualitySummaryService, + AlertSilentWindowService, + OperatorDigestService, +) def dispatch_pending_alerts(limit: int = 50) -> dict: return AlertDispatcherService().dispatch_pending(limit=limit) - def get_alert_deliveries(alert_id: int, limit: int = 20) -> list[dict]: return AlertDeliveryRepository().fetch_recent_deliveries(alert_id=alert_id, limit=limit) @@ -25,5 +28,20 @@ def get_alert_quality_summary( ) +def get_operator_digest( + instrument_id: int, + *, + limit: int = 50, + visual_limit: int = 50, + top_limit: int = 10, +) -> dict: + return OperatorDigestService().build_digest( + instrument_id=instrument_id, + alert_limit=limit, + visual_review_limit=visual_limit, + top_limit=top_limit, + ) + + def get_alert_silent_window_status() -> dict: return AlertSilentWindowService().build_status() diff --git a/src/api/server.py b/src/api/server.py index b1f2952..a1a7505 100644 --- a/src/api/server.py +++ b/src/api/server.py @@ -10,7 +10,12 @@ from src.api.candles import get_recent_candles from src.api.execution_tickets import get_recent_execution_tickets from src.api.health import get_health from src.api.live_operation_audits import get_recent_live_operation_audits -from src.api.alerts import get_alert_deliveries, get_alert_quality_summary, get_alert_silent_window_status +from src.api.alerts import ( + get_alert_deliveries, + get_alert_quality_summary, + get_alert_silent_window_status, + get_operator_digest, +) from src.api.feedback import ( create_alert_feedback, create_visual_review_feedback, @@ -113,6 +118,7 @@ def build_api_dependencies() -> dict: "trading_decision_alerts": get_trading_decision_alerts, "trading_alert_dispatch": run_trading_alert_dispatch, "trading_alert_quality_summary": get_alert_quality_summary, + "trading_alert_operator_digest": get_operator_digest, "trading_alert_silent_window_status": get_alert_silent_window_status, "trading_reconciliation_run": run_trading_reconciliation, "trading_runtime_status": get_trading_runtime_status, @@ -269,6 +275,17 @@ def route_request( limit=limit, visual_limit=visual_limit, ) + if method == "GET" and path == "/trading/alerts/operator-digest": + instrument_id = _require_int(query, "instrument_id") + limit = _get_int(query, "limit", 50) + visual_limit = _get_int(query, "visual_limit", 50) + top_limit = _get_int(query, "top_limit", 10) + return 200, deps["trading_alert_operator_digest"]( + instrument_id=instrument_id, + limit=limit, + visual_limit=visual_limit, + top_limit=top_limit, + ) if method == "POST" and path == "/trading/alerts/dispatch": limit = _get_body_optional_int(body, "limit") or 50 return 200, deps["trading_alert_dispatch"](limit=limit) diff --git a/src/services/alerts/__init__.py b/src/services/alerts/__init__.py index afb1f08..0c271ca 100644 --- a/src/services/alerts/__init__.py +++ b/src/services/alerts/__init__.py @@ -1,5 +1,11 @@ from src.services.alerts.alert_dispatcher_service import AlertDispatcherService from src.services.alerts.alert_quality_summary_service import AlertQualitySummaryService from src.services.alerts.alert_silent_window_service import AlertSilentWindowService +from src.services.alerts.operator_digest_service import OperatorDigestService -__all__ = ["AlertDispatcherService", "AlertQualitySummaryService", "AlertSilentWindowService"] +__all__ = [ + "AlertDispatcherService", + "AlertQualitySummaryService", + "AlertSilentWindowService", + "OperatorDigestService", +] diff --git a/src/services/alerts/operator_digest_service.py b/src/services/alerts/operator_digest_service.py new file mode 100644 index 0000000..a302391 --- /dev/null +++ b/src/services/alerts/operator_digest_service.py @@ -0,0 +1,290 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Optional + +from sqlalchemy.exc import SQLAlchemyError + +from src.repositories.alert_delivery_repository import AlertDeliveryRepository +from src.repositories.alert_feedback_repository import AlertFeedbackRepository +from src.repositories.decision_packet_repository import DecisionPacketRepository +from src.repositories.visual_snapshot_repository import VisualSnapshotRepository +from src.services.alerts.alert_quality_summary_service import AlertQualitySummaryService + + +class OperatorDigestService: + def __init__( + self, + *, + decision_repository: Optional[DecisionPacketRepository] = None, + delivery_repository: Optional[AlertDeliveryRepository] = None, + feedback_repository: Optional[AlertFeedbackRepository] = None, + visual_repository: Optional[VisualSnapshotRepository] = None, + quality_service: Optional[AlertQualitySummaryService] = None, + ) -> None: + self.decision_repository = decision_repository or DecisionPacketRepository() + self.delivery_repository = delivery_repository or AlertDeliveryRepository() + self.feedback_repository = feedback_repository or AlertFeedbackRepository() + self.visual_repository = visual_repository or VisualSnapshotRepository() + self.quality_service = quality_service or AlertQualitySummaryService( + decision_repository=self.decision_repository, + delivery_repository=self.delivery_repository, + feedback_repository=self.feedback_repository, + visual_repository=self.visual_repository, + ) + + def build_digest( + self, + *, + instrument_id: int, + alert_limit: int = 50, + visual_review_limit: int = 50, + top_limit: int = 10, + ) -> dict: + recent_alerts = self.decision_repository.fetch_recent_decision_alerts( + instrument_id=instrument_id, + limit=alert_limit, + ) + alert_ids = [int(item["id"]) for item in recent_alerts if item.get("id") is not None] + warnings: list[str] = [] + deliveries = _safe_fetch( + lambda: self.delivery_repository.fetch_recent_deliveries_for_alerts( + alert_ids, + limit=max(alert_limit * 4, 20), + ), + warning_code="delivery_history_unavailable", + warnings=warnings, + ) + visual_reviews = _safe_fetch( + lambda: self.visual_repository.fetch_recent_reviews_for_instrument( + instrument_id=instrument_id, + limit=visual_review_limit, + ), + warning_code="visual_reviews_unavailable", + warnings=warnings, + ) + alert_feedback = _safe_fetch( + lambda: self.feedback_repository.fetch_feedback_for_alert_ids( + alert_ids, + limit=max(alert_limit * 4, 20), + ), + warning_code="alert_feedback_unavailable", + warnings=warnings, + ) + visual_review_ids = [int(item["id"]) for item in visual_reviews if item.get("id") is not None] + visual_feedback = _safe_fetch( + lambda: self.feedback_repository.fetch_feedback_for_visual_review_ids( + visual_review_ids, + limit=max(visual_review_limit * 4, 20), + ), + warning_code="visual_feedback_unavailable", + warnings=warnings, + ) + feedback_alert_ids = {int(item["alert_id"]) for item in alert_feedback if item.get("alert_id") is not None} + feedback_visual_ids = { + int(item["visual_review_id"]) for item in visual_feedback if item.get("visual_review_id") is not None + } + quality_summary = self.quality_service.build_summary( + instrument_id=instrument_id, + alert_limit=alert_limit, + visual_review_limit=visual_review_limit, + ) + digest = { + "instrument_id": instrument_id, + "generated_ts": datetime.now(timezone.utc).isoformat(), + "status": "completed" if recent_alerts or deliveries or visual_reviews else "idle", + "window": { + "alert_limit": alert_limit, + "visual_review_limit": visual_review_limit, + "top_limit": top_limit, + }, + "summary_counts": { + "recent_alert_count": len(recent_alerts), + "recent_delivery_count": len(deliveries), + "recent_visual_review_count": len(visual_reviews), + "high_value_alert_count": len(_high_value_alerts(recent_alerts)), + "suppressed_delivery_count": len(_suppressed_deliveries(deliveries)), + "failed_or_retry_delivery_count": len(_failed_or_retry_deliveries(deliveries)), + "visual_mismatch_count": len(_visual_mismatches(visual_reviews)), + "pending_feedback_count": len( + _pending_feedback_items( + alerts=recent_alerts, + visual_reviews=visual_reviews, + feedback_alert_ids=feedback_alert_ids, + feedback_visual_ids=feedback_visual_ids, + ) + ), + }, + "high_value_alerts": _alert_items(_high_value_alerts(recent_alerts)[:top_limit]), + "suppressed_alerts": _delivery_items(_suppressed_deliveries(deliveries)[:top_limit]), + "failed_retry_items": _delivery_items(_failed_or_retry_deliveries(deliveries)[:top_limit]), + "visual_mismatch_items": _visual_items(_visual_mismatches(visual_reviews)[:top_limit]), + "pending_feedback_items": _pending_feedback_items( + alerts=recent_alerts, + visual_reviews=visual_reviews, + feedback_alert_ids=feedback_alert_ids, + feedback_visual_ids=feedback_visual_ids, + )[:top_limit], + "quality_flags": list(quality_summary.get("quality_flags") or []), + "quality_summary": quality_summary, + "warnings": warnings + list(quality_summary.get("warnings") or []), + } + digest["review_queue"] = _build_review_queue(digest, top_limit=top_limit) + digest["operator_summary"] = _operator_summary(digest) + return digest + + +def _high_value_alerts(alerts: list[dict]) -> list[dict]: + result = [] + for alert in alerts: + payload = dict(alert.get("payload") or {}) + policy = dict(payload.get("visual_trigger_policy") or {}) + if alert.get("alert_type") in {"candidate_actionable", "ticket_ready", "runtime_degraded"}: + result.append(alert) + elif policy.get("status") == "triggered": + result.append(alert) + return result + + +def _suppressed_deliveries(deliveries: list[dict]) -> list[dict]: + reasons = {"suppressed_noise_policy", "alert_silent_window_active", "workbench_only"} + return [item for item in deliveries if _delivery_reason(item) in reasons] + + +def _failed_or_retry_deliveries(deliveries: list[dict]) -> list[dict]: + return [ + item + for item in deliveries + if str(item.get("status") or "") == "failed" + or _delivery_reason(item) == "retry_exhausted" + or str((item.get("payload") or {}).get("dispatch_mode") or "") == "failed_retry" + ] + + +def _visual_mismatches(visual_reviews: list[dict]) -> list[dict]: + return [item for item in visual_reviews if str(item.get("visual_state") or "") == "mismatch"] + + +def _pending_feedback_items( + *, + alerts: list[dict], + visual_reviews: list[dict], + feedback_alert_ids: set[int], + feedback_visual_ids: set[int], +) -> list[dict]: + items: list[dict] = [] + for alert in alerts: + alert_id = alert.get("id") + if alert_id is not None and int(alert_id) not in feedback_alert_ids: + items.append( + { + "subject_type": "alert", + "subject_id": int(alert_id), + "label": alert.get("alert_type"), + "reason": "feedback_missing", + "created_ts": _iso(alert.get("created_ts")), + } + ) + for review in visual_reviews: + review_id = review.get("id") + if review_id is not None and int(review_id) not in feedback_visual_ids: + items.append( + { + "subject_type": "visual_review", + "subject_id": int(review_id), + "label": review.get("visual_state"), + "reason": "feedback_missing", + "created_ts": _iso(review.get("reviewed_ts")), + } + ) + return items + + +def _build_review_queue(digest: dict, *, top_limit: int) -> list[dict]: + queue: list[dict] = [] + for item in digest.get("failed_retry_items") or []: + queue.append({"priority": "high", "queue_type": "failed_retry", **item}) + for item in digest.get("visual_mismatch_items") or []: + queue.append({"priority": "high", "queue_type": "visual_mismatch", **item}) + for item in digest.get("high_value_alerts") or []: + queue.append({"priority": "medium", "queue_type": "high_value_alert", **item}) + for item in digest.get("pending_feedback_items") or []: + queue.append({"priority": "low", "queue_type": "pending_feedback", **item}) + return queue[:top_limit] + + +def _alert_items(alerts: list[dict]) -> list[dict]: + return [ + { + "alert_id": item.get("id"), + "alert_type": item.get("alert_type"), + "status": item.get("status"), + "channel": item.get("channel"), + "created_ts": _iso(item.get("created_ts")), + "action_recommendation": (item.get("payload") or {}).get("action_recommendation"), + "visual_trigger_types": list((item.get("payload") or {}).get("visual_trigger_types") or []), + } + for item in alerts + ] + + +def _delivery_items(deliveries: list[dict]) -> list[dict]: + return [ + { + "alert_id": item.get("alert_id"), + "delivery_id": item.get("id"), + "channel": item.get("channel"), + "status": item.get("status"), + "reason": _delivery_reason(item), + "attempted_ts": _iso(item.get("attempted_ts")), + } + for item in deliveries + ] + + +def _visual_items(rows: list[dict]) -> list[dict]: + return [ + { + "visual_review_id": item.get("id"), + "watch_run_id": item.get("watch_run_id"), + "trigger_type": item.get("trigger_type"), + "visual_state": item.get("visual_state"), + "visual_confidence": item.get("visual_confidence"), + "reviewed_ts": _iso(item.get("reviewed_ts")), + } + for item in rows + ] + + +def _operator_summary(digest: dict) -> str: + counts = dict(digest.get("summary_counts") or {}) + if digest.get("status") == "idle": + return "Operator digest idle: no recent alert or visual-review items." + return ( + f"Digest alerts={counts.get('recent_alert_count', 0)}, " + f"suppressed={counts.get('suppressed_delivery_count', 0)}, " + f"failed/retry={counts.get('failed_or_retry_delivery_count', 0)}, " + f"visual mismatches={counts.get('visual_mismatch_count', 0)}, " + f"pending feedback={counts.get('pending_feedback_count', 0)}." + ) + + +def _delivery_reason(delivery: dict) -> str: + payload = dict(delivery.get("payload") or {}) + return str(payload.get("reason") or delivery.get("error") or "") + + +def _iso(value) -> str | None: + if value in (None, ""): + return None + if isinstance(value, datetime): + return value.astimezone(timezone.utc).isoformat() + return str(value) + + +def _safe_fetch(fetcher, *, warning_code: str, warnings: list[str]) -> list[dict]: + try: + return list(fetcher() or []) + except SQLAlchemyError: + warnings.append(warning_code) + return [] diff --git a/src/web/app.js b/src/web/app.js index 548688d..7dcf106 100644 --- a/src/web/app.js +++ b/src/web/app.js @@ -553,18 +553,21 @@ function renderAlertHistory(alertRows) { alertHistoryPanel.appendChild(grid); } -function renderAlertQuality(summary) { +function renderAlertQuality(summary, operatorDigest) { clearNode(alertQualityPanel); const delivery = summary?.delivery_summary || {}; const alertFeedback = summary?.alert_feedback_summary || {}; const visualFeedback = summary?.visual_feedback_summary || {}; const reviewSummary = summary?.visual_review_summary || {}; const silentWindow = summary?.silent_window || {}; + const digestCounts = operatorDigest?.summary_counts || {}; + const reviewQueue = operatorDigest?.review_queue || []; alertQualityPanel.appendChild( buildInfoGrid([ ['recent alerts', String(summary?.recent_alert_count ?? 0)], ['recent visual reviews', String(summary?.recent_visual_review_count ?? 0)], ['silent window', `${silentWindow.active ? 'active' : 'inactive'} | ${silentWindow.window_label || 'n/a'}`], + ['operator digest', `${operatorDigest?.status || 'idle'} | queue ${reviewQueue.length}`], ['delivery success', formatPercent(delivery.success_rate)], ['alert usefulness', formatPercent(alertFeedback.positive_rate)], ['visual helpfulness', formatPercent(visualFeedback.positive_rate)], @@ -573,6 +576,20 @@ function renderAlertQuality(summary) { ]) ); const grid = createElement('div', 'cards'); + grid.appendChild( + buildItemCard( + 'Operator Digest', + reviewQueue.length ? 'review' : (operatorDigest?.status || 'idle'), + [ + ['high value alerts', String(digestCounts.high_value_alert_count ?? 0)], + ['suppressed / failed-retry', `${digestCounts.suppressed_delivery_count ?? 0} / ${digestCounts.failed_or_retry_delivery_count ?? 0}`], + ['visual mismatches', String(digestCounts.visual_mismatch_count ?? 0)], + ['pending feedback', String(digestCounts.pending_feedback_count ?? 0)], + ['review queue', formatList(reviewQueue.slice(0, 4).map((item) => `${item.priority || 'n/a'}:${item.queue_type || 'item'}`))], + ['digest summary', operatorDigest?.operator_summary || 'n/a'], + ] + ) + ); grid.appendChild( buildItemCard( 'Delivery', @@ -1296,6 +1313,7 @@ function renderDashboardData({ decisionAlerts, visualReviews, alertQualitySummary, + operatorDigest, decisionPacketId, trustScore, liveReadiness, @@ -1324,7 +1342,7 @@ function renderDashboardData({ renderCandidates(tradeCandidates.items || []); renderTickets(executionTickets.items || []); renderAlertHistory(decisionAlerts); - renderAlertQuality(alertQualitySummary); + renderAlertQuality(alertQualitySummary, operatorDigest); renderVisualReviews(visualReviews, decisionPacketId); renderExecutions(tradingExecutions.items || []); renderOperatorConsole(operatorNotes, reviewPacket); @@ -1340,7 +1358,7 @@ async function refreshDashboard() { const limit = limitInput.value || '20'; try { - const [health, candles, tradingPlan, marketState, runtimeStatus, decisionPacket, decisionAlerts, alertQualitySummary, trustScore, liveReadiness, launchReadiness, sessionView, tradeCandidates, executionTickets, tradingExecutions, journalSummary, operatorNotes, reviewPacket, journalItems, weeklyReview] = await Promise.all([ + const [health, candles, tradingPlan, marketState, runtimeStatus, decisionPacket, decisionAlerts, alertQualitySummary, operatorDigest, trustScore, liveReadiness, launchReadiness, sessionView, tradeCandidates, executionTickets, tradingExecutions, journalSummary, operatorNotes, reviewPacket, journalItems, weeklyReview] = await Promise.all([ fetchJson('/health'), fetchJson(`/candles?${queryString({ instrument_id: instrumentId, timeframe, limit })}`), fetchJson(`/trading/plans/latest?${queryString({ instrument_id: instrumentId })}`).catch(() => null), @@ -1349,6 +1367,7 @@ async function refreshDashboard() { fetchJson(`/trading/decision-packet?${queryString({ instrument_id: instrumentId })}`).catch(() => ({ status: 'unavailable', action_recommendation: 'monitor' })), fetchJson(`/trading/alerts?${queryString({ instrument_id: instrumentId, limit: 10 })}`).catch(() => ({ items: [] })), fetchJson(`/trading/alerts/quality-summary?${queryString({ instrument_id: instrumentId, limit: 50, visual_limit: 50 })}`).catch(() => ({ recent_alert_count: 0, recent_visual_review_count: 0, delivery_summary: {}, alert_feedback_summary: {}, visual_feedback_summary: {}, visual_review_summary: {}, quality_flags: [], operator_summary: 'alert_quality_unavailable' })), + fetchJson(`/trading/alerts/operator-digest?${queryString({ instrument_id: instrumentId, limit: 50, visual_limit: 50, top_limit: 10 })}`).catch(() => ({ status: 'unavailable', summary_counts: {}, review_queue: [], operator_summary: 'operator_digest_unavailable' })), fetchJson(`/trading/trust-score?${queryString({ instrument_id: instrumentId })}`).catch(() => ({ models: [], whitelist: [], observe_list: [], blacklist: [], next_actions: [] })), fetchJson(`/trading/live-readiness?${queryString({ instrument_id: instrumentId, timeframe })}`).catch(() => ({ ready_for_real_live: false, blocking_reasons: ['live_readiness_unavailable'], checklist: [] })), fetchJson(`/trading/launch-readiness?${queryString({ instrument_id: instrumentId, timeframe })}`).catch(() => ({ assisted_trading_ready: false, real_live_ready: false, blocking_reasons: ['launch_readiness_unavailable'], warning_reasons: [], checklist: [], next_actions: [] })), @@ -1376,6 +1395,7 @@ async function refreshDashboard() { decisionPacket, decisionAlerts, alertQualitySummary, + operatorDigest, visualReviews, decisionPacketId, trustScore, diff --git a/tests/unit/test_api_server.py b/tests/unit/test_api_server.py index 68e42f4..41f57c7 100644 --- a/tests/unit/test_api_server.py +++ b/tests/unit/test_api_server.py @@ -86,6 +86,28 @@ class ApiServerRoutingTests(unittest.TestCase): self.assertEqual(payload["recent_alert_count"], 25) self.assertEqual(payload["recent_visual_review_count"], 15) + def test_routes_trading_alert_operator_digest(self) -> None: + status, payload = route_request( + "/trading/alerts/operator-digest", + {"instrument_id": ["1"], "limit": ["25"], "visual_limit": ["15"], "top_limit": ["3"]}, + dependencies={ + "trading_alert_operator_digest": lambda **kwargs: { + "instrument_id": kwargs["instrument_id"], + "alert_limit": kwargs["limit"], + "visual_limit": kwargs["visual_limit"], + "top_limit": kwargs["top_limit"], + "review_queue": [{"queue_type": "failed_retry"}], + } + }, + ) + + self.assertEqual(status, 200) + self.assertEqual(payload["instrument_id"], 1) + self.assertEqual(payload["alert_limit"], 25) + self.assertEqual(payload["visual_limit"], 15) + self.assertEqual(payload["top_limit"], 3) + self.assertEqual(payload["review_queue"][0]["queue_type"], "failed_retry") + def test_routes_trading_alert_silent_window_status(self) -> None: status, payload = route_request( "/trading/alerts/silent-window-status", diff --git a/tests/unit/test_api_static_server.py b/tests/unit/test_api_static_server.py index f90ded9..e81b17f 100644 --- a/tests/unit/test_api_static_server.py +++ b/tests/unit/test_api_static_server.py @@ -101,6 +101,7 @@ class ApiStaticServerTests(unittest.TestCase): "/trading/market-state", "/trading/candidates", "/trading/alerts/quality-summary", + "/trading/alerts/operator-digest", "/trading/visual-watch/cleanup-captures", "/trading/tickets", "/trading/executions", diff --git a/tests/unit/test_operator_digest_service.py b/tests/unit/test_operator_digest_service.py new file mode 100644 index 0000000..3eedf93 --- /dev/null +++ b/tests/unit/test_operator_digest_service.py @@ -0,0 +1,186 @@ +import unittest + +from sqlalchemy.exc import OperationalError + +from src.services.alerts import OperatorDigestService + + +class _DecisionRepository: + def __init__(self, alerts=None) -> None: + self.alerts = alerts if alerts is not None else [ + { + "id": 1, + "alert_type": "candidate_actionable", + "status": "queued", + "channel": "workbench", + "created_ts": "2026-05-12T01:00:00+00:00", + "payload": {"action_recommendation": "confirm_ticket"}, + }, + { + "id": 2, + "alert_type": "observe", + "status": "queued", + "channel": "workbench", + "created_ts": "2026-05-12T00:59:00+00:00", + "payload": { + "visual_trigger_policy": {"status": "triggered"}, + "visual_trigger_types": ["market_degradation_review"], + }, + }, + { + "id": 3, + "alert_type": "observe", + "status": "queued", + "channel": "workbench", + "created_ts": "2026-05-12T00:58:00+00:00", + "payload": {}, + }, + ] + + def fetch_recent_decision_alerts(self, *, instrument_id: int, limit: int = 50): + return self.alerts[:limit] + + +class _DeliveryRepository: + def __init__(self, deliveries=None) -> None: + self.deliveries = deliveries if deliveries is not None else [ + { + "id": 10, + "alert_id": 1, + "channel": "workbench", + "status": "skipped", + "attempted_ts": "2026-05-12T01:01:00+00:00", + "payload": {"reason": "suppressed_noise_policy"}, + }, + { + "id": 11, + "alert_id": 2, + "channel": "webhook", + "status": "failed", + "attempted_ts": "2026-05-12T01:00:30+00:00", + "error": "transport_error", + "payload": {}, + }, + { + "id": 12, + "alert_id": 2, + "channel": "webhook", + "status": "delivered", + "attempted_ts": "2026-05-12T01:00:20+00:00", + "payload": {"dispatch_mode": "failed_retry"}, + }, + ] + + def fetch_recent_deliveries_for_alerts(self, alert_ids, limit: int = 200): + return self.deliveries[:limit] + + +class _FeedbackRepository: + def fetch_feedback_for_alert_ids(self, alert_ids, limit: int = 200): + return [{"alert_id": 1, "label": "useful", "created_ts": "2026-05-12T01:02:00+00:00"}] + + def fetch_feedback_for_visual_review_ids(self, visual_review_ids, limit: int = 200): + return [{"visual_review_id": 32, "label": "visual_helpful", "created_ts": "2026-05-12T01:03:00+00:00"}] + + +class _VisualRepository: + def __init__(self, reviews=None) -> None: + self.reviews = reviews if reviews is not None else [ + { + "id": 31, + "watch_run_id": 21, + "trigger_type": "market_degradation_review", + "visual_state": "mismatch", + "visual_confidence": 0.81, + "reviewed_ts": "2026-05-12T01:04:00+00:00", + }, + { + "id": 32, + "watch_run_id": 22, + "trigger_type": "candidate_actionable_watch", + "visual_state": "confirm", + "visual_confidence": 0.92, + "reviewed_ts": "2026-05-12T01:03:00+00:00", + }, + ] + + def fetch_recent_reviews_for_instrument(self, instrument_id: int, limit: int = 50): + return self.reviews[:limit] + + +class _QualityService: + def __init__(self, warnings=None) -> None: + self.warnings = warnings or [] + + def build_summary(self, *, instrument_id: int, alert_limit: int = 50, visual_review_limit: int = 50): + return { + "instrument_id": instrument_id, + "quality_flags": ["visual_mismatch_present"], + "warnings": self.warnings, + } + + +class OperatorDigestServiceTests(unittest.TestCase): + def test_build_digest_prioritizes_operator_review_queue(self) -> None: + service = OperatorDigestService( + decision_repository=_DecisionRepository(), + delivery_repository=_DeliveryRepository(), + feedback_repository=_FeedbackRepository(), + visual_repository=_VisualRepository(), + quality_service=_QualityService(), + ) + + digest = service.build_digest(instrument_id=1, alert_limit=10, visual_review_limit=10, top_limit=6) + + self.assertEqual(digest["status"], "completed") + self.assertEqual(digest["summary_counts"]["high_value_alert_count"], 2) + self.assertEqual(digest["summary_counts"]["suppressed_delivery_count"], 1) + self.assertEqual(digest["summary_counts"]["failed_or_retry_delivery_count"], 2) + self.assertEqual(digest["summary_counts"]["visual_mismatch_count"], 1) + self.assertEqual(digest["summary_counts"]["pending_feedback_count"], 3) + self.assertIn("visual_mismatch_present", digest["quality_flags"]) + self.assertEqual(digest["review_queue"][0]["queue_type"], "failed_retry") + self.assertIn("visual_mismatch", [item["queue_type"] for item in digest["review_queue"]]) + self.assertIn("pending feedback=3", digest["operator_summary"]) + + def test_build_digest_is_idle_when_no_recent_data_exists(self) -> None: + service = OperatorDigestService( + decision_repository=_DecisionRepository(alerts=[]), + delivery_repository=_DeliveryRepository(deliveries=[]), + feedback_repository=_FeedbackRepository(), + visual_repository=_VisualRepository(reviews=[]), + quality_service=_QualityService(), + ) + + digest = service.build_digest(instrument_id=1) + + self.assertEqual(digest["status"], "idle") + self.assertEqual(digest["summary_counts"]["recent_alert_count"], 0) + self.assertEqual(digest["review_queue"], []) + self.assertIn("idle", digest["operator_summary"]) + + def test_build_digest_degrades_optional_feedback_tables(self) -> None: + class _BrokenFeedbackRepository: + def fetch_feedback_for_alert_ids(self, alert_ids, limit: int = 200): + raise OperationalError("stmt", {}, Exception()) + + def fetch_feedback_for_visual_review_ids(self, visual_review_ids, limit: int = 200): + raise OperationalError("stmt", {}, Exception()) + + service = OperatorDigestService( + decision_repository=_DecisionRepository(), + delivery_repository=_DeliveryRepository(), + feedback_repository=_BrokenFeedbackRepository(), + visual_repository=_VisualRepository(), + quality_service=_QualityService(warnings=["quality_feedback_unavailable"]), + ) + + digest = service.build_digest(instrument_id=1) + + self.assertIn("alert_feedback_unavailable", digest["warnings"]) + self.assertIn("visual_feedback_unavailable", digest["warnings"]) + self.assertIn("quality_feedback_unavailable", digest["warnings"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_run_operator_digest.py b/tests/unit/test_run_operator_digest.py new file mode 100644 index 0000000..a6dcf2f --- /dev/null +++ b/tests/unit/test_run_operator_digest.py @@ -0,0 +1,69 @@ +import contextlib +import io +import json +import unittest +from unittest.mock import patch + +from scripts import run_operator_digest + + +class RunOperatorDigestTests(unittest.TestCase): + def test_main_outputs_json_digest(self) -> None: + payload = { + "status": "completed", + "summary_counts": {"recent_alert_count": 2}, + "review_queue": [{"queue_type": "failed_retry"}], + "operator_summary": "Digest alerts=2.", + } + output = io.StringIO() + with patch("scripts.run_operator_digest.OperatorDigestService") as service_cls: + service_cls.return_value.build_digest.return_value = payload + with contextlib.redirect_stdout(output): + exit_code = run_operator_digest.main( + env={ + "OPERATOR_DIGEST_OUTPUT_JSON": "1", + "OPERATOR_DIGEST_INSTRUMENT_ID": "2", + "OPERATOR_DIGEST_LIMIT": "5", + "OPERATOR_DIGEST_VISUAL_LIMIT": "7", + "OPERATOR_DIGEST_TOP_LIMIT": "3", + } + ) + + self.assertEqual(exit_code, 0) + service_cls.return_value.build_digest.assert_called_once_with( + instrument_id=2, + alert_limit=5, + visual_review_limit=7, + top_limit=3, + ) + self.assertEqual(json.loads(output.getvalue())["review_queue"][0]["queue_type"], "failed_retry") + + def test_main_outputs_text_digest(self) -> None: + payload = { + "status": "completed", + "summary_counts": { + "recent_alert_count": 2, + "suppressed_delivery_count": 1, + "failed_or_retry_delivery_count": 1, + "visual_mismatch_count": 1, + "pending_feedback_count": 3, + }, + "review_queue": [{"priority": "high", "queue_type": "visual_mismatch", "visual_review_id": 31}], + "operator_summary": "Digest alerts=2.", + "warnings": ["visual_feedback_unavailable"], + } + output = io.StringIO() + with patch("scripts.run_operator_digest.OperatorDigestService") as service_cls: + service_cls.return_value.build_digest.return_value = payload + with contextlib.redirect_stdout(output): + exit_code = run_operator_digest.main(env={}) + + self.assertEqual(exit_code, 0) + text = output.getvalue() + self.assertIn("Operator digest status: completed", text) + self.assertIn("Review queue items: 1", text) + self.assertIn("visual_feedback_unavailable", text) + + +if __name__ == "__main__": + unittest.main()