diff --git a/SESSION.md b/SESSION.md index 6bbd095..07d4f9e 100644 --- a/SESSION.md +++ b/SESSION.md @@ -2,7 +2,7 @@ ## Current Goal -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`. +Automation execution-layer P21 is closed locally. P21 adds a runtime incident journal with persisted dedupe/aggregation, recommended actions, API/script access, runtime status summary, and workbench visibility. The default next action is P22 `Maintenance And Retention Rollup`. ## Current Branch @@ -14,8 +14,8 @@ Automation execution-layer P20 is closed locally. P20 adds an operator digest/re ## Automation Status -- state: automation_execution_layer_p20_operator_digest_review_queue_closed -- current_phase: P20 Operator Digest And Review Queue +- state: automation_execution_layer_p21_runtime_incident_journal_closed +- current_phase: P21 Runtime Incident Journal - latest_p7_selector_choice: continue_periodic_analysis_decision_monitoring_with_maturity_rollup - latest_p14_slice: webhook_transport_retry - latest_p15_pxx_tasklist_status: ready @@ -47,7 +47,12 @@ Automation execution-layer P20 is closed locally. P20 adds an operator digest/re - 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_p21_slice: runtime_incident_journal +- latest_runtime_incident_journal_status: implemented +- latest_runtime_incident_journal_route_status: ok +- latest_runtime_incident_journal_workbench_status: visible +- latest_runtime_incident_journal_smoke_status: idle_persisted +- next_execution_layer_phase: P22 Maintenance And Retention Rollup - latest_visual_review_script_status: idle - latest_runtime_visual_watch_task_status: ok - latest_visual_capture_cleanup_task_status: ok @@ -126,6 +131,9 @@ Automation execution-layer P20 is closed locally. P20 adds an operator digest/re - 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. +- Added `runtime_incidents` persistence with upsert aggregation by incident key. +- Added `RuntimeIncidentJournalService`, `scripts/run_runtime_incident_journal.py`, and `GET /trading/runtime-incidents`. +- Extended `runtime_status`, `run_trading_runtime.py`, and the workbench runtime panel with incident summary, severity/source counts, and recommended actions. ## Current Working State @@ -151,6 +159,8 @@ Automation execution-layer P20 is closed locally. P20 adds an operator digest/re - 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. +- P21 runtime incident journal is implemented: runtime/watchdog/blocker/noisy incident signals can be normalized into persisted incidents with severity, source, reason, subject, first/last seen timestamps, occurrence count, and recommended action. +- P21 visibility is implemented: `/trading/runtime-incidents`, `scripts/run_runtime_incident_journal.py`, `runtime_status.incident_summary`, `run_trading_runtime.py`, and the workbench runtime panel expose incident counts and recommendations. - 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`. @@ -160,7 +170,7 @@ Automation execution-layer P20 is closed locally. P20 adds an operator digest/re - 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. -## P20 Blockers +## P21 Blockers - none @@ -181,6 +191,7 @@ Automation execution-layer P20 is closed locally. P20 adds an operator digest/re - 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. +- P21 runtime incident journal remains observability/persistence 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`. @@ -195,11 +206,14 @@ Automation execution-layer P20 is closed locally. P20 adds an operator digest/re - 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. +- P21 local migration applied `043_create_runtime_incidents.sql`. +- P21 full validation passed: 691 unit tests, 69 script static checks, `node --check src/web/app.js`, and runtime smoke blockers=none with `Incident journal: idle / 0`. +- P21 script/API smoke returned idle persisted incident journal state, which is expected when no current runtime incidents are present. - Real-live order submission remains intentionally disabled because the OKX order adapter is still not a real-live submitter. ## Exact Next Step -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. +P21 is locally closed. Default next action is P22 `Maintenance And Retention Rollup`: aggregate retention cleanup, alert delivery/feedback, visual capture, incident journal, and watchdog health into a read-only maintenance summary. ## Validation Commands Used Recently @@ -222,4 +236,7 @@ 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 +D:\ai-exchange\.venv\Scripts\python.exe -m unittest tests.unit.test_runtime_incident_journal_service tests.unit.test_run_runtime_incident_journal tests.unit.test_trading_runtime_service tests.unit.test_api_static_server.ApiStaticServerTests.test_dashboard_uses_canonical_trading_routes +D:\ai-exchange\.venv\Scripts\python.exe scripts\apply_migrations.py +D:\ai-exchange\.venv\Scripts\python.exe scripts\run_runtime_incident_journal.py ``` diff --git a/TASKS.md b/TASKS.md index a3ffe82..9ede5e1 100644 --- a/TASKS.md +++ b/TASKS.md @@ -13,7 +13,7 @@ ## In Progress -- [x] P20 Operator Digest And Review Queue local closure complete +- [x] P21 Runtime Incident Journal local closure complete ## Next Up @@ -31,14 +31,15 @@ - [x] P18 local closure complete - [x] P19 local closure complete - [x] P20 local closure complete -- [ ] Start P21 Runtime Incident Journal +- [x] P21 local closure complete +- [ ] Start P22 Maintenance And Retention Rollup - [ ] 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_p20_operator_digest_review_queue_closed -- [x] current_phase: P20 Operator Digest And Review Queue +- [x] state: automation_execution_layer_p21_runtime_incident_journal_closed +- [x] current_phase: P21 Runtime Incident Journal - [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 @@ -70,7 +71,12 @@ - [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_p21_slice: runtime_incident_journal +- [x] latest_runtime_incident_journal_status: implemented +- [x] latest_runtime_incident_journal_route_status: ok +- [x] latest_runtime_incident_journal_workbench_status: visible +- [x] latest_runtime_incident_journal_smoke_status: idle_persisted +- [x] next_execution_layer_phase: P22 Maintenance And Retention Rollup - [x] latest_visual_review_script_status: idle - [x] latest_runtime_visual_watch_task_status: ok - [x] latest_visual_capture_cleanup_task_status: ok @@ -166,6 +172,13 @@ - [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_incidents` persistence with upsert aggregation by incident key +- [x] Added `RuntimeIncidentJournalService` +- [x] Added `scripts/run_runtime_incident_journal.py` +- [x] Added `GET /trading/runtime-incidents` +- [x] Added `runtime_status.incident_summary` +- [x] Added incident journal visibility to `run_trading_runtime.py` and the workbench runtime panel +- [x] Added focused P21 service, script, API, runtime, 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 @@ -190,6 +203,9 @@ - [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 +- [x] Applied local migration `043_create_runtime_incidents.sql` +- [x] Smoked P21 runtime incident journal script/API with idle persisted state +- [x] Completed P21 full validation: 691 unit tests, 69 script static checks, dashboard JS syntax, runtime incident script/API smoke, and runtime smoke blockers=none ## Blocked @@ -237,6 +253,11 @@ - [x] `automation_execution_layer_p15_pxx_tasklist_ready` - [x] `automation_execution_layer_p15_alert_noise_policy_matrix` - [x] `automation_execution_layer_p16_browser_screenshot_capture_adapter` +- [x] `automation_execution_layer_p17_visual_trigger_roi_router` +- [x] `automation_execution_layer_p18_visual_evidence_media_dispatch` +- [x] `automation_execution_layer_p19_multi_channel_alert_routing` +- [x] `automation_execution_layer_p20_operator_digest_review_queue` +- [x] `automation_execution_layer_p21_runtime_incident_journal` ## P15-PXX Finite Queue @@ -246,7 +267,7 @@ - [x] `P18` Visual Evidence Media Dispatch - [x] `P19` Multi-Channel Alert Routing - [x] `P20` Operator Digest And Review Queue -- [ ] `P21` Runtime Incident Journal +- [x] `P21` Runtime Incident Journal - [ ] `P22` Maintenance And Retention Rollup - [ ] `P23` End-To-End Unattended Harness - [ ] `P24` Remote Smoke And Ops Closure diff --git a/docs/AI_ICT_盯盘告警开发任务清单_v1.md b/docs/AI_ICT_盯盘告警开发任务清单_v1.md index 39bd660..0df7fc8 100644 --- a/docs/AI_ICT_盯盘告警开发任务清单_v1.md +++ b/docs/AI_ICT_盯盘告警开发任务清单_v1.md @@ -410,6 +410,7 @@ candidate -> decision alert -> dispatcher -> 外部消息 -> delivery record - 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` 面板暴露。 +- 2026-05-12 P21 已完成 `runtime incident journal` 本地闭环：runtime/watchdog/blocker/noisy incident 现在可聚合写入 `runtime_incidents`，每条 incident 带 severity、source、reason、first/last seen 与 recommended action，并通过脚本、API、runtime status 和 workbench runtime panel 暴露。 ### 验收 - 非重大错误不打断主循环 diff --git a/docs/AI_ICT_盯盘告警技术设计稿_v1.md b/docs/AI_ICT_盯盘告警技术设计稿_v1.md index 859ad3f..e6a7afe 100644 --- a/docs/AI_ICT_盯盘告警技术设计稿_v1.md +++ b/docs/AI_ICT_盯盘告警技术设计稿_v1.md @@ -114,6 +114,11 @@ 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/runtime/runtime_incident_journal_service.py` +职责：把 runtime status、watchdog、blocker/warning/noisy incident 规范化为可持久化 incident journal。 + +2026-05-12 P21 已落地：`runtime_incidents` 使用 `incident_key` 做去重聚合，并保存 severity、source、reason、subject、first/last seen、occurrence_count、recommended_action 与 payload。`GET /trading/runtime-incidents`、`scripts/run_runtime_incident_journal.py`、`runtime_status.incident_summary` 与 workbench runtime panel 均可查看当前 incident summary；该层只做可观测性与恢复建议，不改变任何交易权限或 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 65a2d8b..dfd3379 100644 --- a/docs/AI_ICT_自动化执行层_P15-PXX闭环任务清单_v1.md +++ b/docs/AI_ICT_自动化执行层_P15-PXX闭环任务清单_v1.md @@ -242,6 +242,10 @@ - 每条 incident 有 severity、source、reason、first_seen、last_seen、recommended_action。 - 环境问题被记录为环境 blocker，不被伪装成代码完成。 +- 2026-05-12 P21 已完成 `Runtime Incident Journal` 本地闭环：新增 `runtime_incidents` 表与 `RuntimeIncidentJournalService`，按 `incident_key` 聚合 runtime/watchdog/blocker/noisy incident，并保留 severity、source、reason、subject、first_seen、last_seen、occurrence_count 与 recommended_action。 +- 2026-05-12 P21 已新增 `scripts/run_runtime_incident_journal.py` 与 `GET /trading/runtime-incidents`，并将 `incident_summary` 接入 runtime status、`run_trading_runtime.py` 和 workbench runtime panel。 +- 2026-05-12 P21 本地闭环已验证：已应用 `043_create_runtime_incidents.sql`，focused P21 tests 通过，`tests/unit` 全量 691 个测试通过，script static checks 69 个脚本通过，`node --check src/web/app.js` 通过；脚本/API smoke 返回 idle persisted state，runtime smoke blockers=none 且显示 `Incident journal: idle / 0`。 + --- ## 12. P22 Maintenance And Retention Rollup diff --git a/migrations/043_create_runtime_incidents.sql b/migrations/043_create_runtime_incidents.sql new file mode 100644 index 0000000..e46a86e --- /dev/null +++ b/migrations/043_create_runtime_incidents.sql @@ -0,0 +1,21 @@ +create table if not exists runtime_incidents ( + id bigserial primary key, + incident_key text not null unique, + severity text not null, + source text not null, + reason text not null, + subject text not null default 'runtime', + status text not null default 'open', + first_seen_ts timestamptz not null, + last_seen_ts timestamptz not null, + occurrence_count integer not null default 1, + recommended_action text not null default '', + last_error text, + payload jsonb not null default '{}'::jsonb +); + +create index if not exists idx_runtime_incidents_last_seen_desc +on runtime_incidents (last_seen_ts desc, id desc); + +create index if not exists idx_runtime_incidents_severity_status +on runtime_incidents (severity, status, last_seen_ts desc); diff --git a/scripts/run_runtime_incident_journal.py b/scripts/run_runtime_incident_journal.py new file mode 100644 index 0000000..de6653c --- /dev/null +++ b/scripts/run_runtime_incident_journal.py @@ -0,0 +1,49 @@ +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.api.trading_runtime import get_trading_runtime_incidents + + +def main(env=None) -> int: + env = os.environ if env is None else env + limit = int(env.get("RUNTIME_INCIDENT_LIMIT", "50")) + refresh = _get_bool(env, "RUNTIME_INCIDENT_REFRESH", True) + output_json = _get_bool(env, "RUNTIME_INCIDENT_OUTPUT_JSON", False) + payload = get_trading_runtime_incidents(limit=limit, refresh=refresh) + if output_json: + print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str)) + else: + print(f"Runtime incident journal status: {payload.get('status', 'unknown')}") + print(f"Persisted: {payload.get('persisted')}") + print(f"Incidents: {payload.get('incident_count', 0)}") + print(f"Severity counts: {json.dumps(payload.get('severity_counts') or {}, sort_keys=True)}") + print(f"Recommended actions: {', '.join(payload.get('recommended_actions') or []) or 'none'}") + latest = payload.get("latest_incident") or {} + if latest: + print( + "Latest: " + f"{latest.get('severity')} {latest.get('source')} " + f"{latest.get('reason')} @ {latest.get('last_seen_ts')}" + ) + if payload.get("warnings"): + print(f"Warnings: {', '.join(payload.get('warnings') or [])}") + return 0 if payload.get("status") != "unavailable" else 2 + + +def _get_bool(env, key: str, default: bool) -> 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/scripts/run_trading_runtime.py b/scripts/run_trading_runtime.py index efbd74e..f09d22c 100644 --- a/scripts/run_trading_runtime.py +++ b/scripts/run_trading_runtime.py @@ -21,8 +21,13 @@ def main(env=None) -> int: if output_json: print(json.dumps(payload, ensure_ascii=False, sort_keys=True)) else: + incident_summary = dict(payload.get("incident_summary") or {}) print(f"Trading runtime: {payload.get('scheduler_state')} / {payload.get('supervisor_state', 'stopped')}") print(f"Blockers: {', '.join(payload.get('current_blockers') or []) or 'none'}") + print( + "Incident journal: " + f"{incident_summary.get('status', 'idle')} / {incident_summary.get('incident_count', 0)}" + ) return 0 if not payload.get("current_blockers") else 2 diff --git a/src/api/__init__.py b/src/api/__init__.py index 36b4ec3..af612c1 100644 --- a/src/api/__init__.py +++ b/src/api/__init__.py @@ -26,6 +26,7 @@ from src.api.trading_runtime import ( get_latest_trading_plan, get_trading_candidate_by_id, get_trading_market_state, + get_trading_runtime_incidents, get_trading_live_readiness, get_trading_launch_readiness, get_trading_operator_notes, @@ -98,6 +99,7 @@ __all__ = [ "get_trading_plan_by_id", "get_trading_runtime_status", "get_trading_runtime_watchdog", + "get_trading_runtime_incidents", "get_trading_trust_score", "get_trading_ticket_by_id", "get_trading_executions", diff --git a/src/api/server.py b/src/api/server.py index a1a7505..c68a66f 100644 --- a/src/api/server.py +++ b/src/api/server.py @@ -59,6 +59,7 @@ from src.api.trading_runtime import ( get_latest_trading_plan, get_trading_candidate_by_id, get_trading_market_state, + get_trading_runtime_incidents, get_trading_operator_notes, get_trading_plan_by_id, get_trading_review_packet, @@ -123,6 +124,7 @@ def build_api_dependencies() -> dict: "trading_reconciliation_run": run_trading_reconciliation, "trading_runtime_status": get_trading_runtime_status, "trading_runtime_watchdog": get_trading_runtime_watchdog, + "trading_runtime_incidents": get_trading_runtime_incidents, "trading_runtime_run_once": run_trading_runtime_once, "trading_trust_score": get_trading_trust_score, "trading_session_view": get_trading_session_view, @@ -255,6 +257,12 @@ def route_request( if method == "GET" and path == "/trading/runtime-watchdog": run_once = _get_single(query, "run_once", "0") return 200, deps["trading_runtime_watchdog"](run_once=str(run_once).strip().lower() in {"1", "true", "yes", "on"}) + if method == "GET" and path == "/trading/runtime-incidents": + refresh = _get_single(query, "refresh", "1") + return 200, deps["trading_runtime_incidents"]( + limit=_get_int(query, "limit", 50), + refresh=str(refresh).strip().lower() in {"1", "true", "yes", "on"}, + ) if method == "POST" and path == "/trading/runtime/run-once": return 200, deps["trading_runtime_run_once"]() if method == "GET" and path == "/trading/decision-packet": diff --git a/src/api/trading_runtime.py b/src/api/trading_runtime.py index 2400a9e..8efc782 100644 --- a/src/api/trading_runtime.py +++ b/src/api/trading_runtime.py @@ -17,6 +17,7 @@ from src.services.operator import OperatorNoteService, ReviewPacketService, Sess from src.services.reconciliation import BrokerExecutionReconciliationService from src.services.readiness import LaunchReadinessService from src.services.runtime import get_trading_runtime_service +from src.services.runtime.runtime_incident_journal_service import RuntimeIncidentJournalService from src.services.trade_plan_candidates import SetupWatcherService from src.services.trading_plans import ACTIVE_PLAN_STATUS, TradingPlanService from src.services.trust import ModelTrustService @@ -381,6 +382,13 @@ def get_trading_runtime_watchdog(run_once: bool = False) -> dict: return dict(payload.get("runtime_watchdog") or {}) +def get_trading_runtime_incidents(*, limit: int = 50, refresh: bool = True) -> dict: + journal = RuntimeIncidentJournalService() + if refresh: + return journal.build_from_runtime_status(get_trading_runtime_service().status(), limit=limit, persist=True) + return journal.build_summary(limit=limit) + + def get_trading_decision_packet(instrument_id: int) -> dict: return DecisionPacketService().build_preview(instrument_id=instrument_id) diff --git a/src/repositories/runtime_incident_repository.py b/src/repositories/runtime_incident_repository.py new file mode 100644 index 0000000..bceca79 --- /dev/null +++ b/src/repositories/runtime_incident_repository.py @@ -0,0 +1,108 @@ +import json + +from sqlalchemy import text + +from src.db.session import SessionLocal + + +class RuntimeIncidentRepository: + def upsert_incident( + self, + *, + incident_key: str, + severity: str, + source: str, + reason: str, + subject: str, + observed_ts, + recommended_action: str, + last_error: str | None = None, + payload: dict | None = None, + ) -> dict: + sql = text( + """ + insert into runtime_incidents ( + incident_key, + severity, + source, + reason, + subject, + first_seen_ts, + last_seen_ts, + occurrence_count, + recommended_action, + last_error, + payload + ) values ( + :incident_key, + :severity, + :source, + :reason, + :subject, + :observed_ts, + :observed_ts, + 1, + :recommended_action, + :last_error, + cast(:payload as jsonb) + ) + on conflict (incident_key) do update set + severity = excluded.severity, + source = excluded.source, + reason = excluded.reason, + subject = excluded.subject, + status = 'open', + last_seen_ts = excluded.last_seen_ts, + occurrence_count = runtime_incidents.occurrence_count + 1, + recommended_action = excluded.recommended_action, + last_error = excluded.last_error, + payload = excluded.payload + returning id, incident_key, severity, source, reason, subject, status, + first_seen_ts, last_seen_ts, occurrence_count, + recommended_action, last_error, payload + """ + ) + params = { + "incident_key": incident_key, + "severity": severity, + "source": source, + "reason": reason, + "subject": subject, + "observed_ts": observed_ts, + "recommended_action": recommended_action, + "last_error": last_error, + "payload": json.dumps(payload or {}, default=str, ensure_ascii=False, sort_keys=True), + } + with SessionLocal.begin() as session: + row = session.execute(sql, params).mappings().one() + return dict(row) + + def fetch_recent_incidents(self, *, limit: int = 50, status: str | None = None) -> list[dict]: + if status: + sql = text( + """ + select id, incident_key, severity, source, reason, subject, status, + first_seen_ts, last_seen_ts, occurrence_count, + recommended_action, last_error, payload + from runtime_incidents + where status = :status + order by last_seen_ts desc, id desc + limit :limit + """ + ) + params = {"status": status, "limit": limit} + else: + sql = text( + """ + select id, incident_key, severity, source, reason, subject, status, + first_seen_ts, last_seen_ts, occurrence_count, + recommended_action, last_error, payload + from runtime_incidents + order by last_seen_ts desc, id desc + limit :limit + """ + ) + params = {"limit": limit} + with SessionLocal() as session: + rows = session.execute(sql, params).mappings().all() + return [dict(row) for row in rows] diff --git a/src/services/runtime/__init__.py b/src/services/runtime/__init__.py index 888b0f4..b7ed171 100644 --- a/src/services/runtime/__init__.py +++ b/src/services/runtime/__init__.py @@ -1,4 +1,5 @@ from src.services.runtime.runtime_task_watchdog_service import RuntimeTaskWatchdogService +from src.services.runtime.runtime_incident_journal_service import RuntimeIncidentJournalService from src.services.runtime.trading_runtime_service import ( TradingRuntimeScheduler, TradingRuntimeSupervisor, @@ -8,6 +9,7 @@ from src.services.runtime.trading_runtime_service import ( __all__ = [ "RuntimeTaskWatchdogService", + "RuntimeIncidentJournalService", "TradingRuntimeScheduler", "TradingRuntimeSupervisor", "get_trading_runtime_service", diff --git a/src/services/runtime/runtime_incident_journal_service.py b/src/services/runtime/runtime_incident_journal_service.py new file mode 100644 index 0000000..7864c4f --- /dev/null +++ b/src/services/runtime/runtime_incident_journal_service.py @@ -0,0 +1,304 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Optional + +from sqlalchemy.exc import SQLAlchemyError + +from src.repositories.runtime_incident_repository import RuntimeIncidentRepository + + +class RuntimeIncidentJournalService: + def __init__(self, *, repository: Optional[RuntimeIncidentRepository] = None) -> None: + self.repository = repository or RuntimeIncidentRepository() + + def record_incident( + self, + *, + severity: str, + source: str, + reason: str, + subject: str = "runtime", + observed_ts=None, + recommended_action: str | None = None, + last_error: str | None = None, + payload: dict | None = None, + ) -> dict: + incident = _normalize_incident( + { + "severity": severity, + "source": source, + "reason": reason, + "subject": subject, + "observed_ts": observed_ts, + "recommended_action": recommended_action, + "last_error": last_error, + "payload": payload or {}, + } + ) + return self.repository.upsert_incident(**incident) + + def build_from_runtime_status(self, runtime_status: dict | None, *, limit: int = 50, persist: bool = True) -> dict: + status = dict(runtime_status or {}) + incidents = _dedupe_incidents(_extract_runtime_incidents(status)) + if not persist: + return _build_summary(incidents[:limit], persisted=False, volatile_count=len(incidents)) + try: + for incident in incidents: + self.repository.upsert_incident(**incident) + rows = self.repository.fetch_recent_incidents(limit=limit) + except SQLAlchemyError: + summary = _build_summary(incidents[:limit], persisted=False, volatile_count=len(incidents)) + summary["status"] = "unavailable" + summary["warnings"].append("runtime_incident_journal_unavailable") + return summary + return _build_summary(rows, persisted=True, volatile_count=len(incidents)) + + def build_summary(self, *, limit: int = 50) -> dict: + try: + rows = self.repository.fetch_recent_incidents(limit=limit) + except SQLAlchemyError: + return { + "status": "unavailable", + "generated_ts": datetime.now(timezone.utc).isoformat(), + "persisted": False, + "incident_count": 0, + "open_incident_count": 0, + "volatile_incident_count": 0, + "severity_counts": {}, + "source_counts": {}, + "recommended_actions": [], + "latest_incident": None, + "items": [], + "warnings": ["runtime_incident_journal_unavailable"], + } + return _build_summary(rows, persisted=True, volatile_count=0) + + +def _extract_runtime_incidents(status: dict) -> list[dict]: + incidents: list[dict] = [] + for item in list(status.get("incidents") or []): + reason = str(item.get("code") or item.get("reason") or "").strip() + if not reason: + continue + incidents.append( + _normalize_incident( + { + "severity": item.get("severity") or _severity_from_runtime_bucket(reason, status), + "source": item.get("task_name") or _source_from_reason(reason), + "reason": reason, + "subject": item.get("symbol") or item.get("subject") or "runtime", + "observed_ts": item.get("observed_ts"), + "last_error": item.get("error"), + "payload": {"runtime_incident": item}, + } + ) + ) + for severity, key in ( + ("blocker", "current_blockers"), + ("warning", "current_warnings"), + ("noisy_incident", "current_noisy_incidents"), + ): + for reason in list(status.get(key) or []): + incidents.append( + _normalize_incident( + { + "severity": severity, + "source": _source_from_reason(reason), + "reason": reason, + "subject": _subject_from_reason(reason), + "payload": {"runtime_bucket": key}, + } + ) + ) + watchdog = dict(status.get("runtime_watchdog") or {}) + for item in list(watchdog.get("stalled_tasks") or []): + reason = f"runtime_task_stalled:{item.get('name', 'unknown')}:{item.get('reason', 'unknown')}" + incidents.append( + _normalize_incident( + { + "severity": "warning", + "source": "runtime_watchdog", + "reason": reason, + "subject": str(item.get("name") or "runtime"), + "observed_ts": watchdog.get("checked_ts"), + "payload": {"stalled_task": item}, + } + ) + ) + return incidents + + +def _normalize_incident(item: dict) -> dict: + severity = _normalize_severity(item.get("severity")) + source = str(item.get("source") or _source_from_reason(item.get("reason")) or "runtime").strip() + reason = str(item.get("reason") or "runtime_incident").strip() + subject = str(item.get("subject") or _subject_from_reason(reason) or "runtime").strip() + observed_ts = _parse_ts(item.get("observed_ts")) or datetime.now(timezone.utc) + recommended_action = str(item.get("recommended_action") or _recommended_action(reason, source)).strip() + return { + "incident_key": f"{source}:{subject}:{reason}", + "severity": severity, + "source": source, + "reason": reason, + "subject": subject, + "observed_ts": observed_ts, + "recommended_action": recommended_action, + "last_error": item.get("last_error") or item.get("error"), + "payload": dict(item.get("payload") or {}), + } + + +def _dedupe_incidents(incidents: list[dict]) -> list[dict]: + by_key: dict[str, dict] = {} + for incident in incidents: + key = incident["incident_key"] + existing = by_key.get(key) + if existing is None: + by_key[key] = incident + continue + if _severity_rank(incident["severity"]) > _severity_rank(existing["severity"]): + existing["severity"] = incident["severity"] + existing["observed_ts"] = max(existing["observed_ts"], incident["observed_ts"]) + if incident.get("last_error"): + existing["last_error"] = incident.get("last_error") + return list(by_key.values()) + + +def _build_summary(rows: list[dict], *, persisted: bool, volatile_count: int) -> dict: + items = [_serialize_incident(row) for row in rows] + severity_counts: dict[str, int] = {} + source_counts: dict[str, int] = {} + recommended_actions: list[str] = [] + for item in items: + severity = str(item.get("severity") or "unknown") + source = str(item.get("source") or "runtime") + severity_counts[severity] = severity_counts.get(severity, 0) + 1 + source_counts[source] = source_counts.get(source, 0) + 1 + action = str(item.get("recommended_action") or "").strip() + if action and action not in recommended_actions: + recommended_actions.append(action) + return { + "status": "completed" if items else "idle", + "generated_ts": datetime.now(timezone.utc).isoformat(), + "persisted": persisted, + "incident_count": len(items), + "open_incident_count": sum(1 for item in items if item.get("status", "open") == "open"), + "volatile_incident_count": volatile_count, + "severity_counts": severity_counts, + "source_counts": source_counts, + "recommended_actions": recommended_actions[:10], + "latest_incident": items[0] if items else None, + "items": items, + "warnings": [], + } + + +def _serialize_incident(row: dict) -> dict: + return { + "id": row.get("id"), + "incident_key": row.get("incident_key"), + "severity": row.get("severity"), + "source": row.get("source"), + "reason": row.get("reason"), + "subject": row.get("subject"), + "status": row.get("status", "open"), + "first_seen_ts": _iso(row.get("first_seen_ts") or row.get("observed_ts")), + "last_seen_ts": _iso(row.get("last_seen_ts") or row.get("observed_ts")), + "occurrence_count": int(row.get("occurrence_count") or 1), + "recommended_action": row.get("recommended_action"), + "last_error": row.get("last_error"), + "payload": dict(row.get("payload") or {}), + } + + +def _recommended_action(reason: str, source: str) -> str: + text = f"{source}:{reason}".lower() + if "runtime_task_stalled" in text: + return "inspect_runtime_task_and_restart_if_stale" + if "alert_dispatch" in text or "retry_exhausted" in text: + return "inspect_alert_transport_policy_and_credentials" + if "visual_capture" in text or "capture_unavailable" in text: + return "inspect_browser_capture_or_visual_artifact_path" + if "channel_disabled" in text or "credential_missing" in text: + return "configure_channel_credentials_or_keep_workbench_only" + if "broker" in text or "okx" in text: + return "check_broker_readonly_connectivity_and_credentials" + if "market_stream" in text or "market_data" in text: + return "check_market_truth_stream_proxy_and_backfill" + if "database" in text or "db_" in text: + return "restore_database_connectivity_before_claiming_runtime_closed" + return "review_runtime_logs_and_clear_after_recovery" + + +def _source_from_reason(reason) -> str: + text = str(reason or "runtime") + if text.startswith("runtime_task_stalled"): + return "runtime_watchdog" + if text.startswith("alert_dispatch"): + return "alert_dispatch" + if text.startswith("visual_capture"): + return "visual_capture" + if text.startswith("visual_watch"): + return "visual_watch" + if text.startswith("broker"): + return "broker_sync" + if text.startswith("reconciliation"): + return "reconciliation" + if text.startswith("market_"): + return "market_state" + if text.startswith("watch_"): + return "watch" + if text.startswith("research_council"): + return "research_council" + if text.startswith("decision_packet"): + return "decision_packet" + return text.split(":", 1)[0] or "runtime" + + +def _subject_from_reason(reason) -> str: + parts = str(reason or "").split(":") + if len(parts) >= 2 and parts[1]: + return parts[1] + return "runtime" + + +def _severity_from_runtime_bucket(reason: str, status: dict) -> str: + if reason in set(status.get("current_blockers") or []): + return "blocker" + if reason in set(status.get("current_warnings") or []): + return "warning" + if reason in set(status.get("current_noisy_incidents") or []): + return "noisy_incident" + return "warning" + + +def _normalize_severity(value) -> str: + text = str(value or "warning").strip() + if text in {"blocker", "warning", "noisy_incident", "info"}: + return text + return "warning" + + +def _severity_rank(value: str) -> int: + return {"info": 0, "noisy_incident": 1, "warning": 2, "blocker": 3}.get(value, 2) + + +def _parse_ts(value) -> datetime | None: + if value in (None, ""): + return None + if isinstance(value, datetime): + return value.astimezone(timezone.utc) if value.tzinfo else value.replace(tzinfo=timezone.utc) + try: + parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except ValueError: + return None + return parsed.astimezone(timezone.utc) if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) + + +def _iso(value) -> str | None: + if value in (None, ""): + return None + if isinstance(value, datetime): + return value.astimezone(timezone.utc).isoformat() + return str(value) diff --git a/src/services/runtime/trading_runtime_service.py b/src/services/runtime/trading_runtime_service.py index 29712c7..1db8fdd 100644 --- a/src/services/runtime/trading_runtime_service.py +++ b/src/services/runtime/trading_runtime_service.py @@ -13,6 +13,7 @@ from src.services.decision import DecisionPacketService from src.services.market_state import MarketStateService, get_market_truth_supervisor from src.services.reconciliation import BrokerExecutionReconciliationService from src.services.research_council import ResearchCouncilService +from src.services.runtime.runtime_incident_journal_service import RuntimeIncidentJournalService from src.services.runtime.runtime_task_watchdog_service import RuntimeTaskWatchdogService from src.services.trade_plan_candidates import SetupWatcherService from src.services.trading_plans import TradingPlanService @@ -66,6 +67,7 @@ class TradingRuntimeScheduler: visual_review_service: Optional[VisualReviewService] = None, visual_capture_retention_service: Optional[VisualCaptureRetentionService] = None, runtime_task_watchdog_service: Optional[RuntimeTaskWatchdogService] = None, + runtime_incident_journal_service: Optional[RuntimeIncidentJournalService] = None, ) -> None: self.market_state_service = market_state_service or MarketStateService(market_truth_supervisor=get_market_truth_supervisor()) self.trading_plan_service = trading_plan_service or TradingPlanService() @@ -83,6 +85,7 @@ class TradingRuntimeScheduler: stale_multiplier=settings.runtime_task_stale_multiplier, min_stuck_seconds=settings.runtime_task_stuck_min_seconds, ) + self.runtime_incident_journal_service = runtime_incident_journal_service or RuntimeIncidentJournalService() self.symbols = [item.strip() for item in settings.trading_runtime_symbols.split(",") if item.strip()] self.bias_tf = settings.trading_runtime_bias_timeframe self.setup_tf = settings.trading_runtime_setup_timeframe @@ -194,7 +197,7 @@ class TradingRuntimeScheduler: "is_degraded": bool(self.current_blockers or current_warnings), "codes": [*self.current_blockers, *current_warnings], } - return { + snapshot = { "enabled": bool(settings.trading_runtime_enabled), "scheduler_state": "running" if settings.trading_runtime_enabled else "disabled", "symbols": list(self.symbols), @@ -214,6 +217,12 @@ class TradingRuntimeScheduler: "latest_market_state_by_symbol": dict(self.latest_market_state_by_symbol), "latest_broker_state_by_symbol": dict(self.latest_broker_state_by_symbol), } + snapshot["incident_summary"] = self.runtime_incident_journal_service.build_from_runtime_status( + snapshot, + limit=50, + persist=False, + ) + return snapshot def _run_plan_task(self, *, symbol: str, instrument_id: int, now: datetime, blockers: list[str]) -> None: task = self.tasks["plan"] diff --git a/src/web/app.js b/src/web/app.js index 7dcf106..4584170 100644 --- a/src/web/app.js +++ b/src/web/app.js @@ -284,12 +284,13 @@ function renderMarketState(payload, executions) { marketStatePanel.appendChild(panel); } -function renderRuntimeStatus(runtimeStatus) { +function renderRuntimeStatus(runtimeStatus, runtimeIncidents) { clearNode(runtimeStatusPanel); const tasks = runtimeStatus?.tasks || {}; const blockers = runtimeStatus?.current_blockers || []; const incidents = runtimeStatus?.incidents || []; const watchdog = runtimeStatus?.runtime_watchdog || {}; + const incidentSummary = runtimeIncidents || runtimeStatus?.incident_summary || {}; const stalledTasks = watchdog?.stalled_tasks || []; const stalledMap = new Map(stalledTasks.map((item) => [item.name, item])); runtimeStatusPanel.appendChild( @@ -302,10 +303,25 @@ function renderRuntimeStatus(runtimeStatus) { ['blockers', formatList(blockers)], ['stalled tasks', formatList(watchdog?.stalled_task_names || [])], ['watchdog checked', formatTimestamp(watchdog?.checked_ts)], + ['incident journal', `${incidentSummary.status || 'idle'} | ${incidentSummary.incident_count ?? 0}`], + ['incident severity', formatJsonInline(incidentSummary.severity_counts || {})], ['last incident', incidents.length ? `${incidents[incidents.length - 1].code} @ ${formatTimestamp(incidents[incidents.length - 1].observed_ts)}` : 'none'], ]) ); const grid = createElement('div', 'cards'); + grid.appendChild( + buildItemCard( + 'Incident Journal', + incidentSummary.open_incident_count ? 'review' : (incidentSummary.status || 'idle'), + [ + ['persisted', String(Boolean(incidentSummary.persisted))], + ['open / volatile', `${incidentSummary.open_incident_count ?? 0} / ${incidentSummary.volatile_incident_count ?? 0}`], + ['sources', formatJsonInline(incidentSummary.source_counts || {})], + ['recommended actions', formatList(incidentSummary.recommended_actions || [])], + ['latest', incidentSummary.latest_incident ? `${incidentSummary.latest_incident.severity || 'unknown'} ${incidentSummary.latest_incident.reason || 'runtime_incident'}` : 'none'], + ] + ) + ); Object.entries(tasks).forEach(([name, task]) => { const stalled = stalledMap.get(name) || null; grid.appendChild( @@ -1309,6 +1325,7 @@ function renderDashboardData({ tradingPlan, marketState, runtimeStatus, + runtimeIncidents, decisionPacket, decisionAlerts, visualReviews, @@ -1332,7 +1349,7 @@ function renderDashboardData({ renderHealth(health); renderOverview(weeklyReview, tradeCandidates.items || [], executionTickets.items || [], tradingExecutions.items || []); renderMarketState(marketState, tradingExecutions.items || []); - renderRuntimeStatus(runtimeStatus); + renderRuntimeStatus(runtimeStatus, runtimeIncidents); renderDecisionPacket(decisionPacket, decisionAlerts); renderTrustScore(trustScore); renderLiveReadiness(liveReadiness); @@ -1358,12 +1375,13 @@ async function refreshDashboard() { const limit = limitInput.value || '20'; try { - 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([ + const [health, candles, tradingPlan, marketState, runtimeStatus, runtimeIncidents, 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), fetchJson(`/trading/market-state?${queryString({ instrument_id: instrumentId, timeframe })}`).catch(() => ({ market_state: {} })), fetchJson('/trading/runtime-status').catch(() => ({ scheduler_state: 'unavailable', tasks: {}, current_blockers: ['runtime_status_unavailable'] })), + fetchJson(`/trading/runtime-incidents?${queryString({ limit: 20, refresh: 0 })}`).catch(() => ({ status: 'unavailable', persisted: false, incident_count: 0, severity_counts: {}, source_counts: {}, recommended_actions: [], items: [] })), 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' })), @@ -1392,6 +1410,7 @@ async function refreshDashboard() { tradingPlan, marketState, runtimeStatus, + runtimeIncidents, decisionPacket, decisionAlerts, alertQualitySummary, diff --git a/tests/unit/test_api_static_server.py b/tests/unit/test_api_static_server.py index e81b17f..40240e7 100644 --- a/tests/unit/test_api_static_server.py +++ b/tests/unit/test_api_static_server.py @@ -102,6 +102,7 @@ class ApiStaticServerTests(unittest.TestCase): "/trading/candidates", "/trading/alerts/quality-summary", "/trading/alerts/operator-digest", + "/trading/runtime-incidents", "/trading/visual-watch/cleanup-captures", "/trading/tickets", "/trading/executions", diff --git a/tests/unit/test_run_runtime_incident_journal.py b/tests/unit/test_run_runtime_incident_journal.py new file mode 100644 index 0000000..73ad814 --- /dev/null +++ b/tests/unit/test_run_runtime_incident_journal.py @@ -0,0 +1,51 @@ +import contextlib +import io +import json +import unittest +from unittest.mock import patch + +from scripts import run_runtime_incident_journal + + +class RunRuntimeIncidentJournalTests(unittest.TestCase): + def test_main_outputs_json_journal(self) -> None: + payload = { + "status": "completed", + "persisted": True, + "incident_count": 1, + "severity_counts": {"blocker": 1}, + "recommended_actions": ["check_market_truth_stream_proxy_and_backfill"], + "latest_incident": {"severity": "blocker", "source": "market_state", "reason": "market_stream_degraded"}, + "items": [], + "warnings": [], + } + output = io.StringIO() + with patch("scripts.run_runtime_incident_journal.get_trading_runtime_incidents", return_value=payload) as journal: + with contextlib.redirect_stdout(output): + exit_code = run_runtime_incident_journal.main( + env={ + "RUNTIME_INCIDENT_OUTPUT_JSON": "1", + "RUNTIME_INCIDENT_LIMIT": "7", + "RUNTIME_INCIDENT_REFRESH": "0", + } + ) + + self.assertEqual(exit_code, 0) + journal.assert_called_once_with(limit=7, refresh=False) + self.assertEqual(json.loads(output.getvalue())["severity_counts"]["blocker"], 1) + + def test_main_returns_nonzero_when_journal_unavailable(self) -> None: + output = io.StringIO() + with patch( + "scripts.run_runtime_incident_journal.get_trading_runtime_incidents", + return_value={"status": "unavailable", "warnings": ["runtime_incident_journal_unavailable"]}, + ): + with contextlib.redirect_stdout(output): + exit_code = run_runtime_incident_journal.main(env={}) + + self.assertEqual(exit_code, 2) + self.assertIn("runtime_incident_journal_unavailable", output.getvalue()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_runtime_incident_journal_service.py b/tests/unit/test_runtime_incident_journal_service.py new file mode 100644 index 0000000..8f45402 --- /dev/null +++ b/tests/unit/test_runtime_incident_journal_service.py @@ -0,0 +1,102 @@ +import unittest + +from sqlalchemy.exc import OperationalError + +from src.services.runtime.runtime_incident_journal_service import RuntimeIncidentJournalService + + +class _IncidentRepository: + def __init__(self) -> None: + self.rows_by_key = {} + self.upserts = [] + + def upsert_incident(self, **kwargs): + self.upserts.append(kwargs) + row = dict(kwargs) + existing = self.rows_by_key.get(kwargs["incident_key"]) + row["id"] = existing.get("id", len(self.rows_by_key) + 1) if existing else len(self.rows_by_key) + 1 + row["status"] = "open" + row["first_seen_ts"] = existing.get("first_seen_ts", kwargs["observed_ts"]) if existing else kwargs["observed_ts"] + row["last_seen_ts"] = kwargs["observed_ts"] + row["occurrence_count"] = int(existing.get("occurrence_count", 0)) + 1 if existing else 1 + self.rows_by_key[kwargs["incident_key"]] = row + return dict(row) + + def fetch_recent_incidents(self, *, limit: int = 50, status=None): + rows = sorted(self.rows_by_key.values(), key=lambda item: str(item["last_seen_ts"]), reverse=True) + return [dict(row) for row in rows[:limit]] + + +class RuntimeIncidentJournalServiceTests(unittest.TestCase): + def test_build_from_runtime_status_persists_deduped_incidents_with_actions(self) -> None: + repository = _IncidentRepository() + service = RuntimeIncidentJournalService(repository=repository) + runtime_status = { + "current_blockers": ["market_stream_degraded:BTC-USDT:market_stream_disconnected"], + "current_warnings": [], + "current_noisy_incidents": ["watch_skipped_market_degraded:BTC-USDT"], + "incidents": [ + { + "task_name": "market_state", + "symbol": "BTC-USDT", + "code": "market_stream_degraded:BTC-USDT:market_stream_disconnected", + "severity": "blocker", + "observed_ts": "2026-05-12T01:00:00+00:00", + } + ], + "runtime_watchdog": { + "checked_ts": "2026-05-12T01:01:00+00:00", + "stalled_tasks": [{"name": "watch", "reason": "running_too_long", "age_seconds": 300}], + }, + } + + summary = service.build_from_runtime_status(runtime_status, limit=10, persist=True) + + self.assertEqual(summary["status"], "completed") + self.assertTrue(summary["persisted"]) + self.assertEqual(len(repository.upserts), 3) + self.assertEqual(summary["severity_counts"]["blocker"], 1) + self.assertEqual(summary["severity_counts"]["warning"], 1) + self.assertEqual(summary["severity_counts"]["noisy_incident"], 1) + self.assertIn("check_market_truth_stream_proxy_and_backfill", summary["recommended_actions"]) + self.assertIn("inspect_runtime_task_and_restart_if_stale", summary["recommended_actions"]) + first = summary["items"][0] + self.assertIn("severity", first) + self.assertIn("source", first) + self.assertIn("reason", first) + self.assertIn("first_seen_ts", first) + self.assertIn("last_seen_ts", first) + self.assertIn("recommended_action", first) + + def test_build_from_runtime_status_is_idle_without_incidents(self) -> None: + service = RuntimeIncidentJournalService(repository=_IncidentRepository()) + + summary = service.build_from_runtime_status({"incidents": [], "runtime_watchdog": {}}, persist=True) + + self.assertEqual(summary["status"], "idle") + self.assertEqual(summary["incident_count"], 0) + self.assertEqual(summary["items"], []) + + def test_build_from_runtime_status_reports_repository_unavailable(self) -> None: + class _BrokenRepository: + def upsert_incident(self, **kwargs): + raise OperationalError("stmt", {}, Exception()) + + def fetch_recent_incidents(self, *, limit: int = 50, status=None): + return [] + + service = RuntimeIncidentJournalService(repository=_BrokenRepository()) + + summary = service.build_from_runtime_status( + {"current_blockers": ["db_unavailable:runtime"], "incidents": []}, + persist=True, + ) + + self.assertEqual(summary["status"], "unavailable") + self.assertFalse(summary["persisted"]) + self.assertEqual(summary["volatile_incident_count"], 1) + self.assertIn("runtime_incident_journal_unavailable", summary["warnings"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_trading_runtime_service.py b/tests/unit/test_trading_runtime_service.py index feea733..bcc55c4 100644 --- a/tests/unit/test_trading_runtime_service.py +++ b/tests/unit/test_trading_runtime_service.py @@ -249,6 +249,7 @@ class TradingRuntimeSchedulerTests(unittest.TestCase): self.assertEqual(payload["tasks"]["alert_dispatch"]["details"]["channel_disabled"], 0) self.assertIn("enabled_channels", payload["tasks"]["alert_dispatch"]["details"]["channel_health"]) self.assertEqual(payload["runtime_watchdog"]["stalled_task_count"], 0) + self.assertEqual(payload["incident_summary"]["status"], "idle") self.assertEqual(payload["latest_broker_state_by_symbol"]["BTC-USDT"]["account"]["equity"], 1000.0) self.assertTrue(payload["tasks"]["watch"]["next_run_ts"]) @@ -274,6 +275,8 @@ class TradingRuntimeSchedulerTests(unittest.TestCase): severities = {item["severity"] for item in payload["incidents"]} self.assertIn("blocker", severities) self.assertIn("noisy_incident", severities) + self.assertGreaterEqual(payload["incident_summary"]["severity_counts"]["blocker"], 1) + self.assertIn("check_market_truth_stream_proxy_and_backfill", payload["incident_summary"]["recommended_actions"]) def test_supervisor_run_once_exposes_supervisor_state(self) -> None: scheduler = TradingRuntimeScheduler( @@ -318,6 +321,7 @@ class TradingRuntimeSchedulerTests(unittest.TestCase): self.assertEqual(payload["runtime_watchdog"]["stalled_task_count"], 1) self.assertIn("runtime_task_stalled:watch:running_too_long", payload["current_warnings"]) + self.assertEqual(payload["incident_summary"]["source_counts"]["runtime_watchdog"], 1) class TradingRuntimeApiTests(unittest.TestCase): @@ -362,6 +366,23 @@ class TradingRuntimeApiTests(unittest.TestCase): self.assertEqual(status, 200) self.assertTrue(payload["run_once"]) + def test_routes_runtime_incidents(self) -> None: + status, payload = route_request( + "/trading/runtime-incidents", + {"limit": ["7"], "refresh": ["0"]}, + dependencies={ + "trading_runtime_incidents": lambda **kwargs: { + "status": "completed", + "limit": kwargs["limit"], + "refresh": kwargs["refresh"], + } + }, + ) + + self.assertEqual(status, 200) + self.assertEqual(payload["limit"], 7) + self.assertFalse(payload["refresh"]) + class TradingRuntimeScriptTests(unittest.TestCase): def test_script_outputs_json(self) -> None: