Setup Supabase Auth and Fastapi middleware.

Backend/api/__init__.py

@@ -0,0 +1,3 @@
+from .app import create_app
+
+app = create_app()

Backend/api/app.py

@@ -0,0 +1,126 @@
+from contextlib import asynccontextmanager
+from pathlib import Path
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse, FileResponse
+from fastapi.staticfiles import StaticFiles
+
+from Backend.core.config import settings
+from Backend.core.events import event_bus
+from Backend.core.logging import setup_logging, get_logger
+from Backend.core.security import SecurityHeadersMiddleware, RateLimitMiddleware, RequestValidationMiddleware
+from Backend.database.connection import init_db, close_db
+from Backend.api.routes import api_router
+
+logger = get_logger(__name__)
+
+STATIC_DIR = Path("static")
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    setup_logging(debug=settings.debug)
+    logger.info("Starting City Issue Resolution Agent")
+    
+    await init_db()
+    logger.info("Database initialized")
+    
+    await event_bus.start()
+    logger.info("Event bus started")
+    
+    
+    from Backend.agents.vision import VisionAgent
+    try:
+        VisionAgent.load_model()
+        logger.info("Vision model loaded")
+    except Exception as e:
+        logger.warning(f"Vision model failed to load: {e}. Running in mock mode.")
+    
+    
+    import asyncio
+    from Backend.database.connection import get_db_context
+    from Backend.agents.escalation.agent import EscalationAgent
+    from Backend.agents.sla.agent import SLAAgent
+    
+    async def run_periodic_checks():
+        while True:
+            try:
+                logger.info("Running periodic SLA and Escalation checks...")
+                async with get_db_context() as db:
+                    
+                    esc_agent = EscalationAgent(db)
+                    await esc_agent.check_all_pending()
+                    
+                    
+                    sla_agent = SLAAgent(db)
+                    await sla_agent.check_all_active()
+            except Exception as e:
+                logger.error(f"Error in background task: {e}")
+            
+            
+            await asyncio.sleep(900)
+            
+    task = asyncio.create_task(run_periodic_checks())
+    
+    yield
+    
+    task.cancel()
+    await event_bus.stop()
+    await close_db()
+    logger.info("Shutdown complete")
+
+
+def create_app() -> FastAPI:
+    app = FastAPI(
+        title="City Issue Resolution Agent",
+        description="Autonomous urban issue detection and resolution platform",
+        version="1.0.0",
+        lifespan=lifespan,
+        root_path="",
+    )
+    
+    # CORS must be added first
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_credentials=False,
+        allow_methods=["*"],
+        allow_headers=["*"],
+        expose_headers=["*"],
+    )
+    
+    app.add_middleware(SecurityHeadersMiddleware)
+    app.add_middleware(RateLimitMiddleware, requests_per_minute=120, burst_limit=20)
+    app.add_middleware(RequestValidationMiddleware)
+
+    
+    settings.local_temp_dir.mkdir(parents=True, exist_ok=True)
+    STATIC_DIR.mkdir(parents=True, exist_ok=True)
+    
+    app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
+    
+    app.include_router(api_router)
+    
+    @app.get("/")
+    async def root():
+        return FileResponse(STATIC_DIR / "flow.html")
+    
+    @app.get("/dashboard")
+    async def dashboard():
+        return FileResponse(STATIC_DIR / "flow.html")
+    
+    @app.exception_handler(ValueError)
+    async def value_error_handler(request: Request, exc: ValueError):
+        return JSONResponse(
+            status_code=400,
+            content={"detail": str(exc)}
+        )
+    
+    @app.exception_handler(Exception)
+    async def general_exception_handler(request: Request, exc: Exception):
+        logger.error(f"Unhandled exception: {exc}", exc_info=True)
+        return JSONResponse(
+            status_code=500,
+            content={"detail": "Internal server error"}
+        )
+    
+    return app

Backend/api/routes/__init__.py

@@ -0,0 +1,16 @@
+from fastapi import APIRouter
+
+from .health import router as health_router
+from .issues import router as issues_router
+from .admin import router as admin_router
+from .flow import router as flow_router
+from .worker import router as worker_router
+
+api_router = APIRouter()
+
+api_router.include_router(health_router, prefix="/health", tags=["Health"])
+api_router.include_router(issues_router, prefix="/issues", tags=["Issues"])
+api_router.include_router(admin_router, prefix="/admin", tags=["Admin"])
+api_router.include_router(flow_router, prefix="/flow", tags=["Agent Flow"])
+api_router.include_router(worker_router, prefix="/worker", tags=["Worker"])
+

Backend/api/routes/flow.py

@@ -0,0 +1,163 @@
+import asyncio
+import json
+from dataclasses import asdict
+from typing import Optional
+from uuid import UUID
+from fastapi import APIRouter, Depends, Query
+from fastapi.responses import StreamingResponse
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from Backend.database.connection import get_db
+from Backend.database.models import Issue, IssueEvent
+from Backend.core.flow_tracker import get_flow_tracker, _active_flows
+
+router = APIRouter()
+
+
+async def event_generator(issue_id: UUID, timeout: int = 300):
+    tracker = get_flow_tracker(issue_id)
+    
+    if not tracker:
+        yield f"data: {json.dumps({'type': 'error', 'message': 'No active flow for this issue'})}\n\n"
+        return
+    
+    queue = tracker.subscribe()
+    
+    try:
+        start_msg = {
+            "type": "connected",
+            "issue_id": str(issue_id),
+            "message": "Connected to agent flow stream",
+            "current_steps": [asdict(s) for s in tracker.flow.steps]
+        }
+        yield f"data: {json.dumps(start_msg)}\n\n"
+        
+        
+        
+        
+        
+        
+        
+        
+        if tracker.flow.status in ["completed", "error"]:
+             yield f"data: {json.dumps({'type': 'flow_' + tracker.flow.status, 'data': tracker.flow.to_dict()})}\n\n"
+             return
+
+        while True:
+            try:
+                message = await asyncio.wait_for(queue.get(), timeout=30)
+                yield f"data: {json.dumps(message)}\n\n"
+                
+                if message.get("type") in ["flow_completed", "flow_error"]:
+                    break
+            except asyncio.TimeoutError:
+                yield f"data: {json.dumps({'type': 'heartbeat'})}\n\n"
+    finally:
+        tracker.unsubscribe(queue)
+
+
+@router.get("/flow/{issue_id}")
+async def stream_agent_flow(issue_id: UUID):
+    return StreamingResponse(
+        event_generator(issue_id),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        }
+    )
+
+
+@router.get("/flow/active")
+async def list_active_flows():
+    return {
+        "active_flows": [
+            {
+                "issue_id": str(issue_id),
+                "status": tracker.flow.status,
+                "steps_count": len(tracker.flow.steps),
+                "started_at": tracker.flow.started_at,
+            }
+            for issue_id, tracker in _active_flows.items()
+        ]
+    }
+
+
+@router.get("/events/{issue_id}")
+async def get_issue_events(
+    issue_id: UUID,
+    limit: int = Query(50, ge=1, le=200),
+    db: AsyncSession = Depends(get_db),
+):
+    query = (
+        select(IssueEvent)
+        .where(IssueEvent.issue_id == issue_id)
+        .order_by(IssueEvent.created_at.asc())
+        .limit(limit)
+    )
+    result = await db.execute(query)
+    events = result.scalars().all()
+    
+    return {
+        "issue_id": str(issue_id),
+        "events": [
+            {
+                "id": str(e.id),
+                "event_type": e.event_type,
+                "agent_name": e.agent_name,
+                "event_data": json.loads(e.event_data) if e.event_data else None,
+                "created_at": e.created_at.isoformat(),
+            }
+            for e in events
+        ]
+    }
+
+
+@router.get("/timeline/{issue_id}")
+async def get_issue_timeline(
+    issue_id: UUID,
+    db: AsyncSession = Depends(get_db),
+):
+    issue = await db.get(Issue, issue_id)
+    if not issue:
+        return {"error": "Issue not found"}
+    
+    query = (
+        select(IssueEvent)
+        .where(IssueEvent.issue_id == issue_id)
+        .order_by(IssueEvent.created_at.asc())
+    )
+    result = await db.execute(query)
+    events = result.scalars().all()
+    
+    timeline = []
+    
+    timeline.append({
+        "timestamp": issue.created_at.isoformat(),
+        "event": "issue_created",
+        "agent": "System",
+        "details": {
+            "latitude": issue.latitude,
+            "longitude": issue.longitude,
+            "description": issue.description,
+        }
+    })
+    
+    for event in events:
+        event_data = json.loads(event.event_data) if event.event_data else {}
+        timeline.append({
+            "timestamp": event.created_at.isoformat(),
+            "event": event.event_type,
+            "agent": event.agent_name or "Unknown",
+            "details": event_data,
+        })
+    
+    return {
+        "issue_id": str(issue_id),
+        "current_state": issue.state,
+        "priority": issue.priority,
+        "is_duplicate": issue.is_duplicate,
+        "timeline": timeline,
+    }

Backend/api/routes/health.py

@@ -0,0 +1,21 @@
+from fastapi import APIRouter
+from sqlalchemy import text
+
+from Backend.database.connection import async_session_factory
+
+router = APIRouter()
+
+
+@router.get("/health")
+async def health_check():
+    return {"status": "healthy", "service": "city-issue-agent"}
+
+
+@router.get("/health/db")
+async def db_health_check():
+    try:
+        async with async_session_factory() as session:
+            await session.execute(text("SELECT 1"))
+        return {"status": "healthy", "database": "connected"}
+    except Exception as e:
+        return {"status": "unhealthy", "database": "disconnected", "error": str(e)}