feat: Cognito Agent Phase 1 & 2 - Agent Loop + Session Management

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

@@ -60,6 +60,7 @@ Settings are loaded in the following order of priority:
 - `app/api/routes/openai_compat.py`: Core streaming and uncertainty calculation logic.
 - `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.
 - `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.
@@ -73,6 +74,13 @@ El backend ahora incluye soporte para agentes capaces de ejecutar herramientas (
   - **Body**: `{ "messages": [...], "cwd": "path/to/repo", "model_params": {} }`
   - **Eventos**: `text_delta`, `tool_call`, `tool_result`, `done`, `error`.
 
+### Sesiones y Persistencia (Fase 2)
+Las sesiones se guardan en `~/.cognito/sessions/` en formato JSONL (append-only) con un índice global en `index.json`.
+
+- **Compactado Automático**: Cuando una sesión supera el umbral de tokens (default 8000), el sistema genera automáticamente un resumen y compacta el historial para liberar ventana de contexto.
+- **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.
+
 ### Herramientas Disponibles
 1. `read`: Lee archivos del sistema (restringido al `cwd`).
 2. `write`: Crea o sobrescribe archivos (requiere `trust`).

very-simplified-stack/cognito-backend/app/api/routes/ai_agents.py

@@ -1,5 +1,6 @@
 import json
 from typing import List, Optional, Any, Dict
+from pathlib import Path
 from fastapi import APIRouter, HTTPException
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel
@@ -14,6 +15,9 @@
 from app.core.tools.bash_tool import BashTool
 from app.core.project_trust import ProjectTrustStore
 from app.core.resource_loader import ResourceLoader
+from app.core.session_manager import SessionManager
+from app.core.compaction import should_compact, compact
+from app.core.events import SessionInfoEvent, TextDeltaEvent, ToolCallEvent, ToolResultEvent, DoneEvent, ErrorEvent
 import logging
 
 # Configure logging
@@ -25,6 +29,7 @@
 class AgentLoopRequest(BaseModel):
     messages: List[Dict[str, Any]]
     cwd: str
+    session_id: Optional[str] = None
     model_params: Optional[Dict[str, Any]] = None
 
 @router.post("/agent", response_model=AIResponse)
@@ -49,50 +54,177 @@ async def run_ai_agent(request: AIRequest):
 @router.post("/agent/loop")
 async def run_agent_loop(request: AgentLoopRequest):
     """
-    Agent Loop endpoint (SSE).
+    Agent Loop endpoint (SSE) with Session Management.
     """
-    logger.info(f"Starting agent loop in {request.cwd}")
+    logger.info(f"Starting agent loop in {request.cwd} | session_id={request.session_id}")
 
+    session_manager = SessionManager()
+    is_new = False
+    session_id = request.session_id
+
+    # 1. Resolver session_id
+    try:
+        if session_id == "latest":
+            session_id = session_manager.continue_recent(request.cwd)
+            if not session_id:
+                session_id = session_manager.create(request.cwd)
+                is_new = True
+        elif session_id:
+            metadata = session_manager.open(session_id)
+            if metadata.cwd != str(Path(request.cwd).resolve()):
+                error_msg = f"CWD mismatch: session {session_id} is for {metadata.cwd}, request is for {request.cwd}"
+                logger.error(error_msg)
+                return StreamingResponse(
+                    iter([f"data: {json.dumps({'type': 'error', 'message': error_msg})}\n\n"]),
+                    media_type="text/event-stream"
+                )
+        else:
+            session_id = session_manager.create(request.cwd)
+            is_new = True
+    except Exception as e:
+        logger.error(f"Session resolution failed: {e}")
+        return StreamingResponse(
+            iter([f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"]),
+            media_type="text/event-stream"
+        )
+
+    # 2. Cargar historial y compactar si es necesario
+    effective_messages = session_manager.get_effective_messages(session_id)
+    if await should_compact(effective_messages):
+        try:
+            last_line = session_manager.get_last_line_index(session_id)
+            summary = await compact(effective_messages, backend_router=backend_router)
+            session_manager.append_compaction(session_id, summary, last_line)
+            effective_messages = session_manager.get_effective_messages(session_id)
+        except Exception as e:
+            logger.warning(f"Compaction failed for session {session_id}, continuing anyway: {e}")
+
+    # 3. Preparar contexto y herramientas
     loader = ResourceLoader(request.cwd)
     trust_store = ProjectTrustStore()
-
     context = ToolContext(
         cwd=request.cwd,
         trusted=trust_store.is_trusted(request.cwd),
         protected_files=loader.get_effective_protected_files()
     )
-
     tools = [ReadTool(), WriteTool(), EditTool(), BashTool()]
 
-    # Inyectar AGENTS.md en el system prompt si existe
+    # Inyectar AGENTS.md
     agents_md = loader.discover_agents_md()
-    messages = list(request.messages)
+    history = list(effective_messages)
     if agents_md:
-        system_msg = next((m for m in messages if m["role"] == "system"), None)
+        system_msg = next((m for m in history if m["role"] == "system"), None)
         if system_msg:
             system_msg["content"] += f"\n\nContext from AGENTS.md:\n{agents_md}"
         else:
-            messages.insert(0, {"role": "system", "content": f"Context from AGENTS.md:\n{agents_md}"})
+            history.insert(0, {"role": "system", "content": f"Context from AGENTS.md:\n{agents_md}"})
+
+    # Mensajes nuevos de la request
+    new_messages = list(request.messages)
+    full_messages_for_loop = history + new_messages
+
+    # REDESIGN: To persist assistant messages correctly, we need to capture them.
+    # The current agent_loop yields events but doesn't return the full objects.
+    async def event_generator_v2():
+        # Persist incoming new messages from request
+        for msg in new_messages:
+            session_manager.append_message(session_id, msg["role"], msg["content"])
+
+        yield f"data: {SessionInfoEvent(session_id=session_id, is_new=is_new).model_dump_json()}\n\n"
+
+        assistant_content = ""
+        current_tool_calls = []
 
-    async def event_generator():
         try:
             async for event in agent_loop(
-                messages=messages,
+                messages=full_messages_for_loop,
                 tools=tools,
                 context=context,
                 backend_router=backend_router,
                 model_params=request.model_params
             ):
+                if isinstance(event, TextDeltaEvent):
+                    assistant_content += event.content
+                elif isinstance(event, ToolCallEvent):
+                    current_tool_calls.append({
+                        "id": event.tool_call_id,
+                        "type": "function",
+                        "function": {"name": event.tool_name, "arguments": json.dumps(event.arguments)}
+                    })
+                elif isinstance(event, ToolResultEvent):
+                    # Before persisting tool result, we MUST persist the assistant message that called it
+                    if assistant_content or current_tool_calls:
+                        session_manager.append_message(
+                            session_id,
+                            role="assistant",
+                            content=assistant_content,
+                            tool_calls=current_tool_calls if current_tool_calls else None
+                        )
+                        assistant_content = ""
+                        current_tool_calls = []
+
+                    session_manager.append_message(
+                        session_id,
+                        role="tool",
+                        content=event.output,
+                        tool_name=event.tool_name,
+                        tool_call_id=event.tool_call_id
+                    )
+                elif isinstance(event, DoneEvent):
+                    if assistant_content or current_tool_calls:
+                        session_manager.append_message(
+                            session_id,
+                            role="assistant",
+                            content=assistant_content,
+                            tool_calls=current_tool_calls if current_tool_calls else None
+                        )
+
                 yield f"data: {event.model_dump_json()}\n\n"
         except Exception as e:
             logger.error(f"Error in agent loop generator: {e}", exc_info=True)
             yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
 
     return StreamingResponse(
-        event_generator(),
+        event_generator_v2(),
         media_type="text/event-stream",
         headers={
             "Cache-Control": "no-cache",
             "X-Accel-Buffering": "no",
         }
     )
+
+@router.get("/agent/sessions")
+async def list_sessions(cwd: Optional[str] = None):
+    """
+    List all sessions, optionally filtered by cwd.
+    """
+    session_manager = SessionManager()
+    return session_manager.list_all(cwd=cwd)
+
+@router.get("/agent/sessions/{session_id}")
+async def get_session(session_id: str):
+    """
+    Get session metadata and effective messages.
+    """
+    session_manager = SessionManager()
+    try:
+        metadata = session_manager.open(session_id)
+        messages = session_manager.get_effective_messages(session_id)
+        return {
+            "metadata": metadata,
+            "messages": messages
+        }
+    except FileNotFoundError:
+        raise HTTPException(status_code=404, detail=f"Session {session_id} not found")
+
+@router.post("/agent/sessions/{session_id}/fork")
+async def fork_session(session_id: str):
+    """
+    Fork an existing session into a new one.
+    """
+    session_manager = SessionManager()
+    try:
+        new_id = session_manager.fork_from(session_id)
+        return {"session_id": new_id}
+    except FileNotFoundError:
+        raise HTTPException(status_code=404, detail=f"Session {session_id} not found")

very-simplified-stack/cognito-backend/app/core/compaction.py

@@ -0,0 +1,74 @@
+import os
+import json
+import logging
+from typing import List, Dict
+
+logger = logging.getLogger(__name__)
+
+# Configurable constants via environment variables
+COMPACTION_THRESHOLD_TOKENS = int(os.getenv("COGNITO_SESSION_COMPACTION_THRESHOLD_TOKENS", "8000"))
+KEEP_LAST_N_MESSAGES = int(os.getenv("COGNITO_SESSION_KEEP_LAST_N_MESSAGES", "6"))
+
+COMPACTION_SYSTEM_PROMPT = """Resume todos los mensajes anteriores de esta conversación de forma concisa.
+ES CRÍTICO que preservas:
+1. Decisiones clave tomadas por el usuario o el asistente.
+2. Archivos que han sido leídos, escritos o editados.
+3. Tareas pendientes que el usuario ha solicitado.
+4. Cualquier restricción o advertencia mencionada (especialmente si se rechazó el acceso a un archivo protegido).
+
+Devuelve únicamente el texto del resumen, sin preámbulos ni explicaciones adicionales."""
+
+async def should_compact(effective_messages: List[Dict], threshold_tokens: int = COMPACTION_THRESHOLD_TOKENS) -> bool:
+    """
+    Heuristic token count: chars // 4.
+    Returns True if threshold is exceeded.
+    """
+    if len(effective_messages) <= KEEP_LAST_N_MESSAGES + 2: # Don't compact very short history
+        return False
+
+    total_chars = 0
+    for msg in effective_messages:
+        content = msg.get("content", "") or ""
+        total_chars += len(content)
+        # Also count tool names and IDs if present
+        if "name" in msg: total_chars += len(msg["name"])
+        if "tool_call_id" in msg: total_chars += len(msg["tool_call_id"])
+
+    estimated_tokens = total_chars // 4
+    return estimated_tokens > threshold_tokens
+
+async def compact(
+    effective_messages: List[Dict],
+    keep_last_n: int = KEEP_LAST_N_MESSAGES,
+    backend_router = None,
+) -> str:
+    """
+    Summarizes messages except the last N using the backend router.
+    """
+    if not backend_router:
+        from app.services.backend_router import backend_router as global_router
+        backend_router = global_router
+
+    # Separate messages to compact and messages to keep
+    to_compact = effective_messages[:-keep_last_n] if keep_last_n > 0 else effective_messages
+
+    # Format to_compact for the prompt
+    history_text = ""
+    for msg in to_compact:
+        role = msg.get("role", "unknown")
+        content = msg.get("content", "")
+        history_text += f"[{role}]: {content}\n\n"
+
+    prompt = f"{COMPACTION_SYSTEM_PROMPT}\n\nHistorial a resumir:\n{history_text}"
+
+    try:
+        logger.info(f"Triggering compaction for {len(to_compact)} messages")
+        result = await backend_router.generate(prompt=prompt)
+        summary = result.get("response", "").strip()
+        if not summary:
+            logger.warning("Compaction generated an empty summary.")
+            return "Resumen no disponible por error del modelo."
+        return summary
+    except Exception as e:
+        logger.error(f"Failed to generate compaction summary: {e}", exc_info=True)
+        raise e

very-simplified-stack/cognito-backend/app/core/events.py

@@ -28,4 +28,9 @@ class ErrorEvent(BaseModel):
     type: Literal["error"] = "error"
     message: str
 
-AgentEvent = Union[TextDeltaEvent, ToolCallEvent, ToolResultEvent, DoneEvent, ErrorEvent]
+class SessionInfoEvent(BaseModel):
+    type: Literal["session_info"] = "session_info"
+    session_id: str
+    is_new: bool
+
+AgentEvent = Union[TextDeltaEvent, ToolCallEvent, ToolResultEvent, DoneEvent, ErrorEvent, SessionInfoEvent]