Upload 182 files

.gitattributes

@@ -38,3 +38,9 @@ backend/reports/simulation_report_0c8fa771-b63c-4986-ba2c-02eb82baeb5b.pdf filte
 backend/reports/simulation_report_89e70e1b-10ed-4fc0-bcaf-b2bbdb309584.pdf filter=lfs diff=lfs merge=lfs -text
 backend/reports/simulation_report_ab587025-827b-489b-9a3a-c1504d2d1770.pdf filter=lfs diff=lfs merge=lfs -text
 backend/reports/simulation_report_aed696fc-a253-42fd-a8c1-186494a1ee2d.pdf filter=lfs diff=lfs merge=lfs -text
+backend/reports/simulation_report_2a0bafc1-f5c0-4f61-b68c-c8cce12152b7.pdf filter=lfs diff=lfs merge=lfs -text
+backend/reports/simulation_report_46f895f7-f35a-4f35-8197-c65c22c7d119.pdf filter=lfs diff=lfs merge=lfs -text
+backend/reports/simulation_report_63b56b63-cca1-4f61-b51e-03a2aa5b81d3.pdf filter=lfs diff=lfs merge=lfs -text
+backend/reports/simulation_report_ca702f0f-e863-4cf1-886f-5e09c2ee9539.pdf filter=lfs diff=lfs merge=lfs -text
+backend/reports/simulation_report_df9ef9a7-fd76-42ec-8c33-240a2a5c95a5.pdf filter=lfs diff=lfs merge=lfs -text
+backend/reports/simulation_report_fb43988b-c0a1-4272-87c0-a126a09ea8b0.pdf filter=lfs diff=lfs merge=lfs -text

Dockerfile

@@ -0,0 +1,32 @@
+FROM python:3.11-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+	PYTHONUNBUFFERED=1 \
+	PIP_NO_CACHE_DIR=1 \
+	PORT=7860 \
+	PYTHONPATH=/app
+
+WORKDIR /app
+
+# Common system packages needed by scientific/ML dependencies.
+RUN apt-get update && apt-get install -y --no-install-recommends \
+	build-essential \
+	curl \
+	libgl1 \
+	libglib2.0-0 \
+	&& rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt ./requirements.txt
+RUN pip install --upgrade pip && pip install -r requirements.txt
+
+# Copy backend and simulation modules used by API/background logic.
+COPY backend ./backend
+COPY simulation ./simulation
+COPY start_hf_services.sh ./start_hf_services.sh
+
+RUN mkdir -p /app/backend/uploads /app/backend/reports
+RUN chmod +x /app/start_hf_services.sh
+
+EXPOSE 7860
+
+CMD ["/app/start_hf_services.sh"]

backend/app/__init__.py

@@ -0,0 +1,4 @@
+"""
+Backend application module
+"""
+from app.main import app

backend/app/config.py

@@ -0,0 +1,89 @@
+"""
+Configuration settings loaded from environment variables
+"""
+import os
+from pydantic_settings import BaseSettings
+from functools import lru_cache
+
+
+# Determine .env path - check both current dir and parent dir
+def find_env_file():
+    """Find .env file in app, backend, or project root directory"""
+    app_dir = os.path.dirname(os.path.abspath(__file__))
+    backend_dir = os.path.dirname(app_dir)
+    project_root_dir = os.path.dirname(backend_dir)
+
+    # Check app directory (this folder)
+    if os.path.exists(os.path.join(app_dir, ".env")):
+        return os.path.join(app_dir, ".env")
+
+    # Check backend directory
+    if os.path.exists(os.path.join(backend_dir, ".env")):
+        return os.path.join(backend_dir, ".env")
+
+    # Check project root directory
+    if os.path.exists(os.path.join(project_root_dir, ".env")):
+        return os.path.join(project_root_dir, ".env")
+    
+    return ".env"
+
+
+class Settings(BaseSettings):
+    # Database
+    database_url: str = "postgresql://agentsociety:dev_password@localhost:5433/agentsociety_db"
+    
+    # Redis
+    redis_url: str = "redis://localhost:6379/0"
+    
+    # MQTT
+    mqtt_broker_host: str = "localhost"
+    mqtt_broker_port: int = 1883
+    mqtt_transport: str = "tcp"
+    mqtt_path: str = ""
+    
+    # ChromaDB
+    chroma_host: str = "localhost"
+    chroma_port: int = 8000
+    chroma_ssl: bool = False
+    
+    # AWS S3 (optional)
+    aws_access_key_id: str = ""
+    aws_secret_access_key: str = ""
+    aws_s3_bucket: str = "agentsociety-videos-dev"
+    aws_region: str = "ap-south-1"
+    
+    # Google Gemini API (used by VLM video analysis)
+    gemini_api_key: str = ""
+    gemini_api_keys: str = ""  # comma-separated keys for rotation
+    
+    # Qwen LLM (HuggingFace Space - Ollama API, used by simulation agents)
+    qwen_api_url: str = "https://vish85521-qwen.hf.space/api/generate"
+    qwen_model_name: str = "qwen3.5:397b-cloud"
+
+    # Hugging Face video storage
+    hf_access_token: str = ""
+    hf_video_repo_id: str = "vish85521/videos"
+    hf_video_repo_type: str = "dataset"
+    hf_video_path_prefix: str = "videos"
+    
+    # Security
+    jwt_secret: str = "change_this_to_a_random_32_character_string"
+    jwt_algorithm: str = "HS256"
+    jwt_expiry_hours: int = 24
+    
+    # Simulation
+    default_num_agents: int = 10
+    default_simulation_days: int = 5
+    
+    # File Storage
+    upload_dir: str = "uploads"
+    
+    class Config:
+        env_file = find_env_file()
+        env_file_encoding = "utf-8"
+
+
+@lru_cache()
+def get_settings() -> Settings:
+    return Settings()
+

backend/app/database.py

@@ -0,0 +1,34 @@
+"""
+Database connection and session management
+"""
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+from app.config import get_settings
+
+settings = get_settings()
+
+# Create engine
+engine = create_engine(
+    settings.database_url,
+    pool_pre_ping=True,
+    pool_size=10,
+    max_overflow=20
+)
+
+# Create session factory
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+# Base class for models
+Base = declarative_base()
+
+
+def get_db():
+    """
+    Dependency that provides database session
+    """
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

backend/app/dependencies.py

@@ -0,0 +1,54 @@
+"""
+FastAPI dependencies for authentication and database access
+"""
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from sqlalchemy.orm import Session
+from app.database import get_db
+from app.models import User
+from app.services.auth_service import decode_access_token
+
+# Security scheme
+security = HTTPBearer()
+
+
+async def get_current_user(
+    credentials: HTTPAuthorizationCredentials = Depends(security),
+    db: Session = Depends(get_db)
+) -> User:
+    """
+    Dependency to get the current authenticated user from JWT token
+    
+    Raises:
+        HTTPException 401: If token is invalid or user not found
+    """
+    token = credentials.credentials
+    
+    # Decode token
+    payload = decode_access_token(token)
+    if payload is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid or expired token",
+            headers={"WWW-Authenticate": "Bearer"}
+        )
+    
+    # Get user ID from token
+    user_id = payload.get("sub")
+    if user_id is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid token payload",
+            headers={"WWW-Authenticate": "Bearer"}
+        )
+    
+    # Fetch user from database
+    user = db.query(User).filter(User.id == user_id).first()
+    if user is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="User not found",
+            headers={"WWW-Authenticate": "Bearer"}
+        )
+    
+    return user

backend/app/main.py

@@ -0,0 +1,117 @@
+"""
+AgentSociety Marketing Platform - FastAPI Backend
+"""
+import os
+import logging
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from contextlib import asynccontextmanager
+from sqlalchemy import text
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Application lifespan events"""
+    from app.config import get_settings
+    settings = get_settings()
+    
+    # Startup
+    logger.info("Starting AgentSociety API...")
+    
+    # Create upload directory
+    os.makedirs(settings.upload_dir, exist_ok=True)
+    
+    # Start results listener (receives simulation results from Ray worker)
+    from app.results_listener import start_results_listener
+    start_results_listener(redis_url=settings.redis_url)
+    logger.info("Results listener started - listening for Ray worker results")
+    
+    yield
+    
+    # Shutdown
+    from app.results_listener import stop_results_listener
+    stop_results_listener()
+    logger.info("Shutting down AgentSociety API...")
+
+
+# Create FastAPI app - disable redirect_slashes to prevent 307 that strips auth headers
+app = FastAPI(
+    title="AgentSociety Marketing Platform",
+    description="AI-powered marketing simulation platform that simulates 1,000+ AI agents reacting to video advertisements",
+    version="1.0.0",
+    lifespan=lifespan,
+    redirect_slashes=False
+)
+
+# CORS configuration - allow any frontend origin.
+origins = ["*"]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=False,
+    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS", "PATCH"],
+    allow_headers=["*"],
+    expose_headers=["*"],
+)
+
+# Register routers
+from app.routers import auth_router, projects_router, simulations_router, agents_router
+app.include_router(auth_router)
+app.include_router(projects_router)
+app.include_router(simulations_router)
+app.include_router(agents_router)
+
+
+@app.get("/")
+async def root():
+    """Health check endpoint"""
+    return {
+        "status": "healthy",
+        "service": "AgentSociety API",
+        "version": "1.0.0"
+    }
+
+
+@app.get("/health")
+async def health_check():
+    """Detailed health check"""
+    from app.config import get_settings
+    settings = get_settings()
+    
+    health = {
+        "api": "healthy",
+        "database": "unknown",
+        "redis": "unknown"
+    }
+    
+    # Check database
+    try:
+        from app.database import engine
+        with engine.connect() as conn:
+            conn.execute(text("SELECT 1"))
+            health["database"] = "healthy"
+    except Exception as e:
+        health["database"] = f"unhealthy: {str(e)}"
+    
+    # Check Redis
+    try:
+        import redis
+        import ssl as ssl_module
+        redis_kwargs = {}
+        if settings.redis_url.startswith("rediss://"):
+            redis_kwargs["ssl_cert_reqs"] = ssl_module.CERT_REQUIRED
+        r = redis.from_url(settings.redis_url, **redis_kwargs)
+        r.ping()
+        health["redis"] = "healthy"
+    except Exception as e:
+        health["redis"] = f"unhealthy: {str(e)}"
+    
+    return health

backend/app/models/__init__.py

@@ -0,0 +1,123 @@
+"""
+SQLAlchemy models for the AgentSociety platform
+"""
+import uuid
+from datetime import datetime
+from sqlalchemy import Column, String, Integer, Float, Text, DateTime, ForeignKey, JSON, BigInteger, Boolean
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.orm import relationship
+from app.database import Base
+
+
+class User(Base):
+    __tablename__ = "users"
+    
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    email = Column(String(255), unique=True, nullable=False, index=True)
+    password_hash = Column(String(255), nullable=False)
+    created_at = Column(DateTime, default=datetime.utcnow)
+    subscription_tier = Column(String(20), default="FREE")
+    
+    # Relationships
+    projects = relationship("Project", back_populates="user", cascade="all, delete-orphan")
+    custom_agents = relationship("CustomAgent", back_populates="user", cascade="all, delete-orphan")
+
+
+class Project(Base):
+    __tablename__ = "projects"
+    
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False)
+    title = Column(String(255), nullable=False)
+    video_path = Column(String(500), nullable=False)
+    video_duration_seconds = Column(Integer, nullable=True)
+    vlm_generated_context = Column(Text, nullable=True)
+    demographic_filter = Column(JSON, nullable=True)
+    status = Column(String(20), default="PENDING")
+    created_at = Column(DateTime, default=datetime.utcnow)
+    
+    # Relationships
+    user = relationship("User", back_populates="projects")
+    simulation_runs = relationship("SimulationRun", back_populates="project", cascade="all, delete-orphan")
+
+
+class SimulationRun(Base):
+    __tablename__ = "simulation_runs"
+    
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    project_id = Column(UUID(as_uuid=True), ForeignKey("projects.id", ondelete="CASCADE"), nullable=False)
+    status = Column(String(20), default="PENDING")
+    num_agents = Column(Integer, default=1000)
+    simulation_days = Column(Integer, default=5)
+    engagement_score = Column(Float, nullable=True)
+    sentiment_breakdown = Column(JSON, nullable=True)
+    map_data = Column(JSON, nullable=True)          # lightweight per-agent coords/opinion/friends
+    agent_states = Column(JSON, nullable=True)       # full agent profile/emotion/reasoning
+    opinion_trajectory = Column(JSON, nullable=True)
+    started_at = Column(DateTime, nullable=True)
+    completed_at = Column(DateTime, nullable=True)
+    error_message = Column(Text, nullable=True)
+    created_at = Column(DateTime, default=datetime.utcnow)
+    
+    # Custom Agents
+    use_custom_agents_only = Column(Boolean, default=False)
+    agent_ids = Column(JSON, nullable=True)
+    
+    # Relationships
+    project = relationship("Project", back_populates="simulation_runs")
+    agent_logs = relationship("AgentLog", back_populates="simulation_run", cascade="all, delete-orphan")
+    risk_flags = relationship("RiskFlag", back_populates="simulation_run", cascade="all, delete-orphan")
+
+
+class AgentLog(Base):
+    __tablename__ = "agent_logs"
+    
+    id = Column(BigInteger, primary_key=True, autoincrement=True)
+    simulation_run_id = Column(UUID(as_uuid=True), ForeignKey("simulation_runs.id", ondelete="CASCADE"), nullable=False)
+    agent_id = Column(String(50), nullable=False)
+    timestamp = Column(DateTime, default=datetime.utcnow)
+    event_type = Column(String(30), nullable=False)
+    event_data = Column(JSON, nullable=False)
+    
+    # Relationships
+    simulation_run = relationship("SimulationRun", back_populates="agent_logs")
+
+
+class RiskFlag(Base):
+    __tablename__ = "risk_flags"
+    
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    simulation_run_id = Column(UUID(as_uuid=True), ForeignKey("simulation_runs.id", ondelete="CASCADE"), nullable=False)
+    flag_type = Column(String(50), nullable=False)
+    severity = Column(String(10), nullable=False)
+    description = Column(Text, nullable=False)
+    affected_demographics = Column(JSON, nullable=True)
+    sample_agent_reactions = Column(JSON, nullable=True)
+    detected_at = Column(DateTime, default=datetime.utcnow)
+    
+    # Relationships
+    simulation_run = relationship("SimulationRun", back_populates="risk_flags")
+
+
+class CustomAgent(Base):
+    __tablename__ = "custom_agents"
+    
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False)
+    name = Column(String(100), nullable=False)
+    age = Column(Integer, nullable=False)
+    gender = Column(String(20), nullable=False)
+    location = Column(String(100), nullable=False)
+    occupation = Column(String(100), nullable=False)
+    education = Column(String(100), nullable=False)
+    income_level = Column(String(50), nullable=False)
+    religion = Column(String(50), nullable=True)
+    ethnicity = Column(String(50), nullable=True)
+    social_media_usage = Column(String(50), nullable=False)
+    political_leaning = Column(String(50), nullable=True)
+    values = Column(JSON, nullable=False)
+    personality_traits = Column(JSON, nullable=False)
+    bio = Column(Text, nullable=True)
+    created_at = Column(DateTime, default=datetime.utcnow)
+    
+    user = relationship("User", back_populates="custom_agents")

backend/app/redis_client.py

@@ -0,0 +1,12 @@
+import redis
+import functools
+import ssl as ssl_module
+from app.config import get_settings
+
+@functools.lru_cache()
+def get_redis_client():
+    settings = get_settings()
+    redis_kwargs = {}
+    if settings.redis_url.startswith("rediss://"):
+        redis_kwargs["ssl_cert_reqs"] = ssl_module.CERT_REQUIRED
+    return redis.from_url(settings.redis_url, **redis_kwargs)

backend/app/results_listener.py

@@ -0,0 +1,239 @@
+"""
+Results Listener - Background thread that listens for simulation results from Ray worker
+
+This runs as a background thread in the FastAPI application, listening to Redis
+for simulation results and updating the database accordingly.
+"""
+import os
+import json
+import logging
+import threading
+from datetime import datetime
+from typing import Optional
+
+import redis
+from sqlalchemy.orm import Session
+
+from app.database import SessionLocal
+from app.models import SimulationRun, AgentLog, RiskFlag
+
+logger = logging.getLogger(__name__)
+
+
+class ResultsListener:
+    """Background listener for simulation results from Ray worker"""
+    
+    def __init__(self, redis_url: str = None, reconnect_delay: int = 2, max_reconnect_delay: int = 30):
+        self.redis_url = redis_url or os.getenv("REDIS_URL", "redis://localhost:6379/0")
+        self.running = False
+        self.thread: Optional[threading.Thread] = None
+        self.reconnect_delay = reconnect_delay
+        self.max_reconnect_delay = max_reconnect_delay
+        
+    def start(self):
+        """Start the listener in a background thread"""
+        if self.running:
+            logger.warning("ResultsListener already running")
+            return
+            
+        self.running = True
+        self.thread = threading.Thread(target=self._listen_loop, daemon=True)
+        self.thread.start()
+        logger.info("ResultsListener started")
+    
+    def stop(self):
+        """Stop the listener"""
+        self.running = False
+        if self.thread:
+            self.thread.join(timeout=5)
+        logger.info("ResultsListener stopped")
+    
+    def _listen_loop(self):
+        """Main listening loop"""
+        import time
+        import ssl as ssl_module
+        
+        current_delay = self.reconnect_delay
+        
+        while self.running:
+            redis_client = None
+            pubsub = None
+            try:
+                redis_kwargs = {}
+                if self.redis_url.startswith("rediss://"):
+                    redis_kwargs["ssl_cert_reqs"] = ssl_module.CERT_REQUIRED
+                redis_client = redis.from_url(self.redis_url, **redis_kwargs)
+                pubsub = redis_client.pubsub()
+                pubsub.subscribe("simulation_results")
+                
+                logger.info("Subscribed to 'simulation_results' channel")
+                
+                # Reset delay on successful connection
+                current_delay = self.reconnect_delay
+                
+                while self.running:
+                    message = pubsub.get_message(timeout=1.0)
+                    if message and message['type'] == 'message':
+                        try:
+                            data = json.loads(message['data'])
+                            self._handle_result(data)
+                        except json.JSONDecodeError as e:
+                            logger.error(f"Invalid JSON in result: {e}")
+                        except Exception as e:
+                            logger.error(f"Error handling result: {e}")
+                            
+            except (redis.exceptions.ConnectionError, OSError) as e:
+                logger.error(f"Redis connection error: {e}. Reconnecting in {current_delay}s...")
+                time.sleep(current_delay)
+                current_delay = min(current_delay * 2, self.max_reconnect_delay)
+            except Exception as e:
+                logger.error(f"ResultsListener unexpected error: {e}. Retrying in {current_delay}s...")
+                time.sleep(current_delay)
+                current_delay = min(current_delay * 2, self.max_reconnect_delay)
+            finally:
+                try:
+                    if pubsub:
+                        pubsub.close()
+                except Exception:
+                    pass
+                try:
+                    if redis_client:
+                        redis_client.close()
+                except Exception:
+                    pass
+    
+    def _handle_result(self, data: dict):
+        """Process a simulation result from Ray worker"""
+        simulation_id = data.get('simulation_id')
+        status = data.get('status', 'FAILED')
+        
+        if not simulation_id:
+            logger.error("Result missing simulation_id")
+            return
+        
+        logger.info(f"Received result for simulation {simulation_id}: {status}")
+        
+        db = SessionLocal()
+        try:
+            simulation = db.query(SimulationRun).filter(
+                SimulationRun.id == simulation_id
+            ).first()
+            
+            if not simulation:
+                logger.error(f"Simulation {simulation_id} not found in database")
+                return
+            
+            if status == 'COMPLETED':
+                results = data.get('results', {})
+                
+                # Update simulation with results
+                simulation.status = "COMPLETED"
+                simulation.completed_at = datetime.utcnow()
+                simulation.engagement_score = results.get('engagement_score', 0)
+                simulation.sentiment_breakdown = results.get('sentiment_breakdown', {})
+                
+                # Save map data and agent states for map visualization
+                simulation.map_data = results.get('map_data', [])
+                simulation.agent_states = results.get('agent_states', [])
+                simulation.opinion_trajectory = results.get('opinion_trajectory', {})
+                
+                db.commit()
+                logger.info(f"Simulation {simulation_id} marked as COMPLETED")
+                
+                # Save agent logs (separate transaction to avoid rollback issues)
+                try:
+                    self._save_agent_logs(db, simulation.id, results.get('agent_logs', []))
+                except Exception as e:
+                    logger.warning(f"Failed to save agent logs: {e}")
+                
+                # Save risk flags
+                try:
+                    self._save_risk_flags(db, simulation.id, results.get('risk_flags', []))
+                except Exception as e:
+                    logger.warning(f"Failed to save risk flags: {e}")
+                
+            else:
+                # Failed simulation
+                simulation.status = "FAILED"
+                simulation.completed_at = datetime.utcnow()
+                simulation.error_message = data.get('error', 'Unknown error')
+                db.commit()
+                logger.info(f"Simulation {simulation_id} marked as FAILED: {data.get('error')}")
+        
+        except Exception as e:
+            logger.error(f"Database error handling result: {e}")
+            db.rollback()
+        finally:
+            db.close()
+    
+    def _save_agent_logs(self, db: Session, simulation_id: str, logs: list):
+        """Save agent logs to database"""
+        # Limit to prevent overwhelming the database
+        total = len(logs)
+        if total > 50:
+            logger.warning(f"Agent log truncation: {total} received, saving only 50")
+            
+        for log_data in logs[:50]:
+            try:
+                event_data = log_data.get('event_data', {})
+                # Ensure JSON serializable
+                if isinstance(event_data, dict):
+                    event_data = json.loads(json.dumps(event_data, default=str))
+                
+                agent_log = AgentLog(
+                    simulation_run_id=simulation_id,
+                    agent_id=str(log_data.get('agent_id', 'unknown')),
+                    event_type=str(log_data.get('event_type', 'UNKNOWN')),
+                    event_data=event_data
+                )
+                db.add(agent_log)
+            except Exception as e:
+                logger.warning(f"Failed to add agent log: {e}")
+        
+        db.commit()
+        logger.info(f"Saved {min(50, len(logs))} agent logs for simulation {simulation_id}")
+    
+    def _save_risk_flags(self, db: Session, simulation_id: str, flags: list):
+        """Save risk flags to database"""
+        for flag_data in flags:
+            try:
+                risk_flag = RiskFlag(
+                    simulation_run_id=simulation_id,
+                    flag_type=str(flag_data.get('flag_type', 'UNKNOWN')),
+                    severity=str(flag_data.get('severity', 'LOW')),
+                    description=str(flag_data.get('description', '')),
+                    affected_demographics=flag_data.get('affected_demographics'),
+                    sample_agent_reactions=flag_data.get('sample_agent_reactions')
+                )
+                db.add(risk_flag)
+            except Exception as e:
+                logger.warning(f"Failed to add risk flag: {e}")
+        
+        db.commit()
+        logger.info(f"Saved {len(flags)} risk flags for simulation {simulation_id}")
+
+
+# Global instance
+_results_listener: Optional[ResultsListener] = None
+
+
+def get_results_listener() -> ResultsListener:
+    """Get or create the global results listener"""
+    global _results_listener
+    if _results_listener is None:
+        _results_listener = ResultsListener()
+    return _results_listener
+
+
+def start_results_listener(redis_url: str = None):
+    """Start the global results listener"""
+    global _results_listener
+    if _results_listener is None:
+        _results_listener = ResultsListener(redis_url=redis_url)
+    _results_listener.start()
+
+
+def stop_results_listener():
+    """Stop the global results listener"""
+    if _results_listener:
+        _results_listener.stop()

backend/app/routers/__init__.py

@@ -0,0 +1,9 @@
+"""
+Router module exports
+"""
+from app.routers.auth import router as auth_router
+from app.routers.projects import router as projects_router
+from app.routers.simulations import router as simulations_router
+from app.routers.agents import router as agents_router
+
+__all__ = ["auth_router", "projects_router", "simulations_router", "agents_router"]