feat(governance): 顯示任務批准邊界與進度彙總

apps/api/src/services/ai_agent_automation_backlog_snapshot.py

@@ -38,6 +38,8 @@ def load_latest_ai_agent_automation_backlog_snapshot(
     _require_schema(payload, _SCHEMA_VERSION, str(latest))
     _require_read_only_boundaries(payload, str(latest))
     _require_rollup_consistency(payload, str(latest))
+    _require_item_approval_boundaries(payload, str(latest))
+    _require_progress_summary_consistency(payload, str(latest))
     return payload
 
 
@@ -67,6 +69,159 @@ def _require_read_only_boundaries(payload: dict[str, Any], label: str) -> None:
 
 def _require_rollup_consistency(payload: dict[str, Any], label: str) -> None:
     items = payload.get("backlog_items") or []
-    total = (payload.get("rollups") or {}).get("total_items")
+    rollups = payload.get("rollups") or {}
+    total = rollups.get("total_items")
     if total != len(items):
         raise ValueError(f"{label}: rollups.total_items must equal backlog_items length")
+
+    expected_by_priority = _count_by(items, "priority")
+    if rollups.get("by_priority") != expected_by_priority:
+        raise ValueError(f"{label}: rollups.by_priority must match backlog_items")
+
+    expected_by_status = _count_by(items, "status")
+    if rollups.get("by_status") != expected_by_status:
+        raise ValueError(f"{label}: rollups.by_status must match backlog_items")
+
+    expected_by_gate = _count_by(items, "gate_status")
+    if rollups.get("by_gate_status") != expected_by_gate:
+        raise ValueError(f"{label}: rollups.by_gate_status must match backlog_items")
+
+    expected_by_owner = _count_by(items, "owner_agent")
+    if rollups.get("by_owner_agent") != expected_by_owner:
+        raise ValueError(f"{label}: rollups.by_owner_agent must match backlog_items")
+
+
+def _require_item_approval_boundaries(payload: dict[str, Any], label: str) -> None:
+    items = payload.get("backlog_items") or []
+    missing = sorted(item.get("item_id") for item in items if not item.get("approval_boundary"))
+    if missing:
+        raise ValueError(f"{label}: backlog_items must include approval_boundary: {missing}")
+
+    mismatched_modes = sorted(
+        item.get("item_id")
+        for item in items
+        if (item.get("approval_boundary") or {}).get("mode") != item.get("gate_status")
+    )
+    if mismatched_modes:
+        raise ValueError(f"{label}: approval_boundary.mode must match gate_status: {mismatched_modes}")
+
+    missing_blocked_actions = sorted(
+        item.get("item_id")
+        for item in items
+        if not (item.get("approval_boundary") or {}).get("blocked_actions")
+    )
+    if missing_blocked_actions:
+        raise ValueError(f"{label}: approval_boundary.blocked_actions must be non-empty: {missing_blocked_actions}")
+
+    rollup = payload.get("item_approval_boundary_rollup") or {}
+    if rollup.get("total_items") != len(items):
+        raise ValueError(f"{label}: item_approval_boundary_rollup.total_items must match backlog_items")
+
+    by_mode: dict[str, int] = {}
+    for item in items:
+        mode = (item.get("approval_boundary") or {}).get("mode")
+        by_mode[mode] = by_mode.get(mode, 0) + 1
+    if rollup.get("by_mode") != by_mode:
+        raise ValueError(f"{label}: item_approval_boundary_rollup.by_mode must match backlog_items")
+
+    explicit_approval = sorted(
+        item.get("item_id")
+        for item in items
+        if (item.get("approval_boundary") or {}).get("mode") != "read_only_allowed"
+    )
+    if sorted(rollup.get("items_requiring_explicit_approval") or []) != explicit_approval:
+        raise ValueError(
+            f"{label}: item_approval_boundary_rollup.items_requiring_explicit_approval must match backlog_items"
+        )
+
+    with_blocked_operations = sorted(
+        item.get("item_id")
+        for item in items
+        if (item.get("approval_boundary") or {}).get("blocked_actions")
+    )
+    if sorted(rollup.get("items_with_blocked_operations") or []) != with_blocked_operations:
+        raise ValueError(
+            f"{label}: item_approval_boundary_rollup.items_with_blocked_operations must match backlog_items"
+        )
+
+
+def _require_progress_summary_consistency(payload: dict[str, Any], label: str) -> None:
+    items = payload.get("backlog_items") or []
+    summary = payload.get("progress_summary") or {}
+    done_items = sum(1 for item in items if item.get("status") == "done")
+    planned_items = sum(1 for item in items if item.get("status") == "planned")
+    total_items = len(items)
+    expected_percent = _percent(done_items, total_items)
+
+    if summary.get("total_items") != total_items:
+        raise ValueError(f"{label}: progress_summary.total_items must match backlog_items")
+    if summary.get("done_items") != done_items:
+        raise ValueError(f"{label}: progress_summary.done_items must match backlog_items")
+    if summary.get("planned_items") != planned_items:
+        raise ValueError(f"{label}: progress_summary.planned_items must match backlog_items")
+    if summary.get("overall_percent") != expected_percent:
+        raise ValueError(f"{label}: progress_summary.overall_percent must match deterministic formula")
+
+    expected_priority_progress = {
+        priority: {
+            "done_items": sum(1 for item in group if item.get("status") == "done"),
+            "total_items": len(group),
+        }
+        for priority, group in _group_by(items, "priority").items()
+    }
+    actual_priority_progress = {
+        row.get("priority"): {
+            "done_items": row.get("done_items"),
+            "total_items": row.get("total_items"),
+            "completion_percent": row.get("completion_percent"),
+        }
+        for row in summary.get("by_priority") or []
+    }
+    for priority, expected in expected_priority_progress.items():
+        actual = actual_priority_progress.get(priority)
+        expected_completion = _percent(expected["done_items"], expected["total_items"])
+        if actual != {**expected, "completion_percent": expected_completion}:
+            raise ValueError(f"{label}: progress_summary.by_priority must match backlog_items")
+
+    expected_workstream_progress = {
+        workstream_id: {
+            "done_items": sum(1 for item in group if item.get("status") == "done"),
+            "total_items": len(group),
+        }
+        for workstream_id, group in _group_by(items, "workstream_id").items()
+    }
+    actual_workstream_progress = {
+        row.get("workstream_id"): {
+            "done_items": row.get("done_items"),
+            "total_items": row.get("total_items"),
+            "completion_percent": row.get("completion_percent"),
+        }
+        for row in summary.get("by_workstream") or []
+    }
+    for workstream_id, expected in expected_workstream_progress.items():
+        actual = actual_workstream_progress.get(workstream_id)
+        expected_completion = _percent(expected["done_items"], expected["total_items"])
+        if actual != {**expected, "completion_percent": expected_completion}:
+            raise ValueError(f"{label}: progress_summary.by_workstream must match backlog_items")
+
+
+def _count_by(items: list[dict[str, Any]], key: str) -> dict[str, int]:
+    counts: dict[str, int] = {}
+    for item in items:
+        value = item.get(key)
+        counts[value] = counts.get(value, 0) + 1
+    return counts
+
+
+def _group_by(items: list[dict[str, Any]], key: str) -> dict[str, list[dict[str, Any]]]:
+    groups: dict[str, list[dict[str, Any]]] = {}
+    for item in items:
+        value = item.get(key)
+        groups.setdefault(value, []).append(item)
+    return groups
+
+
+def _percent(done: int, total: int) -> int:
+    if total == 0:
+        return 0
+    return round((done / total) * 100)

apps/api/src/services/ai_agent_automation_inventory_snapshot.py

@@ -36,6 +36,7 @@ def load_latest_ai_agent_automation_inventory_snapshot(
         raise ValueError(f"{latest}: expected JSON object")
     _require_schema(payload, _SCHEMA_VERSION, str(latest))
     _require_read_only_boundaries(payload, str(latest))
+    _require_task_approval_boundaries(payload, str(latest))
     return payload
 
 
@@ -61,3 +62,57 @@ def _require_read_only_boundaries(payload: dict[str, Any], label: str) -> None:
     allowed = sorted(flag for flag in blocked_flags if boundaries.get(flag) is not False)
     if allowed:
         raise ValueError(f"{label}: approval boundaries must remain false: {allowed}")
+
+
+def _require_task_approval_boundaries(payload: dict[str, Any], label: str) -> None:
+    tasks = payload.get("tasks") or []
+    missing = sorted(task.get("task_id") for task in tasks if not task.get("approval_boundary"))
+    if missing:
+        raise ValueError(f"{label}: tasks must include approval_boundary: {missing}")
+
+    mismatched_modes = sorted(
+        task.get("task_id")
+        for task in tasks
+        if (task.get("approval_boundary") or {}).get("mode") != task.get("gate_status")
+    )
+    if mismatched_modes:
+        raise ValueError(f"{label}: approval_boundary.mode must match gate_status: {mismatched_modes}")
+
+    missing_blocked_actions = sorted(
+        task.get("task_id")
+        for task in tasks
+        if not (task.get("approval_boundary") or {}).get("blocked_actions")
+    )
+    if missing_blocked_actions:
+        raise ValueError(f"{label}: approval_boundary.blocked_actions must be non-empty: {missing_blocked_actions}")
+
+    rollup = payload.get("task_approval_boundary_rollup") or {}
+    if rollup.get("total_tasks") != len(tasks):
+        raise ValueError(f"{label}: task_approval_boundary_rollup.total_tasks must match tasks")
+
+    by_mode: dict[str, int] = {}
+    for task in tasks:
+        mode = (task.get("approval_boundary") or {}).get("mode")
+        by_mode[mode] = by_mode.get(mode, 0) + 1
+    if rollup.get("by_mode") != by_mode:
+        raise ValueError(f"{label}: task_approval_boundary_rollup.by_mode must match tasks")
+
+    explicit_approval = sorted(
+        task.get("task_id")
+        for task in tasks
+        if (task.get("approval_boundary") or {}).get("mode") != "read_only_allowed"
+    )
+    if sorted(rollup.get("tasks_requiring_explicit_approval") or []) != explicit_approval:
+        raise ValueError(
+            f"{label}: task_approval_boundary_rollup.tasks_requiring_explicit_approval must match tasks"
+        )
+
+    with_blocked_operations = sorted(
+        task.get("task_id")
+        for task in tasks
+        if (task.get("approval_boundary") or {}).get("blocked_actions")
+    )
+    if sorted(rollup.get("tasks_with_blocked_operations") or []) != with_blocked_operations:
+        raise ValueError(
+            f"{label}: task_approval_boundary_rollup.tasks_with_blocked_operations must match tasks"
+        )