theo-agent-dashboard

sessions.py (11113B)

---

 """Session routes — proxy to Hermes API with local topic enrichment."""
 
 from __future__ import annotations
 
 import logging
 from typing import Any
 
 from fastapi import APIRouter, HTTPException, Request
 
 from app.config import get_config
 from app.hermes_client import (
     HermesClient,
     _normalize_message,
     normalize_messages,
     _normalize_session,
     _unwrap_hermes_response,
 )
 from app.models import SessionCreate, SessionUpdate
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/sessions", tags=["sessions"])
 
 
 def get_hermes_client():  # pragma: no mutate: block
     """Build a Hermes client from the current config."""
     cfg = get_config()
     return HermesClient(base_url=cfg.hermes_base_url, api_key=cfg.hermes_api_key)
 
 
 async def _proxy_call(client: HermesClient, coro: Any) -> Any:  # pragma: no mutate: block
     """Await a client coroutine, wrapping errors as 502."""
     try:
         return await coro
     except Exception as exc:
         logger.exception("Hermes API error")
         raise HTTPException(status_code=502, detail=f"Hermes: {exc}") from exc
 
 
 async def _enrich_sessions(
     sessions: list[dict], request: Request,
 ) -> list[dict]:
     """Attach topic_id, topic_name, and archived status from the local DB."""
     db = getattr(request.app.state, "db", None)
     if db is None:
         return sessions
     topic_map = await _load_topic_map(db)
     archived = await _load_archived_set(db)
     return [_attach_topic(_attach_archived(s, archived), topic_map) for s in sessions]
 
 
 async def _load_topic_map(db: Any) -> dict[str, tuple[str, str]]:
     """Map session_id → (topic_id, topic_name) from topic_sessions."""
     try:
         rows = await db.fetch_all(
             "SELECT ts.session_id, t.id, t.name "
             "FROM topic_sessions ts "
             "JOIN topics t ON t.id = ts.topic_id"
         )
         return {r[0]: (r[1], r[2]) for r in rows}
     except Exception:
         return {}
 
 
 async def _load_archived_set(db: Any) -> set[str]:
     """Return set of archived session IDs from the DB."""
     try:
         rows = await db.fetch_all("SELECT session_id FROM archived_sessions")
         return {r[0] for r in rows}
     except Exception:
         return set()
 
 
 def _attach_topic(session: dict, topic_map: dict) -> dict:  # pragma: no mutate: block
     """Add topic fields to a session dict if available."""
     sid = session.get("id", "")
     if sid in topic_map:
         tid, tname = topic_map[sid]
         session["topic_id"] = tid
         session["topic_name"] = tname
     return session
 
 
 def _attach_archived(session: dict, archived: set[str]) -> dict:
     """Add archived: true/false to a session dict."""
     session["archived"] = session.get("id", "") in archived
     return session
 
 
 def _extract_sessions(data: Any) -> list[dict]:
     """Extract a list of normalized sessions from a Hermes response.
 
     Hermes returns ``{"object": "list", "data": [...]}``.
     Also handles legacy ``{"sessions": [...]}`` and bare ``[...]``.
     """
     if data is None:
         return []
     if isinstance(data, list):
         return [_normalize_session(s) for s in data]
     if isinstance(data, dict):
         raw_list = data.get("data") or data.get("sessions") or []
         if isinstance(raw_list, list):
             return [_normalize_session(s) for s in raw_list]
     return []
 
 
 def _extract_single_session(data: Any) -> dict:
     """Extract a single normalized session from a Hermes response.
 
     Hermes returns ``{"object": "hermes.session", "session": {...}}``.
     """
     if data is None:
         return {}
     if isinstance(data, dict):
         inner = data.get("session") or data
         if isinstance(inner, dict):
             return _normalize_session(inner)
     return {}
 
 
 @router.get("")
 async def list_sessions(request: Request) -> dict:  # pragma: no mutate: block
     """List all sessions, enriched with local topic info."""
     client = get_hermes_client()
     data = await _proxy_call(client, client.list_sessions())
     sessions = _extract_sessions(data)
     enriched = await _enrich_sessions(sessions, request)
     return {"sessions": enriched}
 
 
 @router.post("")
 async def create_session(payload: SessionCreate, request: Request) -> dict:  # pragma: no mutate: block
     """Create a session via Hermes, optionally assign to a topic."""
     client = get_hermes_client()
     result = await _proxy_call(client, client.create_session(name=payload.name))
     if result is None:
         raise HTTPException(status_code=502, detail="Hermes returned no data")
     session = _extract_single_session(result)
     await _assign_topic(request, session.get("id"), payload.topic_id)
     return session
 
 
 async def _assign_topic(  # pragma: no mutate: block
     request: Request, session_id: str | None, topic_id: str | None,
 ) -> None:  # pragma: no mutate: block
     """Insert a row into topic_sessions if both IDs are present."""
     if not session_id or not topic_id:
         return
     db = getattr(request.app.state, "db", None)
     if db is None:
         return
     try:
         await db.execute(
             "INSERT INTO topic_sessions (topic_id, session_id) VALUES (?, ?)",
             (topic_id, session_id),
         )
     except Exception:
         logger.warning("Failed to assign topic %s to session %s", topic_id, session_id)
 
 
 @router.patch("/{session_id}")
 async def update_session(  # pragma: no mutate: block
     session_id: str, payload: SessionUpdate, request: Request,
 ) -> dict:  # pragma: no mutate: block
     """Update session fields via Hermes."""
     client = get_hermes_client()
     fields: dict[str, Any] = {
         k: v for k, v in payload.model_dump().items() if v is not None
     }
     result = await _proxy_call(
         client, client.update_session(session_id, **fields),
     )
     if result is None:
         raise HTTPException(status_code=502, detail="Hermes returned no data")
     return _extract_single_session(result)
 
 
 @router.delete("/{session_id}", status_code=204)
 async def delete_session(session_id: str) -> None:  # pragma: no mutate: block
     """Delete a session via Hermes."""
     client = get_hermes_client()
     await _proxy_call(client, client.delete_session(session_id))
 
 
 @router.get("/{session_id}/messages")
 async def get_messages(session_id: str, request: Request) -> dict:  # pragma: no mutate: block
     """Get messages for a session, enriched with attachment metadata."""
     client = get_hermes_client()
     data = await _proxy_call(client, client.get_messages(session_id))
     if data is None:
         messages: list = []
     else:
         # Hermes returns {"object": "list", "data": [...]}
         raw_list = (
             data.get("data", []) if isinstance(data, dict)
             else data if isinstance(data, list)
             else []
         )
         messages = normalize_messages(raw_list) if isinstance(raw_list, list) else []
 
     # Fetch attachments for this session from local DB
     attachments = await _load_session_attachments(request, session_id)
     return {"messages": messages, "attachments": attachments}
 
 
 async def _load_session_attachments(  # pragma: no mutate: block
     request: Request, session_id: str,
 ) -> list[dict]:
     """Return attachment metadata for a session, or empty list if unavailable."""
     db = getattr(request.app.state, "db", None)
     if db is None:
         return []
     try:
         rows = await db.fetch_all(
             "SELECT id, filename, content_type, size_bytes, created_at "
             "FROM attachments WHERE session_id = ? ORDER BY created_at",
             (session_id,),
         )
         return [
             {
                 "id": r[0],
                 "filename": r[1],
                 "content_type": r[2],
                 "size_bytes": r[3],
                 "created_at": r[4],
             }
             for r in rows
         ]
     except Exception:
         return []
 
 
 @router.post("/{session_id}/fork")
 async def fork_session(session_id: str, request: Request) -> dict:  # pragma: no mutate: block
     """Fork a session via Hermes and store branch locally."""
     client = get_hermes_client()
     result = await _proxy_call(client, client.fork_session(session_id))
     if result is None:
         raise HTTPException(status_code=502, detail="Hermes returned no data")
     session = _extract_single_session(result)
     session["parent_id"] = session_id
     await _store_branch(request, session.get("id"), session_id)
     return session
 
 
 async def _store_branch(  # pragma: no mutate: block
     request: Request, child_id: str | None, parent_id: str,
 ) -> None:  # pragma: no mutate: block
     """Insert a row into the branches table."""
     if not child_id:
         return
     db = getattr(request.app.state, "db", None)
     if db is None:
         return
     try:
         await db.execute(
             "INSERT INTO branches (session_id, parent_session_id, "
             "fork_message_index, created_at) VALUES (?, ?, 0, strftime('%s','now'))",
             (child_id, parent_id),
         )
     except Exception:
         logger.warning(
             "Failed to store branch %s for session %s", child_id, parent_id,
         )
 
 
 @router.post("/{session_id}/archive")
 async def archive_session(session_id: str, request: Request) -> dict:
     """Mark a session as archived in the local DB."""
     db = getattr(request.app.state, "db", None)
     if db is None:
         raise HTTPException(status_code=500, detail="Database not available")
     try:
         await db.execute(
             "INSERT OR IGNORE INTO archived_sessions (session_id) VALUES (?)",
             (session_id,),
         )
     except Exception as exc:
         raise HTTPException(status_code=500, detail=str(exc)) from exc
     return {"session_id": session_id, "archived": True}
 
 
 @router.delete("/{session_id}/archive", status_code=204)
 async def unarchive_session(session_id: str, request: Request) -> None:
     """Remove archived status for a session."""
     db = getattr(request.app.state, "db", None)
     if db is None:
         raise HTTPException(status_code=500, detail="Database not available")
     try:
         await db.execute(
             "DELETE FROM archived_sessions WHERE session_id = ?",
             (session_id,),
         )
     except Exception as exc:
         raise HTTPException(status_code=500, detail=str(exc)) from exc
 
 
 @router.get("/{session_id}/branches")
 async def list_branches(session_id: str, request: Request) -> dict:  # pragma: no mutate: block
     """List child sessions forked from this session."""
     db = getattr(request.app.state, "db", None)
     if db is None:
         return {"branches": []}
     try:
         rows = await db.fetch_all(
             "SELECT session_id, parent_session_id, fork_message_index "
             "FROM branches WHERE parent_session_id = ?",
             (session_id,),
         )
         branches = [
             {
                 "session_id": r[0],
                 "parent_session_id": r[1],
                 "fork_message_index": r[2],
             }
             for r in rows
         ]
         return {"branches": branches}
     except Exception:
         return {"branches": []}