Axlfc · Axlfc · Jul 9, 2026 · Jul 9, 2026
diff --git a/very-simplified-stack/cognito-backend/README.md b/very-simplified-stack/cognito-backend/README.md
@@ -62,6 +62,8 @@ Settings are loaded in the following order of priority:
 - `app/core/agent_loop.py`: Turning text generation into a tool-executing agent loop.
 - `app/core/session_manager.py`: Persistence and history management for AI sessions.
 - `cli/cognito_cli.py`: Python CLI client for Cognito Agent.
+- `app/core/extensions/`: System for loading and managing extensions.
+- `app/services/escalation_routing.py`: Uncertainty-based subtask escalation mapping.
 - `test-voice-api.ps1`: The main PowerShell profile script containing `cog` and `cogt`.
 - `Install-CognitoProfile.ps1`: Installer for the PowerShell environment.
 - `config.example.json`: Template for the user configuration file.
@@ -109,6 +111,21 @@ El backend incluye un cliente ligero en Python con tres modos de operación:
 - **Project Trust**: Las herramientas de escritura y ejecución requieren que el directorio haya sido marcado como confiable.
 - **AGENTS.md**: Si existe en el raíz del `cwd`, se inyecta automáticamente como contexto del sistema.
 
+### Sistema de Extensiones (Fase 4)
+El sistema permite extender el agente sin modificar el código fuente mediante ficheros Python cargados en runtime.
+
+- **Niveles de Carga**: Global (`~/.cognito/extensions/`), Configurado (`config.json`), y Local al Proyecto (`.cognito/extensions/`).
+- **Capacidades**: Registrar herramientas nuevas, backends, intents del orquestador, y suscribirse a eventos (hooks).
+- **Seguridad**: Las extensiones locales requieren que el proyecto sea marcado como confiable para extensiones (`set_extensions_trusted`).
+  - ⚠️ **ADVERTENCIA**: Marcar un repo con `extensions_trusted=True` concede a ese código el mismo nivel de acceso que el propio proceso del backend. No es un sandbox.
+
+### Escalado Adaptativo (Fase 5)
+El orquestador (`cognito-orchestrator`) ahora puede detectar si una subtarea se ha generado con alta incertidumbre y reintentarla automáticamente con un modelo de mayor capacidad.
+
+- **Umbral de Escalado**: configurable vía `COGNITO_ESCALATION_UNCERTAINTY_THRESHOLD` (default: 0.6).
+- **Mapeo de Escalado**: Definido en `app/services/escalation_routing.py`. Axel debe revisar este archivo para asegurar que los modelos de destino están disponibles en su entorno.
+- **Transparencia**: El escalado es automático y se registra en los logs del servidor. La respuesta final incluye metadatos sobre qué subtareas fueron escaladas.
+
 ## 🧪 Testing
 
 To test the uncertainty features:

diff --git a/very-simplified-stack/cognito-backend/app/core/uncertainty.py b/very-simplified-stack/cognito-backend/app/core/uncertainty.py
@@ -37,3 +37,11 @@ def compute_uncertainty(logprob_data: Any) -> Optional[float]:
     except Exception as e:
         logger.error("[Uncertainty] Error computing entropy: %s", e)
         return None
+
+def aggregate_uncertainty(per_token_values: list[float]) -> Optional[float]:
+    """Media aritmética simple de las incertidumbres por token de una respuesta completa.
+    Devuelve None si la lista está vacía (el backend no devolvió logprobs).
+    Se usa la media y no el máximo para no disparar escalado por un único token ruidoso."""
+    if not per_token_values:
+        return None
+    return sum(per_token_values) / len(per_token_values)
diff --git a/very-simplified-stack/cognito-backend/app/services/backend_client.py b/very-simplified-stack/cognito-backend/app/services/backend_client.py
@@ -15,6 +15,7 @@
 from typing import Any, AsyncGenerator, Dict, List, Optional, Set
 
 from app.services.backend_registry import BackendConfig, BackendType
+from app.core.uncertainty import compute_uncertainty, aggregate_uncertainty
 
 logger = logging.getLogger(__name__)
 
@@ -56,6 +57,35 @@ async def generate(
         else:
             return await self._call_openai(prompt, model_params)
 
+    async def generate_with_uncertainty(
+        self,
+        prompt: str,
+        model_params: Optional[Dict[str, Any]] = None,
+    ) -> tuple[str, Optional[float]]:
+        """
+        Executes generation and aggregates uncertainty from all tokens.
+        Returns (full_text, aggregated_uncertainty).
+        """
+        full_text = ""
+        uncertainties = []
+
+        if self.config.backend_type == BackendType.OLLAMA:
+            async for chunk in self._stream_ollama(prompt, model_params):
+                token = chunk.get("token", "")
+                full_text += token
+                u = compute_uncertainty(chunk.get("logprobs"))
+                if u is not None:
+                    uncertainties.append(u)
+        else:
+            async for chunk in self._stream_openai(prompt, model_params):
+                token = chunk.get("token", "")
+                full_text += token
+                u = compute_uncertainty(chunk.get("logprobs"))
+                if u is not None:
+                    uncertainties.append(u)
+
+        return full_text, aggregate_uncertainty(uncertainties)
+
     async def generate_stream(
         self,
         prompt: str,

diff --git a/very-simplified-stack/cognito-backend/app/services/escalation_routing.py b/very-simplified-stack/cognito-backend/app/services/escalation_routing.py
@@ -0,0 +1,20 @@
+"""
+Mapeo de escalado: qué backend+modelo usar como reintento cuando la incertidumbre
+agregada de una subtarea supera el umbral. Es un punto de partida con los modelos
+que ya aparecen en MODEL_ROUTING (app/services/semantic_orchestrator.py) — no se
+inventan modelos nuevos que no estén confirmados como disponibles en el stack real.
+
+Axel: revisa y ajusta estas entradas según qué modelos tengas realmente disponibles.
+No todos los intents tienen un target de escalado por defecto — traducción, visión y
+edge son tareas especializadas donde "escalar" a un modelo de texto genérico rompería
+la tarea en vez de mejorarla, así que se dejan sin target (no se escalan).
+"""
+from typing import Dict
+
+ESCALATION_ROUTING: Dict[str, Dict[str, str]] = {
+    "fast": {"backend": "ollama-local", "model": "qwen3.5:9b"},
+    "general": {"backend": "ollama-local", "model": "phi4:latest"},
+    "coding": {"backend": "ollama-local", "model": "phi4:latest"},
+    "analysis": {"backend": "ollama-local", "model": "phi4:latest"},
+    # reasoning, translation, vision, edge: sin entrada -> no se escalan.
+}
diff --git a/very-simplified-stack/cognito-backend/app/services/semantic_orchestrator.py b/very-simplified-stack/cognito-backend/app/services/semantic_orchestrator.py
@@ -25,16 +25,25 @@
 import asyncio
 import logging
 import re
+import os
 import xml.etree.ElementTree as ET
 from dataclasses import dataclass
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Set
 
 from app.services.backend_client import BackendClient
 from app.services.backend_registry import BackendConfig, BackendType, BACKENDS_BY_PRIORITY
 from app.models.ai import AIRequest, AIResponse
+from app.services.escalation_routing import ESCALATION_ROUTING
 
 logger = logging.getLogger(__name__)
 
+# Escalation settings
+ESCALATION_ENABLED = os.environ.get("COGNITO_ESCALATION_ENABLED", "true").lower() == "true"
+ESCALATION_THRESHOLD = float(os.environ.get("COGNITO_ESCALATION_UNCERTAINTY_THRESHOLD", "0.6"))
+
+# In-memory tracking of backends/models that don't return logprobs
+_NO_LOGPROBS_WARNED: Set[tuple[str, str]] = set()
+
 
 # ── Routing table: intent → (backend_name, model) ────────────────────────────
 # Edit here to reassign tasks to different nodes.
@@ -209,6 +218,7 @@ def __init__(self, configs: List[BackendConfig], extra_routing: Optional[Dict[st
         self._client_map = _build_client_map(configs)
         self.routing = {**MODEL_ROUTING, **(extra_routing or {})}
         self._orchestrator_client = self._build_orchestrator_client()
+        self._subtask_metadata: Dict[str, Dict[str, Any]] = {} # Track metadata like escalation
 
     def add_intent_route(self, intent: str, backend_name: str, model: str) -> None:
         """Adds a new intent route at runtime."""
@@ -267,22 +277,33 @@ async def _execute_plan(self, plan: RoutingPlan, original_input: str) -> Dict[st
 
             # Execute ready tasks concurrently
             async def _run(task: SubTask) -> tuple[str, str]:
-                client = _resolve_route(task.intent, self._client_map, self.routing)
+                # Resolve original route
+                route = self.routing.get(task.intent) or self.routing.get("general")
+                backend_name = route["backend"]
+                model_name = route["model"]
+
                 # Enrich the input slice with context from upstream tasks
-                context = ""
+                context_str = ""
                 for dep_id in task.depends_on:
-                    context += f"\n<context from_task='{dep_id}'>{results[dep_id]}</context>"
+                    context_str += f"\n<context from_task='{dep_id}'>{results[dep_id]}</context>"
 
                 full_prompt = task.input_slice or original_input
-                if context:
-                    full_prompt = f"{context}\n\n{full_prompt}"
+                if context_str:
+                    full_prompt = f"{context_str}\n\n{full_prompt}"
 
-                logger.info(
-                    "[Orchestrator] Task %s → %s (%s) | intent=%s",
-                    task.id, client.config.name, client.config.model, task.intent,
+                # Execute with escalation logic
+                text, final_backend, final_model, escalated = await self._execute_subtask_with_escalation(
+                    task, task.intent, backend_name, model_name, full_prompt
                 )
-                res = await client.generate(prompt=full_prompt)
-                return task.id, res.get("response", "")
+
+                # Store metadata for process() to use
+                self._subtask_metadata[task.id] = {
+                    "backend": final_backend,
+                    "model": final_model,
+                    "escalated": escalated
+                }
+
+                return task.id, text
 
             batch_results = await asyncio.gather(*[_run(t) for t in ready])
             for task_id, response in batch_results:
@@ -292,6 +313,55 @@ async def _run(task: SubTask) -> tuple[str, str]:
 
         return results
 
+    async def _execute_subtask_with_escalation(
+        self, task: SubTask, intent: str, backend_name: str, model_name: str, prompt: str, already_escalated: bool = False
+    ) -> tuple[str, str, str, bool]:
+        client = _resolve_route(intent if not already_escalated else "general", self._client_map, {**self.routing, "general": {"backend": backend_name, "model": model_name}})
+        # Actually, _resolve_route is simpler: it just needs intent and client_map.
+        # But we want to force a specific backend/model.
+
+        import copy
+        target_client = self._client_map.get(backend_name)
+        if target_client:
+            cfg = copy.copy(target_client.config)
+            cfg.model = model_name
+            client = BackendClient(cfg)
+        else:
+            client = next(iter(self._client_map.values()))
+
+        if not ESCALATION_ENABLED:
+            res = await client.generate(prompt=prompt)
+            return res.get("response", ""), backend_name, model_name, False
+
+        text, uncertainty = await client.generate_with_uncertainty(prompt)
+
+        if uncertainty is None:
+            cache_key = (backend_name, model_name)
+            if cache_key not in _NO_LOGPROBS_WARNED and intent in ESCALATION_ROUTING:
+                logger.warning(
+                    "Backend %s / modelo %s no devuelve logprobs — el escalado no podrá activarse "
+                    "para este backend/modelo aunque el intent lo tenga configurado",
+                    backend_name, model_name
+                )
+                _NO_LOGPROBS_WARNED.add(cache_key)
+
+        if (
+            uncertainty is not None
+            and uncertainty > ESCALATION_THRESHOLD
+            and not already_escalated
+            and intent in ESCALATION_ROUTING
+        ):
+            target = ESCALATION_ROUTING[intent]
+            logger.info(
+                "Subtask %s (intent=%s) escalado de %s/%s a %s/%s: incertidumbre %.2f > umbral %.2f",
+                task.id, intent, backend_name, model_name, target["backend"], target["model"], uncertainty, ESCALATION_THRESHOLD,
+            )
+            return await self._execute_subtask_with_escalation(
+                task, intent, target["backend"], target["model"], prompt, already_escalated=True
+            )
+
+        return text, backend_name, model_name, already_escalated
+
     # ── Public interface ───────────────────────────────────────────────────────
 
     async def process(self, request: AIRequest) -> AIResponse:
@@ -328,16 +398,17 @@ async def process(self, request: AIRequest) -> AIResponse:
         duration_ms = (time.perf_counter() - start) * 1000
 
         # Build routing metadata for observability
-        routing_info = [
-            {
+        routing_info = []
+        for t in plan.subtasks:
+            meta = self._subtask_metadata.get(t.id, {})
+            routing_info.append({
                 "task_id":     t.id,
                 "description": t.description,
                 "intent":      t.intent,
-                "backend":     self.routing.get(t.intent, self.routing["general"])["backend"],
-                "model":       self.routing.get(t.intent, self.routing["general"])["model"],
-            }
-            for t in plan.subtasks
-        ]
+                "backend":     meta.get("backend", self.routing.get(t.intent, self.routing["general"])["backend"]),
+                "model":       meta.get("model", self.routing.get(t.intent, self.routing["general"])["model"]),
+                "escalated":   meta.get("escalated", False),
+            })
 
         return AIResponse(
             response=final_response,

diff --git a/very-simplified-stack/cognito-backend/tests/test_backend_client_uncertainty.py b/very-simplified-stack/cognito-backend/tests/test_backend_client_uncertainty.py
@@ -0,0 +1,76 @@
+import pytest
+import json
+import httpx
+from unittest.mock import AsyncMock, MagicMock
+from app.services.backend_client import BackendClient
+from app.services.backend_registry import BackendConfig, BackendType
+
+@pytest.mark.asyncio
+async def test_generate_with_uncertainty_ollama(respx_mock):
+    config = BackendConfig(
+        name="test-ollama",
+        base_url="http://ollama:11434",
+        backend_type=BackendType.OLLAMA,
+        model="test-model",
+        priority=1
+    )
+    client = BackendClient(config)
+
+    # Mock stream response for /api/generate (Ollama native)
+    # Logprobs must have multiple candidates to result in non-zero uncertainty
+    sse_lines = [
+        json.dumps({
+            "response": "Hello",
+            "done": False,
+            "logprobs": [{
+                "top_logprobs": [
+                    {"token": "Hello", "logprob": -0.1},
+                    {"token": "Hi", "logprob": -1.5}
+                ]
+            }]
+        }),
+        json.dumps({
+            "response": " world",
+            "done": True,
+            "logprobs": [{
+                "top_logprobs": [
+                    {"token": " world", "logprob": -0.2},
+                    {"token": " earth", "logprob": -1.2}
+                ]
+            }]
+        })
+    ]
+    respx_mock.post("http://ollama:11434/api/generate").return_value = httpx.Response(
+        200, content="\n".join(sse_lines) + "\n"
+    )
+
+    text, uncertainty = await client.generate_with_uncertainty("test prompt")
+
+    assert text == "Hello world"
+    assert uncertainty is not None
+    assert uncertainty > 0
+
+@pytest.mark.asyncio
+async def test_generate_with_uncertainty_no_logprobs(respx_mock):
+    config = BackendConfig(
+        name="test-openai",
+        base_url="http://openai:8000",
+        backend_type=BackendType.OPENAI,
+        model="test-model",
+        priority=1
+    )
+    client = BackendClient(config)
+
+    # OpenAI style stream
+    sse_content = (
+        "data: {\"choices\": [{\"delta\": {\"content\": \"Hello\"}}]}\n\n"
+        "data: [DONE]\n\n"
+    )
+    respx_mock.post("http://openai:8000/v1/chat/completions").return_value = httpx.Response(
+        200, content=sse_content
+    )
+
+    text, uncertainty = await client.generate_with_uncertainty("test prompt")
+
+    assert text == "Hello"
+    assert uncertainty is None