feat(delivery): expose p0 cicd baseline readiness

apps/api/src/services/delivery_closure_workbench.py

@@ -21,6 +21,9 @@ from src.services.backup_dr_readiness_matrix import (
 from src.services.gitea_workflow_runner_health import (
     load_latest_gitea_workflow_runner_health,
 )
+from src.services.p0_cicd_baseline_source_readiness import (
+    load_latest_p0_cicd_baseline_source_readiness,
+)
 from src.services.runtime_surface_inventory import (
     load_latest_runtime_surface_inventory,
 )
@@ -33,6 +36,7 @@ def load_delivery_closure_workbench() -> dict[str, Any]:
     status_cleanup = load_latest_awoooi_status_cleanup_dashboard()
     production_deploy = load_latest_awoooi_production_deploy_readback_blocker()
     github = _load_github_private_backup_evidence_gate()
+    cicd_baseline = load_latest_p0_cicd_baseline_source_readiness()
     gitea = load_latest_gitea_workflow_runner_health()
     runtime = load_latest_runtime_surface_inventory()
     backup = load_latest_backup_dr_readiness_matrix()
@@ -40,6 +44,7 @@ def load_delivery_closure_workbench() -> dict[str, Any]:
         status_cleanup=status_cleanup,
         production_deploy=production_deploy,
         github=github,
+        cicd_baseline=cicd_baseline,
         gitea=gitea,
         runtime=runtime,
         backup=backup,
@@ -51,6 +56,7 @@ def build_delivery_closure_workbench(
     status_cleanup: dict[str, Any],
     production_deploy: dict[str, Any],
     github: dict[str, Any],
+    cicd_baseline: dict[str, Any],
     gitea: dict[str, Any],
     runtime: dict[str, Any],
     backup: dict[str, Any],
@@ -65,6 +71,8 @@ def build_delivery_closure_workbench(
         github_preflight.get("internal_governance_writeback")
         or github.get("internal_governance_writeback")
     )
+    cicd_baseline_readback = _dict(cicd_baseline.get("readback"))
+    cicd_baseline_rollups = _dict(cicd_baseline.get("rollups"))
     production_deploy_readback = _dict(production_deploy.get("readback"))
     production_deploy_rollups = _dict(production_deploy.get("rollups"))
     gitea_status = _dict(gitea.get("program_status"))
@@ -410,6 +418,51 @@ def build_delivery_closure_workbench(
                 or _first_target_action(github.get("targets"))
             ),
         },
+        {
+            "id": "cicd_baseline",
+            "source_id": "p0_cicd_baseline_source_readiness",
+            "completion_percent": _percent(
+                cicd_baseline_rollups.get("source_readiness_percent")
+            ),
+            "status": str(cicd_baseline.get("status") or "unknown"),
+            "blocker_count": _int(
+                cicd_baseline_rollups.get("missing_required_source_count")
+            ),
+            "metric": {
+                "kind": "source_readiness",
+                "workplan_id": str(
+                    cicd_baseline_readback.get("workplan_id") or "P0-004"
+                ),
+                "required_source_count": _int(
+                    cicd_baseline_rollups.get("required_source_count")
+                ),
+                "present_required_source_count": _int(
+                    cicd_baseline_rollups.get("present_required_source_count")
+                ),
+                "missing_required_source_count": _int(
+                    cicd_baseline_rollups.get("missing_required_source_count")
+                ),
+                "source_readiness_percent": _int(
+                    cicd_baseline_rollups.get("source_readiness_percent")
+                ),
+                "blocked_source_ids": _strings(
+                    cicd_baseline_rollups.get("blocked_source_ids")
+                ),
+                "workflow_modification_allowed": _dict(
+                    cicd_baseline.get("operation_boundaries")
+                ).get("workflow_modification_allowed")
+                is True,
+                "workflow_trigger_allowed": _dict(
+                    cicd_baseline.get("operation_boundaries")
+                ).get("workflow_trigger_allowed")
+                is True,
+                "safe_next_step": str(
+                    cicd_baseline_readback.get("safe_next_step") or ""
+                ),
+            },
+            "href": "/deployments",
+            "next_action": _first_string(cicd_baseline.get("next_actions")),
+        },
         {
             "id": "gitea",
             "source_id": "gitea_ci_cd",
@@ -521,6 +574,7 @@ def build_delivery_closure_workbench(
         _source_status("status_cleanup", status_cleanup),
         _source_status("production_deploy_readback", production_deploy),
         _source_status("github_private_backup", github),
+        _source_status("p0_cicd_baseline_source_readiness", cicd_baseline),
         _source_status("gitea_ci_cd", gitea),
         _source_status("runtime_surface", runtime),
         _source_status("backup_dr", backup),
@@ -569,6 +623,28 @@ def build_delivery_closure_workbench(
                 "workflow_trigger_authorized"
             )
             is True,
+            "p0_cicd_baseline_status": str(cicd_baseline.get("status") or ""),
+            "p0_cicd_baseline_workplan_id": str(
+                cicd_baseline_readback.get("workplan_id") or ""
+            ),
+            "p0_cicd_baseline_source_readiness_percent": _int(
+                cicd_baseline_rollups.get("source_readiness_percent")
+            ),
+            "p0_cicd_baseline_required_source_count": _int(
+                cicd_baseline_rollups.get("required_source_count")
+            ),
+            "p0_cicd_baseline_present_required_source_count": _int(
+                cicd_baseline_rollups.get("present_required_source_count")
+            ),
+            "p0_cicd_baseline_missing_required_source_count": _int(
+                cicd_baseline_rollups.get("missing_required_source_count")
+            ),
+            "p0_cicd_baseline_blocked_source_ids": _strings(
+                cicd_baseline_rollups.get("blocked_source_ids")
+            ),
+            "p0_cicd_baseline_safe_next_step": str(
+                cicd_baseline_readback.get("safe_next_step") or ""
+            ),
             "production_deploy_status": str(production_deploy.get("status") or ""),
             "production_deploy_source_control_main_ready": production_deploy_rollups.get(
                 "source_control_main_ready"

apps/api/src/services/p0_cicd_baseline_source_readiness.py

@@ -0,0 +1,133 @@
+"""P0-004 CI/CD baseline source readiness.
+
+Loads the committed source-readiness snapshot for the dev/prod CI/CD baseline.
+This is read-only: it checks committed source paths and does not modify
+workflows, trigger Gitea, create repos, sync refs, read secrets, or touch hosts.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from src.services.snapshot_paths import default_operations_dir, resolve_repo_root
+
+_DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
+_SNAPSHOT_FILE = "p0-cicd-baseline-source-readiness.snapshot.json"
+_SCHEMA_VERSION = "p0_cicd_baseline_source_readiness_v1"
+
+
+def load_latest_p0_cicd_baseline_source_readiness(
+    operations_dir: Path | None = None,
+) -> dict[str, Any]:
+    """Load the committed P0-004 CI/CD baseline source readiness snapshot."""
+    directory = operations_dir or _DEFAULT_OPERATIONS_DIR
+    path = directory / _SNAPSHOT_FILE
+    with path.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+
+    if not isinstance(payload, dict):
+        raise ValueError(f"{path}: expected JSON object")
+    repo_root = resolve_repo_root(path)
+    _enrich_source_presence(payload, repo_root)
+    _require_schema(payload, str(path))
+    _require_operation_boundaries(payload, str(path))
+    _require_rollup_consistency(payload, str(path))
+    return payload
+
+
+def _enrich_source_presence(payload: dict[str, Any], repo_root: Path) -> None:
+    sources = _list(payload.get("required_sources"))
+    required_sources = [source for source in sources if _dict(source).get("required") is True]
+    present_ids: list[str] = []
+    missing_ids: list[str] = []
+    for source in required_sources:
+        item = _dict(source)
+        relative_path = str(item.get("path") or "")
+        present = bool(relative_path) and (repo_root / relative_path).is_file()
+        item["present"] = present
+        if present:
+            present_ids.append(str(item.get("id") or relative_path))
+        else:
+            missing_ids.append(str(item.get("id") or relative_path))
+
+    required_count = len(required_sources)
+    present_count = len(present_ids)
+    missing_count = len(missing_ids)
+    rollups = _dict(payload.get("rollups"))
+    rollups["required_source_count"] = required_count
+    rollups["present_required_source_count"] = present_count
+    rollups["missing_required_source_count"] = missing_count
+    rollups["source_readiness_percent"] = round(
+        present_count / max(required_count, 1) * 100
+    )
+    rollups["blocked_source_ids"] = missing_ids
+    rollups["hard_blocker_count"] = len(_list(payload.get("blockers")))
+    rollups["next_action_count"] = len(_list(payload.get("next_actions")))
+    payload["status"] = (
+        "ready_for_template_copy_apply_gate"
+        if missing_count == 0
+        else "blocked_required_sources_missing"
+    )
+
+
+def _require_schema(payload: dict[str, Any], label: str) -> None:
+    actual = payload.get("schema_version")
+    if actual != _SCHEMA_VERSION:
+        raise ValueError(
+            f"{label}: expected schema_version={_SCHEMA_VERSION}, got {actual!r}"
+        )
+
+
+def _require_operation_boundaries(payload: dict[str, Any], label: str) -> None:
+    boundaries = _dict(payload.get("operation_boundaries"))
+    if boundaries.get("read_only_api_allowed") is not True:
+        raise ValueError(f"{label}: read_only_api_allowed must be true")
+    blocked_flags = {
+        "workflow_modification_allowed",
+        "workflow_trigger_allowed",
+        "repo_creation_allowed",
+        "refs_sync_allowed",
+        "github_api_allowed",
+        "host_or_k8s_write_allowed",
+        "secret_read_allowed",
+        "raw_session_or_sqlite_read_allowed",
+    }
+    allowed = sorted(flag for flag in blocked_flags if boundaries.get(flag) is not False)
+    if allowed:
+        raise ValueError(f"{label}: operation boundaries must remain false: {allowed}")
+
+
+def _require_rollup_consistency(payload: dict[str, Any], label: str) -> None:
+    sources = [
+        source
+        for source in _list(payload.get("required_sources"))
+        if _dict(source).get("required") is True
+    ]
+    blockers = _list(payload.get("blockers"))
+    next_actions = _list(payload.get("next_actions"))
+    rollups = _dict(payload.get("rollups"))
+    missing_ids = [
+        str(_dict(source).get("id") or _dict(source).get("path") or "")
+        for source in sources
+        if _dict(source).get("present") is not True
+    ]
+    if rollups.get("required_source_count") != len(sources):
+        raise ValueError(f"{label}: required_source_count must match sources")
+    if rollups.get("missing_required_source_count") != len(missing_ids):
+        raise ValueError(f"{label}: missing_required_source_count mismatch")
+    if rollups.get("blocked_source_ids") != missing_ids:
+        raise ValueError(f"{label}: blocked_source_ids must match missing sources")
+    if rollups.get("hard_blocker_count") != len(blockers):
+        raise ValueError(f"{label}: hard_blocker_count must match blockers")
+    if rollups.get("next_action_count") != len(next_actions):
+        raise ValueError(f"{label}: next_action_count must match next_actions")
+
+
+def _dict(value: Any) -> dict[str, Any]:
+    return value if isinstance(value, dict) else {}
+
+
+def _list(value: Any) -> list[Any]:
+    return value if isinstance(value, list) else []