Close P8 visual watch automation loop

scripts/deploy_with_password.py

@@ -0,0 +1,186 @@
+from __future__ import annotations
+
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from typing import Optional
+import argparse
+import os
+import tarfile
+import time
+import urllib.request
+
+try:
+    import paramiko
+except ImportError as exc:  # pragma: no cover - local tooling guard
+    raise SystemExit("paramiko_missing: install paramiko in the local venv before deploying") from exc
+
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_HOST = "192.168.3.90"
+DEFAULT_USER = "root"
+DEFAULT_REMOTE_PATH = "/root/deploy/ai-ict"
+DEFAULT_REMOTE_ARCHIVE = "/root/ai-ict-deploy.tgz"
+DEFAULT_SMOKE_BASE_URL = "http://192.168.3.90:8000"
+DEPLOY_PATHS = [
+    "Dockerfile",
+    ".dockerignore",
+    "docker-compose.yml",
+    "pyproject.toml",
+    "README.md",
+    ".env.example",
+    "docs",
+    "SESSION.md",
+    "TASKS.md",
+    "src",
+    "scripts",
+    "migrations",
+    "data",
+]
+
+
+def main(argv: Optional[list[str]] = None, env=None) -> int:
+    env = os.environ if env is None else env
+    args = _parse_args(argv, env)
+    if not args.password:
+        raise SystemExit("missing_password: set AI_ICT_DEPLOY_PASSWORD or pass --password")
+
+    with TemporaryDirectory() as tmp_dir:
+        archive_path = Path(tmp_dir) / "ai-ict-deploy.tgz"
+        create_archive(archive_path)
+        client = connect_ssh(host=args.host, user=args.user, password=args.password, port=args.port)
+        try:
+            run_remote(client, f"mkdir -p {args.remote_path}")
+            upload_file(client, archive_path, args.remote_archive)
+            run_remote(
+                client,
+                build_deploy_command(
+                    remote_path=args.remote_path,
+                    remote_archive=args.remote_archive,
+                ),
+                timeout=600,
+            )
+        finally:
+            client.close()
+
+    smoke = run_http_smoke(args.smoke_base_url, attempts=args.smoke_attempts, sleep_seconds=args.smoke_sleep_seconds)
+    print(f"deploy_complete host={args.host} remote_path={args.remote_path}")
+    print(f"smoke_complete endpoints={len(smoke)}")
+    return 0
+
+
+def create_archive(archive_path: Path) -> None:
+    with tarfile.open(archive_path, "w:gz") as archive:
+        for relative in DEPLOY_PATHS:
+            path = ROOT / relative
+            if not path.exists():
+                continue
+            archive.add(path, arcname=relative, filter=_tar_filter)
+
+
+def connect_ssh(*, host: str, user: str, password: str, port: int):
+    client = paramiko.SSHClient()
+    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+    client.connect(
+        hostname=host,
+        port=port,
+        username=user,
+        password=password,
+        look_for_keys=False,
+        allow_agent=False,
+        timeout=20,
+        banner_timeout=20,
+        auth_timeout=20,
+    )
+    return client
+
+
+def upload_file(client, local_path: Path, remote_path: str) -> None:
+    with client.open_sftp() as sftp:
+        sftp.put(str(local_path), remote_path)
+
+
+def run_remote(client, command: str, *, timeout: int = 120) -> str:
+    stdin, stdout, stderr = client.exec_command(command, timeout=timeout)
+    exit_status = stdout.channel.recv_exit_status()
+    out = stdout.read().decode("utf-8", errors="replace")
+    err = stderr.read().decode("utf-8", errors="replace")
+    if exit_status != 0:
+        raise RuntimeError(f"remote_command_failed:{exit_status}\n{command}\nSTDOUT:\n{out}\nSTDERR:\n{err}")
+    return out
+
+
+def build_deploy_command(*, remote_path: str, remote_archive: str) -> str:
+    remote_path_q = _sh_quote(remote_path)
+    remote_archive_q = _sh_quote(remote_archive)
+    env_backup = "/root/ai-ict-deploy.env.bak"
+    return " && ".join(
+        [
+            f"mkdir -p {remote_path_q}",
+            f"if [ -f {remote_path_q}/.env ]; then cp {remote_path_q}/.env {_sh_quote(env_backup)}; fi",
+            f"rm -rf {remote_path_q}/*",
+            f"tar -xzf {remote_archive_q} -C {remote_path_q}",
+            f"if [ -f {_sh_quote(env_backup)} ]; then mv {_sh_quote(env_backup)} {remote_path_q}/.env; fi",
+            f"cd {remote_path_q}",
+            "((docker compose down || docker-compose down) >/dev/null 2>&1 || true)",
+            "((docker compose up -d --build) || (docker-compose up -d --build))",
+            "((docker compose ps) || (docker-compose ps))",
+        ]
+    )
+
+
+def run_http_smoke(base_url: str, *, attempts: int = 10, sleep_seconds: float = 3.0) -> list[dict]:
+    endpoints = [
+        "/health",
+        "/trading/runtime-status",
+        "/trading/research-council/optimization-cycle/latest?instrument_id=1",
+        "/trading/research-council/optimization-cycles?instrument_id=1&limit=1",
+        "/trading/research-council/optimization-review?instrument_id=1",
+        "/trading/review-packet?instrument_id=1",
+    ]
+    results = []
+    for endpoint in endpoints:
+        url = f"{base_url.rstrip('/')}{endpoint}"
+        last_error = None
+        for attempt in range(1, attempts + 1):
+            try:
+                with urllib.request.urlopen(url, timeout=20) as response:
+                    body = response.read(4096)
+                results.append({"url": url, "status": response.status, "bytes": len(body), "attempt": attempt})
+                break
+            except Exception as exc:  # pragma: no cover - exercised by live deploys
+                last_error = exc
+                if attempt < attempts:
+                    time.sleep(sleep_seconds)
+        else:
+            raise RuntimeError(f"http_smoke_failed:{url}:{last_error}")
+    return results
+
+
+def _tar_filter(info: tarfile.TarInfo):
+    normalized = info.name.replace("\\", "/")
+    ignored_parts = {".git", ".venv", "__pycache__", ".pytest_cache", "node_modules"}
+    if any(part in ignored_parts for part in normalized.split("/")):
+        return None
+    return info
+
+
+def _parse_args(argv: Optional[list[str]], env) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Deploy ai-exchange over password SSH without expect.")
+    parser.add_argument("--host", default=env.get("AI_ICT_DEPLOY_HOST", DEFAULT_HOST))
+    parser.add_argument("--port", type=int, default=int(env.get("AI_ICT_DEPLOY_PORT", "22")))
+    parser.add_argument("--user", default=env.get("AI_ICT_DEPLOY_USER", DEFAULT_USER))
+    parser.add_argument("--password", default=env.get("AI_ICT_DEPLOY_PASSWORD", ""))
+    parser.add_argument("--remote-path", default=env.get("AI_ICT_DEPLOY_REMOTE_PATH", DEFAULT_REMOTE_PATH))
+    parser.add_argument("--remote-archive", default=env.get("AI_ICT_DEPLOY_REMOTE_ARCHIVE", DEFAULT_REMOTE_ARCHIVE))
+    parser.add_argument("--smoke-base-url", default=env.get("AI_ICT_DEPLOY_SMOKE_BASE_URL", DEFAULT_SMOKE_BASE_URL))
+    parser.add_argument("--smoke-attempts", type=int, default=int(env.get("AI_ICT_DEPLOY_SMOKE_ATTEMPTS", "10")))
+    parser.add_argument("--smoke-sleep-seconds", type=float, default=float(env.get("AI_ICT_DEPLOY_SMOKE_SLEEP_SECONDS", "3")))
+    return parser.parse_args(argv)
+
+
+def _sh_quote(value: str) -> str:
+    return "'" + value.replace("'", "'\"'\"'") + "'"
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/fix_pg_hba.exp

@@ -5,6 +5,7 @@ from pathlib import Path
 hba = Path('/var/lib/pgsql/16/data/pg_hba.conf')
 lines = [
     'host ai_ict yuqei 172.18.0.0/16 scram-sha-256\n',
+    'host ai_ict yuqei 172.19.0.0/16 scram-sha-256\n',
     'host ai_ict yuqei 172.20.0.0/16 scram-sha-256\n',
     'host ai_ict yuqei 172.21.0.0/16 scram-sha-256\n',
     'host ai_ict yuqei 172.22.0.0/16 scram-sha-256\n',

scripts/run_alert_dispatcher.py

@@ -0,0 +1,42 @@
+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.config.settings import settings
+from src.services.alerts import AlertDispatcherService
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    limit = int(env.get("ALERT_DISPATCH_LIMIT", str(settings.alert_dispatch_max_per_run)))
+    output_json = _get_bool(env, "ALERT_DISPATCH_OUTPUT_JSON", False)
+    payload = AlertDispatcherService().dispatch_pending(limit=limit)
+    if output_json:
+        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
+    else:
+        print(f"Alert dispatch attempted: {payload.get('attempted', 0)}")
+        items = payload.get("items") or []
+        delivered = sum(1 for item in items if item.get("status") == "delivered")
+        failed = sum(1 for item in items if item.get("status") == "failed")
+        skipped = sum(1 for item in items if item.get("status") == "skipped")
+        print(f"Delivered: {delivered} / Failed: {failed} / Skipped: {skipped}")
+    items = payload.get("items") or []
+    return 0 if all(item.get("status") != "failed" for item in items) else 2
+
+
+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())

scripts/run_analysis_decision_monitor.py

@@ -0,0 +1,292 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Optional
+import json
+import os
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+
+DEFAULT_BASE_URL = "http://192.168.3.90:8000"
+DEFAULT_TIMEOUT_SECONDS = 30
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    base_url = str(env.get("ANALYSIS_DECISION_MONITOR_BASE_URL") or DEFAULT_BASE_URL)
+    instrument_id = int(env.get("ANALYSIS_DECISION_MONITOR_INSTRUMENT_ID", "1"))
+    timeframe = str(env.get("ANALYSIS_DECISION_MONITOR_TIMEFRAME") or "1m")
+    timeout_seconds = float(env.get("ANALYSIS_DECISION_MONITOR_TIMEOUT_SECONDS", str(DEFAULT_TIMEOUT_SECONDS)))
+    run_p6_tick = _get_bool(env, "ANALYSIS_DECISION_MONITOR_RUN_P6_TICK", True)
+    output_json = _get_bool(env, "ANALYSIS_DECISION_MONITOR_OUTPUT_JSON", False)
+
+    try:
+        payload = run_monitor(
+            base_url=base_url,
+            instrument_id=instrument_id,
+            timeframe=timeframe,
+            timeout_seconds=timeout_seconds,
+            run_p6_tick=run_p6_tick,
+        )
+    except Exception as exc:
+        payload = {
+            "passed": False,
+            "generated_ts": datetime.now(timezone.utc).isoformat(),
+            "instrument_id": instrument_id,
+            "timeframe": timeframe,
+            "base_url": base_url.rstrip("/"),
+            "blockers": ["analysis_decision_monitor_exception"],
+            "warnings": [],
+            "error_type": exc.__class__.__name__,
+            "error": str(exc),
+        }
+
+    if output_json:
+        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
+    else:
+        print(f"Analysis/decision monitor passed: {payload.get('passed')}")
+        print(f"Blockers: {', '.join(payload.get('blockers') or []) or 'none'}")
+        print(f"Warnings: {', '.join(payload.get('warnings') or []) or 'none'}")
+    return 0 if payload.get("passed") else 2
+
+
+def run_monitor(
+    *,
+    base_url: str,
+    instrument_id: int,
+    timeframe: str,
+    timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
+    run_p6_tick: bool = True,
+) -> dict:
+    base = base_url.rstrip("/")
+    payloads = {
+        "health": _http_get_json(f"{base}/health", timeout_seconds=timeout_seconds),
+        "launch_readiness": _http_get_json(
+            f"{base}/trading/launch-readiness?{urllib.parse.urlencode({'instrument_id': instrument_id, 'timeframe': timeframe})}",
+            timeout_seconds=timeout_seconds,
+        ),
+        "broker_state": _http_get_json(
+            f"{base}/trading/broker-state?{urllib.parse.urlencode({'instrument_id': instrument_id})}",
+            timeout_seconds=timeout_seconds,
+        ),
+        "runtime_status": _http_get_json(f"{base}/trading/runtime-status", timeout_seconds=timeout_seconds),
+        "decision_packet": _http_get_json(
+            f"{base}/trading/decision-packet?{urllib.parse.urlencode({'instrument_id': instrument_id, 'timeframe': timeframe})}",
+            timeout_seconds=timeout_seconds,
+        ),
+        "review_packet": _http_get_json(
+            f"{base}/trading/review-packet?{urllib.parse.urlencode({'instrument_id': instrument_id})}",
+            timeout_seconds=timeout_seconds,
+        ),
+    }
+    if run_p6_tick:
+        payloads["p6_governance_tick"] = _http_post_json(
+            f"{base}/trading/research-council/governance-tick",
+            {
+                "instrument_id": instrument_id,
+                "limit": 5,
+                "stale_after_hours": 24,
+                "auto_run_stale_cycle": False,
+                "force_cycle": False,
+                "timeframe": timeframe,
+            },
+            timeout_seconds=timeout_seconds,
+        )
+    return evaluate_analysis_decision_monitor_payloads(
+        payloads=payloads,
+        base_url=base,
+        instrument_id=instrument_id,
+        timeframe=timeframe,
+    )
+
+
+def evaluate_analysis_decision_monitor_payloads(
+    *,
+    payloads: dict,
+    base_url: str = DEFAULT_BASE_URL,
+    instrument_id: int = 1,
+    timeframe: str = "1m",
+) -> dict:
+    checks: list[dict] = []
+
+    health = dict(payloads.get("health") or {})
+    launch = dict(payloads.get("launch_readiness") or {})
+    broker = dict(payloads.get("broker_state") or {})
+    runtime = dict(payloads.get("runtime_status") or {})
+    decision = dict(payloads.get("decision_packet") or {})
+    review = dict(payloads.get("review_packet") or {})
+    p6_tick = payloads.get("p6_governance_tick")
+    p6_tick = dict(p6_tick or {}) if p6_tick is not None else None
+
+    launch_warnings = list(launch.get("warning_reasons") or [])
+    launch_next_actions = list(launch.get("next_actions") or [])
+    market_state = _first_market_state(runtime) or dict(launch.get("market_state") or {})
+    stream_code = _path(market_state, "stream_state", "code")
+    freshness_code = _path(market_state, "freshness", "code")
+    gap_code = _path(market_state, "data_gap_state", "code")
+
+    _add_check(checks, "health_ok", health.get("app") == "ai-ict", detail=f"app={health.get('app')}")
+    _add_check(
+        checks,
+        "launch_no_blockers",
+        not list(launch.get("blocking_reasons") or []),
+        detail=f"blocking_reasons={launch.get('blocking_reasons') or []}",
+    )
+    _add_check(checks, "launch_analysis_only_ready", bool(launch.get("analysis_only_ready")))
+    _add_check(checks, "launch_restricted_decision_support_ready", bool(launch.get("restricted_decision_support_ready")))
+    _add_check(checks, "operator_soft_rollout_ready", bool(launch.get("operator_soft_rollout_ready")))
+    _add_check(
+        checks,
+        "real_trade_sample_gate_deferred",
+        launch.get("real_trade_sample_gate_required") is False
+        and launch.get("real_trade_sample_gate_status") == "deferred_analysis_decision_only",
+        detail=(
+            f"required={launch.get('real_trade_sample_gate_required')}, "
+            f"status={launch.get('real_trade_sample_gate_status')}"
+        ),
+    )
+    _add_check(
+        checks,
+        "real_trade_sample_not_current_action",
+        "real_trade_sample_verified" not in launch_warnings
+        and "wait_for_and_verify_real_fill_sample" not in launch_next_actions,
+        detail=f"warnings={launch_warnings}, next_actions={launch_next_actions}",
+    )
+    _add_check(checks, "broker_state_ok", broker.get("status") == "broker_state_ok", detail=f"status={broker.get('status')}")
+    _add_check(
+        checks,
+        "runtime_health_ok",
+        _path(runtime, "runtime_health", "level") == "ok"
+        and not list(runtime.get("current_blockers") or [])
+        and not bool(_path(runtime, "degraded_mode_state", "is_degraded")),
+        detail=(
+            f"health={_path(runtime, 'runtime_health', 'level')}, "
+            f"blockers={runtime.get('current_blockers') or []}, "
+            f"degraded={_path(runtime, 'degraded_mode_state', 'is_degraded')}"
+        ),
+    )
+    _add_check(checks, "market_stream_connected", stream_code == "market_stream_connected", detail=f"stream={stream_code}")
+    _add_check(checks, "market_data_fresh", freshness_code == "market_data_fresh", detail=f"freshness={freshness_code}")
+    _add_check(checks, "market_gap_clear", gap_code == "market_data_gap_clear", detail=f"gap={gap_code}")
+    _add_check(
+        checks,
+        "decision_restricted_support",
+        decision.get("action_recommendation") == "restricted_decision_support",
+        detail=f"action={decision.get('action_recommendation')}",
+    )
+    _add_check(
+        checks,
+        "decision_observe_only",
+        decision.get("operator_mode") == "observe_only",
+        detail=f"operator_mode={decision.get('operator_mode')}",
+    )
+    _add_check(
+        checks,
+        "decision_execution_not_allowed",
+        decision.get("execution_allowed") in {None, False}
+        and "execution_assist_not_allowed" in set(decision.get("why_allowed") or []),
+        detail=f"execution_allowed={decision.get('execution_allowed')}, why_allowed={decision.get('why_allowed') or []}",
+    )
+    _add_check(checks, "review_packet_present", bool(review.get("decision_packet")))
+    _add_check(checks, "review_trust_score_present", bool(review.get("trust_score")))
+    _add_check(checks, "review_research_summary_present", bool(review.get("research_council_summary")))
+    if p6_tick is not None:
+        _add_check(
+            checks,
+            "p6_governance_tick_noop",
+            p6_tick.get("status") == "no_op_healthy"
+            and not bool(p6_tick.get("requires_cycle_run"))
+            and not bool(p6_tick.get("requires_bounded_slice")),
+            detail=(
+                f"status={p6_tick.get('status')}, "
+                f"requires_cycle_run={p6_tick.get('requires_cycle_run')}, "
+                f"requires_bounded_slice={p6_tick.get('requires_bounded_slice')}"
+            ),
+        )
+
+    blockers = [item["code"] for item in checks if not item["passed"] and item["severity"] == "blocker"]
+    warnings = [item["code"] for item in checks if not item["passed"] and item["severity"] == "warning"]
+    return {
+        "passed": not blockers,
+        "generated_ts": datetime.now(timezone.utc).isoformat(),
+        "base_url": base_url.rstrip("/"),
+        "instrument_id": instrument_id,
+        "timeframe": timeframe,
+        "phase": "analysis_decision_only",
+        "blockers": blockers,
+        "warnings": warnings,
+        "checks": checks,
+        "observed": {
+            "broker_status": broker.get("status"),
+            "broker_fills": len(list(broker.get("fills") or [])),
+            "runtime_health": _path(runtime, "runtime_health", "level"),
+            "market_stream": stream_code,
+            "market_freshness": freshness_code,
+            "market_gap": gap_code,
+            "launch_mode": launch.get("operator_soft_rollout_mode"),
+            "real_trade_sample_gate_required": launch.get("real_trade_sample_gate_required"),
+            "real_trade_sample_gate_status": launch.get("real_trade_sample_gate_status"),
+            "decision_action": decision.get("action_recommendation"),
+            "decision_operator_mode": decision.get("operator_mode"),
+            "decision_trust_classification": decision.get("trust_classification"),
+            "p6_governance_tick_status": p6_tick.get("status") if p6_tick else None,
+        },
+    }
+
+
+def _http_get_json(url: str, *, timeout_seconds: float) -> dict:
+    with urllib.request.urlopen(url, timeout=timeout_seconds) as response:
+        return json.loads(response.read().decode("utf-8"))
+
+
+def _http_post_json(url: str, payload: dict, *, timeout_seconds: float) -> dict:
+    body = json.dumps(payload).encode("utf-8")
+    request = urllib.request.Request(
+        url,
+        data=body,
+        method="POST",
+        headers={"Content-Type": "application/json"},
+    )
+    with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
+        return json.loads(response.read().decode("utf-8"))
+
+
+def _add_check(checks: list[dict], code: str, passed: bool, *, detail: str = "", severity: str = "blocker") -> None:
+    checks.append({"code": code, "passed": bool(passed), "severity": severity, "detail": detail})
+
+
+def _first_market_state(runtime: dict) -> dict:
+    latest = dict(runtime.get("latest_market_state_by_symbol") or {})
+    for value in latest.values():
+        if isinstance(value, dict):
+            return value
+    return {}
+
+
+def _path(payload: dict, *parts: str):
+    current = payload
+    for part in parts:
+        if not isinstance(current, dict):
+            return None
+        current = current.get(part)
+    return current
+
+
+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())

scripts/run_decision_packet.py

@@ -18,7 +18,7 @@ def main(env=None) -> int:
     output_json = _get_bool(env, "DECISION_PACKET_OUTPUT_JSON", False)
     payload = DecisionPacketService().build_persist_and_alert(instrument_id=instrument_id)
     if output_json:
-        print(json.dumps(payload, ensure_ascii=False, sort_keys=True))
+        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
     else:
         print(f"Decision packet: {payload.get('status')} / {payload.get('action_recommendation')}")
         print(f"Why blocked: {', '.join(payload.get('why_blocked') or []) or 'none'}")

scripts/run_launch_readiness.py

@@ -19,7 +19,7 @@ def main(env=None) -> int:
     output_json = _get_bool(env, "LAUNCH_READINESS_OUTPUT_JSON", False)
     payload = LaunchReadinessService().build_summary(instrument_id=instrument_id, timeframe=timeframe)
     if output_json:
-        print(json.dumps(payload, ensure_ascii=False, sort_keys=True))
+        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
     else:
         print(f"Assisted trading ready: {payload.get('assisted_trading_ready')}")
         print(f"Real live ready: {payload.get('real_live_ready')}")

scripts/run_research_council.py

@@ -0,0 +1,70 @@
+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 sqlalchemy.exc import SQLAlchemyError
+
+from src.services.research_council import ResearchCouncilService
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    instrument_id = int(env.get("RESEARCH_COUNCIL_INSTRUMENT_ID", "1"))
+    timeframe = env.get("RESEARCH_COUNCIL_TIMEFRAME", "1m")
+    model_code = env.get("RESEARCH_COUNCIL_MODEL_CODE", "OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE")
+    output_json = _get_bool(env, "RESEARCH_COUNCIL_OUTPUT_JSON", False)
+    checkpoint_status = _get_bool(env, "RESEARCH_COUNCIL_CHECKPOINT_STATUS", False)
+    try:
+        service = ResearchCouncilService()
+        if checkpoint_status:
+            payload = service.get_resume_state(instrument_id=instrument_id)
+        else:
+            payload = service.build_and_persist(
+                instrument_id=instrument_id,
+                timeframe=timeframe,
+                model_code=model_code,
+            )
+    except SQLAlchemyError as exc:
+        failure = {
+            "passed": False,
+            "instrument_id": instrument_id,
+            "timeframe": timeframe,
+            "model_code": model_code,
+            "error_type": exc.__class__.__name__,
+            "error": str(exc),
+        }
+        _print_payload(failure, output_json=output_json)
+        return 2
+    payload["passed"] = True
+    _print_payload(payload, output_json=output_json)
+    return 0
+
+
+def _print_payload(payload: dict, *, output_json: bool) -> None:
+    if output_json:
+        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
+        return
+    if payload.get("can_resume") is not None:
+        print(f"Research council resume state for instrument {payload.get('instrument_id')}")
+        print(f"Can resume: {payload.get('can_resume')}")
+        return
+    print(f"Research council run completed for instrument {payload.get('instrument_id')}")
+    print(f"Run id: {payload.get('id', 'n/a')}")
+
+
+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())

scripts/run_research_council_arena.py

@@ -0,0 +1,75 @@
+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.research_council import ResearchCouncilService
+from src.services.trust import ModelTrustService
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    instrument_id = int(env.get("RESEARCH_COUNCIL_INSTRUMENT_ID", "1"))
+    timeframe = env.get("RESEARCH_COUNCIL_TIMEFRAME", "1m")
+    output_json = _get_bool(env, "RESEARCH_COUNCIL_OUTPUT_JSON", False)
+    model_codes = _resolve_model_codes(env=env, instrument_id=instrument_id)
+    service = ResearchCouncilService()
+    rows = []
+    for model_code in model_codes:
+        preview = service.build_preview(instrument_id=instrument_id, timeframe=timeframe, model_code=model_code)
+        decision = dict(preview.get("decision") or {})
+        context = dict(preview.get("context") or {})
+        rows.append(
+            {
+                "model_code": model_code,
+                "overall_stance": decision.get("overall_stance"),
+                "committee_confidence": decision.get("committee_confidence"),
+                "follow_up_check_count": len(list(decision.get("follow_up_checks") or [])),
+                "disagreement_count": len(list(decision.get("disagreement_points") or [])),
+                "lesson_count": len(list(context.get("compact_lessons") or [])),
+            }
+        )
+    payload = {
+        "instrument_id": instrument_id,
+        "timeframe": timeframe,
+        "rows": rows,
+        "passed": bool(rows),
+    }
+    _print_payload(payload, output_json=output_json)
+    return 0 if rows else 2
+
+
+def _resolve_model_codes(*, env, instrument_id: int) -> list[str]:
+    explicit = [item.strip() for item in str(env.get("RESEARCH_COUNCIL_ARENA_MODEL_CODES") or "").split(",") if item.strip()]
+    if explicit:
+        return explicit
+    trust_summary = ModelTrustService().build_summary(instrument_id=instrument_id)
+    models = [str(item.get("model_code") or "").strip() for item in list(trust_summary.get("models") or []) if str(item.get("model_code") or "").strip()]
+    if models:
+        return models[:3]
+    return ["OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE"]
+
+
+def _print_payload(payload: dict, *, output_json: bool) -> None:
+    if output_json:
+        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
+        return
+    print(f"Research council arena for instrument {payload.get('instrument_id')}")
+    print(f"Rows: {len(payload.get('rows') or [])}")
+
+
+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())

scripts/run_research_council_backfill_lessons.py

@@ -0,0 +1,45 @@
+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.research_council import ResearchCouncilService
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    instrument_id = int(env.get("RESEARCH_COUNCIL_INSTRUMENT_ID", "1"))
+    limit = int(env.get("RESEARCH_COUNCIL_LESSON_LIMIT", "100"))
+    output_json = _get_bool(env, "RESEARCH_COUNCIL_OUTPUT_JSON", False)
+    payload = ResearchCouncilService().backfill_lessons(instrument_id=instrument_id, limit=limit)
+    payload["passed"] = True
+    _print_payload(payload, output_json=output_json)
+    return 0
+
+
+def _print_payload(payload: dict, *, output_json: bool) -> None:
+    if output_json:
+        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
+        return
+    totals = dict(payload.get("totals") or {})
+    print(f"Research council lessons backfill completed for instrument {payload.get('instrument_id')}")
+    print(
+        f"Processed={totals.get('requested', 0)} created={totals.get('created', 0)} updated={totals.get('updated', 0)}"
+    )
+
+
+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())

scripts/run_research_council_eval.py

@@ -0,0 +1,96 @@
+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.research_council import ResearchCouncilService
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    instrument_id = int(env.get("RESEARCH_COUNCIL_INSTRUMENT_ID", "1"))
+    limit = int(env.get("RESEARCH_COUNCIL_EVAL_LIMIT", "20"))
+    output_json = _get_bool(env, "RESEARCH_COUNCIL_OUTPUT_JSON", False)
+    runs = ResearchCouncilService().get_recent_runs(instrument_id=instrument_id, limit=limit)
+    payload = build_eval_summary(instrument_id=instrument_id, runs=runs)
+    payload["passed"] = bool(payload.get("run_count"))
+    _print_payload(payload, output_json=output_json)
+    return 0 if payload["passed"] else 2
+
+
+def build_eval_summary(*, instrument_id: int, runs: list[dict]) -> dict:
+    if not runs:
+        return {
+            "instrument_id": instrument_id,
+            "run_count": 0,
+            "error": "no_research_council_runs",
+        }
+    baseline_counts = []
+    follow_up_counts = []
+    question_counts = []
+    disagreement_count = 0
+    lesson_usage_count = 0
+    session_scores: dict[str, list[float]] = {}
+    setup_scores: dict[str, list[float]] = {}
+
+    for row in runs:
+        payload = dict(row.get("payload") or {})
+        context = dict(payload.get("context") or {})
+        decision = dict(payload.get("decision") or {})
+        candidate = dict(context.get("latest_candidate") or {})
+        trust_policy = dict(context.get("trust_policy") or {})
+        risk_result = dict(context.get("risk_result") or {})
+        baseline = len(list(candidate.get("missing_confirmations") or [])) + len(list(trust_policy.get("blocking_reasons") or [])) + len(list(risk_result.get("reasons") or []))
+        baseline_counts.append(baseline)
+        follow_up_counts.append(len(list(decision.get("follow_up_checks") or [])))
+        question_counts.append(len(list(decision.get("operator_questions") or [])))
+        disagreement_count += 1 if list(decision.get("disagreement_points") or []) else 0
+        lessons = list(context.get("compact_lessons") or [])
+        if lessons:
+            lesson_usage_count += 1
+        session_code = str(candidate.get("session_code") or "UNKNOWN")
+        setup_type = str(candidate.get("model_code") or "UNKNOWN")
+        session_scores.setdefault(session_code, []).append(float(decision.get("committee_confidence") or 0.0))
+        setup_scores.setdefault(setup_type, []).append(float(decision.get("committee_confidence") or 0.0))
+
+    return {
+        "instrument_id": instrument_id,
+        "run_count": len(runs),
+        "avg_baseline_issue_count": _avg(baseline_counts),
+        "avg_follow_up_check_count": _avg(follow_up_counts),
+        "avg_operator_question_count": _avg(question_counts),
+        "guidance_delta": round(_avg(follow_up_counts) - _avg(baseline_counts), 4),
+        "disagreement_hint_rate": round(disagreement_count / len(runs), 4),
+        "lesson_usage_rate": round(lesson_usage_count / len(runs), 4),
+        "session_confidence": {key: round(_avg(values), 4) for key, values in session_scores.items()},
+        "setup_confidence": {key: round(_avg(values), 4) for key, values in setup_scores.items()},
+    }
+
+
+def _avg(values: list[float]) -> float:
+    return round(sum(values) / len(values), 4) if values else 0.0
+
+
+def _print_payload(payload: dict, *, output_json: bool) -> None:
+    if output_json:
+        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
+        return
+    print(f"Research council eval for instrument {payload.get('instrument_id')}")
+    print(f"Runs: {payload.get('run_count')}")
+
+
+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())

scripts/run_research_council_governance_tick.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from pathlib import Path
+import json
+import os
+import sys
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from src.services.research_council import CouncilGovernanceTickService
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    instrument_id = int(env.get("RESEARCH_COUNCIL_INSTRUMENT_ID", "1"))
+    limit = int(env.get("RESEARCH_COUNCIL_REVIEW_LIMIT", "5"))
+    stale_after_hours = float(env.get("RESEARCH_COUNCIL_STALE_AFTER_HOURS", "24"))
+    auto_run_stale_cycle = _get_bool(env, "RESEARCH_COUNCIL_GOVERNANCE_AUTO_RUN_STALE", False)
+    force_cycle = _get_bool(env, "RESEARCH_COUNCIL_GOVERNANCE_FORCE_CYCLE", False)
+    timeframe = env.get("RESEARCH_COUNCIL_TIMEFRAME", "1m")
+    model_code = env.get("RESEARCH_COUNCIL_MODEL_CODE", "OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE")
+    replay_file = str(env.get("RESEARCH_COUNCIL_REPLAY_FILE") or "").strip() or None
+    eval_limit = int(env.get("RESEARCH_COUNCIL_EVAL_LIMIT", "20"))
+    lesson_limit = int(env.get("RESEARCH_COUNCIL_LESSON_LIMIT", "100"))
+    cycle_key = str(env.get("RESEARCH_COUNCIL_OPTIMIZATION_CYCLE_KEY") or "").strip() or None
+    output_json = _get_bool(env, "RESEARCH_COUNCIL_OUTPUT_JSON", False)
+    payload = CouncilGovernanceTickService().run_tick(
+        instrument_id=instrument_id,
+        limit=limit,
+        stale_after_hours=stale_after_hours,
+        auto_run_stale_cycle=auto_run_stale_cycle,
+        force_cycle=force_cycle,
+        timeframe=timeframe,
+        model_code=model_code,
+        replay_file=replay_file,
+        eval_limit=eval_limit,
+        lesson_limit=lesson_limit,
+        cycle_key=cycle_key,
+    )
+    _print_payload(payload, output_json=output_json)
+    return 0 if payload.get("passed") else 2
+
+
+def _print_payload(payload: dict, *, output_json: bool) -> None:
+    if output_json:
+        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
+        return
+    print(f"Research council governance tick {payload.get('status')} for instrument {payload.get('instrument_id')}")
+    print(f"Health: {payload.get('effective_health_state')}")
+    print(f"Actions: {', '.join(payload.get('recommended_actions') or [])}")
+
+
+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())

scripts/run_research_council_optimization_loop.py

@@ -0,0 +1,62 @@
+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.research_council import CouncilOptimizationService
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    instrument_id = int(env.get("RESEARCH_COUNCIL_INSTRUMENT_ID", "1"))
+    timeframe = env.get("RESEARCH_COUNCIL_TIMEFRAME", "1m")
+    model_code = env.get("RESEARCH_COUNCIL_MODEL_CODE", "OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE")
+    replay_file = str(env.get("RESEARCH_COUNCIL_REPLAY_FILE") or "").strip() or None
+    eval_limit = int(env.get("RESEARCH_COUNCIL_EVAL_LIMIT", "20"))
+    lesson_limit = int(env.get("RESEARCH_COUNCIL_LESSON_LIMIT", "100"))
+    cycle_key = str(env.get("RESEARCH_COUNCIL_OPTIMIZATION_CYCLE_KEY") or "").strip() or None
+    arena_model_codes = [
+        item.strip()
+        for item in str(env.get("RESEARCH_COUNCIL_ARENA_MODEL_CODES") or "").split(",")
+        if item.strip()
+    ]
+    output_json = _get_bool(env, "RESEARCH_COUNCIL_OUTPUT_JSON", False)
+
+    payload = CouncilOptimizationService().run_cycle(
+        instrument_id=instrument_id,
+        timeframe=timeframe,
+        model_code=model_code,
+        replay_file=replay_file,
+        eval_limit=eval_limit,
+        lesson_limit=lesson_limit,
+        arena_model_codes=arena_model_codes or None,
+        cycle_key=cycle_key,
+    )
+    _print_payload(payload, output_json=output_json)
+    return 0 if payload.get("passed") else 2
+
+
+def _print_payload(payload: dict, *, output_json: bool) -> None:
+    if output_json:
+        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
+        return
+    print(f"Research council optimization cycle {payload.get('status')} for instrument {payload.get('instrument_id')}")
+    print(f"Cycle: {payload.get('cycle_key')}")
+    print(f"Next actions: {', '.join(payload.get('next_actions') or [])}")
+
+
+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())

scripts/run_research_council_optimization_review.py

@@ -0,0 +1,48 @@
+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.research_council import CouncilOptimizationReviewService
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    instrument_id = int(env.get("RESEARCH_COUNCIL_INSTRUMENT_ID", "1"))
+    limit = int(env.get("RESEARCH_COUNCIL_REVIEW_LIMIT", "5"))
+    stale_after_hours = float(env.get("RESEARCH_COUNCIL_STALE_AFTER_HOURS", "24"))
+    output_json = _get_bool(env, "RESEARCH_COUNCIL_OUTPUT_JSON", False)
+    payload = CouncilOptimizationReviewService().build_review(
+        instrument_id=instrument_id,
+        limit=limit,
+        stale_after_hours=stale_after_hours,
+    )
+    payload["passed"] = bool(payload.get("present")) and payload.get("health_state") != "missing"
+    _print_payload(payload, output_json=output_json)
+    return 0 if payload["passed"] else 2
+
+
+def _print_payload(payload: dict, *, output_json: bool) -> None:
+    if output_json:
+        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
+        return
+    print(f"Research council optimization review for instrument {payload.get('instrument_id')}")
+    print(f"Health: {payload.get('health_state')}")
+    print(f"Summary: {payload.get('operator_summary')}")
+
+
+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())

scripts/run_research_council_replay.py

@@ -0,0 +1,71 @@
+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.backtest import detect_confirmed_swings_for_replay, iter_historical_replay, load_replay_candles
+from src.services.research_council import ResearchCouncilService
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    instrument_id = int(env.get("RESEARCH_COUNCIL_INSTRUMENT_ID", "1"))
+    timeframe = env.get("RESEARCH_COUNCIL_TIMEFRAME", "1m")
+    model_code = env.get("RESEARCH_COUNCIL_MODEL_CODE", "OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE")
+    replay_file = str(env.get("RESEARCH_COUNCIL_REPLAY_FILE") or "").strip()
+    output_json = _get_bool(env, "RESEARCH_COUNCIL_OUTPUT_JSON", False)
+    service = ResearchCouncilService()
+    payload = {
+        "instrument_id": instrument_id,
+        "timeframe": timeframe,
+        "model_code": model_code,
+        "resume_state": service.get_resume_state(instrument_id=instrument_id),
+        "preview": service.build_preview(instrument_id=instrument_id, timeframe=timeframe, model_code=model_code),
+        "replay": None,
+    }
+    if replay_file:
+        payload["replay"] = build_replay_summary(replay_file)
+    payload["passed"] = True
+    _print_payload(payload, output_json=output_json)
+    return 0
+
+
+def build_replay_summary(path: str) -> dict:
+    candles = load_replay_candles(path)
+    frames = list(iter_historical_replay(candles))
+    swings = detect_confirmed_swings_for_replay(candles)
+    return {
+        "path": path,
+        "candle_count": len(candles),
+        "frame_count": len(frames),
+        "confirmed_swing_count": len(swings),
+        "first_ts": candles[0].get("ts_open") if candles else None,
+        "last_ts": candles[-1].get("ts_close") if candles else None,
+    }
+
+
+def _print_payload(payload: dict, *, output_json: bool) -> None:
+    if output_json:
+        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
+        return
+    preview = dict(payload.get("preview") or {})
+    decision = dict(preview.get("decision") or {})
+    print(f"Research council replay summary for instrument {payload.get('instrument_id')}")
+    print(f"Overall stance: {decision.get('overall_stance')}")
+
+
+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())

scripts/run_visual_review.py

@@ -0,0 +1,84 @@
+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.visual_watch import VisualReviewService
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    output_json = _get_bool(env, "VISUAL_REVIEW_OUTPUT_JSON", False)
+    service = VisualReviewService()
+    watch_run_id = _get_optional_int(env, "VISUAL_REVIEW_WATCH_RUN_ID")
+    if watch_run_id is not None:
+        payload = service.review_watch_run(
+            watch_run_id=watch_run_id,
+            snapshot_id=_get_optional_int(env, "VISUAL_REVIEW_SNAPSHOT_ID"),
+            image_path=_get_optional_value(env, "VISUAL_REVIEW_IMAGE_PATH"),
+            source_name=_get_optional_value(env, "VISUAL_REVIEW_SOURCE_NAME"),
+            visual_state=_get_optional_value(env, "VISUAL_REVIEW_STATE"),
+            visual_confidence=_get_optional_float(env, "VISUAL_REVIEW_CONFIDENCE"),
+            visual_notes=_get_optional_value(env, "VISUAL_REVIEW_NOTES") or "",
+        )
+    else:
+        payload = service.review_pending_watch_runs(
+            limit=_get_optional_int(env, "VISUAL_REVIEW_LIMIT") or 1,
+            instrument_id=_get_optional_int(env, "VISUAL_REVIEW_INSTRUMENT_ID"),
+            trigger_type=_get_optional_value(env, "VISUAL_REVIEW_TRIGGER_TYPE"),
+            image_path=_get_optional_value(env, "VISUAL_REVIEW_IMAGE_PATH"),
+            source_name=_get_optional_value(env, "VISUAL_REVIEW_SOURCE_NAME"),
+        )
+    if output_json:
+        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
+    else:
+        print(f"Visual review status: {payload.get('status')}")
+        print(f"Attempted: {payload.get('attempted', 1)} / Reviewed: {payload.get('reviewed', 1)}")
+        for item in payload.get("items") or []:
+            print(
+                f"- watch_run_id={item.get('watch_run_id')} "
+                f"state={item.get('visual_state')} snapshot_id={item.get('snapshot_id')}"
+            )
+            if item.get("snapshot_image_path"):
+                print(f"  image_path={item.get('snapshot_image_path')}")
+            if item.get("snapshot_artifact_path"):
+                print(f"  artifact_path={item.get('snapshot_artifact_path')}")
+    return 0 if payload.get("status") not in {"failed"} else 2
+
+
+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"}
+
+
+def _get_optional_int(env, key: str):
+    value = env.get(key)
+    if value is None or value == "":
+        return None
+    return int(value)
+
+
+def _get_optional_float(env, key: str):
+    value = env.get(key)
+    if value is None or value == "":
+        return None
+    return float(value)
+
+
+def _get_optional_value(env, key: str):
+    value = env.get(key)
+    if value is None or value == "":
+        return None
+    return value
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/run_visual_snapshot_capture.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from pathlib import Path
+import json
+import os
+import sys
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from src.api.visual_watch import capture_visual_watch_snapshot
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    output_json = _get_bool(env, "VISUAL_CAPTURE_OUTPUT_JSON", False)
+    payload = capture_visual_watch_snapshot(
+        watch_run_id=_get_optional_int(env, "VISUAL_CAPTURE_WATCH_RUN_ID"),
+        instrument_id=_get_optional_int(env, "VISUAL_CAPTURE_INSTRUMENT_ID"),
+        trigger_type=_get_optional_value(env, "VISUAL_CAPTURE_TRIGGER_TYPE"),
+        source_name=_get_optional_value(env, "VISUAL_CAPTURE_SOURCE_NAME"),
+        output_dir=_get_optional_value(env, "VISUAL_CAPTURE_OUTPUT_DIR"),
+    )
+    if output_json:
+        print(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str))
+    else:
+        print(f"Visual snapshot status: {payload.get('status')}")
+        print(f"Watch run: {payload.get('watch_run_id')}")
+        if payload.get("snapshot_id") is not None:
+            print(f"Snapshot id: {payload.get('snapshot_id')}")
+        if payload.get("image_path"):
+            print(f"Image path: {payload.get('image_path')}")
+        if payload.get("artifact_path"):
+            print(f"Artifact path: {payload.get('artifact_path')}")
+        if payload.get("reason"):
+            print(f"Reason: {payload.get('reason')}")
+    return 0 if payload.get("status") not in {"failed"} else 2
+
+
+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"}
+
+
+def _get_optional_int(env, key: str):
+    value = env.get(key)
+    if value is None or value == "":
+        return None
+    return int(value)
+
+
+def _get_optional_value(env, key: str):
+    value = env.get(key)
+    if value is None or value == "":
+        return None
+    return value
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/run_webhook_capture.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+from pathlib import Path
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+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))
+
+CAPTURE_FILE = ROOT / "tmp_webhook_capture.jsonl"
+
+
+class CaptureHandler(BaseHTTPRequestHandler):
+    def do_POST(self) -> None:
+        try:
+            length = int(self.headers.get("Content-Length", "0"))
+        except ValueError:
+            length = 0
+        raw_body = self.rfile.read(length) if length else b""
+        record = {
+            "path": self.path,
+            "headers": {key: value for key, value in self.headers.items()},
+            "body": _decode_json(raw_body),
+        }
+        with CAPTURE_FILE.open("a", encoding="utf-8") as handle:
+            handle.write(json.dumps(record, ensure_ascii=False) + "\n")
+        payload = json.dumps({"ok": True, "captured": True}).encode("utf-8")
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(payload)))
+        self.send_header("x-request-id", "local-capture")
+        self.end_headers()
+        self.wfile.write(payload)
+
+    def log_message(self, format: str, *args) -> None:
+        return
+
+
+def main(env=None) -> int:
+    env = os.environ if env is None else env
+    host = env.get("WEBHOOK_CAPTURE_HOST", "127.0.0.1")
+    port = int(env.get("WEBHOOK_CAPTURE_PORT", "8787"))
+    server = ThreadingHTTPServer((host, port), CaptureHandler)
+    print(f"Webhook capture listening on http://{host}:{port}")
+    server.serve_forever()
+    return 0
+
+
+def _decode_json(raw_body: bytes):
+    if not raw_body:
+        return {}
+    try:
+        return json.loads(raw_body.decode("utf-8"))
+    except Exception:
+        return {"raw": raw_body.decode("utf-8", errors="replace")}
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())