feat(telegram): consume alert registry in ai loop context

apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py

@@ -22,11 +22,16 @@ from src.services.ai_agent_log_controlled_writeback_dispatch import (
     EXECUTOR_ROUTE,
     OPERATION_TYPE,
 )
+from src.services.platform_operator_service import list_ai_alert_card_delivery_readback
 
 SCHEMA_VERSION = "ai_agent_log_controlled_writeback_consumer_readback_v1"
+TELEGRAM_ALERT_CONTEXT_SCHEMA_VERSION = (
+    "telegram_alert_ai_loop_consumer_context_readback_v1"
+)
 DEFAULT_PROJECT_ID = "awoooi"
 CONSUMER_OPERATION_TYPE = "log_controlled_writeback_consumed"
 CONSUMER_EXECUTOR_ROUTE = "ai_agent_metadata_writeback_consumer"
+TELEGRAM_ALERT_CONTEXT_PER_PAGE = 6
 _TARGETS = ("km", "rag", "playbook", "mcp", "verifier", "ai_agent")
 _CONSUMER_SURFACES = {
     "km": "knowledge_memory_context",
@@ -59,7 +64,14 @@ async def load_latest_ai_agent_log_controlled_writeback_consumer_readback(
             error_type=type(exc).__name__,
         )
 
-    return _build_consumer_readback(rows=rows, consumer_rows=consumer_rows)
+    telegram_alert_learning_context = await _load_telegram_alert_learning_context(
+        project_id=project_id
+    )
+    return _build_consumer_readback(
+        rows=rows,
+        consumer_rows=consumer_rows,
+        telegram_alert_learning_context=telegram_alert_learning_context,
+    )
 
 
 async def _load_consumer_rows_with_retry(
@@ -161,6 +173,7 @@ def _build_consumer_readback(
     *,
     rows: list[dict[str, Any]],
     consumer_rows: list[dict[str, Any]],
+    telegram_alert_learning_context: dict[str, Any],
 ) -> dict[str, Any]:
     consumer_receipts = _consumer_receipts_by_dispatch_id(consumer_rows)
     bindings = [_consumer_binding(row, consumer_receipts) for row in rows]
@@ -251,13 +264,37 @@ def _build_consumer_readback(
             "ai_agent_context_receipt_write_count": _target_write_count(
                 bindings, "ai_agent"
             ),
+            "telegram_alert_learning_context_readback_ready": (
+                telegram_alert_learning_context.get("status")
+                == "ai_loop_agent_context_receipt_ready"
+            ),
+            "telegram_alert_learning_context_receipt_count": _safe_int(
+                telegram_alert_learning_context.get("context_receipt_count")
+            ),
+            "telegram_alert_learning_ai_agent_context_receipt_count": _safe_int(
+                telegram_alert_learning_context.get(
+                    "ai_agent_context_receipt_count"
+                )
+            ),
+            "telegram_alert_learning_ready_target_count": _safe_int(
+                telegram_alert_learning_context.get("ready_target_count")
+            ),
+            "telegram_alert_learning_target_count": _safe_int(
+                telegram_alert_learning_context.get("target_count")
+            ),
             "runtime_target_write_performed": runtime_target_write_performed,
         },
+        "telegram_alert_learning_context": telegram_alert_learning_context,
         "active_blockers": active_blockers,
         "operation_boundaries": {
             "consumer_readback_only": True,
             "metadata_ledger_read_performed": True,
             "consumer_apply_receipt_read_performed": True,
+            "telegram_alert_registry_metadata_read_performed": (
+                telegram_alert_learning_context["operation_boundaries"][
+                    "metadata_read_performed"
+                ]
+            ),
             "runtime_target_write_performed": runtime_target_write_performed,
             "km_write_performed": False,
             "rag_index_write_performed": False,
@@ -335,8 +372,17 @@ def _fallback_consumer_readback(
             "mcp_context_receipt_write_count": 0,
             "verifier_context_receipt_write_count": 0,
             "ai_agent_context_receipt_write_count": 0,
+            "telegram_alert_learning_context_readback_ready": False,
+            "telegram_alert_learning_context_receipt_count": 0,
+            "telegram_alert_learning_ai_agent_context_receipt_count": 0,
+            "telegram_alert_learning_ready_target_count": 0,
+            "telegram_alert_learning_target_count": 0,
             "runtime_target_write_performed": False,
         },
+        "telegram_alert_learning_context": _telegram_alert_learning_context_unavailable(
+            project_id=project_id,
+            error_type=error_type,
+        ),
         "active_blockers": [
             "log_controlled_writeback_consumer_readback_db_unavailable"
         ],
@@ -344,6 +390,7 @@ def _fallback_consumer_readback(
             "consumer_readback_only": True,
             "metadata_ledger_read_performed": False,
             "consumer_apply_receipt_read_performed": False,
+            "telegram_alert_registry_metadata_read_performed": False,
             "runtime_target_write_performed": False,
             "km_write_performed": False,
             "rag_index_write_performed": False,
@@ -553,6 +600,224 @@ def _consumer_receipts_by_dispatch_id(
     return receipts
 
 
+async def _load_telegram_alert_learning_context(
+    *,
+    project_id: str,
+) -> dict[str, Any]:
+    """Read Telegram alert learning registry as AI Loop consumer context."""
+    try:
+        readback = await list_ai_alert_card_delivery_readback(
+            project_id=project_id,
+            page=1,
+            per_page=TELEGRAM_ALERT_CONTEXT_PER_PAGE,
+            refresh=True,
+        )
+    except Exception as exc:  # pragma: no cover - live DB pool pressure
+        logger.warning(
+            "telegram_alert_learning_context_readback_unavailable",
+            project_id=project_id,
+            error_type=type(exc).__name__,
+        )
+        return _telegram_alert_learning_context_unavailable(
+            project_id=project_id,
+            error_type=type(exc).__name__,
+        )
+    return _telegram_alert_learning_context_from_alert_cards(readback)
+
+
+def _telegram_alert_learning_context_from_alert_cards(
+    readback: Mapping[str, Any],
+) -> dict[str, Any]:
+    registry = _dict(readback.get("learning_registry"))
+    summary = _dict(readback.get("summary"))
+    project_id = str(
+        registry.get("project_id")
+        or summary.get("project_id")
+        or DEFAULT_PROJECT_ID
+    )
+    items = [_dict(item) for item in _list(readback.get("items"))]
+    context_receipts = _telegram_alert_context_receipts(items)
+    target_count = _safe_int(registry.get("target_count") or len(_TARGETS))
+    ready_target_count = _safe_int(registry.get("ready_target_count"))
+    ai_agent_context_receipt_count = sum(
+        1 for receipt in context_receipts if receipt["target"] == "ai_agent"
+    )
+    ready = (
+        registry.get("status") == "learning_registry_ready"
+        and ready_target_count == target_count
+        and target_count > 0
+        and ai_agent_context_receipt_count > 0
+    )
+
+    return {
+        "schema_version": TELEGRAM_ALERT_CONTEXT_SCHEMA_VERSION,
+        "status": (
+            "ai_loop_agent_context_receipt_ready"
+            if ready
+            else "waiting_telegram_alert_learning_registry"
+        ),
+        "priority_work_item_id": "CIR-P0-TG-001",
+        "project_id": project_id,
+        "source_endpoint": "/api/v1/platform/runs/ai-alert-cards",
+        "source_registry_schema_version": registry.get("schema_version"),
+        "source_registry_status": registry.get("status"),
+        "consumer_context_route": "ai_loop_agent.telegram_alert_learning_context",
+        "target_count": target_count,
+        "ready_target_count": ready_target_count,
+        "context_receipt_count": len(context_receipts),
+        "ai_agent_context_receipt_count": ai_agent_context_receipt_count,
+        "next_action": (
+            "verify_telegram_alert_learning_context_receipts_with_ai_loop_post_apply_verifier"
+            if ready
+            else "wait_for_telegram_alert_learning_registry_before_context_receipt"
+        ),
+        "context_receipts": context_receipts,
+        "active_blockers": (
+            []
+            if ready
+            else ["telegram_alert_learning_registry_not_ready_for_ai_loop_context"]
+        ),
+        "operation_boundaries": {
+            "metadata_read_performed": True,
+            "db_or_log_receipt_read_performed": bool(items),
+            "runtime_target_write_performed": False,
+            "km_write_performed": False,
+            "rag_index_write_performed": False,
+            "playbook_trust_write_performed": False,
+            "mcp_tool_call_performed": False,
+            "agent_runtime_action_performed": False,
+            "telegram_send_performed": False,
+            "raw_payload_included": False,
+            "secret_value_collection_allowed": False,
+            "github_api_used": False,
+        },
+    }
+
+
+def _telegram_alert_context_receipts(
+    items: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    receipts: list[dict[str, Any]] = []
+    for item in items:
+        learning_refs = _dict(item.get("learning_writeback_refs"))
+        message_id = str(item.get("message_id") or learning_refs.get("receipt_id") or "")
+        run_id = str(item.get("run_id") or learning_refs.get("run_id") or "")
+        for binding_value in _list(item.get("learning_registry_bindings")):
+            binding = _dict(binding_value)
+            target = str(binding.get("target") or "")
+            if target not in _TARGETS:
+                continue
+            ref = str(binding.get("ref") or "")
+            status = str(binding.get("status") or "")
+            ready = status == "ready_for_consumer_context" and bool(ref)
+            receipts.append({
+                "receipt_id": (
+                    "telegram_alert_learning_context::"
+                    f"{message_id or 'unknown-message'}::{target}"
+                ),
+                "source_receipt_id": str(learning_refs.get("receipt_id") or message_id),
+                "source_run_id": run_id,
+                "source_message_id": message_id,
+                "work_item_id": str(
+                    binding.get("work_item_id")
+                    or learning_refs.get("work_item_id")
+                    or "CIR-P0-TG-001"
+                ),
+                "target": target,
+                "consumer_surface": str(binding.get("consumer_surface") or ""),
+                "source_ref": ref,
+                "ai_agent_context_ref": str(
+                    learning_refs.get("ai_agent_context_ref") or ""
+                ),
+                "event_type": str(item.get("event_type") or ""),
+                "lane": str(item.get("lane") or ""),
+                "status": (
+                    "ready_for_ai_loop_context" if ready else "missing_source_ref"
+                ),
+                "metadata_only": True,
+                "target_write_performed": False,
+                "raw_payload_included": False,
+                "target_selector": {
+                    "project_id": str(item.get("project_id") or DEFAULT_PROJECT_ID),
+                    "run_id": run_id,
+                    "message_id": message_id,
+                    "target": target,
+                    "source_ref": ref,
+                },
+                "source_of_truth_diff": {
+                    "current_state": "telegram_alert_learning_registry_readable",
+                    "desired_state": "ai_loop_agent_context_receipt_available",
+                    "delta_kind": f"telegram_alert_{target}_context_binding",
+                    "raw_payload_included": False,
+                },
+                "check_mode": {
+                    "enabled": True,
+                    "checks": [
+                        "ai_alert_card_delivery_receipt_present",
+                        "learning_registry_binding_ready",
+                        "metadata_only_raw_payload_absent",
+                        "ai_agent_context_ref_present",
+                    ],
+                },
+                "rollback": {
+                    "required": True,
+                    "rollback_ref": (
+                        "rollback://telegram-alert-learning-context/"
+                        f"{message_id or 'unknown-message'}/{target}"
+                    ),
+                    "strategy": "ignore_context_receipt_and_wait_for_next_delivery_readback",
+                },
+                "post_apply_verifier": {
+                    "required": True,
+                    "verifier_refs": _unique([
+                        str(learning_refs.get("post_verifier_ref") or ""),
+                        str(binding.get("ref") or ""),
+                    ]),
+                },
+            })
+    return receipts
+
+
+def _telegram_alert_learning_context_unavailable(
+    *,
+    project_id: str,
+    error_type: str,
+) -> dict[str, Any]:
+    return {
+        "schema_version": TELEGRAM_ALERT_CONTEXT_SCHEMA_VERSION,
+        "status": "blocked_telegram_alert_learning_registry_readback_unavailable",
+        "priority_work_item_id": "CIR-P0-TG-001",
+        "project_id": project_id,
+        "source_endpoint": "/api/v1/platform/runs/ai-alert-cards",
+        "source_registry_schema_version": None,
+        "source_registry_status": "unavailable",
+        "consumer_context_route": "ai_loop_agent.telegram_alert_learning_context",
+        "target_count": len(_TARGETS),
+        "ready_target_count": 0,
+        "context_receipt_count": 0,
+        "ai_agent_context_receipt_count": 0,
+        "next_action": "retry_telegram_alert_learning_registry_readback",
+        "context_receipts": [],
+        "active_blockers": [
+            f"telegram_alert_learning_registry_readback_unavailable:{error_type}"
+        ],
+        "operation_boundaries": {
+            "metadata_read_performed": False,
+            "db_or_log_receipt_read_performed": False,
+            "runtime_target_write_performed": False,
+            "km_write_performed": False,
+            "rag_index_write_performed": False,
+            "playbook_trust_write_performed": False,
+            "mcp_tool_call_performed": False,
+            "agent_runtime_action_performed": False,
+            "telegram_send_performed": False,
+            "raw_payload_included": False,
+            "secret_value_collection_allowed": False,
+            "github_api_used": False,
+        },
+    }
+
+
 def _result_rows(result: Any) -> list[dict[str, Any]]:
     mappings = getattr(result, "mappings", None)
     if callable(mappings):
@@ -574,16 +839,39 @@ def _row_mapping(row: Any) -> dict[str, Any]:
 
 def _strings(value: Any) -> list[str]:
     if isinstance(value, list):
-        return [str(item) for item in value]
+        return [str(item) for item in value if str(item)]
     if isinstance(value, tuple):
-        return [str(item) for item in value]
+        return [str(item) for item in value if str(item)]
     return []
 
 
+def _dict(value: Any) -> dict[str, Any]:
+    if isinstance(value, Mapping):
+        return dict(value)
+    return {}
+
+
+def _list(value: Any) -> list[Any]:
+    if isinstance(value, list):
+        return value
+    if isinstance(value, tuple):
+        return list(value)
+    return []
+
+
+def _safe_int(value: Any) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return 0
+
+
 def _unique(values: list[str]) -> list[str]:
     seen = set()
     result = []
     for value in values:
+        if not value:
+            continue
         if value in seen:
             continue
         seen.add(value)

apps/api/src/services/awoooi_priority_work_order_readback.py

@@ -340,10 +340,11 @@ _COMMANDER_INSERTED_REQUIREMENT_WORK_ITEMS: list[dict[str, Any]] = [
             "已改為 AWOOI API / TelegramGateway receipt path；目前明確讀回 workflow direct send 0、"
             "ops script direct gap 0、API direct gap 0；/runs/ai-alert-cards 已補 KM / PlayBook / RAG / MCP "
             "learning_writeback_refs read model，並投影 KM / PlayBook / RAG / MCP / verifier / AI Agent "
-            "learning_registry consumer bindings。"
+            "learning_registry consumer bindings；agent-log-controlled-writeback-consumer-readback 已把同一批 "
+            "Telegram alert registry 轉成 AI Loop Agent consumer context receipt。"
         ),
-        "acceptance": "Telegram alert matrix 顯示 total、DB receipt、AI route、controlled queue、manual/default gap、direct send gap、learning refs 與 registry readback。",
-        "next_action": "把 Telegram alert learning registry 接到 AI Loop Agent consumer context receipt，讓後續判斷可直接重用這些 receipt。",
+        "acceptance": "Telegram alert matrix 顯示 total、DB receipt、AI route、controlled queue、manual/default gap、direct send gap、learning refs、registry readback 與 AI Loop consumer context receipt。",
+        "next_action": "把 Telegram alert AI Loop context receipts 接 post-apply verifier，讓後續判斷可直接重用這些 receipt。",
         "mapped_workplan_id": "P0-006",
     },
     {
@@ -3809,6 +3810,7 @@ def apply_ai_automation_live_closure_readbacks(
     consumer = _dict(consumer_readback)
     consumer_rollups = _dict(consumer.get("rollups"))
     consumer_control = _dict(consumer.get("controlled_consume"))
+    telegram_context = _dict(consumer.get("telegram_alert_learning_context"))
     consumer_active_blockers = _strings(consumer.get("active_blockers"))
     if consumer:
         context_receipt_counts = {
@@ -3851,6 +3853,27 @@ def apply_ai_automation_live_closure_readbacks(
         summary["ai_automation_consumer_apply_route"] = str(
             consumer_control.get("consumer_apply_route") or ""
         )
+        summary["ai_automation_telegram_alert_context_status"] = str(
+            telegram_context.get("status") or ""
+        )
+        summary["ai_automation_telegram_alert_context_readback_ready"] = bool(
+            consumer_rollups.get("telegram_alert_learning_context_readback_ready")
+            is True
+        )
+        summary["ai_automation_telegram_alert_context_receipt_count"] = _int(
+            consumer_rollups.get("telegram_alert_learning_context_receipt_count")
+        )
+        summary["ai_automation_telegram_alert_ai_agent_context_receipt_count"] = _int(
+            consumer_rollups.get(
+                "telegram_alert_learning_ai_agent_context_receipt_count"
+            )
+        )
+        summary["ai_automation_telegram_alert_context_ready_target_count"] = _int(
+            consumer_rollups.get("telegram_alert_learning_ready_target_count")
+        )
+        summary["ai_automation_telegram_alert_context_target_count"] = _int(
+            consumer_rollups.get("telegram_alert_learning_target_count")
+        )
         summary["ai_automation_consumer_active_blockers"] = consumer_active_blockers
         rollups["ai_automation_consumer_active_blocker_count"] = len(
             consumer_active_blockers
@@ -3861,6 +3884,12 @@ def apply_ai_automation_live_closure_readbacks(
         rollups["ai_automation_consumer_target_context_receipt_write_count"] = (
             summary["ai_automation_consumer_target_context_receipt_write_count"]
         )
+        rollups["ai_automation_telegram_alert_context_receipt_count"] = summary[
+            "ai_automation_telegram_alert_context_receipt_count"
+        ]
+        rollups["ai_automation_telegram_alert_ai_agent_context_receipt_count"] = (
+            summary["ai_automation_telegram_alert_ai_agent_context_receipt_count"]
+        )
         state["ai_automation_consumer_readback_ready"] = summary[
             "ai_automation_consumer_readback_ready"
         ]
@@ -3939,6 +3968,15 @@ def _apply_ai_automation_node_receipts(payload: dict[str, Any]) -> None:
     consumer_context_counts = _dict(
         summary.get("ai_automation_consumer_context_receipt_counts")
     )
+    telegram_alert_context_ready = bool(
+        summary.get("ai_automation_telegram_alert_context_readback_ready") is True
+    )
+    telegram_alert_context_receipt_count = _int(
+        summary.get("ai_automation_telegram_alert_context_receipt_count")
+    )
+    telegram_alert_ai_agent_context_receipt_count = _int(
+        summary.get("ai_automation_telegram_alert_ai_agent_context_receipt_count")
+    )
     consumer_apply_route = str(summary.get("ai_automation_consumer_apply_route") or "")
     normalizer_ready = bool(
         queue_fields
@@ -3948,7 +3986,8 @@ def _apply_ai_automation_node_receipts(payload: dict[str, Any]) -> None:
     )
     candidate_ready = bool(safe_next_action_id or executor_dispatch_ready)
     execution_boundary_ready = bool(
-        consumer_runtime_target_write_performed
+        telegram_alert_context_ready
+        or consumer_runtime_target_write_performed
         or executor_runtime_dispatch_performed
         or executor_dispatch_ready
     )
@@ -3958,9 +3997,15 @@ def _apply_ai_automation_node_receipts(payload: dict[str, Any]) -> None:
         or consumer_post_apply_verifier_ref_count > 0
     )
     learning_writeback_ready = bool(
-        consumer_readback_ready
-        and consumer_context_receipt_write_count >= consumer_ready_target_count
-        and consumer_ready_target_count > 0
+        (
+            consumer_readback_ready
+            and consumer_context_receipt_write_count >= consumer_ready_target_count
+            and consumer_ready_target_count > 0
+        )
+        or (
+            telegram_alert_context_ready
+            and telegram_alert_ai_agent_context_receipt_count > 0
+        )
     )
 
     base_refs = [
@@ -4055,10 +4100,15 @@ def _apply_ai_automation_node_receipts(payload: dict[str, Any]) -> None:
                 "no_secret_no_runtime_secret",
                 f"consumer_ready_targets:{consumer_ready_target_count}",
                 f"context_receipts:{consumer_context_receipt_write_count}",
+                f"telegram_alert_context_receipts:{telegram_alert_context_receipt_count}",
+                (
+                    "telegram_alert_ai_agent_context_receipts:"
+                    f"{telegram_alert_ai_agent_context_receipt_count}"
+                ),
                 f"consumer_apply_route:{consumer_apply_route}",
             ],
             "verifier_result": "controlled_consumer_context_receipt_readback"
-            if consumer_runtime_target_write_performed
+            if consumer_runtime_target_write_performed or telegram_alert_context_ready
             else (
                 "controlled_executor_dispatch_ready"
                 if executor_dispatch_ready
@@ -4103,6 +4153,7 @@ def _apply_ai_automation_node_receipts(payload: dict[str, Any]) -> None:
                 f"rag_context_receipt:{_int(consumer_context_counts.get('rag'))}",
                 f"playbook_context_receipt:{_int(consumer_context_counts.get('playbook'))}",
                 f"mcp_context_receipt:{_int(consumer_context_counts.get('mcp'))}",
+                f"telegram_alert_ai_agent_context_receipt:{telegram_alert_ai_agent_context_receipt_count}",
             ],
             "verifier_result": "consumer_context_receipt_writeback_ready"
             if learning_writeback_ready

apps/api/src/services/telegram_alert_ai_automation_matrix.py

@@ -87,6 +87,9 @@ def load_latest_telegram_alert_ai_automation_matrix(
     learning_registry_present = bool(
         source_proof["ai_alert_card_learning_registry_readback_present"]
     )
+    ai_loop_context_present = bool(
+        source_proof["telegram_alert_ai_loop_consumer_context_readback_present"]
+    )
 
     direct_gap_count = int(egress_summary["direct_bot_api_call_count"])
     direct_gap_file_count = int(egress_summary["direct_bot_api_file_count"])
@@ -273,6 +276,46 @@ def load_latest_telegram_alert_ai_automation_matrix(
                 "/api/v1/platform/runs/ai-alert-cards",
             ],
         },
+        {
+            "surface_id": "telegram_ai_loop_agent_consumer_context",
+            "priority_work_item_id": "CIR-P0-TG-001",
+            "surface_kind": "ai_loop_agent_consumer_context",
+            "status": (
+                "ai_loop_context_readback_present"
+                if ai_loop_context_present
+                else "ai_loop_context_readback_missing"
+            ),
+            "known_item_count": source_proof[
+                "telegram_alert_ai_loop_consumer_context_readback_count"
+            ],
+            "db_or_log_receipt": "ready_ai_alert_card_registry_receipts",
+            "ai_route": (
+                "ready_ai_loop_agent_context_reuse"
+                if ai_loop_context_present
+                else "missing_ai_loop_agent_context_reuse"
+            ),
+            "controlled_queue": (
+                "ready_metadata_only_context_receipt"
+                if ai_loop_context_present
+                else "missing_context_receipt"
+            ),
+            "post_verifier": (
+                "ready_ai_loop_context_source_contract_present"
+                if ai_loop_context_present
+                else "missing_ai_loop_context_source_contract"
+            ),
+            "learning_writeback": (
+                "ready_agent_context_receipt_readback"
+                if ai_loop_context_present
+                else "missing_agent_context_receipt_readback"
+            ),
+            "manual_default_gap_count": 0,
+            "direct_send_gap_count": 0,
+            "evidence_refs": [
+                "apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py",
+                "/api/v1/agents/agent-log-controlled-writeback-consumer-readback",
+            ],
+        },
     ]
 
     next_priority_action = _next_priority_action(
@@ -281,6 +324,7 @@ def load_latest_telegram_alert_ai_automation_matrix(
         direct_gap_count=direct_gap_count,
         learning_writeback_refs_present=learning_writeback_refs_present,
         learning_registry_present=learning_registry_present,
+        ai_loop_context_present=ai_loop_context_present,
     )
     next_controlled_actions = _ordered_controlled_actions(next_priority_action)
     summary = _build_summary(
@@ -338,6 +382,12 @@ def load_latest_telegram_alert_ai_automation_matrix(
             (
                 (
                     (
+                        "Manual/default semantics are only acceptable for critical/break-glass or historical evidence; "
+                        "Telegram alert registry is now consumable by AI Loop Agent context, "
+                        "so the next P0 step is post-apply verifier receipt."
+                    )
+                    if ai_loop_context_present
+                    else (
                         "Manual/default semantics are only acceptable for critical/break-glass or historical evidence; "
                         "KM / PlayBook / RAG / MCP / verifier / AI Agent registry readback is present, "
                         "so the next P0 step is AI Loop Agent consumer context."
@@ -489,6 +539,10 @@ def _inspect_source_contract(repo_root: Path) -> dict[str, int | bool]:
             "apps/api/src/services/platform_operator_service.py",
             "ai_agent_context_ref",
         ),
+        "telegram_alert_ai_loop_consumer_context_readback_present": (
+            "apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py",
+            "telegram_alert_ai_loop_consumer_context_readback_v1",
+        ),
     }
     result: dict[str, int | bool] = {}
     for key, (relative_path, marker) in source_checks.items():
@@ -503,6 +557,10 @@ def _inspect_source_contract(repo_root: Path) -> dict[str, int | bool]:
             result["ai_alert_card_learning_registry_readback_count"] = (
                 1 if present else 0
             )
+        if key == "telegram_alert_ai_loop_consumer_context_readback_present":
+            result["telegram_alert_ai_loop_consumer_context_readback_count"] = (
+                1 if present else 0
+            )
 
     required = [key for key, value in result.items() if key.endswith("_present") and value is not True]
     if required:
@@ -529,6 +587,7 @@ def _next_priority_action(
     direct_gap_count: int,
     learning_writeback_refs_present: bool,
     learning_registry_present: bool,
+    ai_loop_context_present: bool,
 ) -> dict[str, Any]:
     if api_direct_gap_count > 0:
         return {
@@ -580,14 +639,24 @@ def _next_priority_action(
             "requires_runtime_send": False,
             "post_verifier": "KM / PlayBook / RAG / MCP registry readback includes CIR-P0-TG-001 alert-card receipt refs.",
         }
+    if not ai_loop_context_present:
+        return {
+            "action_id": "consume_telegram_alert_learning_registry_in_ai_loop_agent_context",
+            "scope": "KM / PlayBook / RAG / MCP / verifier registry -> AI Loop Agent consumer context",
+            "priority": "P0",
+            "why": "Telegram alert learning registry is readable; the next closure is AI Loop Agent consumer context reuse and apply receipt.",
+            "requires_secret": False,
+            "requires_runtime_send": False,
+            "post_verifier": "AI Loop Agent context readback references the CIR-P0-TG-001 Telegram alert learning registry.",
+        }
     return {
-        "action_id": "consume_telegram_alert_learning_registry_in_ai_loop_agent_context",
-        "scope": "KM / PlayBook / RAG / MCP / verifier registry -> AI Loop Agent consumer context",
+        "action_id": "verify_telegram_alert_learning_context_receipts_with_ai_loop_post_apply_verifier",
+        "scope": "AI Loop Agent Telegram alert consumer context -> post-apply verifier receipt",
         "priority": "P0",
-        "why": "Telegram alert learning registry is readable; the next closure is AI Loop Agent consumer context reuse and apply receipt.",
+        "why": "Telegram alert learning registry is readable by AI Loop Agent context; the next closure is verifier receipt and controlled apply reuse.",
         "requires_secret": False,
         "requires_runtime_send": False,
-        "post_verifier": "AI Loop Agent context readback references the CIR-P0-TG-001 Telegram alert learning registry.",
+        "post_verifier": "AI Loop post-apply verifier reads Telegram alert context receipts and reports reusable decision context.",
     }
 
 
@@ -653,6 +722,9 @@ def _build_summary(
         "ai_alert_card_learning_registry_readback_count": source_proof[
             "ai_alert_card_learning_registry_readback_count"
         ],
+        "telegram_alert_ai_loop_consumer_context_readback_count": source_proof[
+            "telegram_alert_ai_loop_consumer_context_readback_count"
+        ],
         "next_priority_action_id": next_priority_action["action_id"],
         "next_priority_work_item_id": "CIR-P0-TG-001",
     }

apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py

@@ -59,6 +59,18 @@ class _FailingContext:
         return False
 
 
+@pytest.fixture(autouse=True)
+def _stub_telegram_alert_card_readback(monkeypatch):
+    async def fake_alert_card_readback(**_kwargs):
+        return _telegram_alert_card_readback()
+
+    monkeypatch.setattr(
+        consumer_module,
+        "list_ai_alert_card_delivery_readback",
+        fake_alert_card_readback,
+    )
+
+
 def _ledger_rows() -> list[dict]:
     return [
         {
@@ -118,6 +130,71 @@ def _consumer_receipt_rows() -> list[dict]:
     ]
 
 
+def _telegram_alert_card_readback() -> dict:
+    return {
+        "summary": {
+            "project_id": "awoooi",
+            "learning_registry_target_count": 6,
+            "learning_registry_ready_target_count": 6,
+            "learning_registry_missing_target_count": 0,
+            "learning_registry_next_action": (
+                "consume_telegram_alert_learning_registry_in_ai_loop_agent_context"
+            ),
+        },
+        "items": [
+            {
+                "message_id": "msg-telegram-1",
+                "run_id": "run-telegram-1",
+                "project_id": "awoooi",
+                "event_type": "wazuh_dashboard_api_readback_degraded",
+                "lane": "siem_observability_readback_degraded",
+                "learning_writeback_refs": {
+                    "receipt_id": "msg-telegram-1",
+                    "run_id": "run-telegram-1",
+                    "work_item_id": "CIR-P0-TG-001",
+                    "post_verifier_ref": (
+                        "verifier://awoooi/telegram-alert-delivery-receipt/"
+                        "run-telegram-1"
+                    ),
+                    "ai_agent_context_ref": (
+                        "ai-agent://awoooi/telegram-alert-learning-context/"
+                        "run-telegram-1/msg-telegram-1"
+                    ),
+                },
+                "learning_registry_bindings": [
+                    {
+                        "target": target,
+                        "status": "ready_for_consumer_context",
+                        "consumer_surface": surface,
+                        "ref": f"{target}://awoooi/telegram-alert/msg-telegram-1",
+                        "work_item_id": "CIR-P0-TG-001",
+                        "target_write_performed": False,
+                        "raw_payload_included": False,
+                    }
+                    for target, surface in [
+                        ("km", "knowledge_memory_context"),
+                        ("rag", "rag_index_candidate_context"),
+                        ("playbook", "playbook_trust_candidate_context"),
+                        ("mcp", "mcp_audit_feedback_context"),
+                        ("verifier", "post_apply_verifier_feedback_context"),
+                        ("ai_agent", "autonomous_runtime_decision_context"),
+                    ]
+                ],
+            }
+        ],
+        "learning_registry": {
+            "schema_version": "ai_alert_card_learning_registry_readback_v1",
+            "status": "learning_registry_ready",
+            "project_id": "awoooi",
+            "target_count": 6,
+            "ready_target_count": 6,
+            "missing_target_count": 0,
+            "ready_receipt_ref_count": 6,
+            "consumer_context_route": "ai_loop_agent.telegram_alert_learning_context",
+        },
+    }
+
+
 @pytest.mark.asyncio
 async def test_log_controlled_writeback_consumer_loader_reads_live_ledger(monkeypatch):
     fake_db = _FakeDb(_ledger_rows())
@@ -237,6 +314,38 @@ def _assert_consumer_readback(payload: dict):
     assert payload["rollups"]["mcp_consumer_binding_count"] == 1
     assert payload["rollups"]["verifier_consumer_binding_count"] == 1
     assert payload["rollups"]["ai_agent_consumer_binding_count"] == 1
+    assert payload["rollups"]["telegram_alert_learning_context_readback_ready"] is True
+    assert payload["rollups"]["telegram_alert_learning_context_receipt_count"] == 6
+    assert (
+        payload["rollups"]["telegram_alert_learning_ai_agent_context_receipt_count"]
+        == 1
+    )
+    assert payload["rollups"]["telegram_alert_learning_ready_target_count"] == 6
+
+    telegram_context = payload["telegram_alert_learning_context"]
+    assert telegram_context["schema_version"] == (
+        "telegram_alert_ai_loop_consumer_context_readback_v1"
+    )
+    assert telegram_context["status"] == "ai_loop_agent_context_receipt_ready"
+    assert telegram_context["consumer_context_route"] == (
+        "ai_loop_agent.telegram_alert_learning_context"
+    )
+    assert telegram_context["next_action"] == (
+        "verify_telegram_alert_learning_context_receipts_with_ai_loop_post_apply_verifier"
+    )
+    assert telegram_context["context_receipt_count"] == 6
+    assert telegram_context["ai_agent_context_receipt_count"] == 1
+    ai_agent_receipt = [
+        receipt
+        for receipt in telegram_context["context_receipts"]
+        if receipt["target"] == "ai_agent"
+    ][0]
+    assert ai_agent_receipt["status"] == "ready_for_ai_loop_context"
+    assert ai_agent_receipt["work_item_id"] == "CIR-P0-TG-001"
+    assert ai_agent_receipt["metadata_only"] is True
+    assert ai_agent_receipt["target_write_performed"] is False
+    assert ai_agent_receipt["check_mode"]["enabled"] is True
+    assert ai_agent_receipt["rollback"]["required"] is True
 
     target_rollups = {item["target"]: item for item in payload["target_rollups"]}
     assert set(target_rollups) == {"km", "rag", "playbook", "mcp", "verifier", "ai_agent"}
@@ -264,6 +373,7 @@ def _assert_consumer_readback(payload: dict):
     boundaries = payload["operation_boundaries"]
     assert boundaries["consumer_readback_only"] is True
     assert boundaries["metadata_ledger_read_performed"] is True
+    assert boundaries["telegram_alert_registry_metadata_read_performed"] is True
     assert boundaries["km_write_performed"] is False
     assert boundaries["rag_index_write_performed"] is False
     assert boundaries["playbook_trust_write_performed"] is False

apps/api/tests/test_telegram_alert_ai_automation_matrix_api.py

@@ -32,7 +32,7 @@ def test_telegram_alert_ai_automation_matrix_loader_returns_gap_matrix():
     }
 
     summary = payload["summary"]
-    assert summary["telegram_alert_surface_count"] == 6
+    assert summary["telegram_alert_surface_count"] == 7
     assert summary["known_direct_send_gap_count"] == 0
     assert summary["known_direct_send_gap_file_count"] == 0
     assert summary["workflow_direct_send_gap_count"] == 0
@@ -54,8 +54,9 @@ def test_telegram_alert_ai_automation_matrix_loader_returns_gap_matrix():
     assert summary["ai_alert_card_delivery_readback_endpoint_count"] == 1
     assert summary["ai_alert_card_learning_writeback_refs_count"] == 1
     assert summary["ai_alert_card_learning_registry_readback_count"] == 1
+    assert summary["telegram_alert_ai_loop_consumer_context_readback_count"] == 1
     assert summary["next_priority_action_id"] == (
-        "consume_telegram_alert_learning_registry_in_ai_loop_agent_context"
+        "verify_telegram_alert_learning_context_receipts_with_ai_loop_post_apply_verifier"
     )
 
     matrix = {item["surface_id"]: item for item in payload["matrix"]}
@@ -88,6 +89,12 @@ def test_telegram_alert_ai_automation_matrix_loader_returns_gap_matrix():
     assert matrix["telegram_alert_learning_registry_readback"]["learning_writeback"] == (
         "ready_km_rag_playbook_mcp_verifier_ai_agent_registry"
     )
+    assert matrix["telegram_ai_loop_agent_consumer_context"]["status"] == (
+        "ai_loop_context_readback_present"
+    )
+    assert matrix["telegram_ai_loop_agent_consumer_context"]["ai_route"] == (
+        "ready_ai_loop_agent_context_reuse"
+    )
 
     assert payload["direct_send_gap_refs"] == []
     assert not any(
@@ -131,6 +138,10 @@ def test_telegram_matrix_source_contract_supports_container_layout(tmp_path):
         "ai_agent_context_ref\n",
         encoding="utf-8",
     )
+    (tmp_path / "src/services/ai_agent_log_controlled_writeback_consumer_readback.py").write_text(
+        "telegram_alert_ai_loop_consumer_context_readback_v1\n",
+        encoding="utf-8",
+    )
 
     proof = _inspect_source_contract(tmp_path)
 
@@ -145,6 +156,8 @@ def test_telegram_matrix_source_contract_supports_container_layout(tmp_path):
     assert proof["ai_alert_card_learning_registry_readback_present"] is True
     assert proof["ai_alert_card_learning_registry_readback_count"] == 1
     assert proof["ai_alert_card_ai_agent_context_ref_present"] is True
+    assert proof["telegram_alert_ai_loop_consumer_context_readback_present"] is True
+    assert proof["telegram_alert_ai_loop_consumer_context_readback_count"] == 1
 
 
 def test_telegram_alert_ai_automation_matrix_endpoint_returns_readback():