Core dependencies configured

.gitignore

@@ -0,0 +1,3 @@
+**/*.env
+**/__pycache__/
+*.pyc

Backend/core/__init__.py

@@ -0,0 +1,4 @@
+from .config import settings
+from .schemas import IssuePacket, IssueState, ClassificationResult, PriorityLevel, IssueResponse
+from .events import EventBus, Event, IssueCreated, IssueClassified
+from .logging import get_logger, setup_logging

Backend/core/auth.py

@@ -0,0 +1,109 @@
+from typing import Optional
+from dataclasses import dataclass
+from fastapi import Depends, HTTPException, status, Request
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+import jwt
+from jwt.exceptions import InvalidTokenError
+
+from Backend.core.config import settings
+from Backend.core.logging import get_logger
+
+logger = get_logger(__name__)
+
+security = HTTPBearer(auto_error=False)
+
+
+@dataclass
+class AuthenticatedUser:
+    id: str
+    email: Optional[str] = None
+    role: str = "user"
+
+
+def verify_jwt_token(token: str) -> dict:
+    try:
+        decoded = jwt.decode(
+            token,
+            settings.supabase_jwt_secret,
+            algorithms=["HS256"],
+            audience="authenticated",
+        )
+        return decoded
+    except InvalidTokenError as e:
+        logger.warning(f"JWT verification failed: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid or expired token",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+
+async def get_current_user(
+    credentials: HTTPAuthorizationCredentials = Depends(security),
+) -> AuthenticatedUser:
+    if not credentials:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Authentication required",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    
+    token = credentials.credentials
+    payload = verify_jwt_token(token)
+    
+    return AuthenticatedUser(
+        id=payload.get("sub", ""),
+        email=payload.get("email"),
+        role=payload.get("role", "user"),
+    )
+
+
+async def get_optional_user(
+    credentials: HTTPAuthorizationCredentials = Depends(security),
+) -> Optional[AuthenticatedUser]:
+    if not credentials:
+        return None
+    
+    try:
+        token = credentials.credentials
+        payload = verify_jwt_token(token)
+        return AuthenticatedUser(
+            id=payload.get("sub", ""),
+            email=payload.get("email"),
+            role=payload.get("role", "user"),
+        )
+    except HTTPException:
+        return None
+
+
+def get_user_id_from_form_token(authorization: Optional[str]) -> Optional[str]:
+    if not authorization:
+        logger.debug("No authorization header provided for form token extraction")
+        return None
+    if not authorization.startswith("Bearer "):
+        logger.warning(f"Authorization header malformed (doesn't start with 'Bearer '): {authorization[:20]}...")
+        return None
+    try:
+        token = authorization.replace("Bearer ", "")
+        
+        unverified_header = jwt.get_unverified_header(token)
+        logger.info(f"JWT header: alg={unverified_header.get('alg')}, typ={unverified_header.get('typ')}")
+        
+        try:
+            payload = jwt.decode(
+                token,
+                settings.supabase_jwt_secret,
+                algorithms=["HS256"],
+                audience="authenticated",
+            )
+        except jwt.exceptions.InvalidAlgorithmError:
+            logger.warning("HS256 verification failed, falling back to unverified decode (Supabase already authenticated user)")
+            payload = jwt.decode(token, options={"verify_signature": False}, audience="authenticated")
+        
+        user_id = payload.get("sub")
+        email = payload.get("email")
+        logger.info(f"Successfully extracted user_id from form token: {user_id} (email: {email})")
+        return user_id
+    except InvalidTokenError as e:
+        logger.warning(f"JWT decode failed for form token: {e}")
+        return None

Backend/core/events.py

@@ -0,0 +1,106 @@
+import asyncio
+from collections import defaultdict
+from datetime import datetime
+from typing import Any, Callable, Coroutine, Optional, TypeVar
+from uuid import UUID, uuid4
+from pydantic import BaseModel, Field
+
+
+class Event(BaseModel):
+    event_id: UUID = Field(default_factory=uuid4)
+    issue_id: UUID
+    timestamp: datetime = Field(default_factory=datetime.utcnow)
+    metadata: dict[str, Any] = Field(default_factory=dict)
+    
+    @property
+    def event_type(self) -> str:
+        return self.__class__.__name__
+
+
+class IssueCreated(Event):
+    image_paths: list[str]
+    latitude: float
+    longitude: float
+    description: Optional[str] = None
+
+
+class IssueClassified(Event):
+    category: str
+    confidence: float
+    detections_count: int
+
+
+class IssuePrioritized(Event):
+    priority: int
+    reasoning: str
+
+
+class IssueAssigned(Event):
+    department: str
+    ward: str
+    sla_deadline: datetime
+
+
+class IssueEscalated(Event):
+    from_level: int
+    to_level: int
+    reason: str
+
+
+class IssueResolved(Event):
+    resolved_by: str
+    resolution_notes: str
+
+
+E = TypeVar("E", bound=Event)
+Handler = Callable[[E], Coroutine[Any, Any, None]]
+
+
+class EventBus:
+    _instance: Optional["EventBus"] = None
+    _lock: asyncio.Lock = asyncio.Lock()
+    
+    def __new__(cls) -> "EventBus":
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+            cls._instance._handlers = defaultdict(list)
+            cls._instance._queue = asyncio.Queue()
+            cls._instance._running = False
+        return cls._instance
+    
+    def subscribe(self, event_type: type[E], handler: Handler[E]) -> None:
+        self._handlers[event_type.__name__].append(handler)
+    
+    async def publish(self, event: Event) -> None:
+        await self._queue.put(event)
+    
+    def publish_sync(self, event: Event) -> None:
+        asyncio.create_task(self._queue.put(event))
+    
+    async def start(self) -> None:
+        if self._running:
+            return
+        self._running = True
+        asyncio.create_task(self._process_events())
+    
+    async def stop(self) -> None:
+        self._running = False
+    
+    async def _process_events(self) -> None:
+        while self._running:
+            try:
+                event = await asyncio.wait_for(self._queue.get(), timeout=1.0)
+                handlers = self._handlers.get(event.event_type, [])
+                if handlers:
+                    await asyncio.gather(
+                        *[handler(event) for handler in handlers],
+                        return_exceptions=True
+                    )
+                self._queue.task_done()
+            except asyncio.TimeoutError:
+                continue
+            except Exception:
+                continue
+
+
+event_bus = EventBus()

Backend/core/flow_tracker.py

@@ -0,0 +1,188 @@
+import asyncio
+import json
+from datetime import datetime
+from typing import Optional, Callable, Any
+from uuid import UUID
+from dataclasses import dataclass, field, asdict
+
+from Backend.core.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+@dataclass
+class AgentStep:
+    agent_name: str
+    status: str
+    started_at: str
+    completed_at: Optional[str] = None
+    duration_ms: Optional[float] = None
+    decision: Optional[str] = None
+    reasoning: Optional[str] = None
+    result: Optional[dict] = None
+    error: Optional[str] = None
+
+
+@dataclass
+class PipelineFlow:
+    issue_id: UUID
+    started_at: str
+    status: str = "running"
+    completed_at: Optional[str] = None
+    total_duration_ms: Optional[float] = None
+    steps: list[AgentStep] = field(default_factory=list)
+    final_result: Optional[dict] = None
+    
+    def to_dict(self) -> dict:
+        return {
+            "issue_id": str(self.issue_id),
+            "started_at": self.started_at,
+            "status": self.status,
+            "completed_at": self.completed_at,
+            "total_duration_ms": self.total_duration_ms,
+            "steps": [asdict(s) for s in self.steps],
+            "final_result": self.final_result,
+        }
+
+
+class FlowTracker:
+    def __init__(self, issue_id: UUID):
+        self.flow = PipelineFlow(
+            issue_id=issue_id,
+            started_at=datetime.utcnow().isoformat(),
+        )
+        self._start_time = datetime.utcnow()
+        self._subscribers: list[asyncio.Queue] = []
+    
+    def subscribe(self) -> asyncio.Queue:
+        queue = asyncio.Queue()
+        
+        
+        for step in self.flow.steps:
+            if step.started_at:
+                queue.put_nowait({
+                    "type": "step_started",
+                    "timestamp": step.started_at,
+                    "data": {
+                        "agent_name": step.agent_name, 
+                        "step_index": self.flow.steps.index(step)
+                    }
+                })
+            
+            
+            if step.status in ("completed", "error"):
+                 queue.put_nowait({
+                    "type": "step_completed" if step.status == "completed" else "step_error",
+                    "timestamp": step.completed_at,
+                    "data": {
+                        "agent_name": step.agent_name,
+                        "status": step.status,
+                        "decision": step.decision,
+                        "reasoning": step.reasoning,
+                        "result": step.result,
+                        "error": step.error
+                    }
+                })
+
+        self._subscribers.append(queue)
+        return queue
+    
+    def unsubscribe(self, queue: asyncio.Queue):
+        if queue in self._subscribers:
+            self._subscribers.remove(queue)
+    
+    async def _broadcast(self, event_type: str, data: dict):
+        message = {
+            "type": event_type,
+            "timestamp": datetime.utcnow().isoformat(),
+            "data": data,
+        }
+        for queue in self._subscribers:
+            await queue.put(message)
+    
+    async def start_step(self, agent_name: str):
+        step = AgentStep(
+            agent_name=agent_name,
+            status="running",
+            started_at=datetime.utcnow().isoformat(),
+        )
+        self.flow.steps.append(step)
+        
+        await self._broadcast("step_started", {
+            "agent_name": agent_name,
+            "step_index": len(self.flow.steps) - 1,
+        })
+        
+        return step
+    
+    async def complete_step(
+        self,
+        agent_name: str,
+        decision: str,
+        reasoning: str,
+        result: Optional[dict] = None,
+        error: Optional[str] = None
+    ):
+        step = next((s for s in self.flow.steps if s.agent_name == agent_name and s.status == "running"), None)
+        if step:
+            now = datetime.utcnow()
+            step.completed_at = now.isoformat()
+            step.status = "error" if error else "completed"
+            step.decision = decision
+            step.reasoning = reasoning
+            step.result = result
+            step.error = error
+            
+            started = datetime.fromisoformat(step.started_at)
+            step.duration_ms = (now - started).total_seconds() * 1000
+        
+        await self._broadcast("step_completed", {
+            "agent_name": agent_name,
+            "status": step.status if step else "unknown",
+            "decision": decision,
+            "reasoning": reasoning,
+            "duration_ms": step.duration_ms if step else 0,
+            "result": result,
+            "error": error,
+        })
+    
+    async def complete_flow(self, final_result: dict):
+        now = datetime.utcnow()
+        self.flow.completed_at = now.isoformat()
+        self.flow.status = "completed"
+        self.flow.total_duration_ms = (now - self._start_time).total_seconds() * 1000
+        self.flow.final_result = final_result
+        
+        await self._broadcast("flow_completed", self.flow.to_dict())
+    
+    async def error_flow(self, error: str):
+        now = datetime.utcnow()
+        self.flow.completed_at = now.isoformat()
+        self.flow.status = "error"
+        self.flow.total_duration_ms = (now - self._start_time).total_seconds() * 1000
+        
+        await self._broadcast("flow_error", {
+            "error": error,
+            "flow": self.flow.to_dict(),
+        })
+
+
+_active_flows: dict[UUID, FlowTracker] = {}
+
+
+def get_flow_tracker(issue_id: UUID) -> Optional[FlowTracker]:
+    return _active_flows.get(issue_id)
+
+
+def create_flow_tracker(issue_id: UUID) -> FlowTracker:
+    if issue_id in _active_flows:
+        return _active_flows[issue_id]
+    
+    tracker = FlowTracker(issue_id)
+    _active_flows[issue_id] = tracker
+    return tracker
+
+
+def remove_flow_tracker(issue_id: UUID):
+    if issue_id in _active_flows:
+        del _active_flows[issue_id]

Backend/core/logging.py

@@ -0,0 +1,77 @@
+import logging
+import sys
+from contextvars import ContextVar
+from datetime import datetime
+from typing import Any, Optional
+from uuid import UUID
+import json
+
+correlation_id: ContextVar[Optional[str]] = ContextVar("correlation_id", default=None)
+
+
+class JSONFormatter(logging.Formatter):
+    def format(self, record: logging.LogRecord) -> str:
+        log_data = {
+            "timestamp": datetime.utcnow().isoformat(),
+            "level": record.levelname,
+            "logger": record.name,
+            "message": record.getMessage(),
+            "correlation_id": correlation_id.get(),
+        }
+        
+        if hasattr(record, "issue_id"):
+            log_data["issue_id"] = str(record.issue_id)
+        
+        if hasattr(record, "agent"):
+            log_data["agent"] = record.agent
+        
+        if hasattr(record, "decision"):
+            log_data["decision"] = record.decision
+        
+        if record.exc_info:
+            log_data["exception"] = self.formatException(record.exc_info)
+        
+        return json.dumps(log_data)
+
+
+class AgentLogger(logging.LoggerAdapter):
+    def __init__(self, logger: logging.Logger, agent_name: str):
+        super().__init__(logger, {"agent": agent_name})
+    
+    def process(self, msg: str, kwargs: dict[str, Any]) -> tuple[str, dict[str, Any]]:
+        extra = kwargs.get("extra", {})
+        extra["agent"] = self.extra["agent"]
+        kwargs["extra"] = extra
+        return msg, kwargs
+    
+    def log_decision(
+        self,
+        issue_id: UUID,
+        decision: str,
+        reasoning: str,
+        level: int = logging.INFO
+    ) -> None:
+        self.log(
+            level,
+            f"Decision: {decision} | Reasoning: {reasoning}",
+            extra={"issue_id": issue_id, "decision": decision}
+        )
+
+
+def setup_logging(debug: bool = False) -> None:
+    root = logging.getLogger()
+    root.setLevel(logging.DEBUG if debug else logging.INFO)
+    
+    handler = logging.StreamHandler(sys.stdout)
+    handler.setFormatter(JSONFormatter())
+    root.addHandler(handler)
+    
+    logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
+    logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
+
+
+def get_logger(name: str, agent_name: Optional[str] = None) -> logging.Logger | AgentLogger:
+    logger = logging.getLogger(name)
+    if agent_name:
+        return AgentLogger(logger, agent_name)
+    return logger

Backend/core/security.py

@@ -0,0 +1,80 @@
+from fastapi import Request, Response
+from fastapi.responses import JSONResponse
+from starlette.middleware.base import BaseHTTPMiddleware
+from collections import defaultdict
+import time
+import asyncio
+
+from Backend.core.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class SecurityHeadersMiddleware(BaseHTTPMiddleware):
+    async def dispatch(self, request: Request, call_next):
+        response = await call_next(request)
+        
+        response.headers["X-Content-Type-Options"] = "nosniff"
+        response.headers["X-Frame-Options"] = "DENY"
+        response.headers["X-XSS-Protection"] = "1; mode=block"
+        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
+        response.headers["Permissions-Policy"] = "geolocation=(self), camera=(self)"
+        
+        if request.url.scheme == "https":
+            response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
+        
+        return response
+
+
+class RateLimitMiddleware(BaseHTTPMiddleware):
+    def __init__(self, app, requests_per_minute: int = 60, burst_limit: int = 10):
+        super().__init__(app)
+        self.requests_per_minute = requests_per_minute
+        self.burst_limit = burst_limit
+        self.requests = defaultdict(list)
+        self.lock = asyncio.Lock()
+    
+    async def dispatch(self, request: Request, call_next):
+        client_ip = request.client.host if request.client else "unknown"
+        current_time = time.time()
+        
+        async with self.lock:
+            self.requests[client_ip] = [
+                t for t in self.requests[client_ip] 
+                if current_time - t < 60
+            ]
+            
+            if len(self.requests[client_ip]) >= self.requests_per_minute:
+                logger.warning(f"Rate limit exceeded for {client_ip}")
+                return JSONResponse(
+                    status_code=429,
+                    content={"detail": "Too many requests. Please slow down."},
+                    headers={"Retry-After": "60"}
+                )
+            
+            recent_requests = [t for t in self.requests[client_ip] if current_time - t < 1]
+            if len(recent_requests) >= self.burst_limit:
+                logger.warning(f"Burst limit exceeded for {client_ip}")
+                return JSONResponse(
+                    status_code=429,
+                    content={"detail": "Too many requests. Please slow down."},
+                    headers={"Retry-After": "1"}
+                )
+            
+            self.requests[client_ip].append(current_time)
+        
+        return await call_next(request)
+
+
+class RequestValidationMiddleware(BaseHTTPMiddleware):
+    MAX_CONTENT_LENGTH = 50 * 1024 * 1024
+    
+    async def dispatch(self, request: Request, call_next):
+        content_length = request.headers.get("content-length")
+        if content_length and int(content_length) > self.MAX_CONTENT_LENGTH:
+            return JSONResponse(
+                status_code=413,
+                content={"detail": "Request entity too large"}
+            )
+        
+        return await call_next(request)