Spaces:
Sleeping
Sleeping
File size: 3,039 Bytes
cbb1b1a dc0c45b cbb1b1a dc0c45b cbb1b1a dc0c45b 4070852 dc0c45b cbb1b1a dc0c45b cbb1b1a dc0c45b cbb1b1a dc0c45b cbb1b1a dc0c45b cbb1b1a dc0c45b cbb1b1a 4070852 cbb1b1a dc0c45b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | """
Session registry — maps session_id → Session (agent + conversation history).
In the initial implementation this is a module-level dict (in-process).
Replace with Redis or another distributed store for horizontal scaling.
Each session holds:
agent : GUIDEAgent instance (stateful CMA)
history : ordered list of {role, content} turn dicts
created_at : ISO-8601 UTC timestamp string
"""
from __future__ import annotations
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import TYPE_CHECKING
from fastapi import HTTPException
if TYPE_CHECKING:
from src.agent.agent import GUIDEAgent
# ---------------------------------------------------------------------------
# Session container
# ---------------------------------------------------------------------------
@dataclass
class Session:
agent: "GUIDEAgent"
history: list[dict] = field(default_factory=list)
# Privacy audit trail — append-only log of every event where data either
# left the process (outbound to Anthropic) or was processed locally.
audit: list[dict] = field(default_factory=list)
created_at: str = field(
default_factory=lambda: datetime.now(timezone.utc).isoformat()
)
_sessions: dict[str, Session] = {}
# ---------------------------------------------------------------------------
# Registry operations
# ---------------------------------------------------------------------------
def create_session() -> str:
"""
Create a new GUIDEAgent session, register it, and return the session_id.
The session_id is a random UUID-4 string.
"""
from src.agent.agent import GUIDEAgent # local import avoids circular deps
session_id = str(uuid.uuid4())
_sessions[session_id] = Session(agent=GUIDEAgent(session_id))
return session_id
def get_session(session_id: str) -> Session:
"""
Return the Session for *session_id*.
Raises HTTP 404 (not KeyError) so callers in route handlers can let the
exception propagate directly to FastAPI without extra wrapping.
"""
session = _sessions.get(session_id)
if session is None:
raise HTTPException(
status_code=404,
detail=f"Session '{session_id}' not found. "
"Create one with POST /api/session/create.",
)
return session
def append_history(session_id: str, role: str, content: str) -> None:
"""Append a turn to the session's conversation history."""
_sessions[session_id].history.append({"role": role, "content": content})
def append_audit(session_id: str, entry: dict) -> None:
"""Append an event to the session's privacy audit trail."""
_sessions[session_id].audit.append(entry)
def delete_session(session_id: str) -> None:
"""Remove *session_id* from the registry (idempotent)."""
_sessions.pop(session_id, None)
def session_count() -> int:
"""Return the number of active sessions (used by the health endpoint)."""
return len(_sessions)
|