feat(iwooos): show controlled cd lane live metric proof

apps/api/src/services/awoooi_priority_work_order_readback.py

@@ -9,6 +9,8 @@ from __future__ import annotations
 
 import json
 import re
+import urllib.error
+import urllib.request
 from datetime import datetime
 from pathlib import Path
 from typing import Any
@@ -19,7 +21,15 @@ from src.services.snapshot_paths import default_operations_dir
 _DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
 _SNAPSHOT_FILE = "awoooi-priority-work-order-readback.snapshot.json"
 _SCHEMA_VERSION = "awoooi_priority_work_order_readback_v1"
+_CONTROLLED_CD_LANE_LIVE_METRIC_SCHEMA_VERSION = (
+    "controlled_cd_lane_live_metric_readback_v1"
+)
+_CONTROLLED_CD_LANE_NODE_EXPORTER_URL = "http://192.168.0.110:9100/metrics"
 _SHA_RE = re.compile(r"^[0-9a-f]{40}$")
+_METRIC_LINE_RE = re.compile(
+    r"^(?P<name>[a-zA-Z_:][a-zA-Z0-9_:]*)(?:\{[^}]*\})?\s+"
+    r"(?P<value>[-+]?(?:[0-9]+(?:\.[0-9]*)?|\.[0-9]+)(?:[eE][-+]?[0-9]+)?)$"
+)
 
 
 def load_latest_awoooi_priority_work_order_readback(
@@ -41,6 +51,141 @@ def load_latest_awoooi_priority_work_order_readback(
     return payload
 
 
+def load_controlled_cd_lane_live_metric_readback(
+    metrics_url: str = _CONTROLLED_CD_LANE_NODE_EXPORTER_URL,
+    timeout_seconds: float = 3.0,
+) -> dict[str, Any]:
+    """Read 110 node_exporter text metrics for the controlled CD lane.
+
+    This is a read-only HTTP metrics probe. It does not SSH, read runner
+    registration contents, touch Docker/systemd, or perform runtime writes.
+    """
+    try:
+        request = urllib.request.Request(metrics_url, method="GET")
+        with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
+            raw = response.read(2_000_000).decode("utf-8", errors="replace")
+    except (OSError, TimeoutError, urllib.error.URLError) as exc:
+        return {
+            "schema_version": _CONTROLLED_CD_LANE_LIVE_METRIC_SCHEMA_VERSION,
+            "status": "controlled_cd_lane_live_metric_unavailable",
+            "source": "node_exporter_text_metrics",
+            "source_url": metrics_url,
+            "metadata_only": True,
+            "active_blockers": ["controlled_cd_lane_live_metric_unavailable"],
+            "error_class": exc.__class__.__name__,
+            "readback": {
+                "enforcer_metric_present": False,
+                "controlled_drain_active_allowed": False,
+                "controlled_drain_staging_allowed": False,
+                "enforcer_last_run_fresh": False,
+            },
+        }
+
+    metrics = _parse_prometheus_text_metrics(raw)
+    readback = _build_controlled_cd_lane_metric_readback(metrics)
+    active_blockers: list[str] = []
+    if not readback["enforcer_metric_present"]:
+        active_blockers.append("controlled_cd_lane_enforcer_metric_missing")
+    if not readback["enforcer_last_run_fresh"]:
+        active_blockers.append("controlled_cd_lane_enforcer_metric_stale")
+    if not readback["controlled_drain_active_allowed"]:
+        active_blockers.append("controlled_cd_lane_active_not_allowed_yet")
+    if not readback["controlled_drain_staging_allowed"]:
+        active_blockers.append("controlled_cd_lane_staging_not_allowed_yet")
+
+    return {
+        "schema_version": _CONTROLLED_CD_LANE_LIVE_METRIC_SCHEMA_VERSION,
+        "status": (
+            "controlled_cd_lane_live_metric_readback_ready"
+            if readback["enforcer_metric_present"]
+            else "controlled_cd_lane_live_metric_missing"
+        ),
+        "source": "node_exporter_text_metrics",
+        "source_url": metrics_url,
+        "metadata_only": True,
+        "active_blockers": active_blockers,
+        "readback": readback,
+    }
+
+
+def apply_controlled_cd_lane_live_metric_readback(
+    payload: dict[str, Any],
+    live_metric_readback: dict[str, Any],
+) -> None:
+    """Overlay live controlled CD lane metric truth onto P0-006 readback."""
+    readback = _dict(live_metric_readback.get("readback"))
+    state = _dict(payload.setdefault("mainline_execution_state", {}))
+    active_blockers = _strings(live_metric_readback.get("active_blockers"))
+
+    metric_fields = {
+        "controlled_cd_lane_live_metric_status": str(
+            live_metric_readback.get("status") or "unknown"
+        ),
+        "controlled_cd_lane_live_metric_source": str(
+            live_metric_readback.get("source") or "node_exporter_text_metrics"
+        ),
+        "controlled_cd_lane_live_metric_metadata_only": bool(
+            live_metric_readback.get("metadata_only") is True
+        ),
+        "controlled_cd_lane_live_enforcer_metric_present": bool(
+            readback.get("enforcer_metric_present") is True
+        ),
+        "controlled_cd_lane_live_enforcer_last_run_timestamp": _float(
+            readback.get("enforcer_last_run_timestamp")
+        ),
+        "controlled_cd_lane_live_enforcer_last_run_fresh": bool(
+            readback.get("enforcer_last_run_fresh") is True
+        ),
+        "controlled_cd_lane_live_enforcer_apply_performed": _int(
+            readback.get("enforcer_apply_performed")
+        ),
+        "controlled_cd_lane_live_active_job_containers": _int(
+            readback.get("active_job_containers")
+        ),
+        "controlled_cd_lane_live_lane_process_count": _int(
+            readback.get("lane_process_count")
+        ),
+        "controlled_cd_lane_live_root_restore_sources_left": _int(
+            readback.get("root_restore_sources_left")
+        ),
+        "controlled_cd_lane_live_controlled_drain_active_allowed": bool(
+            readback.get("controlled_drain_active_allowed") is True
+        ),
+        "controlled_cd_lane_live_controlled_drain_staging_allowed": bool(
+            readback.get("controlled_drain_staging_allowed") is True
+        ),
+        "controlled_cd_lane_live_host_load5_per_core": _float(
+            readback.get("host_load5_per_core")
+        ),
+        "controlled_cd_lane_live_gitea_actions_active_container_count": _int(
+            readback.get("gitea_actions_active_container_count")
+        ),
+        "controlled_cd_lane_live_gitea_actions_active_process_count": _int(
+            readback.get("gitea_actions_active_process_count")
+        ),
+        "controlled_cd_lane_live_metric_blockers": active_blockers,
+    }
+    state.update(metric_fields)
+
+    rollups = _dict(payload.setdefault("rollups", {}))
+    rollups.update(metric_fields)
+    rollups["controlled_cd_lane_live_metric_blocker_count"] = len(active_blockers)
+
+    summary = _dict(payload.setdefault("summary", {}))
+    summary.update(metric_fields)
+    summary["controlled_cd_lane_live_metric_blocker_count"] = len(active_blockers)
+
+    for item in _list(payload.get("in_progress_or_blocked_in_priority_order")):
+        workplan = _dict(item)
+        if workplan.get("workplan_id") != "P0-006":
+            continue
+        evidence = _dict(workplan.setdefault("evidence", {}))
+        evidence.update(metric_fields)
+        evidence["controlled_cd_lane_live_metric_blocker_count"] = len(
+            active_blockers
+        )
+
+
 def apply_stockplatform_public_api_runtime_readback(
     payload: dict[str, Any],
     runtime_readback: dict[str, Any],
@@ -2116,6 +2261,87 @@ def _require_mainline_consistency(payload: dict[str, Any], label: str) -> None:
         raise ValueError(f"{label}: next_execution_order must start with P0-006")
 
 
+def _parse_prometheus_text_metrics(raw: str) -> dict[str, float]:
+    metrics: dict[str, float] = {}
+    allowed_names = {
+        "awoooi_host_gitea_actions_active_container_count",
+        "awoooi_host_gitea_actions_active_process_count",
+        "awoooi_host_load5_per_core",
+        "awoooi_runner_failclosed_enforcer_active_job_containers",
+        "awoooi_runner_failclosed_enforcer_apply_performed",
+        "awoooi_runner_failclosed_enforcer_controlled_drain_active_allowed",
+        "awoooi_runner_failclosed_enforcer_controlled_drain_staging_allowed",
+        "awoooi_runner_failclosed_enforcer_lane_process_count",
+        "awoooi_runner_failclosed_enforcer_last_run_timestamp",
+        "awoooi_runner_failclosed_enforcer_root_restore_sources_left",
+    }
+    for line in raw.splitlines():
+        if not line or line.startswith("#"):
+            continue
+        match = _METRIC_LINE_RE.match(line.strip())
+        if not match:
+            continue
+        name = match.group("name")
+        if name not in allowed_names:
+            continue
+        try:
+            metrics[name] = float(match.group("value"))
+        except ValueError:
+            continue
+    return metrics
+
+
+def _build_controlled_cd_lane_metric_readback(
+    metrics: dict[str, float],
+) -> dict[str, Any]:
+    last_run = metrics.get(
+        "awoooi_runner_failclosed_enforcer_last_run_timestamp", 0.0
+    )
+    now_seconds = datetime.now(ZoneInfo("Asia/Taipei")).timestamp()
+    enforcer_metric_present = (
+        "awoooi_runner_failclosed_enforcer_last_run_timestamp" in metrics
+    )
+    return {
+        "enforcer_metric_present": enforcer_metric_present,
+        "enforcer_last_run_timestamp": last_run,
+        "enforcer_last_run_age_seconds": max(0, int(now_seconds - last_run))
+        if last_run
+        else 0,
+        "enforcer_last_run_fresh": bool(last_run and now_seconds - last_run <= 900),
+        "enforcer_apply_performed": _int(
+            metrics.get("awoooi_runner_failclosed_enforcer_apply_performed")
+        ),
+        "active_job_containers": _int(
+            metrics.get("awoooi_runner_failclosed_enforcer_active_job_containers")
+        ),
+        "lane_process_count": _int(
+            metrics.get("awoooi_runner_failclosed_enforcer_lane_process_count")
+        ),
+        "root_restore_sources_left": _int(
+            metrics.get("awoooi_runner_failclosed_enforcer_root_restore_sources_left")
+        ),
+        "controlled_drain_active_allowed": _int(
+            metrics.get(
+                "awoooi_runner_failclosed_enforcer_controlled_drain_active_allowed"
+            )
+        )
+        == 1,
+        "controlled_drain_staging_allowed": _int(
+            metrics.get(
+                "awoooi_runner_failclosed_enforcer_controlled_drain_staging_allowed"
+            )
+        )
+        == 1,
+        "host_load5_per_core": _float(metrics.get("awoooi_host_load5_per_core")),
+        "gitea_actions_active_container_count": _int(
+            metrics.get("awoooi_host_gitea_actions_active_container_count")
+        ),
+        "gitea_actions_active_process_count": _int(
+            metrics.get("awoooi_host_gitea_actions_active_process_count")
+        ),
+    }
+
+
 def _dict(value: Any) -> dict[str, Any]:
     return value if isinstance(value, dict) else {}
 
@@ -2131,6 +2357,13 @@ def _int(value: Any) -> int:
         return 0
 
 
+def _float(value: Any) -> float:
+    try:
+        return float(str(value))
+    except (TypeError, ValueError):
+        return 0.0
+
+
 def _strings(value: Any) -> list[str]:
     return [str(item) for item in _list(value)]