feat(work-items): surface AI runbook loop coverage

apps/api/src/services/awoooi_priority_work_order_readback.py

@@ -84,6 +84,16 @@ _AI_AUTOMATION_NODE_RECEIPT_REQUIRED_FIELDS = [
     "km_writeback_ref",
     "playbook_trust_writeback_ref",
 ]
+_AI_CONTROLLED_REPAIR_LOOP_REQUIRED_FIELDS = [
+    "target_selector",
+    "source_truth_diff",
+    "candidate_action",
+    "check_mode_or_dry_run",
+    "controlled_apply_boundary",
+    "rollback_or_no_write",
+    "post_verifier",
+    "km_playbook_trust_writeback",
+]
 _COMMANDER_INSERTED_REQUIREMENT_WORK_ITEMS: list[dict[str, Any]] = [
     {
         "id": "CIR-P0-001",
@@ -6688,6 +6698,307 @@ def _apply_ai_automation_node_receipts(payload: dict[str, Any]) -> None:
     summary["ai_automation_node_receipt_metadata_only"] = True
     summary["ai_automation_node_receipt_lanes"] = lanes
     summary["ai_automation_node_receipt_work_items"] = work_items
+    _apply_ai_controlled_repair_loop_runbook_matrix(payload)
+
+
+def _contains_any(value: str, needles: tuple[str, ...]) -> bool:
+    normalized = value.lower()
+    return any(needle.lower() in normalized for needle in needles)
+
+
+def _apply_ai_controlled_repair_loop_runbook_matrix(
+    payload: dict[str, Any],
+) -> None:
+    items = [
+        _dict(item)
+        for item in payload.get("commander_inserted_requirement_work_items") or []
+        if str(_dict(item).get("priority") or "") == "P0"
+    ]
+    receipts = [_dict(item) for item in payload.get("ai_automation_node_receipts") or []]
+    ready_receipt_work_items = {
+        str(item.get("work_item_id") or "")
+        for item in receipts
+        if str(item.get("node_status") or "") == "ready"
+    }
+    learning_receipt_ready = any(
+        str(item.get("node_id") or "") == "learning_writeback"
+        and str(item.get("node_status") or "") == "ready"
+        for item in receipts
+    )
+    post_verifier_receipt_ready = any(
+        str(item.get("node_id") or "") == "post_verifier"
+        and str(item.get("node_status") or "") == "ready"
+        for item in receipts
+    )
+
+    rows: list[dict[str, Any]] = []
+    for item in sorted(items, key=lambda raw: _int(raw.get("order"))):
+        item_id = str(item.get("id") or "")
+        lane = str(item.get("lane") or "")
+        status = str(item.get("status") or "")
+        text = " ".join(
+            str(item.get(key) or "")
+            for key in (
+                "id",
+                "lane",
+                "request",
+                "normalized_work_item",
+                "current_state",
+                "acceptance",
+                "next_action",
+                "mapped_workplan_id",
+                "visible_surface",
+            )
+        )
+        break_glass_required = lane == "secret_safety" or _contains_any(
+            text,
+            (
+                "secret plaintext",
+                "plaintext secret",
+                "private key",
+                "cookie",
+                "authorization header",
+                "明文",
+                "密碼",
+                "敏感值",
+            ),
+        )
+        fields = {
+            "target_selector": bool(
+                item.get("mapped_workplan_id")
+                or item.get("visible_surface")
+                or item_id
+            ),
+            "source_truth_diff": _contains_any(
+                text,
+                (
+                    "gitea",
+                    "source",
+                    "repo",
+                    "inventory",
+                    "matrix",
+                    "snapshot",
+                    "evidence",
+                    "readback",
+                    "production",
+                    "證據",
+                    "讀回",
+                    "正式環境",
+                ),
+            ),
+            "candidate_action": bool(item.get("next_action"))
+            or item_id in ready_receipt_work_items,
+            "check_mode_or_dry_run": _contains_any(
+                text,
+                (
+                    "check-mode",
+                    "check mode",
+                    "dry-run",
+                    "dry run",
+                    "verifier",
+                    "hidden prompt",
+                    "metadata verifier",
+                    "non-secret",
+                    "只跑",
+                    "乾跑",
+                    "驗證",
+                ),
+            ),
+            "controlled_apply_boundary": bool(
+                break_glass_required
+                or _contains_any(
+                    text,
+                    (
+                        "controlled",
+                        "ai",
+                        "fail-closed",
+                        "no-write",
+                        "no_secret",
+                        "non-secret",
+                        "不讀 secret",
+                        "受控",
+                        "不恢復 generic",
+                    ),
+                )
+            ),
+            "rollback_or_no_write": bool(
+                break_glass_required
+                or _contains_any(
+                    text,
+                    (
+                        "rollback",
+                        "no-write",
+                        "no write",
+                        "no runtime write",
+                        "不入檔",
+                        "不重印",
+                        "不保存",
+                        "不建立",
+                        "不恢復",
+                        "fail-closed",
+                    ),
+                )
+            ),
+            "post_verifier": bool(
+                post_verifier_receipt_ready
+                or _contains_any(
+                    text,
+                    (
+                        "verifier",
+                        "readback",
+                        "smoke",
+                        "api",
+                        "production",
+                        "deploy marker",
+                        "metadata verifier",
+                        "驗證",
+                        "讀回",
+                        "正式環境",
+                    ),
+                )
+            ),
+            "km_playbook_trust_writeback": bool(
+                learning_receipt_ready
+                or _contains_any(
+                    text,
+                    (
+                        "km",
+                        "rag",
+                        "mcp",
+                        "playbook",
+                        "trust",
+                        "writeback",
+                        "learning",
+                        "學習",
+                        "回寫",
+                    ),
+                )
+            ),
+        }
+        ready_field_count = sum(1 for ready in fields.values() if ready)
+        missing_fields = [
+            field
+            for field in _AI_CONTROLLED_REPAIR_LOOP_REQUIRED_FIELDS
+            if not fields[field]
+        ]
+        controlled_apply_allowed = not break_glass_required and status not in {
+            "deferred",
+        }
+        if break_glass_required:
+            coverage_state = "break_glass"
+        elif not missing_fields:
+            coverage_state = "ready"
+        elif ready_field_count > 0:
+            coverage_state = "partial"
+        else:
+            coverage_state = "queued"
+
+        rows.append(
+            {
+                "work_item_id": item_id,
+                "priority": "P0",
+                "order": _int(item.get("order")),
+                "status": status,
+                "lane": lane,
+                "mapped_workplan_id": str(item.get("mapped_workplan_id") or ""),
+                "coverage_state": coverage_state,
+                "ready_field_count": ready_field_count,
+                "required_field_count": len(_AI_CONTROLLED_REPAIR_LOOP_REQUIRED_FIELDS),
+                "missing_fields": missing_fields,
+                "target_selector_ready": fields["target_selector"],
+                "source_truth_diff_ready": fields["source_truth_diff"],
+                "candidate_action_ready": fields["candidate_action"],
+                "check_mode_or_dry_run_ready": fields["check_mode_or_dry_run"],
+                "controlled_apply_boundary_ready": fields[
+                    "controlled_apply_boundary"
+                ],
+                "controlled_apply_allowed": controlled_apply_allowed,
+                "break_glass_required": break_glass_required,
+                "rollback_or_no_write_ready": fields["rollback_or_no_write"],
+                "post_verifier_ready": fields["post_verifier"],
+                "km_playbook_trust_writeback_ready": fields[
+                    "km_playbook_trust_writeback"
+                ],
+                "manual_terminal_allowed": False,
+                "next_action": str(item.get("next_action") or ""),
+            }
+        )
+
+    required_total = len(rows) * len(_AI_CONTROLLED_REPAIR_LOOP_REQUIRED_FIELDS)
+    ready_total = sum(_int(row.get("ready_field_count")) for row in rows)
+    state_counts = {
+        state: sum(1 for row in rows if row["coverage_state"] == state)
+        for state in ("ready", "partial", "queued", "break_glass")
+    }
+    next_row = next(
+        (
+            row
+            for row in rows
+            if row["coverage_state"] not in {"ready", "break_glass"}
+            and row["status"] not in {"done", "deferred"}
+        ),
+        next(
+            (
+                row
+                for row in rows
+                if row["coverage_state"] not in {"ready", "break_glass"}
+            ),
+            rows[0] if rows else {},
+        ),
+    )
+
+    matrix = {
+        "schema_version": "ai_controlled_repair_loop_runbook_matrix_v1",
+        "status": "ai_controlled_repair_loop_runbook_matrix_ready",
+        "metadata_only": True,
+        "source": _COMMANDER_INSERTED_REQUIREMENT_SOURCE,
+        "required_fields": _AI_CONTROLLED_REPAIR_LOOP_REQUIRED_FIELDS,
+        "manual_as_default_terminal_allowed": False,
+        "hard_blockers_preserved": [
+            "secret_plaintext",
+            "destructive_data_change",
+            "reboot_or_node_drain",
+            "force_push_or_repo_ref_delete",
+            "paid_provider_route_change",
+        ],
+        "rows": rows,
+        "summary": {
+            "p0_runbook_count": len(rows),
+            "ready_runbook_count": state_counts["ready"],
+            "partial_runbook_count": state_counts["partial"],
+            "queued_runbook_count": state_counts["queued"],
+            "break_glass_runbook_count": state_counts["break_glass"],
+            "controlled_apply_allowed_count": sum(
+                1 for row in rows if row["controlled_apply_allowed"] is True
+            ),
+            "manual_terminal_count": sum(
+                1 for row in rows if row["manual_terminal_allowed"] is True
+            ),
+            "required_field_total": required_total,
+            "ready_field_total": ready_total,
+            "completion_percent": round(
+                (ready_total / required_total) * 100
+            )
+            if required_total
+            else 0,
+            "next_work_item_id": str(next_row.get("work_item_id") or ""),
+            "next_missing_fields": next_row.get("missing_fields") or [],
+        },
+    }
+
+    payload["ai_controlled_repair_loop_runbook_matrix"] = matrix
+
+    summary = _dict(payload.setdefault("summary", {}))
+    rollups = _dict(payload.setdefault("rollups", {}))
+    for key, value in matrix["summary"].items():
+        summary[f"ai_controlled_repair_loop_{key}"] = value
+        rollups[f"ai_controlled_repair_loop_{key}"] = value
+    summary["ai_controlled_repair_loop_schema_version"] = matrix["schema_version"]
+    summary["ai_controlled_repair_loop_metadata_only"] = True
+    summary["ai_controlled_repair_loop_manual_as_default_terminal_allowed"] = False
+    rollups["ai_controlled_repair_loop_schema_version"] = matrix["schema_version"]
+    rollups["ai_controlled_repair_loop_metadata_only"] = True
+    rollups["ai_controlled_repair_loop_manual_as_default_terminal_allowed"] = False
 
 
 def _mark_runtime_generated_at(payload: dict[str, Any]) -> None: