Upload 17 files

Dockerfile

@@ -0,0 +1,46 @@
+# Hugging Face Spaces (Docker) - FastAPI backend
+# Build context: this folder
+
+FROM python:3.11-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+WORKDIR /code
+
+# System deps: build tools + Postgres client libs (psycopg2) + minimal OpenCV runtime deps
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends \
+        build-essential \
+        gcc \
+        libpq-dev \
+        libgl1 \
+        libglib2.0-0 \
+    && rm -rf /var/lib/apt/lists/*
+
+# Install Python dependencies.
+# Note: This repo folder doesn't include requirements.txt/pyproject.toml, so we install pinned-at-runtime deps directly.
+RUN pip install --no-cache-dir --upgrade pip \
+    && pip install --no-cache-dir \
+        fastapi \
+        "uvicorn[standard]" \
+        sqlalchemy \
+        psycopg2-binary \
+        pydantic \
+        pydantic-settings \
+        bcrypt \
+        "python-jose[cryptography]" \
+        redis \
+        celery \
+        aiofiles \
+        python-multipart \
+        google-genai \
+        opencv-python-headless
+
+# Copy your FastAPI package into /code/app so imports like `app.main:app` work.
+COPY . /code/app
+
+# Hugging Face Spaces sets PORT; default to 7860 (HF convention) if not present.
+EXPOSE 7860
+
+CMD ["sh", "-c", "uvicorn app.main:app --host 0.0.0.0 --port ${PORT:-7860}"]

__init__.py

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

config.py

@@ -0,0 +1,78 @@
+"""
+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 current or parent directory"""
+    current_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+    
+    # Check backend directory
+    if os.path.exists(os.path.join(current_dir, ".env")):
+        return os.path.join(current_dir, ".env")
+    
+    # Check parent directory (agent-society-platform)
+    parent_dir = os.path.dirname(current_dir)
+    if os.path.exists(os.path.join(parent_dir, ".env")):
+        return os.path.join(parent_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-doc.hf.space/api/generate"
+    qwen_model_name: str = "qwen3.5:397b-cloud"
+    
+    # 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()
+

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()

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

main.py

@@ -0,0 +1,121 @@
+"""
+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 - explicit origins required when credentials=True
+origins = [
+    "http://localhost:3000",
+    "http://127.0.0.1:3000",
+    "http://localhost:8000",
+    "http://127.0.0.1:8000",
+]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS", "PATCH"],
+    allow_headers=["*"],
+    expose_headers=["*"],
+)
+
+# Register routers
+from app.routers import auth_router, projects_router, simulations_router
+app.include_router(auth_router)
+app.include_router(projects_router)
+app.include_router(simulations_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

models/__init__.py

@@ -0,0 +1,93 @@
+"""
+SQLAlchemy models for the AgentSociety platform
+"""
+import uuid
+from datetime import datetime
+from sqlalchemy import Column, String, Integer, Float, Text, DateTime, ForeignKey, JSON, BigInteger
+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")
+
+
+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)
+    virality_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
+    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)
+    
+    # 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")

results_listener.py

@@ -0,0 +1,210 @@
+"""
+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):
+        self.redis_url = redis_url or os.getenv("REDIS_URL", "redis://localhost:6379/0")
+        self.running = False
+        self.thread: Optional[threading.Thread] = None
+        
+    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"""
+        try:
+            import ssl as ssl_module
+            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")
+            
+            for message in pubsub.listen():
+                if not self.running:
+                    break
+                
+                if 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}")
+            
+            pubsub.unsubscribe()
+            
+        except Exception as e:
+            logger.error(f"ResultsListener error: {e}")
+            self.running = False
+    
+    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.virality_score = results.get('virality_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', [])
+                
+                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
+        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()

routers/__init__.py

@@ -0,0 +1,8 @@
+"""
+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
+
+__all__ = ["auth_router", "projects_router", "simulations_router"]

routers/auth.py

@@ -0,0 +1,68 @@
+"""
+Authentication routes - register, login, and user info
+"""
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+from app.database import get_db
+from app.models import User
+from app.schemas import UserCreate, UserLogin, UserResponse, TokenResponse
+from app.services.auth_service import hash_password, verify_password, create_access_token
+from app.dependencies import get_current_user
+
+router = APIRouter(prefix="/auth", tags=["Authentication"])
+
+
+@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
+async def register(user_data: UserCreate, db: Session = Depends(get_db)):
+    """
+    Register a new user account
+    """
+    # Check if email already exists
+    existing_user = db.query(User).filter(User.email == user_data.email).first()
+    if existing_user:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Email already registered"
+        )
+    
+    # Create new user
+    hashed_pw = hash_password(user_data.password)
+    new_user = User(
+        email=user_data.email,
+        password_hash=hashed_pw
+    )
+    
+    db.add(new_user)
+    db.commit()
+    db.refresh(new_user)
+    
+    return new_user
+
+
+@router.post("/login", response_model=TokenResponse)
+async def login(credentials: UserLogin, db: Session = Depends(get_db)):
+    """
+    Login and receive JWT access token
+    """
+    # Find user by email
+    user = db.query(User).filter(User.email == credentials.email).first()
+    
+    if not user or not verify_password(credentials.password, user.password_hash):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid email or password",
+            headers={"WWW-Authenticate": "Bearer"}
+        )
+    
+    # Create access token
+    access_token = create_access_token(data={"sub": str(user.id)})
+    
+    return {"access_token": access_token, "token_type": "bearer"}
+
+
+@router.get("/me", response_model=UserResponse)
+async def get_me(current_user: User = Depends(get_current_user)):
+    """
+    Get current user information
+    """
+    return current_user

routers/projects.py

@@ -0,0 +1,167 @@
+"""
+Project management routes - create, list, and get projects
+"""
+import os
+import uuid
+import aiofiles
+from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Form, status
+from sqlalchemy.orm import Session
+from typing import List, Optional
+from app.database import get_db
+from app.models import User, Project
+from app.schemas import ProjectCreate, ProjectResponse, ProjectListResponse
+from app.dependencies import get_current_user
+from app.config import get_settings
+
+settings = get_settings()
+router = APIRouter(prefix="/projects", tags=["Projects"])
+
+# Allowed video extensions
+ALLOWED_EXTENSIONS = {".mp4", ".mov", ".avi", ".webm", ".mkv"}
+MAX_FILE_SIZE = 500 * 1024 * 1024  # 500MB
+
+
+@router.post("", response_model=ProjectResponse, status_code=status.HTTP_201_CREATED)
+async def create_project(
+    title: str = Form(...),
+    demographic_filter: Optional[str] = Form(None),
+    video: UploadFile = File(...),
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db)
+):
+    """
+    Create a new project with video upload
+    """
+    # Validate file extension
+    file_ext = os.path.splitext(video.filename)[1].lower()
+    if file_ext not in ALLOWED_EXTENSIONS:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Invalid file type. Allowed types: {', '.join(ALLOWED_EXTENSIONS)}"
+        )
+    
+    # Create upload directory if it doesn't exist
+    upload_dir = os.path.join(settings.upload_dir, str(current_user.id))
+    os.makedirs(upload_dir, exist_ok=True)
+    
+    # Generate unique filename
+    file_id = str(uuid.uuid4())
+    filename = f"{file_id}{file_ext}"
+    file_path = os.path.join(upload_dir, filename)
+    
+    # Save file
+    try:
+        async with aiofiles.open(file_path, 'wb') as f:
+            # Read in chunks to handle large files
+            total_size = 0
+            while chunk := await video.read(1024 * 1024):  # 1MB chunks
+                total_size += len(chunk)
+                if total_size > MAX_FILE_SIZE:
+                    os.remove(file_path)
+                    raise HTTPException(
+                        status_code=status.HTTP_400_BAD_REQUEST,
+                        detail=f"File too large. Maximum size: {MAX_FILE_SIZE // (1024*1024)}MB"
+                    )
+                await f.write(chunk)
+    except Exception as e:
+        if os.path.exists(file_path):
+            os.remove(file_path)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Failed to save file: {str(e)}"
+        )
+    
+    # Parse demographic filter
+    demo_filter = None
+    if demographic_filter:
+        import json
+        try:
+            demo_filter = json.loads(demographic_filter)
+        except json.JSONDecodeError:
+            pass
+    
+    # Create project record
+    project = Project(
+        user_id=current_user.id,
+        title=title,
+        video_path=file_path,
+        demographic_filter=demo_filter,
+        status="PENDING"
+    )
+    
+    db.add(project)
+    db.commit()
+    db.refresh(project)
+    
+    # Queue VLM processing task (async)
+    from app.tasks import process_video_task
+    process_video_task.delay(str(project.id))
+    
+    return project
+
+
+@router.get("", response_model=List[ProjectListResponse])
+async def list_projects(
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db)
+):
+    """
+    List all projects for the current user
+    """
+    projects = db.query(Project).filter(
+        Project.user_id == current_user.id
+    ).order_by(Project.created_at.desc()).all()
+    
+    return projects
+
+
+@router.get("/{project_id}", response_model=ProjectResponse)
+async def get_project(
+    project_id: str,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db)
+):
+    """
+    Get a specific project by ID
+    """
+    project = db.query(Project).filter(
+        Project.id == project_id,
+        Project.user_id == current_user.id
+    ).first()
+    
+    if not project:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Project not found"
+        )
+    
+    return project
+
+
+@router.delete("/{project_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_project(
+    project_id: str,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db)
+):
+    """
+    Delete a project and its video file
+    """
+    project = db.query(Project).filter(
+        Project.id == project_id,
+        Project.user_id == current_user.id
+    ).first()
+    
+    if not project:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Project not found"
+        )
+    
+    # Delete video file
+    if os.path.exists(project.video_path):
+        os.remove(project.video_path)
+    
+    # Delete project record
+    db.delete(project)
+    db.commit()

routers/simulations.py

@@ -0,0 +1,328 @@
+"""
+Simulation management routes - start, status, and results
+"""
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+from typing import List
+from datetime import datetime
+from app.database import get_db
+from app.models import User, Project, SimulationRun, RiskFlag, AgentLog
+from app.schemas import (
+    SimulationCreate, 
+    SimulationResponse, 
+    SimulationStatusResponse,
+    SimulationResultsResponse,
+    RiskFlagResponse,
+    MapDataResponse,
+    AgentDetailResponse,
+    AgentProfileData,
+)
+from app.dependencies import get_current_user
+from app.config import get_settings
+
+settings = get_settings()
+router = APIRouter(prefix="/simulations", tags=["Simulations"])
+
+
+@router.post("/{project_id}/start", response_model=SimulationResponse)
+async def start_simulation(
+    project_id: str,
+    config: SimulationCreate = None,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db)
+):
+    """
+    Start a new simulation for a project
+    """
+    # Verify project exists and belongs to user
+    project = db.query(Project).filter(
+        Project.id == project_id,
+        Project.user_id == current_user.id
+    ).first()
+    
+    if not project:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Project not found"
+        )
+    
+    # Check if project is ready (VLM processing complete)
+    if project.status != "READY":
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Project is not ready for simulation. Current status: {project.status}"
+        )
+    
+    # Use default config if not provided
+    if config is None:
+        config = SimulationCreate()
+    
+    # Create simulation run record
+    simulation = SimulationRun(
+        project_id=project.id,
+        num_agents=config.num_agents,
+        simulation_days=config.simulation_days,
+        status="PENDING"
+    )
+    
+    db.add(simulation)
+    db.commit()
+    db.refresh(simulation)
+    
+    # Start simulation task (async)
+    from app.tasks import run_simulation_task
+    run_simulation_task.delay(str(simulation.id))
+    
+    return simulation
+
+
+@router.get("/{simulation_id}", response_model=SimulationResponse)
+async def get_simulation(
+    simulation_id: str,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db)
+):
+    """
+    Get simulation details
+    """
+    simulation = db.query(SimulationRun).join(Project).filter(
+        SimulationRun.id == simulation_id,
+        Project.user_id == current_user.id
+    ).first()
+    
+    if not simulation:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Simulation not found"
+        )
+    
+    return simulation
+
+
+@router.get("/{simulation_id}/status", response_model=SimulationStatusResponse)
+async def get_simulation_status(
+    simulation_id: str,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db)
+):
+    """
+    Get current simulation status (for polling)
+    """
+    simulation = db.query(SimulationRun).join(Project).filter(
+        SimulationRun.id == simulation_id,
+        Project.user_id == current_user.id
+    ).first()
+    
+    if not simulation:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Simulation not found"
+        )
+    
+    # Calculate progress for running simulations
+    progress = None
+    current_day = None
+    active_agents = None
+    
+    if simulation.status == "RUNNING":
+        # Get progress from Redis cache if available
+        import redis
+        import ssl as ssl_module
+        try:
+            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)
+            cached = r.get(f"sim:{simulation_id}:status")
+            if cached:
+                import json
+                status_data = json.loads(cached)
+                progress = status_data.get("progress")
+                current_day = status_data.get("current_day")
+                active_agents = status_data.get("active_agents")
+        except:
+            pass
+    elif simulation.status == "COMPLETED":
+        progress = 100
+    
+    return SimulationStatusResponse(
+        id=simulation.id,
+        status=simulation.status,
+        progress=progress,
+        current_day=current_day,
+        active_agents=active_agents
+    )
+
+
+@router.get("/{simulation_id}/results", response_model=SimulationResultsResponse)
+async def get_simulation_results(
+    simulation_id: str,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db)
+):
+    """
+    Get full simulation results with risk flags
+    """
+    simulation = db.query(SimulationRun).join(Project).filter(
+        SimulationRun.id == simulation_id,
+        Project.user_id == current_user.id
+    ).first()
+    
+    if not simulation:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Simulation not found"
+        )
+    
+    if simulation.status != "COMPLETED":
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Simulation has not completed yet"
+        )
+    
+    # Get risk flags
+    risk_flags = db.query(RiskFlag).filter(
+        RiskFlag.simulation_run_id == simulation.id
+    ).order_by(RiskFlag.severity.desc()).all()
+    
+    # Get sample agent logs
+    agent_sample = db.query(AgentLog).filter(
+        AgentLog.simulation_run_id == simulation.id,
+        AgentLog.event_type.in_(["BOYCOTT", "ENDORSEMENT"])
+    ).limit(10).all()
+    
+    sample_data = [
+        {
+            "agent_id": log.agent_id,
+            "event_type": log.event_type,
+            "event_data": log.event_data,
+            "timestamp": log.timestamp.isoformat()
+        }
+        for log in agent_sample
+    ]
+    
+    return SimulationResultsResponse(
+        simulation=simulation,
+        risk_flags=risk_flags,
+        agent_sample=sample_data
+    )
+
+
+@router.get("/project/{project_id}", response_model=List[SimulationResponse])
+async def list_project_simulations(
+    project_id: str,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db)
+):
+    """
+    List all simulations for a project
+    """
+    project = db.query(Project).filter(
+        Project.id == project_id,
+        Project.user_id == current_user.id
+    ).first()
+    
+    if not project:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Project not found"
+        )
+    
+    simulations = db.query(SimulationRun).filter(
+        SimulationRun.project_id == project_id
+    ).order_by(SimulationRun.created_at.desc()).all()
+    
+    return simulations
+
+
+@router.get("/{simulation_id}/map-data", response_model=MapDataResponse)
+async def get_simulation_map_data(
+    simulation_id: str,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db)
+):
+    """
+    Get lightweight map data for all agents in a completed simulation.
+    Returns coordinates, opinion and friends for each agent.
+    """
+    simulation = db.query(SimulationRun).join(Project).filter(
+        SimulationRun.id == simulation_id,
+        Project.user_id == current_user.id
+    ).first()
+    
+    if not simulation:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Simulation not found"
+        )
+    
+    if simulation.status != "COMPLETED":
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Simulation has not completed yet"
+        )
+    
+    map_data = simulation.map_data or []
+    return MapDataResponse(map_data=map_data)
+
+
+@router.get("/{simulation_id}/agents/{agent_id}", response_model=AgentDetailResponse)
+async def get_agent_detail(
+    simulation_id: str,
+    agent_id: str,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db)
+):
+    """
+    Get full detail for a specific agent in a simulation.
+    Returns profile, emotion, opinion, reasoning and friends.
+    """
+    simulation = db.query(SimulationRun).join(Project).filter(
+        SimulationRun.id == simulation_id,
+        Project.user_id == current_user.id
+    ).first()
+    
+    if not simulation:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Simulation not found"
+        )
+    
+    if simulation.status != "COMPLETED":
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Simulation has not completed yet"
+        )
+    
+    # Find the agent in agent_states
+    agent_states = simulation.agent_states or []
+    agent_data = None
+    for state in agent_states:
+        if state.get("agent_id") == agent_id:
+            agent_data = state
+            break
+    
+    if not agent_data:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Agent {agent_id} not found in simulation"
+        )
+    
+    profile_raw = agent_data.get("profile", {})
+    return AgentDetailResponse(
+        agent_id=agent_data["agent_id"],
+        coordinates=agent_data.get("coordinates", [0, 0]),
+        opinion=agent_data.get("opinion", "NEUTRAL"),
+        emotion=agent_data.get("emotion", "neutral"),
+        emotion_intensity=agent_data.get("emotion_intensity", 0),
+        reasoning=agent_data.get("reasoning", ""),
+        friends=agent_data.get("friends", []),
+        profile=AgentProfileData(
+            age=profile_raw.get("age"),
+            gender=profile_raw.get("gender"),
+            location=profile_raw.get("location"),
+            occupation=profile_raw.get("occupation"),
+            education=profile_raw.get("education"),
+            values=profile_raw.get("values", []),
+        ),
+    )

schemas.py

@@ -0,0 +1,159 @@
+"""
+Pydantic schemas for request/response validation
+"""
+from pydantic import BaseModel, EmailStr, Field
+from typing import Optional, Dict, Any, List
+from datetime import datetime
+from uuid import UUID
+
+
+# ----- User Schemas -----
+class UserCreate(BaseModel):
+    email: EmailStr
+    password: str = Field(min_length=8, max_length=100)
+
+
+class UserLogin(BaseModel):
+    email: EmailStr
+    password: str
+
+
+class UserResponse(BaseModel):
+    id: UUID
+    email: str
+    subscription_tier: str
+    created_at: datetime
+
+    class Config:
+        from_attributes = True
+
+
+class TokenResponse(BaseModel):
+    access_token: str
+    token_type: str = "bearer"
+
+
+# ----- Project Schemas -----
+class DemographicFilter(BaseModel):
+    age_range: Optional[List[int]] = None
+    location: Optional[str] = None
+    gender: Optional[str] = None
+    values: Optional[List[str]] = None
+
+
+class ProjectCreate(BaseModel):
+    title: str = Field(min_length=1, max_length=200)
+    demographic_filter: Optional[DemographicFilter] = None
+
+
+class ProjectResponse(BaseModel):
+    id: UUID
+    title: str
+    video_path: str
+    video_duration_seconds: Optional[int]
+    vlm_generated_context: Optional[str]
+    demographic_filter: Optional[Dict[str, Any]]
+    status: str
+    created_at: datetime
+
+    class Config:
+        from_attributes = True
+
+
+class ProjectListResponse(BaseModel):
+    id: UUID
+    title: str
+    status: str
+    created_at: datetime
+
+    class Config:
+        from_attributes = True
+
+
+# ----- Simulation Schemas -----
+class SimulationCreate(BaseModel):
+    num_agents: int = Field(default=10, ge=1, le=10000)
+    simulation_days: int = Field(default=5, ge=1, le=30)
+
+
+class SentimentBreakdown(BaseModel):
+    positive: int = 0
+    neutral: int = 0
+    negative: int = 0
+
+
+class RiskFlagResponse(BaseModel):
+    id: UUID
+    flag_type: str
+    severity: str
+    description: str
+    affected_demographics: Optional[Dict[str, Any]]
+    sample_agent_reactions: Optional[List[Dict[str, Any]]]
+    detected_at: datetime
+
+    class Config:
+        from_attributes = True
+
+
+class SimulationResponse(BaseModel):
+    id: UUID
+    project_id: UUID
+    status: str
+    num_agents: int
+    simulation_days: int
+    virality_score: Optional[float]
+    sentiment_breakdown: Optional[Dict[str, int]]
+    started_at: Optional[datetime]
+    completed_at: Optional[datetime]
+    error_message: Optional[str]
+    created_at: datetime
+
+    class Config:
+        from_attributes = True
+
+
+class SimulationStatusResponse(BaseModel):
+    id: UUID
+    status: str
+    progress: Optional[int] = None
+    current_day: Optional[int] = None
+    active_agents: Optional[int] = None
+
+
+class SimulationResultsResponse(BaseModel):
+    simulation: SimulationResponse
+    risk_flags: List[RiskFlagResponse]
+    agent_sample: Optional[List[Dict[str, Any]]] = None
+
+
+# ----- Map Visualization Schemas -----
+class MapAgentData(BaseModel):
+    agent_id: str
+    coordinates: List[float]
+    opinion: str
+    friends: List[str]
+
+
+class MapDataResponse(BaseModel):
+    map_data: List[MapAgentData]
+
+
+class AgentProfileData(BaseModel):
+    age: Optional[int] = None
+    gender: Optional[str] = None
+    location: Optional[str] = None
+    occupation: Optional[str] = None
+    education: Optional[str] = None
+    values: List[str] = []
+
+
+class AgentDetailResponse(BaseModel):
+    agent_id: str
+    coordinates: List[float]
+    opinion: str
+    emotion: str
+    emotion_intensity: float = 0
+    reasoning: str = ""
+    friends: List[str] = []
+    profile: AgentProfileData
+

services/__init__.py

@@ -0,0 +1,13 @@
+"""
+Service module exports
+"""
+from app.services.auth_service import hash_password, verify_password, create_access_token
+from app.services.vlm_service import process_video, analyze_video_with_gemini
+
+__all__ = [
+    "hash_password", 
+    "verify_password", 
+    "create_access_token",
+    "process_video",
+    "analyze_video_with_gemini"
+]

services/auth_service.py

@@ -0,0 +1,79 @@
+"""
+Authentication service - password hashing and JWT token management
+"""
+from datetime import datetime, timedelta
+from typing import Optional
+import bcrypt
+from jose import JWTError, jwt
+from app.config import get_settings
+
+settings = get_settings()
+
+
+def hash_password(password: str) -> str:
+    """Hash a password using bcrypt directly (passlib-free, bcrypt>=4.0 compatible)"""
+    password_bytes = password.encode("utf-8")
+    salt = bcrypt.gensalt()
+    hashed = bcrypt.hashpw(password_bytes, salt)
+    return hashed.decode("utf-8")
+
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    """Verify a password against its bcrypt hash"""
+    try:
+        return bcrypt.checkpw(
+            plain_password.encode("utf-8"),
+            hashed_password.encode("utf-8")
+        )
+    except Exception:
+        return False
+
+
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
+    """
+    Create a JWT access token
+
+    Args:
+        data: Payload to encode (should include 'sub' with user ID)
+        expires_delta: Token expiry time
+
+    Returns:
+        Encoded JWT token
+    """
+    to_encode = data.copy()
+
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(hours=settings.jwt_expiry_hours)
+
+    to_encode.update({"exp": expire})
+
+    encoded_jwt = jwt.encode(
+        to_encode,
+        settings.jwt_secret,
+        algorithm=settings.jwt_algorithm
+    )
+
+    return encoded_jwt
+
+
+def decode_access_token(token: str) -> Optional[dict]:
+    """
+    Decode and validate a JWT token
+
+    Args:
+        token: JWT token string
+
+    Returns:
+        Token payload if valid, None otherwise
+    """
+    try:
+        payload = jwt.decode(
+            token,
+            settings.jwt_secret,
+            algorithms=[settings.jwt_algorithm]
+        )
+        return payload
+    except JWTError:
+        return None

services/vlm_service.py

@@ -0,0 +1,136 @@
+"""
+Video processing service using Google Gemini Files API
+Uploads entire video instead of extracting frames for better analysis
+"""
+import os
+import time
+import logging
+from typing import Tuple
+from google import genai
+from app.config import get_settings
+
+settings = get_settings()
+logger = logging.getLogger(__name__)
+
+# Configure Gemini client
+_client = None
+
+def get_client():
+    """Get or create Gemini client"""
+    global _client
+    if _client is None:
+        _client = genai.Client(api_key=settings.gemini_api_key)
+    return _client
+
+
+def get_video_duration_cv2(video_path: str) -> int:
+    """Get video duration in seconds using OpenCV"""
+    try:
+        import cv2
+        video = cv2.VideoCapture(video_path)
+        fps = video.get(cv2.CAP_PROP_FPS)
+        frame_count = video.get(cv2.CAP_PROP_FRAME_COUNT)
+        video.release()
+        
+        if fps > 0:
+            return int(frame_count / fps)
+    except Exception as e:
+        logger.warning(f"Could not get video duration: {e}")
+    return 0
+
+
+def analyze_video_with_gemini(video_path: str) -> str:
+    """
+    Analyze video using Gemini Files API
+    
+    This uploads the entire video file to Gemini for comprehensive analysis,
+    which provides better context than frame extraction.
+    
+    Args:
+        video_path: Local path to video file
+    
+    Returns:
+        Scene descriptions and analysis
+    """
+    try:
+        client = get_client()
+        
+        # Step 1: Upload video to Gemini
+        logger.info(f"Uploading video to Gemini: {video_path}")
+        uploaded_file = client.files.upload(file=video_path)
+        logger.info(f"Uploaded file: {uploaded_file.name}")
+        
+        # Step 2: Wait for processing (videos need processing time)
+        max_wait = 120  # Maximum 2 minutes wait
+        wait_time = 0
+        
+        while uploaded_file.state == "PROCESSING":
+            if wait_time >= max_wait:
+                logger.error("Video processing timeout")
+                return "Video processing timeout. Please try with a shorter video."
+            
+            logger.info(f"Waiting for video processing... ({wait_time}s)")
+            time.sleep(10)
+            wait_time += 10
+            uploaded_file = client.files.get(name=uploaded_file.name)
+        
+        if uploaded_file.state == "FAILED":
+            logger.error("Video processing failed on Gemini side")
+            return "Video processing failed. Please try a different video format."
+        
+        logger.info("Video processing complete, analyzing content...")
+        
+        # Step 3: Analyze with Gemini
+        prompt = """Analyze this video advertisement in detail. Describe:
+
+1. **Visual Elements**: People shown (ages, genders, relationships), settings, products
+2. **Cultural Context**: Any cultural symbols, traditions, or practices shown
+3. **Narrative**: What story is being told? What emotions are evoked?
+4. **Target Audience**: Who appears to be the intended audience?
+5. **Potential Sensitivities**: Any elements that might be controversial or offensive to certain groups (religious, political, cultural, age-related)
+6. **Message**: What is the main message or call to action?
+
+Be specific and detailed. Focus on elements that could affect how different demographic groups might react to this advertisement."""
+
+        response = client.models.generate_content(
+            model="gemini-3.1-flash-lite-preview",
+            contents=[uploaded_file, prompt]
+        )
+        
+        # Step 4: Clean up uploaded file (optional - files expire automatically)
+        try:
+            client.files.delete(name=uploaded_file.name)
+            logger.info("Cleaned up uploaded video file")
+        except:
+            pass  # File cleanup is optional
+        
+        if response.text:
+            logger.info(f"Generated analysis: {len(response.text)} characters")
+            return response.text
+        else:
+            return "No analysis could be generated for this video."
+        
+    except Exception as e:
+        logger.error(f"Gemini video analysis failed: {e}")
+        return f"Video analysis failed: {str(e)}"
+
+
+def process_video(video_path: str) -> Tuple[str, int]:
+    """
+    Full video processing pipeline using Gemini Files API
+    
+    Args:
+        video_path: Path to video file
+    
+    Returns:
+        Tuple of (scene_descriptions, duration_seconds)
+    """
+    logger.info(f"Processing video: {video_path}")
+    
+    # Get duration
+    duration = get_video_duration_cv2(video_path)
+    
+    # Analyze with Gemini Files API
+    descriptions = analyze_video_with_gemini(video_path)
+    
+    return descriptions, duration

tasks.py

@@ -0,0 +1,182 @@
+"""
+Celery tasks for background processing
+"""
+import os
+import sys
+import logging
+from celery import Celery
+from datetime import datetime
+from app.config import get_settings
+
+# Add simulation directory to path (it's a sibling of backend)
+backend_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+project_root = os.path.dirname(backend_dir)
+simulation_path = os.path.join(project_root, "simulation")
+if simulation_path not in sys.path:
+    sys.path.insert(0, project_root)
+
+settings = get_settings()
+logger = logging.getLogger(__name__)
+
+# Initialize Celery
+celery_app = Celery(
+    "agentsociety",
+    broker=settings.redis_url,
+    backend=settings.redis_url
+)
+
+import ssl as ssl_module
+
+celery_app.conf.update(
+    task_serializer="json",
+    accept_content=["json"],
+    result_serializer="json",
+    timezone="UTC",
+    enable_utc=True,
+    task_track_started=True,
+    task_time_limit=3600,  # 1 hour max
+    broker_connection_retry_on_startup=True,  # silence CPendingDeprecationWarning in Celery 5/6
+)
+
+# Enable SSL for rediss:// connections (e.g. Upstash)
+if settings.redis_url.startswith("rediss://"):
+    celery_app.conf.update(
+        broker_use_ssl={"ssl_cert_reqs": ssl_module.CERT_REQUIRED},
+        redis_backend_use_ssl={"ssl_cert_reqs": ssl_module.CERT_REQUIRED},
+    )
+
+
+@celery_app.task(bind=True)
+def process_video_task(self, project_id: str):
+    """
+    Background task to process video with VLM
+    """
+    from app.database import SessionLocal
+    from app.models import Project
+    from app.services.vlm_service import process_video
+    
+    db = SessionLocal()
+    
+    try:
+        # Get project
+        project = db.query(Project).filter(Project.id == project_id).first()
+        if not project:
+            logger.error(f"Project not found: {project_id}")
+            return {"error": "Project not found"}
+        
+        # Update status
+        project.status = "PROCESSING"
+        db.commit()
+        
+        logger.info(f"Processing video for project {project_id}")
+        
+        # Process video
+        descriptions, duration = process_video(project.video_path)
+        
+        # Update project with results
+        project.vlm_generated_context = descriptions
+        project.video_duration_seconds = duration
+        project.status = "READY"
+        db.commit()
+        
+        logger.info(f"Video processing complete for project {project_id}")
+        
+        return {
+            "project_id": project_id,
+            "status": "READY",
+            "duration": duration
+        }
+        
+    except Exception as e:
+        logger.error(f"Video processing failed for project {project_id}: {e}")
+        try:
+            project = db.query(Project).filter(Project.id == project_id).first()
+            if project:
+                project.status = "FAILED"
+                db.commit()
+        except:
+            pass
+        return {"error": str(e)}
+    finally:
+        db.close()
+
+
+@celery_app.task(bind=True)
+def run_simulation_task(self, simulation_id: str):
+    """
+    Background task to queue simulation for Ray worker
+    
+    This task:
+    1. Validates the simulation and project
+    2. Publishes request to Redis 'simulation_requests' channel
+    3. Ray worker (separate process) handles the actual simulation
+    4. Results listener updates the database when complete
+    """
+    from app.database import SessionLocal
+    from app.models import SimulationRun, Project
+    import json
+    import redis
+    
+    db = SessionLocal()
+    redis_kwargs = {}
+    if settings.redis_url.startswith("rediss://"):
+        redis_kwargs["ssl_cert_reqs"] = ssl_module.CERT_REQUIRED
+    redis_client = redis.from_url(settings.redis_url, **redis_kwargs)
+    
+    try:
+        # Get simulation
+        simulation = db.query(SimulationRun).filter(SimulationRun.id == simulation_id).first()
+        if not simulation:
+            logger.error(f"Simulation not found: {simulation_id}")
+            return {"error": "Simulation not found"}
+        
+        # Get project
+        project = db.query(Project).filter(Project.id == simulation.project_id).first()
+        if not project or not project.vlm_generated_context:
+            logger.error(f"Project not ready for simulation: {simulation.project_id}")
+            simulation.status = "FAILED"
+            simulation.error_message = "Project video analysis not complete"
+            db.commit()
+            return {"error": "Project not ready"}
+        
+        # Update status to RUNNING (Ray worker will process)
+        simulation.status = "RUNNING"
+        simulation.started_at = datetime.utcnow()
+        db.commit()
+        
+        logger.info(f"Sending simulation {simulation_id} to Ray worker")
+        
+        # Publish request to Redis for Ray worker
+        request = {
+            "simulation_id": str(simulation.id),
+            "project_id": str(project.id),
+            "ad_content": project.vlm_generated_context,
+            "demographic_filter": project.demographic_filter,
+            "num_agents": simulation.num_agents,
+            "simulation_days": simulation.simulation_days
+        }
+        
+        redis_client.publish("simulation_requests", json.dumps(request))
+        
+        logger.info(f"Simulation {simulation_id} published to Ray worker queue")
+        
+        return {
+            "simulation_id": simulation_id,
+            "status": "RUNNING",
+            "message": "Simulation sent to Ray worker for processing"
+        }
+        
+    except Exception as e:
+        logger.error(f"Failed to queue simulation {simulation_id}: {e}")
+        try:
+            simulation = db.query(SimulationRun).filter(SimulationRun.id == simulation_id).first()
+            if simulation:
+                simulation.status = "FAILED"
+                simulation.error_message = str(e)
+                simulation.completed_at = datetime.utcnow()
+                db.commit()
+        except:
+            pass
+        return {"error": str(e)}
+    finally:
+        db.close()