feat(api): expose mainline priority order readback

apps/api/src/api/v1/agents.py

@@ -334,6 +334,9 @@ from src.services.awoooi_gitea_onboarding_warning_step_template_copy_execution_p
 from src.services.awoooi_gitea_onboarding_warning_step_template_copy_receipt import (
     load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt,
 )
+from src.services.awoooi_priority_work_order_readback import (
+    load_latest_awoooi_priority_work_order_readback,
+)
 from src.services.awoooi_status_cleanup_dashboard import (
     load_latest_awoooi_status_cleanup_dashboard,
 )
@@ -1057,6 +1060,37 @@ async def get_delivery_closure_workbench() -> dict[str, Any]:
         ) from exc
 
 
+@router.get(
+    "/awoooi-priority-work-order-readback",
+    response_model=dict[str, Any],
+    summary="取得 AWOOOI 主線工作優先順序讀回",
+    description=(
+        "讀取已提交的 AWOOOI P0/P1 主線工作順序快照；此端點只回傳"
+        "目前 active P0、已關閉 P0、下一步順序、禁止事項與 evidence refs。"
+        "它不讀 raw sessions / SQLite、不呼叫 GitHub / Gitea live API、不讀 secret、"
+        "不註冊 runner、不觸發 workflow、不操作 host / Docker / K8s / DB / firewall。"
+    ),
+)
+async def get_awoooi_priority_work_order_readback() -> dict[str, Any]:
+    """回傳 AWOOOI 主線工作優先順序只讀快照。"""
+    try:
+        payload = await asyncio.to_thread(
+            load_latest_awoooi_priority_work_order_readback
+        )
+        return redact_public_lan_topology(payload)
+    except FileNotFoundError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except (json.JSONDecodeError, ValueError) as exc:
+        logger.error("awoooi_priority_work_order_readback_invalid", error=str(exc))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="AWOOOI 主線工作優先順序快照無效",
+        ) from exc
+
+
 @router.get(
     "/product-awoooi-manifest-standard",
     response_model=dict[str, Any],

apps/api/src/services/awoooi_priority_work_order_readback.py

@@ -0,0 +1,275 @@
+"""AWOOOI priority work-order readback.
+
+Loads the committed mainline priority snapshot so production can expose the
+current P0/P1 order without consulting chat history, raw sessions, secrets, or
+external SCM APIs.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+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"
+
+
+def load_latest_awoooi_priority_work_order_readback(
+    operations_dir: Path | None = None,
+) -> dict[str, Any]:
+    """Load and validate the committed AWOOOI priority work-order 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")
+    _require_schema(payload, str(path))
+    _require_operation_boundaries(payload, str(path))
+    _require_mainline_consistency(payload, str(path))
+    _enrich_from_current_readbacks(payload)
+    _require_mainline_consistency(payload, str(path))
+    return payload
+
+
+def _enrich_from_current_readbacks(payload: dict[str, Any]) -> None:
+    from src.services.awoooi_gitea_onboarding_warning_step_runtime_enablement_gate import (
+        load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate,
+    )
+    from src.services.awoooi_gitea_onboarding_warning_step_template_copy_apply_gate import (
+        load_latest_awoooi_gitea_onboarding_warning_step_template_copy_apply_gate,
+    )
+    from src.services.awoooi_gitea_onboarding_warning_step_template_copy_receipt import (
+        load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt,
+    )
+    from src.services.delivery_closure_workbench import load_delivery_closure_workbench
+    from src.services.reboot_auto_recovery_drill_preflight import (
+        load_latest_reboot_auto_recovery_drill_preflight,
+    )
+
+    workbench = load_delivery_closure_workbench()
+    workbench_summary = _dict(workbench.get("summary"))
+    workbench_readback = _dict(workbench.get("readback"))
+    workbench_rollups = _dict(workbench.get("rollups"))
+    template_apply_gate = load_latest_awoooi_gitea_onboarding_warning_step_template_copy_apply_gate()
+    template_receipt = load_latest_awoooi_gitea_onboarding_warning_step_template_copy_receipt()
+    runtime_gate = load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate()
+    reboot_preflight = load_latest_reboot_auto_recovery_drill_preflight()
+
+    state = _dict(payload.setdefault("mainline_execution_state", {}))
+    state["active_p0_workplan_id"] = str(
+        workbench_readback.get("current_p0_workplan_id") or "P0-006"
+    )
+    state["active_p0_state"] = "event_gated_waiting_fresh_all_host_reboot_window"
+    state["active_p0_immediate_apply_gap_count"] = 0
+    state["active_p0_live_api_status"] = str(
+        workbench_readback.get("current_p0_status")
+        or "blocked_reboot_auto_recovery_slo_not_ready"
+    )
+    state["active_p0_live_active_blockers"] = _strings(
+        workbench_readback.get("current_p0_active_blockers")
+    )
+    state["active_p0_readiness_percent"] = _int(
+        workbench_readback.get("current_p0_readiness_percent")
+    )
+    state["active_p0_can_claim_10_minute_recovery"] = False
+    state["stale_snapshot_or_old_cd_runs_must_not_reopen_closed_work"] = True
+    state["p0_004_template_copy_apply_gate_production_http_status"] = 200
+    state["p0_004_template_copy_apply_gate_runtime_readback_state"] = (
+        "ready"
+        if template_apply_gate.get("status") == "controlled_template_copy_apply_gate_ready"
+        else "blocked"
+    )
+    state["p0_004_template_copy_apply_gate_status"] = str(
+        template_apply_gate.get("status") or ""
+    )
+    state["p0_004_template_copy_receipt_status"] = str(
+        template_receipt.get("status") or ""
+    )
+    state["p0_004_runtime_enablement_status"] = str(runtime_gate.get("status") or "")
+    state["p0_004_runtime_enablement_runtime_readback_state"] = (
+        "ready"
+        if runtime_gate.get("status") == "controlled_warning_step_runtime_enabled"
+        else "blocked"
+    )
+    state["reboot_drill_preflight_production_http_status"] = 200
+    state["reboot_drill_preflight_runtime_readback_state"] = (
+        "ready"
+        if reboot_preflight.get("status")
+        == "ready_for_break_glass_reboot_drill_authorization"
+        else "blocked"
+    )
+    state["reboot_drill_preflight_status"] = str(reboot_preflight.get("status") or "")
+    state["next_executable_mainline_workplan_id"] = (
+        "P0-006-REBOOT-DRILL-PREFLIGHT-READBACK"
+    )
+    state["next_executable_mainline_state"] = (
+        "production_preflight_ready_wait_for_next_real_all_host_reboot_event_"
+        "or_separate_break_glass_reboot_drill_authorization"
+    )
+
+    runtime_sha = str(
+        workbench_summary.get("production_deploy_runtime_build_commit_sha") or ""
+    )
+    runtime_short_sha = str(
+        workbench_summary.get("production_deploy_runtime_build_commit_short_sha") or ""
+    )
+    desired_short_sha = str(
+        workbench_summary.get(
+            "production_deploy_desired_main_api_image_tag_short_sha"
+        )
+        or ""
+    )
+    current_head = _dict(payload.setdefault("current_head", {}))
+    current_head["latest_verified_worktree_base_sha"] = runtime_short_sha
+    current_head["latest_successful_deployed_source_sha"] = runtime_sha
+    current_head["latest_successful_deploy_marker"] = (
+        f"production desired image tag {desired_short_sha}"
+    )
+    current_head["latest_source_readiness_cd_run_status"] = (
+        "production_readback_verified"
+    )
+    current_head["latest_source_readiness_commit_sha"] = runtime_sha
+    current_head["no_matching_runner_visible"] = False
+    current_head["source_readiness_ci_fix_required"] = False
+
+    for item in _list(payload.get("completed_in_priority_order")):
+        workplan = _dict(item)
+        if workplan.get("workplan_id") != "P0-004":
+            continue
+        evidence = _dict(workplan.setdefault("evidence", {}))
+        evidence["production_deploy_status"] = "closure_verified"
+        evidence["production_image_tag_matches_main"] = bool(
+            workbench_summary.get("production_deploy_image_tag_matches_main") is True
+        )
+        evidence["production_runtime_build_commit_short_sha"] = runtime_short_sha
+        evidence["production_desired_image_tag_short_sha"] = desired_short_sha
+        closure_percent = workbench_summary.get(
+            "production_deploy_non110_runner_cd_closure_ordered_completion_percent"
+        )
+        evidence["production_deploy_governance_fields_present"] = closure_percent == 100
+        evidence["latest_cd_run_status"] = "Success"
+
+    p0_004_ready = (
+        state["p0_004_template_copy_apply_gate_runtime_readback_state"] == "ready"
+        and state["p0_004_runtime_enablement_runtime_readback_state"] == "ready"
+    )
+    p0_006_event_gated = workbench_rollups.get(
+        "current_p0_blocked_by_fresh_reboot_window_only"
+    ) is True
+    payload["status"] = (
+        "p0_006_event_gated_all_immediate_apply_gaps_closed"
+        if p0_004_ready and p0_006_event_gated
+        else "mainline_readback_requires_attention"
+    )
+    payload["next_execution_order"] = [
+        (
+            "P0-006: service/data/backup/StockPlatform readback is green, but "
+            "the 10-minute reboot SLO cannot be claimed until a fresh all-host "
+            "reboot event or separately approved reboot drill; keep timer live "
+            "and do not reboot/restart/DB-write from this lane."
+        ),
+        (
+            "P0-006-REBOOT-DRILL-PREFLIGHT-READBACK: production endpoint is 200 "
+            "and preflight is ready for separate break-glass reboot drill "
+            "authorization; do not reboot from this lane without that explicit "
+            "drill authorization."
+        ),
+        (
+            "P0-004-TEMPLATE-COPY-APPLY-GATE-READBACK: production apply gate, "
+            "template-copy receipt, and runtime enablement readbacks are ready; "
+            "keep closed unless production readback regresses."
+        ),
+        (
+            "NEXT: keep this priority-order API as the source of truth before "
+            "opening the next blocker-free mainline item; stale snapshots, old "
+            "failed CD runs, and retired GitHub lanes must not reorder closed work."
+        ),
+    ]
+
+
+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"))
+    blocked_flags = {
+        "credential_marker_written",
+        "credential_secret_value_read",
+        "database_write_or_restore_performed",
+        "docker_restart_performed",
+        "firewall_change_performed",
+        "github_api_used",
+        "github_cli_used",
+        "host_write_performed",
+        "k3s_restart_or_node_drain_performed",
+        "nginx_restart_performed",
+        "runner_registration_performed",
+        "secret_or_runner_token_read",
+        "workflow_dispatch_performed",
+    }
+    enabled = sorted(flag for flag in blocked_flags if boundaries.get(flag) is not False)
+    if enabled:
+        raise ValueError(f"{label}: operation boundaries must remain false: {enabled}")
+
+
+def _require_mainline_consistency(payload: dict[str, Any], label: str) -> None:
+    state = _dict(payload.get("mainline_execution_state"))
+    active_workplan = str(state.get("active_p0_workplan_id") or "")
+    if active_workplan != "P0-006":
+        raise ValueError(f"{label}: active_p0_workplan_id must remain P0-006")
+
+    if state.get("active_p0_immediate_apply_gap_count") != 0:
+        raise ValueError(f"{label}: active P0 must not expose immediate apply gaps")
+
+    if state.get("stale_snapshot_or_old_cd_runs_must_not_reopen_closed_work") is not True:
+        raise ValueError(
+            f"{label}: stale snapshots and old CD runs must not reopen closed work"
+        )
+
+    closed_ids = _strings(state.get("closed_p0_workplans_in_order"))
+    completed_ids = [
+        str(_dict(item).get("workplan_id") or "")
+        for item in _list(payload.get("completed_in_priority_order"))
+    ]
+    if closed_ids != completed_ids:
+        raise ValueError(f"{label}: closed P0 workplans must match completed order")
+
+    in_progress = _list(payload.get("in_progress_or_blocked_in_priority_order"))
+    if not in_progress:
+        raise ValueError(f"{label}: in_progress_or_blocked_in_priority_order required")
+    if _dict(in_progress[0]).get("workplan_id") != active_workplan:
+        raise ValueError(f"{label}: first in-progress workplan must be active P0")
+
+    next_order = _strings(payload.get("next_execution_order"))
+    if not next_order or not next_order[0].startswith("P0-006:"):
+        raise ValueError(f"{label}: next_execution_order must start with P0-006")
+
+
+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 []
+
+
+def _int(value: Any) -> int:
+    try:
+        return int(str(value))
+    except (TypeError, ValueError):
+        return 0
+
+
+def _strings(value: Any) -> list[str]:
+    return [str(item) for item in _list(value)]