feat: Cognito Agent Phase 3 - Python CLI Client

very-simplified-stack/cognito-backend/README.md

@@ -61,6 +61,7 @@ Settings are loaded in the following order of priority:
 - `app/services/backend_client.py`: Unified async client for Ollama and OpenAI backends.
 - `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.
 - `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.
@@ -81,6 +82,22 @@ Las sesiones se guardan en `~/.cognito/sessions/` en formato JSONL (append-only)
 - **Continuidad**: Usa `session_id: "latest"` para continuar automáticamente la conversación más reciente en el `cwd` actual.
 - **Forking**: Permite clonar una sesión existente para explorar ramas alternativas sin perder el historial original.
 
+### CLI de Python (Fase 3)
+El backend incluye un cliente ligero en Python con tres modos de operación:
+
+- **Modo `print`** (default): Salida interactiva con colores ANSI TrueColor por incertidumbre.
+  ```bash
+  python -m cli.cognito_cli "Explica la fotosíntesis" --session-id latest
+  ```
+- **Modo `json`**: Salida NDJSON para integración con otros scripts.
+  ```bash
+  python -m cli.cognito_cli "Lista archivos" --mode json
+  ```
+- **Modo `rpc`**: JSON-RPC 2.0 sobre stdin/stdout, para integración con procesos de larga duración.
+  ```bash
+  python -m cli.cognito_cli --mode rpc
+  ```
+
 ### Herramientas Disponibles
 1. `read`: Lee archivos del sistema (restringido al `cwd`).
 2. `write`: Crea o sobrescribe archivos (requiere `trust`).

very-simplified-stack/cognito-backend/cli/cognito_cli.py

@@ -0,0 +1,91 @@
+import sys
+import asyncio
+import argparse
+import os
+import logging
+from cli.config import load_config
+from cli.http_client import CognitoClient
+from cli.modes.print_mode import print_mode
+from cli.modes.json_mode import json_mode
+from cli.modes.rpc_mode import rpc_mode
+
+# Configure logging to stderr
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    stream=sys.stderr
+)
+logger = logging.getLogger("cognito-cli")
+
+async def main():
+    parser = argparse.ArgumentParser(description="Cognito CLI Client")
+    parser.add_argument("prompt", nargs="?", help="The prompt to send. If missing, reads from stdin.")
+    parser.add_argument("--mode", choices=["print", "json", "rpc"], default="print", help="Output mode (default: print)")
+    parser.add_argument("--endpoint", help="Server endpoint URL")
+    parser.add_argument("--threshold", type=float, help="Uncertainty threshold")
+    parser.add_argument("--no-color", action="store_true", help="Disable colors")
+    parser.add_argument("--timeout", type=float, help="Network timeout in seconds")
+    parser.add_argument("--cwd", help="Workspace directory (default: current directory)")
+    parser.add_argument("--session-id", help="Session ID to continue (or 'latest')")
+    parser.add_argument("--color-mode", choices=["full", "threshold", "none"], help="Color mode")
+
+    args = parser.parse_args()
+
+    config = load_config(
+        cli_endpoint=args.endpoint,
+        cli_threshold=args.threshold,
+        cli_no_color=args.no_color,
+        cli_timeout=args.timeout,
+        cli_color_mode=args.color_mode
+    )
+
+    cwd = os.path.realpath(args.cwd or os.getcwd())
+
+    try:
+        async with CognitoClient(config.endpoint, config.timeout) as client:
+            if args.mode == "rpc":
+                return await rpc_mode(client, config)
+
+            # For print and json mode, we need a prompt
+            prompt = args.prompt
+            if not prompt:
+                if not sys.stdin.isatty():
+                    prompt = sys.stdin.read()
+
+            if not prompt:
+                sys.stderr.write("Error: No prompt provided.\n")
+                return 1
+
+            messages = [{"role": "user", "content": prompt}]
+            event_iterator = client.agent_loop(
+                messages=messages,
+                cwd=cwd,
+                session_id=args.session_id
+            )
+
+            if args.mode == "json":
+                return await json_mode(event_iterator, config)
+            else: # print
+                return await print_mode(event_iterator, config)
+
+    except Exception as e:
+        if args.mode == "rpc":
+            # For RPC, we already handle errors inside rpc_mode mostly,
+            # but if it fails before starting the loop:
+            print(json.dumps({
+                "jsonrpc": "2.0",
+                "error": {"code": -32000, "message": "Internal error", "data": {"detail": str(e)}},
+                "id": None
+            }))
+        else:
+            # Check for network errors specifically
+            import httpx
+            if isinstance(e, (httpx.ConnectError, httpx.TimeoutException)):
+                sys.stderr.write(f"Error: no se pudo conectar con {config.endpoint}: {str(e)}\n")
+                return 2
+            else:
+                sys.stderr.write(f"Error: {str(e)}\n")
+        return 1
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))

very-simplified-stack/cognito-backend/cli/config.py

@@ -0,0 +1,71 @@
+import os
+import json
+from pathlib import Path
+from dataclasses import dataclass
+from typing import Optional
+
+@dataclass
+class CognitoConfig:
+    endpoint: str
+    uncertainty_threshold: float
+    enable_uncertainty: bool
+    color_mode: str
+    timeout: float
+    no_color: bool = False
+
+def load_config(
+    cli_endpoint: Optional[str] = None,
+    cli_threshold: Optional[float] = None,
+    cli_no_color: bool = False,
+    cli_timeout: Optional[float] = None,
+    cli_color_mode: Optional[str] = None
+) -> CognitoConfig:
+    # 1. Defaults
+    config = {
+        "Endpoint": "http://localhost:8000",
+        "UncertaintyThreshold": 0.55,
+        "EnableUncertainty": True,
+        "ColorMode": "full",
+        "Timeout": 120.0
+    }
+
+    # 2. config.json
+    config_path = Path.home() / ".cognito" / "config.json"
+    if config_path.exists():
+        try:
+            with open(config_path, "r") as f:
+                file_config = json.load(f)
+                for k, v in file_config.items():
+                    if k in config:
+                        config[k] = v
+        except Exception:
+            pass
+
+    # 3. Environment variables
+    if os.getenv("COGNITO_ENDPOINT"):
+        config["Endpoint"] = os.getenv("COGNITO_ENDPOINT")
+    if os.getenv("COGNITO_UNCERTAINTY_THRESHOLD"):
+        config["UncertaintyThreshold"] = float(os.getenv("COGNITO_UNCERTAINTY_THRESHOLD"))
+    if os.getenv("COGNITO_ENABLE_UNCERTAINTY"):
+        config["EnableUncertainty"] = os.getenv("COGNITO_ENABLE_UNCERTAINTY").lower() != "false"
+    if os.getenv("COGNITO_COLOR_MODE"):
+        config["ColorMode"] = os.getenv("COGNITO_COLOR_MODE")
+
+    # 4. CLI arguments (highest priority)
+    if cli_endpoint:
+        config["Endpoint"] = cli_endpoint
+    if cli_threshold is not None:
+        config["UncertaintyThreshold"] = cli_threshold
+    if cli_timeout is not None:
+        config["Timeout"] = cli_timeout
+    if cli_color_mode:
+        config["ColorMode"] = cli_color_mode
+
+    return CognitoConfig(
+        endpoint=config["Endpoint"],
+        uncertainty_threshold=config["UncertaintyThreshold"],
+        enable_uncertainty=config["EnableUncertainty"],
+        color_mode=config["ColorMode"],
+        timeout=config["Timeout"],
+        no_color=cli_no_color or config["ColorMode"] == "none"
+    )

very-simplified-stack/cognito-backend/cli/http_client.py

@@ -0,0 +1,74 @@
+import json
+import httpx
+import logging
+from typing import AsyncIterator, Dict, List, Optional, Any
+
+logger = logging.getLogger(__name__)
+
+class CognitoClient:
+    def __init__(self, endpoint: str, timeout: float = 120.0):
+        self.endpoint = endpoint.rstrip("/")
+        self.timeout = httpx.Timeout(timeout, connect=timeout, read=timeout)
+        self._client = httpx.AsyncClient(timeout=self.timeout)
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        await self._client.aclose()
+
+    async def agent_loop(
+        self,
+        messages: List[Dict[str, Any]],
+        cwd: str,
+        session_id: Optional[str] = None,
+        model_params: Optional[Dict[str, Any]] = None,
+    ) -> AsyncIterator[Dict[str, Any]]:
+        url = f"{self.endpoint}/api/agent/loop"
+        payload = {
+            "messages": messages,
+            "cwd": cwd,
+            "session_id": session_id,
+            "model_params": model_params
+        }
+
+        async with self._client.stream("POST", url, json=payload) as response:
+            if response.status_code != 200:
+                # Handle non-200 responses as errors
+                try:
+                    error_data = await response.aread()
+                    detail = json.loads(error_data).get("detail", str(error_data))
+                except:
+                    detail = response.reason_phrase
+                raise RuntimeError(f"Server returned {response.status_code}: {detail}")
+
+            async for line in response.aiter_lines():
+                if not line.strip():
+                    continue
+                if line.startswith("data: "):
+                    data_str = line[6:].strip()
+                    try:
+                        yield json.loads(data_str)
+                    except json.JSONDecodeError:
+                        logger.warning(f"Failed to parse SSE data: {data_str}")
+
+    async def list_sessions(self, cwd: Optional[str] = None) -> List[Dict[str, Any]]:
+        params = {"cwd": cwd} if cwd else {}
+        resp = await self._client.get(f"{self.endpoint}/api/agent/sessions", params=params)
+        resp.raise_for_status()
+        return resp.json()
+
+    async def get_session(self, session_id: str) -> Dict[str, Any]:
+        resp = await self._client.get(f"{self.endpoint}/api/agent/sessions/{session_id}")
+        resp.raise_for_status()
+        return resp.json()
+
+    async def fork_session(self, session_id: str) -> str:
+        resp = await self._client.post(f"{self.endpoint}/api/agent/sessions/{session_id}/fork")
+        resp.raise_for_status()
+        return resp.json()["session_id"]
+
+    async def health(self) -> Dict[str, Any]:
+        resp = await self._client.get(f"{self.endpoint}/health")
+        resp.raise_for_status()
+        return resp.json()

very-simplified-stack/cognito-backend/cli/modes/json_mode.py

@@ -0,0 +1,17 @@
+import sys
+import json
+from cli.config import CognitoConfig
+
+async def json_mode(event_iterator, config: CognitoConfig):
+    exit_code = 0
+    async for event in event_iterator:
+        sys.stdout.write(json.dumps(event) + "\n")
+        sys.stdout.flush()
+
+        if event.get("type") == "done":
+            if event.get("stop_reason") != "end_turn":
+                exit_code = 1
+        elif event.get("type") == "error":
+            exit_code = 1
+
+    return exit_code

very-simplified-stack/cognito-backend/cli/modes/print_mode.py

@@ -0,0 +1,71 @@
+import sys
+from cli.config import CognitoConfig
+from typing import Dict, Any
+
+RESET = "\x1b[0m"
+
+def get_uncertainty_color(score: float) -> str:
+    if score < 0.5:
+        # blue (100, 200, 255) -> amber (255, 200, 60)
+        t = score / 0.5
+        r = int(100 + t * (255 - 100))
+        g = 200
+        b = int(255 + t * (60 - 255))
+    else:
+        # amber (255, 200, 60) -> red (255, 60, 40)
+        t = (score - 0.5) / 0.5
+        r = 255
+        g = int(200 + t * (60 - 200))
+        b = int(60 + t * (40 - 60))
+    return f"\x1b[38;2;{r};{g};{b}m"
+
+async def print_mode(event_iterator, config: CognitoConfig):
+    async for event in event_iterator:
+        event_type = event.get("type")
+
+        if event_type == "session_info":
+            sys.stderr.write(f"[session: {event.get('session_id')}]\n")
+            sys.stderr.flush()
+
+        elif event_type == "text_delta":
+            content = event.get("content", "")
+            uncertainty = event.get("uncertainty")
+
+            if not config.no_color and config.enable_uncertainty and uncertainty is not None:
+                if config.color_mode == "threshold":
+                    if uncertainty >= config.uncertainty_threshold:
+                        sys.stdout.write(f"\x1b[31m{content}{RESET}")
+                    else:
+                        sys.stdout.write(content)
+                else: # full
+                    color = get_uncertainty_color(uncertainty)
+                    sys.stdout.write(f"{color}{content}{RESET}")
+            else:
+                sys.stdout.write(content)
+            sys.stdout.flush()
+
+        elif event_type == "tool_call":
+            sys.stderr.write(f"→ ejecutando {event.get('tool_name')}({event.get('arguments')})\n")
+            sys.stderr.flush()
+
+        elif event_type == "tool_result":
+            output = event.get("output", "")
+            size = len(output)
+            sys.stderr.write(f"← {event.get('tool_name')}: {size} bytes\n")
+            sys.stderr.flush()
+
+        elif event_type == "done":
+            stop_reason = event.get("stop_reason")
+            if stop_reason != "end_turn":
+                sys.stderr.write(f"[done: {stop_reason}]\n")
+                if event.get("error_message"):
+                    sys.stderr.write(f"Error: {event.get('error_message')}\n")
+                sys.stderr.flush()
+            return 0 if stop_reason == "end_turn" else 1
+
+        elif event_type == "error":
+            sys.stderr.write(f"Error: {event.get('message')}\n")
+            sys.stderr.flush()
+            return 1
+
+    return 0