feat(api): expose mainline priority order readback

.gitea/workflows/cd.yaml

@@ -358,6 +358,8 @@ jobs:
                 ;;
               apps/api/src/services/awoooi_onboarding_source_contracts.py)
                 ;;
+              apps/api/src/services/awoooi_priority_work_order_readback.py)
+                ;;
               apps/api/src/services/awoooi_product_onboarding_guard.py)
                 ;;
               apps/api/src/services/p0_cicd_baseline_source_readiness.py)
@@ -438,6 +440,8 @@ jobs:
                 ;;
               apps/api/tests/test_awoooi_production_deploy_readback_blocker.py)
                 ;;
+              apps/api/tests/test_awoooi_priority_work_order_readback_api.py)
+                ;;
               apps/api/tests/e2e_network_test.py)
                 ;;
               apps/api/tests/test_p0_cicd_baseline_source_readiness_api.py)
@@ -627,6 +631,7 @@ jobs:
               src/services/awoooi_new_product_onboarding_page_model.py \
               src/services/awoooi_onboarding_reminder_contract.py \
               src/services/awoooi_onboarding_source_contracts.py \
+              src/services/awoooi_priority_work_order_readback.py \
               src/services/awoooi_product_onboarding_guard.py \
               src/services/p0_cicd_baseline_source_readiness.py \
               src/services/product_awoooi_manifest_standard.py \
@@ -671,6 +676,7 @@ jobs:
               tests/test_reboot_auto_recovery_slo_scorecard_api.py \
               tests/test_iwooos_security_operating_system.py \
               tests/test_awoooi_production_deploy_readback_blocker.py \
+              tests/test_awoooi_priority_work_order_readback_api.py \
               tests/e2e_network_test.py::TestHMACVerification::test_valid_hmac_signature \
               tests/test_p0_cicd_baseline_source_readiness_api.py \
               tests/test_product_awoooi_manifest_standard_api.py \

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)]

apps/api/tests/test_awoooi_priority_work_order_readback_api.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from src.api.v1.agents import router
+from src.services.awoooi_priority_work_order_readback import (
+    load_latest_awoooi_priority_work_order_readback,
+)
+
+_REPO_ROOT = Path(__file__).resolve().parents[3]
+_SNAPSHOT_PATH = (
+    _REPO_ROOT
+    / "docs"
+    / "operations"
+    / "awoooi-priority-work-order-readback.snapshot.json"
+)
+
+
+def test_awoooi_priority_work_order_readback_loader_returns_mainline_order():
+    payload = load_latest_awoooi_priority_work_order_readback()
+
+    assert payload["schema_version"] == "awoooi_priority_work_order_readback_v1"
+    assert payload["mainline_execution_state"]["active_p0_workplan_id"] == "P0-006"
+    assert payload["mainline_execution_state"]["active_p0_immediate_apply_gap_count"] == 0
+    assert payload["mainline_execution_state"][
+        "stale_snapshot_or_old_cd_runs_must_not_reopen_closed_work"
+    ] is True
+    assert payload["next_execution_order"][0].startswith("P0-006:")
+    assert payload["operation_boundaries"]["github_api_used"] is False
+    assert payload["operation_boundaries"]["github_cli_used"] is False
+    assert payload["operation_boundaries"]["secret_or_runner_token_read"] is False
+    assert payload["operation_boundaries"]["workflow_dispatch_performed"] is False
+    assert payload["operation_boundaries"]["host_write_performed"] is False
+
+
+def test_awoooi_priority_work_order_readback_endpoint_returns_snapshot():
+    app = FastAPI()
+    app.include_router(router, prefix="/api/v1")
+    client = TestClient(app)
+
+    response = client.get("/api/v1/agents/awoooi-priority-work-order-readback")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["status"] == "p0_006_event_gated_all_immediate_apply_gaps_closed"
+    assert data["mainline_execution_state"]["active_p0_workplan_id"] == "P0-006"
+    assert data["mainline_execution_state"]["p0_004_template_copy_apply_gate_runtime_readback_state"] == "ready"
+    assert data["mainline_execution_state"]["reboot_drill_preflight_runtime_readback_state"] == "ready"
+    assert data["next_execution_order"][0].startswith("P0-006:")
+    assert "do not reboot" in data["next_execution_order"][0]
+
+
+def test_awoooi_priority_work_order_readback_rejects_reordered_active_p0(tmp_path):
+    operations_dir = tmp_path / "docs" / "operations"
+    operations_dir.mkdir(parents=True)
+    payload = json.loads(_SNAPSHOT_PATH.read_text(encoding="utf-8"))
+    payload["mainline_execution_state"]["active_p0_workplan_id"] = "P1-OPENCLAW"
+    (operations_dir / _SNAPSHOT_PATH.name).write_text(
+        json.dumps(payload),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="active_p0_workplan_id"):
+        load_latest_awoooi_priority_work_order_readback(operations_dir)

docs/LOGBOOK.md

@@ -1,3 +1,14 @@
+## 2026-06-30 — 08:55 P0 mainline priority-order production readback
+
+**照主線完成的實作**：
+- Gitea CD `#4012` 對 `88ac9c33d fix(ci): include deploy readback test in narrow cd profile` 已轉 `Success`；Gitea main 已有 deploy marker `b7bedc996 chore(cd): deploy 88ac9c3 [skip ci]`。
+- Production `/api/v1/agents/delivery-closure-workbench` 已讀回 `production_deploy_runtime_build_commit_short_sha=88ac9c33da`、`production_deploy_desired_main_api_image_tag_short_sha=88ac9c33da`、`production_deploy_desired_main_api_image_tag_readback_status=ok`、`production_deploy_image_tag_matches_main=true`。
+- Production `/api/v1/agents/gitea-workflow-runner-owner-attestation-request` 回 `200`，`/api/v1/agents/agent-log-controlled-writeback-consumer-readback` 回 `200 controlled_writeback_consumer_readback_ready`。
+- 新增 GET `/api/v1/agents/awoooi-priority-work-order-readback`，把 P0/P1 主線排序從本機 snapshot 提升為 production 可讀回 API；loader 會用 committed Delivery Workbench / P0-004 / P0-006 loaders 動態補齊目前狀態，避免舊 CD run 或舊 snapshot 重新打亂優先順序。
+- 目前 active P0 仍是 `P0-006`；P0-003 / P0-005 closed，P0-004 apply gate / template receipt / runtime enablement ready。P0-006 只剩 fresh all-host reboot event 或另行 break-glass reboot drill authorization，不得把一般「繼續」當成重啟授權。
+
+**邊界**：未 workflow_dispatch，未重啟主機，未 restart Docker / K3s / Nginx / DB / firewall，未寫 K8s / DB，未讀 secret / token / raw sessions / SQLite / `.env`，未使用 GitHub / `gh` / GitHub API。
+
 ## 2026-06-30 — 08:19 API bootstrap controlled startup unblock
 
 **照主線完成的實作**：
@@ -50030,6 +50041,7 @@ production browser smoke:
 **狀態**：
 - Gitea `main` 已有 `49c02e5b3` API / Workbench source 與 `6083cb71a` CD deploy marker，但 production 兩個新 API route 仍回 404，公開 Gitea Actions 顯示 CD #4003 `Failure`。
 - 既有 production deploy readback 只用 runtime `AWOOOI_BUILD_COMMIT_SHA` 覆寫 source/main 欄位，可能讓舊 production image 自稱 `production_image_tag_matches_main=true`。
+- 補充：CD #4010 在 deploy marker 前失敗；根因是新測試檔 `apps/api/tests/test_awoooi_production_deploy_readback_blocker.py` 未納入 controlled-runtime 白名單，導致窄發布掉入 full/B5。已補白名單與 focused pytest 清單。
 
 **完成內容**：
 - `awoooi_production_deploy_readback_blocker` 改為比對 runtime `AWOOOI_BUILD_COMMIT_SHA` 與 GitOps desired env `AWOOOI_DESIRED_API_IMAGE_TAG`；缺少 desired env 或不一致時 fail-closed 成 blocker。

ops/runner/test_cd_controlled_runtime_profile.py

@@ -108,6 +108,19 @@ def test_p0_onboarding_readiness_sources_stay_on_controlled_runtime_profile() ->
     assert "tests/test_p0_cicd_baseline_source_readiness_api.py" in text
 
 
+def test_priority_work_order_readback_stays_on_controlled_runtime_profile() -> None:
+    text = _workflow_text()
+    expected_sources = [
+        "docs/operations/awoooi-priority-work-order-readback.snapshot.json)",
+        "apps/api/src/services/awoooi_priority_work_order_readback.py)",
+        "apps/api/tests/test_awoooi_priority_work_order_readback_api.py)",
+        "src/services/awoooi_priority_work_order_readback.py",
+        "tests/test_awoooi_priority_work_order_readback_api.py",
+    ]
+    for source in expected_sources:
+        assert source in text
+
+
 def test_iwooos_security_operation_api_stays_on_controlled_runtime_profile() -> None:
     text = _workflow_text()
     expected_sources = [