fix(agents): fallback learning loop from consumer receipts

apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py

@@ -304,6 +304,57 @@ async def test_log_controlled_writeback_consumer_loader_uses_direct_fallback(
     assert payload["operation_boundaries"]["metadata_ledger_read_performed"] is True
 
 
+@pytest.mark.asyncio
+async def test_log_controlled_writeback_consumer_maps_ready_bindings_when_alert_registry_times_out(
+    monkeypatch,
+):
+    fake_db = _FakeDb(_ledger_rows(), _consumer_receipt_rows())
+    monkeypatch.setattr(
+        consumer_module,
+        "get_db_context",
+        lambda project_id: _FakeContext(fake_db),
+    )
+
+    async def alert_registry_timeout(**_kwargs):
+        raise TimeoutError("alert registry unavailable")
+
+    monkeypatch.setattr(
+        consumer_module,
+        "list_ai_alert_card_delivery_readback",
+        alert_registry_timeout,
+    )
+
+    payload = await load_latest_ai_agent_log_controlled_writeback_consumer_readback()
+
+    assert payload["status"] == "controlled_writeback_consumer_readback_ready"
+    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
+    )
+    context = payload["telegram_alert_learning_context"]
+    assert context["status"] == "ai_loop_agent_context_receipt_ready"
+    assert context["source_registry_status"] == (
+        "fallback_log_controlled_writeback_consumer_context"
+    )
+    assert context["active_blockers"] == []
+    assert {receipt["target"] for receipt in context["context_receipts"]} == {
+        "km",
+        "rag",
+        "playbook",
+        "mcp",
+        "verifier",
+        "ai_agent",
+    }
+    assert all(
+        receipt["status"] == "ready_for_ai_loop_context"
+        for receipt in context["context_receipts"]
+    )
+    assert context["operation_boundaries"]["runtime_target_write_performed"] is True
+    assert context["operation_boundaries"]["raw_payload_included"] is False
+
+
 def test_log_controlled_writeback_consumer_direct_rows_parse_json_text():
     dispatch_row = _ledger_rows()[0]
     consumer_row = _consumer_receipt_rows()[0]