Close P23 unattended harness

SESSION.md

@@ -2,7 +2,7 @@
 
 ## Current Goal
 
-Automation execution-layer P22 is closed locally. P22 adds a read-only maintenance summary that rolls up visual capture retention, alert delivery/feedback health, runtime watchdog state, and incident journal counts into P23-ready assertions. The default next action is P23 `End-To-End Unattended Harness`.
+Automation execution-layer P23 is closed locally. P23 adds a repeatable analysis/decision-only unattended harness that validates P15-P22 key paths and emits a single blockers/warnings/passed-checks report for local and future remote smoke. The default next action is P24 `Remote Smoke And Ops Closure`.
 
 ## Current Branch
 
@@ -14,8 +14,8 @@ Automation execution-layer P22 is closed locally. P22 adds a read-only maintenan
 
 ## Automation Status
 
-- state: automation_execution_layer_p22_maintenance_retention_rollup_closed
+- state: automation_execution_layer_p23_end_to_end_unattended_harness_closed
-- current_phase: P22 Maintenance And Retention Rollup
+- current_phase: P23 End-To-End Unattended Harness
 - latest_p7_selector_choice: continue_periodic_analysis_decision_monitoring_with_maturity_rollup
 - latest_p14_slice: webhook_transport_retry
 - latest_p15_pxx_tasklist_status: ready
@@ -57,7 +57,11 @@ Automation execution-layer P22 is closed locally. P22 adds a read-only maintenan
 - latest_maintenance_summary_route_status: ok
 - latest_maintenance_summary_workbench_status: visible
 - latest_maintenance_summary_smoke_status: warning_feedback_sample_health
-- next_execution_layer_phase: P23 End-To-End Unattended Harness
+- latest_p23_slice: end_to_end_unattended_harness
+- latest_unattended_harness_status: implemented
+- latest_unattended_harness_route_status: ok
+- latest_unattended_harness_smoke_status: passed_9_checks
+- next_execution_layer_phase: P24 Remote Smoke And Ops Closure
 - latest_visual_review_script_status: idle
 - latest_runtime_visual_watch_task_status: ok
 - latest_visual_capture_cleanup_task_status: ok
@@ -142,6 +146,8 @@ Automation execution-layer P22 is closed locally. P22 adds a read-only maintenan
 - Added `MaintenanceSummaryService` for read-only storage/delivery/feedback/runtime maintenance rollups.
 - Added `scripts/run_maintenance_summary.py` and `GET /trading/maintenance-summary`.
 - Extended the workbench runtime panel with maintenance status, blockers/warnings, actions, and P23 assertion visibility.
+- Added `UnattendedHarnessService`, `scripts/run_unattended_harness.py`, and `GET /trading/unattended-harness`.
+- The harness validates policy suppression, dispatch retry contract, runtime watchdog, visual capture fallback, alert quality feedback rollup, visual media/channel rollup, operator digest, maintenance summary, and runtime environment blockers without using external channels or real trades.
 
 ## Current Working State
 
@@ -171,6 +177,8 @@ Automation execution-layer P22 is closed locally. P22 adds a read-only maintenan
 - 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.
 - P22 maintenance retention rollup is implemented: visual capture retention dry-run stats, alert delivery health, feedback sample health, runtime watchdog, and incident journal health are aggregated into `checks`, `blockers`, `warnings`, `maintenance_actions`, and `p23_assertions`.
 - P22 visibility is implemented: `/trading/maintenance-summary`, `scripts/run_maintenance_summary.py`, and the workbench runtime panel expose maintenance state without deleting business data.
+- P23 unattended harness is implemented: a single local command/API call returns mode, passed checks, warnings, blockers, and trimmed payloads across P15-P22 key paths.
+- P23 is analysis/decision-only: `external_channels_used=false` and `real_trade_required=false`.
 - 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`.
@@ -180,7 +188,7 @@ Automation execution-layer P22 is closed locally. P22 adds a read-only maintenan
 - 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.
 
-## P22 Blockers
+## P23 Blockers
 
 - none
 
@@ -203,6 +211,7 @@ Automation execution-layer P22 is closed locally. P22 adds a read-only maintenan
 - 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.
 - P22 maintenance summary remains read-only and uses visual capture retention dry-run statistics only; it must not delete business data or alter execution permission, live readiness, trust bucket, risk governor, or operator confirmation.
+- P23 unattended harness remains a local/remote smoke verifier only and must not use external channels, require real trades, or alter 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`.
@@ -222,11 +231,13 @@ Automation execution-layer P22 is closed locally. P22 adds a read-only maintenan
 - P21 script/API smoke returned idle persisted incident journal state, which is expected when no current runtime incidents are present.
 - P22 full validation passed: 696 unit tests, 70 script static checks, and `node --check src/web/app.js`.
 - P22 script/API smoke returned `status=warning` with `warnings=['feedback.sample_health']`, blockers=[], automatic_actions=[], and P23 assertions visible.
+- P23 full validation passed: 701 unit tests, 71 script static checks, and `node --check src/web/app.js`.
+- P23 script/API smoke returned `status=passed`, passed_checks=9, warnings=[], blockers=[], `external_channels_used=false`, and `real_trade_required=false`.
 - Real-live order submission remains intentionally disabled because the OKX order adapter is still not a real-live submitter.
 
 ## Exact Next Step
 
-P22 is locally closed. Default next action is P23 `End-To-End Unattended Harness`: build a repeatable local harness covering P15-P22 key paths and producing a single JSON report with blockers, warnings, and passed checks.
+P23 is locally closed. Default next action is P24 `Remote Smoke And Ops Closure`: deploy current HEAD when the remote environment is reachable, apply migrations, and smoke `/health`, runtime status, review packet, alert quality, visual watch, maintenance summary, and the P23 harness.
 
 ## Validation Commands Used Recently
 
@@ -254,4 +265,6 @@ D:\ai-exchange\.venv\Scripts\python.exe scripts\apply_migrations.py
 D:\ai-exchange\.venv\Scripts\python.exe scripts\run_runtime_incident_journal.py
 D:\ai-exchange\.venv\Scripts\python.exe -m unittest tests.unit.test_maintenance_summary_service tests.unit.test_run_maintenance_summary tests.unit.test_api_server.ApiServerRoutingTests.test_routes_maintenance_summary tests.unit.test_api_static_server.ApiStaticServerTests.test_dashboard_uses_canonical_trading_routes
 D:\ai-exchange\.venv\Scripts\python.exe scripts\run_maintenance_summary.py
+D:\ai-exchange\.venv\Scripts\python.exe -m unittest tests.unit.test_unattended_harness_service tests.unit.test_run_unattended_harness tests.unit.test_api_server.ApiServerRoutingTests.test_routes_unattended_harness
+D:\ai-exchange\.venv\Scripts\python.exe scripts\run_unattended_harness.py
 ```

TASKS.md

@@ -13,7 +13,7 @@
 
 ## In Progress
 
-- [x] P22 Maintenance And Retention Rollup local closure complete
+- [x] P23 End-To-End Unattended Harness local closure complete
 
 ## Next Up
 
@@ -33,14 +33,15 @@
 - [x] P20 local closure complete
 - [x] P21 local closure complete
 - [x] P22 local closure complete
-- [ ] Start P23 End-To-End Unattended Harness
+- [x] P23 local closure complete
+- [ ] Start P24 Remote Smoke And Ops Closure
 - [ ] 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_p22_maintenance_retention_rollup_closed
+- [x] state: automation_execution_layer_p23_end_to_end_unattended_harness_closed
-- [x] current_phase: P22 Maintenance And Retention Rollup
+- [x] current_phase: P23 End-To-End Unattended Harness
 - [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
@@ -82,7 +83,11 @@
 - [x] latest_maintenance_summary_route_status: ok
 - [x] latest_maintenance_summary_workbench_status: visible
 - [x] latest_maintenance_summary_smoke_status: warning_feedback_sample_health
-- [x] next_execution_layer_phase: P23 End-To-End Unattended Harness
+- [x] latest_p23_slice: end_to_end_unattended_harness
+- [x] latest_unattended_harness_status: implemented
+- [x] latest_unattended_harness_route_status: ok
+- [x] latest_unattended_harness_smoke_status: passed_9_checks
+- [x] next_execution_layer_phase: P24 Remote Smoke And Ops Closure
 - [x] latest_visual_review_script_status: idle
 - [x] latest_runtime_visual_watch_task_status: ok
 - [x] latest_visual_capture_cleanup_task_status: ok
@@ -190,6 +195,10 @@
 - [x] Added `GET /trading/maintenance-summary`
 - [x] Added maintenance rollup visibility to the workbench runtime panel
 - [x] Added focused P22 maintenance service, script, API, and dashboard route tests
+- [x] Added `UnattendedHarnessService`
+- [x] Added `scripts/run_unattended_harness.py`
+- [x] Added `GET /trading/unattended-harness`
+- [x] Added focused P23 harness service, script, and API 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
@@ -219,6 +228,8 @@
 - [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
 - [x] Smoked P22 maintenance summary script/API with warning-only feedback sample health and no blockers
 - [x] Completed P22 full validation: 696 unit tests, 70 script static checks, dashboard JS syntax, maintenance summary script/API smoke
+- [x] Smoked P23 unattended harness script/API with 9 passed checks, no warnings, no blockers, and analysis/decision-only mode
+- [x] Completed P23 full validation: 701 unit tests, 71 script static checks, dashboard JS syntax, unattended harness script/API smoke
 
 ## Blocked
 
@@ -283,7 +294,7 @@
 - [x] `P20` Operator Digest And Review Queue
 - [x] `P21` Runtime Incident Journal
 - [x] `P22` Maintenance And Retention Rollup
-- [ ] `P23` End-To-End Unattended Harness
+- [x] `P23` End-To-End Unattended Harness
 - [ ] `P24` Remote Smoke And Ops Closure
 - [ ] `PXX` Automation Execution Closure Gate
 

docs/AI_ICT_盯盘告警开发任务清单_v1.md

@@ -412,6 +412,7 @@ candidate -> decision alert -> dispatcher -> 外部消息 -> delivery record
 - 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 暴露。
 - 2026-05-12 P22 已完成 `maintenance / retention rollup` 本地闭环：maintenance summary 以只读方式聚合 visual capture retention dry-run、delivery/feedback、runtime watchdog 与 incident journal，输出 P23 harness 可复用的 checks/blockers/warnings/p23_assertions，不执行未纳入 retention policy 的业务数据删除。
+- 2026-05-12 P23 已完成 `end-to-end unattended harness` 本地闭环：`UnattendedHarnessService`、脚本与 API 会在 analysis/decision-only 模式下一次性验证 P15-P22 关键路径，输出单一 JSON blockers/warnings/passed-checks 报告，不使用外部通道、不要求真实成交。
 
 ### 验收
 - 非重大错误不打断主循环

docs/AI_ICT_盯盘告警技术设计稿_v1.md

@@ -124,6 +124,11 @@
 
 2026-05-12 P22 已落地：maintenance summary 会 dry-run visual capture retention，读取 alert quality summary、runtime watchdog 与 runtime incident journal，输出 `checks`、`blockers`、`warnings`、`maintenance_actions` 与 `p23_assertions`。该 summary 不执行删除；自动清理动作只作为可用 action 暴露，未纳入 retention policy 的业务数据不会被删除。
 
+#### `src/services/maintenance/unattended_harness_service.py`
+职责：把 P15-P22 的自动化执行层关键能力聚合成可一条命令运行的 unattended smoke harness。
+
+2026-05-12 P23 已落地：unattended harness 以 analysis/decision-only 模式验证 policy suppression、dispatch retry contract、runtime watchdog、visual capture fallback、feedback rollup、visual media/channel rollup、operator digest、maintenance summary 与 runtime environment，并输出单一 JSON 报告。该 harness 不使用外部通道、不要求真实成交，环境不可用时以 environment blocker 暴露，不改变 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

@@ -291,6 +291,12 @@
 - 环境依赖不可用时输出明确 blocker，不误报为业务失败。
 - 可作为 P24 远端 smoke 的主要入口。
 
+### 本地闭环记录
+
+- 2026-05-12 P23 已新增 `UnattendedHarnessService`，以 analysis/decision-only 模式验证 policy suppression、dispatch retry contract、runtime watchdog、visual capture fallback、feedback rollup、visual media/channel rollup、operator digest、maintenance summary 与 runtime environment。
+- 2026-05-12 P23 已新增 `scripts/run_unattended_harness.py` 与 `GET /trading/unattended-harness`，输出单一 JSON 报告，包含 `checks`、`passed_checks`、`warnings`、`blockers`、`external_channels_used=false` 与 `real_trade_required=false`。
+- 2026-05-12 P23 本地闭环已验证：focused P23 tests 通过，`tests/unit` 全量 701 个测试通过，script static checks 71 个脚本通过，`node --check src/web/app.js` 通过；脚本/API smoke 均返回 `status=passed`、passed_checks=9、warnings=[]、blockers=[]。
+
 ---
 
 ## 14. P24 Remote Smoke And Ops Closure

scripts/run_unattended_harness.py

@@ -0,0 +1,38 @@
+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.harness import run_unattended_harness
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    payload = run_unattended_harness(
+        instrument_id=int(env.get("UNATTENDED_HARNESS_INSTRUMENT_ID", "1")),
+    )
+    if _get_bool(env, "UNATTENDED_HARNESS_OUTPUT_JSON", False):
+        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
+    else:
+        print(f"Unattended harness status: {payload.get('status', 'unknown')}")
+        print(f"Passed checks: {', '.join(payload.get('passed_checks') or []) or 'none'}")
+        print(f"Warnings: {', '.join(payload.get('warnings') or []) or 'none'}")
+        print(f"Blockers: {', '.join(payload.get('blockers') or []) or 'none'}")
+    return 0 if not payload.get("blockers") 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())

src/api/__init__.py

@@ -5,6 +5,7 @@ from src.api.broker_state import get_broker_state
 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.harness import run_unattended_harness
 from src.api.live_operation_audits import get_recent_live_operation_audits
 from src.api.maintenance import get_maintenance_summary
 from src.api.server import route_request, serve_api
@@ -85,6 +86,7 @@ from src.api.visual_watch import (
 
 __all__ = [
     "get_health",
+    "run_unattended_harness",
     "get_broker_state",
     "get_recent_candles",
     "get_recent_execution_tickets",

src/api/harness.py

@@ -0,0 +1,5 @@
+from src.services.maintenance import UnattendedHarnessService
+
+
+def run_unattended_harness(*, instrument_id: int = 1) -> dict:
+    return UnattendedHarnessService().run_harness(instrument_id=instrument_id)

src/api/server.py

@@ -9,6 +9,7 @@ from src.api.broker_state import get_broker_state
 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.harness import run_unattended_harness
 from src.api.live_operation_audits import get_recent_live_operation_audits
 from src.api.maintenance import get_maintenance_summary
 from src.api.alerts import (
@@ -105,6 +106,7 @@ from src.services.runtime import start_trading_runtime_service
 def build_api_dependencies() -> dict:
     return {
         "health": get_health,
+        "unattended_harness": run_unattended_harness,
         "broker_state": get_broker_state,
         "candles": get_recent_candles,
         "execution_tickets": get_recent_execution_tickets,
@@ -208,6 +210,8 @@ def route_request(
 
     if method == "GET" and path == "/health":
         return 200, deps["health"]()
+    if method == "GET" and path == "/trading/unattended-harness":
+        return 200, deps["unattended_harness"](instrument_id=_get_int(query, "instrument_id", 1))
     if method == "GET" and path == "/trading/broker-state":
         instrument_id = _require_int(query, "instrument_id")
         return 200, deps["broker_state"](instrument_id=instrument_id)

src/services/maintenance/__init__.py

@@ -10,9 +10,11 @@ from src.services.maintenance.historical_compatibility import (
     summarize_liquidity_duplicate_groups,
 )
 from src.services.maintenance.maintenance_summary_service import MaintenanceSummaryService
+from src.services.maintenance.unattended_harness_service import UnattendedHarnessService
 
 __all__ = [
     "MaintenanceSummaryService",
+    "UnattendedHarnessService",
     "build_historical_compatibility_summary",
     "classify_liquidity_duplicate_groups",
     "derive_historical_fvg_meta",

src/services/maintenance/unattended_harness_service.py

@@ -0,0 +1,228 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from typing import Callable, Optional
+
+from src.config.settings import settings
+from src.services.alerts import AlertQualitySummaryService, OperatorDigestService
+from src.services.alerts.delivery_policy_service import evaluate_alert_delivery_policy
+from src.services.maintenance.maintenance_summary_service import MaintenanceSummaryService
+from src.services.runtime.runtime_task_watchdog_service import RuntimeTaskWatchdogService
+from src.services.visual_watch import VisualSnapshotCaptureService
+
+
+class UnattendedHarnessService:
+    def __init__(
+        self,
+        *,
+        maintenance_summary_service: Optional[MaintenanceSummaryService] = None,
+        alert_quality_service: Optional[AlertQualitySummaryService] = None,
+        operator_digest_service: Optional[OperatorDigestService] = None,
+        visual_capture_service: Optional[VisualSnapshotCaptureService] = None,
+        runtime_status_provider: Optional[Callable[[], dict]] = None,
+    ) -> None:
+        self.maintenance_summary_service = maintenance_summary_service or MaintenanceSummaryService()
+        self.alert_quality_service = alert_quality_service or AlertQualitySummaryService()
+        self.operator_digest_service = operator_digest_service or OperatorDigestService()
+        self.visual_capture_service = visual_capture_service or VisualSnapshotCaptureService(browser_capture_enabled=False)
+        self.runtime_status_provider = runtime_status_provider
+
+    def run_harness(self, *, instrument_id: int = 1) -> dict:
+        checks = []
+        checks.append(_check_policy_suppression())
+        checks.append(_check_dispatch_retry_contract())
+        checks.append(_check_runtime_watchdog_contract())
+        checks.append(self._check_visual_capture_fallback())
+        alert_quality = self._safe_payload(
+            "alert_quality_summary",
+            lambda: self.alert_quality_service.build_summary(instrument_id=instrument_id),
+        )
+        checks.append(_check_alert_quality_rollup(alert_quality))
+        checks.append(_check_visual_media_and_channel_rollup(alert_quality))
+        operator_digest = self._safe_payload(
+            "operator_digest",
+            lambda: self.operator_digest_service.build_digest(instrument_id=instrument_id, top_limit=5),
+        )
+        checks.append(_check_operator_digest(operator_digest))
+        maintenance = self._safe_payload(
+            "maintenance_summary",
+            lambda: self.maintenance_summary_service.build_summary(instrument_id=instrument_id),
+        )
+        checks.append(_check_maintenance_summary(maintenance))
+        runtime_status = self._safe_payload("runtime_status", self._load_runtime_status)
+        checks.append(_check_runtime_environment(runtime_status))
+
+        blockers = []
+        warnings = []
+        passed_checks = []
+        for check in checks:
+            if check["status"] == "passed":
+                passed_checks.append(check["name"])
+            elif check["status"] == "warning":
+                warnings.append(check["name"])
+            elif check["status"] == "blocked":
+                blockers.append(check["name"])
+        return {
+            "status": "blocked" if blockers else "passed",
+            "generated_ts": datetime.now(timezone.utc).isoformat(),
+            "instrument_id": instrument_id,
+            "mode": "analysis_decision_only",
+            "external_channels_used": False,
+            "real_trade_required": False,
+            "checks": checks,
+            "passed_checks": passed_checks,
+            "warnings": warnings,
+            "blockers": blockers,
+            "payloads": {
+                "alert_quality_summary": _trim_payload(alert_quality),
+                "operator_digest": _trim_payload(operator_digest),
+                "maintenance_summary": _trim_payload(maintenance),
+                "runtime_status": _trim_payload(runtime_status),
+            },
+        }
+
+    def _safe_payload(self, name: str, fetcher) -> dict:
+        try:
+            return dict(fetcher() or {})
+        except Exception as exc:
+            return {"status": "unavailable", "error": str(exc), "source": name}
+
+    def _load_runtime_status(self) -> dict:
+        if self.runtime_status_provider is not None:
+            return dict(self.runtime_status_provider() or {})
+        from src.services.runtime import get_trading_runtime_service
+
+        return get_trading_runtime_service().status()
+
+    def _check_visual_capture_fallback(self) -> dict:
+        payload = self._safe_payload(
+            "visual_capture_fallback",
+            lambda: self.visual_capture_service.capture_watch_run_snapshot(
+                watch_run={},
+                browser_capture_enabled=False,
+            ),
+        )
+        if payload.get("status") == "ignored" and payload.get("reason") in {"watch_run_missing", "watch_run_id_missing"}:
+            return _passed("visual_capture_fallback", "missing watch run returns safe ignored status")
+        return _blocked("visual_capture_fallback", f"unexpected visual capture fallback: {payload}")
+
+
+def _check_policy_suppression() -> dict:
+    policy = evaluate_alert_delivery_policy(
+        {
+            "id": 0,
+            "alert_type": "candidate_blocked",
+            "status": "queued",
+            "payload": {"severity": "low", "operator_mode": "observe_only"},
+        },
+        channel="webhook",
+        silent_window_service=_InactiveSilentWindowService(),
+    )
+    if policy.get("skip") and policy.get("reason") == "suppressed_noise_policy":
+        return _passed("policy_suppression", "low-priority external alert is suppressed by noise policy")
+    return _blocked("policy_suppression", f"unexpected policy result: {policy}")
+
+
+def _check_dispatch_retry_contract() -> dict:
+    if bool(settings.alert_retry_failed_enabled) and int(settings.alert_retry_max_attempts or 0) > 0:
+        return _passed("dispatch_retry_contract", "failed-alert retry is enabled with bounded attempts")
+    return _blocked("dispatch_retry_contract", "failed-alert retry is disabled or unbounded")
+
+
+def _check_runtime_watchdog_contract() -> dict:
+    now = datetime.now(timezone.utc)
+    stale_ts = (now - timedelta(seconds=600)).isoformat()
+    payload = RuntimeTaskWatchdogService(stale_multiplier=1, min_stuck_seconds=1).build_watchdog(
+        {
+            "watch": {
+                "interval_seconds": 1,
+                "last_status": "running",
+                "last_started_ts": stale_ts,
+                "run_count": 1,
+            }
+        },
+        checked_ts=now,
+    )
+    if payload.get("stalled_task_count") == 1:
+        return _passed("runtime_watchdog_contract", "synthetic stale task is detected")
+    return _blocked("runtime_watchdog_contract", f"watchdog did not detect stale task: {payload}")
+
+
+def _check_alert_quality_rollup(payload: dict) -> dict:
+    required = {"delivery_summary", "alert_feedback_summary", "visual_feedback_summary", "visual_review_summary"}
+    missing = sorted(required - set(payload.keys()))
+    if missing:
+        return _blocked("feedback_rollup", f"alert quality summary missing {missing}")
+    return _passed("feedback_rollup", "alert and visual feedback rollups are present")
+
+
+def _check_visual_media_and_channel_rollup(payload: dict) -> dict:
+    delivery = dict(payload.get("delivery_summary") or {})
+    visual = dict(payload.get("visual_review_summary") or {})
+    required_delivery = {"media_attached_count", "media_linked_count", "media_unavailable_count", "channel_status_counts"}
+    missing_delivery = sorted(required_delivery - set(delivery.keys()))
+    if missing_delivery:
+        return _blocked("visual_media_channel_rollup", f"delivery summary missing {missing_delivery}")
+    if "trigger_type_counts" not in visual:
+        return _blocked("visual_media_channel_rollup", "visual review trigger counts missing")
+    return _passed("visual_media_channel_rollup", "media/channel/visual trigger rollups are present")
+
+
+def _check_operator_digest(payload: dict) -> dict:
+    if "review_queue" in payload and "summary_counts" in payload:
+        return _passed("operator_digest", "operator digest exposes summary counts and review queue")
+    return _blocked("operator_digest", f"operator digest unavailable: {payload}")
+
+
+def _check_maintenance_summary(payload: dict) -> dict:
+    assertions = dict(payload.get("p23_assertions") or {})
+    blockers = list(payload.get("blockers") or [])
+    required = {
+        "storage_visible",
+        "delivery_visible",
+        "feedback_visible",
+        "runtime_visible",
+        "incident_journal_visible",
+        "no_unscoped_deletion",
+        "retention_policy_dry_run_only",
+    }
+    missing = sorted(item for item in required if not assertions.get(item))
+    if missing:
+        return _blocked("maintenance_summary", f"maintenance assertions failed: {missing}")
+    if blockers:
+        return _blocked("maintenance_summary", f"maintenance blockers: {blockers}")
+    return _passed("maintenance_summary", "P22 maintenance assertions are available")
+
+
+def _check_runtime_environment(payload: dict) -> dict:
+    blockers = list(payload.get("current_blockers") or [])
+    if blockers:
+        return _blocked("runtime_environment", f"runtime blockers: {blockers}", environment=True)
+    if payload.get("status") == "unavailable":
+        return _blocked("runtime_environment", f"runtime status unavailable: {payload}", environment=True)
+    return _passed("runtime_environment", "runtime status has no current blockers")
+
+
+def _passed(name: str, detail: str) -> dict:
+    return {"name": name, "status": "passed", "detail": detail}
+
+
+def _blocked(name: str, detail: str, *, environment: bool = False) -> dict:
+    payload = {"name": name, "status": "blocked", "detail": detail}
+    if environment:
+        payload["blocker_type"] = "environment"
+    return payload
+
+
+def _trim_payload(payload: dict) -> dict:
+    result = dict(payload or {})
+    if "items" in result and isinstance(result["items"], list):
+        result["items"] = result["items"][:5]
+    if "checks" in result and isinstance(result["checks"], list):
+        result["checks"] = result["checks"][:10]
+    return result
+
+
+class _InactiveSilentWindowService:
+    def build_status(self, now=None) -> dict:
+        return {"active": False, "enabled": True, "window_label": "harness_inactive"}

tests/unit/test_api_server.py

@@ -40,6 +40,17 @@ class ApiServerRoutingTests(unittest.TestCase):
         self.assertEqual(status, 200)
         self.assertEqual(payload["id"], 9)
 
+    def test_routes_unattended_harness(self) -> None:
+        status, payload = route_request(
+            "/trading/unattended-harness",
+            {"instrument_id": ["2"]},
+            dependencies={"unattended_harness": lambda **kwargs: {"status": "passed", "instrument_id": kwargs["instrument_id"]}},
+        )
+
+        self.assertEqual(status, 200)
+        self.assertEqual(payload["status"], "passed")
+        self.assertEqual(payload["instrument_id"], 2)
+
     def test_routes_trading_market_state(self) -> None:
         status, payload = route_request(
             "/trading/market-state",

tests/unit/test_run_unattended_harness.py

@@ -0,0 +1,46 @@
+import contextlib
+import io
+import json
+import unittest
+from unittest.mock import patch
+
+from scripts import run_unattended_harness
+
+
+class RunUnattendedHarnessTests(unittest.TestCase):
+    def test_main_outputs_json_report(self) -> None:
+        payload = {
+            "status": "passed",
+            "passed_checks": ["policy_suppression"],
+            "warnings": [],
+            "blockers": [],
+        }
+        output = io.StringIO()
+        with patch("scripts.run_unattended_harness.run_unattended_harness", return_value=payload) as harness:
+            with contextlib.redirect_stdout(output):
+                exit_code = run_unattended_harness.main(
+                    env={"UNATTENDED_HARNESS_OUTPUT_JSON": "1", "UNATTENDED_HARNESS_INSTRUMENT_ID": "2"}
+                )
+
+        self.assertEqual(exit_code, 0)
+        harness.assert_called_once_with(instrument_id=2)
+        self.assertEqual(json.loads(output.getvalue())["status"], "passed")
+
+    def test_main_returns_nonzero_for_blockers(self) -> None:
+        payload = {
+            "status": "blocked",
+            "passed_checks": [],
+            "warnings": [],
+            "blockers": ["runtime_environment"],
+        }
+        output = io.StringIO()
+        with patch("scripts.run_unattended_harness.run_unattended_harness", return_value=payload):
+            with contextlib.redirect_stdout(output):
+                exit_code = run_unattended_harness.main(env={})
+
+        self.assertEqual(exit_code, 2)
+        self.assertIn("runtime_environment", output.getvalue())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_unattended_harness_service.py

@@ -0,0 +1,90 @@
+import unittest
+
+from src.services.maintenance import UnattendedHarnessService
+
+
+class _MaintenanceSummaryService:
+    def __init__(self, payload=None) -> None:
+        self.payload = payload or {
+            "status": "healthy",
+            "blockers": [],
+            "warnings": [],
+            "p23_assertions": {
+                "storage_visible": True,
+                "delivery_visible": True,
+                "feedback_visible": True,
+                "runtime_visible": True,
+                "incident_journal_visible": True,
+                "no_unscoped_deletion": True,
+                "retention_policy_dry_run_only": True,
+            },
+        }
+
+    def build_summary(self, *, instrument_id: int):
+        return dict(self.payload)
+
+
+class _AlertQualityService:
+    def build_summary(self, *, instrument_id: int):
+        return {
+            "delivery_summary": {
+                "media_attached_count": 0,
+                "media_linked_count": 0,
+                "media_unavailable_count": 0,
+                "channel_status_counts": {},
+            },
+            "alert_feedback_summary": {"count": 1},
+            "visual_feedback_summary": {"count": 1},
+            "visual_review_summary": {"trigger_type_counts": {}},
+        }
+
+
+class _OperatorDigestService:
+    def build_digest(self, *, instrument_id: int, top_limit: int = 5):
+        return {"summary_counts": {}, "review_queue": []}
+
+
+class _VisualCaptureService:
+    def capture_watch_run_snapshot(self, **kwargs):
+        return {"status": "ignored", "reason": "watch_run_missing"}
+
+
+class UnattendedHarnessServiceTests(unittest.TestCase):
+    def test_run_harness_passes_analysis_decision_only_contract(self) -> None:
+        service = UnattendedHarnessService(
+            maintenance_summary_service=_MaintenanceSummaryService(),
+            alert_quality_service=_AlertQualityService(),
+            operator_digest_service=_OperatorDigestService(),
+            visual_capture_service=_VisualCaptureService(),
+            runtime_status_provider=lambda: {"current_blockers": [], "runtime_health": {"level": "ok"}},
+        )
+
+        payload = service.run_harness(instrument_id=1)
+
+        self.assertEqual(payload["status"], "passed")
+        self.assertFalse(payload["external_channels_used"])
+        self.assertFalse(payload["real_trade_required"])
+        self.assertIn("policy_suppression", payload["passed_checks"])
+        self.assertIn("runtime_watchdog_contract", payload["passed_checks"])
+        self.assertIn("maintenance_summary", payload["passed_checks"])
+        self.assertEqual(payload["blockers"], [])
+
+    def test_run_harness_reports_environment_blockers(self) -> None:
+        service = UnattendedHarnessService(
+            maintenance_summary_service=_MaintenanceSummaryService(),
+            alert_quality_service=_AlertQualityService(),
+            operator_digest_service=_OperatorDigestService(),
+            visual_capture_service=_VisualCaptureService(),
+            runtime_status_provider=lambda: {"current_blockers": ["broker_state_unavailable"], "runtime_health": {"level": "blocked"}},
+        )
+
+        payload = service.run_harness(instrument_id=1)
+
+        self.assertEqual(payload["status"], "blocked")
+        self.assertIn("runtime_environment", payload["blockers"])
+        runtime_check = [item for item in payload["checks"] if item["name"] == "runtime_environment"][0]
+        self.assertEqual(runtime_check["blocker_type"], "environment")
+
+
+if __name__ == "__main__":
+    unittest.main()