File size: 3,154 Bytes
6062397 | 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 98 99 100 101 102 103 104 105 106 107 108 109 110 | """Session management for FixMyNeighborhood.
Provides:
- Session-scoped orchestrator management
- Multi-user isolation via gr.State
- Session ID generation
"""
import hashlib
import time
from typing import Optional
from config import ANTHROPIC_API_KEY
from agents.controller import AutonomousController
def generate_session_id(orchestrator: Optional[AutonomousController] = None) -> str:
"""
Generate a unique session ID for rate limiting and audit.
Args:
orchestrator: Optional orchestrator instance for stable ID
Returns:
16-character hex session ID
"""
if orchestrator:
seed = str(id(orchestrator))
else:
seed = str(time.time())
return hashlib.sha256(seed.encode()).hexdigest()[:16]
def get_or_create_orchestrator(
session_orchestrator: Optional[AutonomousController]
) -> Optional[AutonomousController]:
"""
Get existing session orchestrator or create new one.
Uses gr.State to maintain orchestrator per browser session:
- Same user (same tab): Reuses orchestrator for multi-turn context
- Different users (different tabs/browsers): Separate orchestrators
Args:
session_orchestrator: Existing orchestrator from gr.State (None if first request)
Returns:
Orchestrator instance for this session, or None if API key not configured
"""
if session_orchestrator is not None:
return session_orchestrator
if not ANTHROPIC_API_KEY:
return None
try:
return AutonomousController()
except Exception as e:
print(f"Orchestrator creation error: {e}")
return None
class SessionManager:
"""
Manages session state for multi-user isolation.
In Gradio, each browser tab gets its own gr.State values.
This class provides additional session management utilities.
"""
def __init__(self):
self._active_sessions: dict = {}
def create_session(self) -> str:
"""Create a new session and return its ID."""
session_id = generate_session_id()
self._active_sessions[session_id] = {
"created_at": time.time(),
"last_activity": time.time(),
}
return session_id
def update_activity(self, session_id: str) -> None:
"""Update last activity time for a session."""
if session_id in self._active_sessions:
self._active_sessions[session_id]["last_activity"] = time.time()
def cleanup_stale_sessions(self, max_age_seconds: float = 3600) -> int:
"""
Remove sessions older than max_age.
Args:
max_age_seconds: Maximum session age in seconds (default 1 hour)
Returns:
Number of sessions cleaned up
"""
now = time.time()
stale = [
sid for sid, data in self._active_sessions.items()
if now - data["last_activity"] > max_age_seconds
]
for sid in stale:
del self._active_sessions[sid]
return len(stale)
def get_active_count(self) -> int:
"""Get count of active sessions."""
return len(self._active_sessions)
|