yuqei/ai-exchange
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Close P20 operator digest review queue

Browse Source
This commit is contained in:
Codex
2026-05-12 08:43:12 +08:00
parent e53ef895ae
commit 03cd2aa915
16 changed files with 758 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
SESSION.md
+21 -6
View File
@@ -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
```
TASKS.md
+19 -6
View File
@@ -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
docs/AI_ICT_盯盘告警开发任务清单_v1.md
+1
View File
@@ -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` 面板暴露。
### 验收
- 非重大错误不打断主循环
docs/AI_ICT_盯盘告警技术设计稿_v1.md
+5
View File
@@ -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`
职责：
docs/AI_ICT_自动化执行层_P15-PXX闭环任务清单_v1.md
+4
View File
@@ -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
scripts/run_operator_digest.py
+63
View File
@@ -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())
src/api/__init__.py
+8 -1
View File
@@ -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",
src/api/alerts.py
+21 -3
View File
@@ -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()
src/api/server.py
+18 -1
View File
@@ -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)
src/services/alerts/__init__.py
+7 -1
View File
@@ -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",
]
src/services/alerts/operator_digest_service.py
+290
View File
@@ -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 []
src/web/app.js
+23 -3
View File
@@ -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,
tests/unit/test_api_server.py
+22
View File
@@ -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",
tests/unit/test_api_static_server.py
+1
View File
@@ -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",
tests/unit/test_operator_digest_service.py
+186
View File
@@ -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()
tests/unit/test_run_operator_digest.py
+69
View File
@@ -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()