Backend v1 finished

Backend/agents/__init__.py

@@ -0,0 +1,6 @@
+from .vision import VisionAgent
+from .geoDeduplicate import GeoDeduplicateAgent
+from .priority import PriorityAgent
+from .routing import RoutingAgent
+from .escalation import EscalationAgent
+from .notification import NotificationAgent

Backend/agents/escalation/__init__.py

@@ -0,0 +1 @@
+from .agent import EscalationAgent, IssueEscalated

Backend/agents/escalation/agent.py

@@ -0,0 +1,178 @@
+import json
+from datetime import datetime
+from typing import Optional
+from uuid import UUID
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+import google.generativeai as genai
+
+from Backend.core.events import event_bus, Event
+from Backend.core.logging import get_logger
+from Backend.core.config import settings
+from Backend.database.models import Issue, IssueEvent, Escalation, Department, Member
+from Backend.orchestration.base import BaseAgent
+
+logger = get_logger(__name__, agent_name="EscalationAgent")
+
+if settings.gemini_api_key:
+    genai.configure(api_key=settings.gemini_api_key)
+
+
+class IssueEscalated(Event):
+    from_level: int
+    to_level: int
+    reason: str
+    hours_overdue: float
+
+
+class EscalationAgent(BaseAgent):
+    def __init__(self, db: AsyncSession):
+        super().__init__("EscalationAgent")
+        self.db = db
+        if settings.gemini_api_key:
+            self.model = genai.GenerativeModel('gemma-3-27b-it')
+        else:
+            self.model = None
+    
+    async def should_escalate(self, issue: Issue) -> tuple[bool, int, str]:
+        if not issue.sla_deadline:
+            return False, 0, "No SLA deadline set"
+        
+        if not self.model:
+            return False, 0, "Gemini API not configured"
+        
+        now = datetime.utcnow()
+        hours_since_creation = (now - issue.created_at).total_seconds() / 3600
+        hours_until_deadline = (issue.sla_deadline - now).total_seconds() / 3600
+        
+        prompt = f"""Analyze civic issue escalation:
+
+Issue State: {issue.state}
+Priority: {issue.priority} (1=Critical, 2=High, 3=Medium, 4=Low)
+Current Escalation Level: {issue.escalation_level}
+Hours Since Creation: {hours_since_creation:.1f}
+Hours Until Deadline: {hours_until_deadline:.1f}
+Category: {issue.description[:100] if issue.description else 'N/A'}
+
+Determine if escalation is needed. Consider:
+- SLA breach (negative deadline hours)
+- Priority urgency
+- Time criticality
+
+Return ONLY valid JSON:
+{{"should_escalate": true/false, "new_level": 0-3, "reason": "max 80 chars"}}"""
+        
+        try:
+            response = self.model.generate_content(prompt)
+            result = json.loads(response.text.replace("```json", "").replace("```", "").strip())
+            return result.get("should_escalate", False), result.get("new_level", issue.escalation_level), result.get("reason", "Analysis completed")
+        except Exception as e:
+            logger.error(f"Gemini escalation analysis failed: {e}")
+            return False, issue.escalation_level, "Analysis error"
+    
+    async def get_escalation_targets(self, issue: Issue) -> list[str]:
+        targets = []
+        
+        if issue.department_id:
+            query = select(Department).where(Department.id == issue.department_id)
+            result = await self.db.execute(query)
+            dept = result.scalar_one_or_none()
+            if dept and dept.escalation_email:
+                targets.append(dept.escalation_email)
+        
+        if issue.assigned_member_id:
+            query = select(Member).where(Member.id == issue.assigned_member_id)
+            result = await self.db.execute(query)
+            member = result.scalar_one_or_none()
+            if member:
+                targets.append(member.email)
+        
+        return targets
+    
+    async def process_issue(self, issue_id: UUID) -> dict:
+        issue = await self.db.get(Issue, issue_id)
+        if not issue:
+            return {"error": "Issue not found"}
+        
+        if issue.state in ["resolved", "verified", "closed"]:
+            return {"skipped": True, "reason": "Issue already resolved"}
+        
+        should_esc, new_level, reason = await self.should_escalate(issue)
+        
+        if not should_esc:
+            return {"escalated": False, "reason": reason}
+        
+        old_level = issue.escalation_level
+        issue.escalation_level = new_level
+        issue.escalated_at = datetime.utcnow()
+        issue.state = "escalated"
+        
+        targets = await self.get_escalation_targets(issue)
+        
+        escalation = Escalation(
+            issue_id=issue_id,
+            from_level=old_level,
+            to_level=new_level,
+            reason=reason,
+            escalated_by="EscalationAgent",
+            notified_emails=",".join(targets) if targets else None,
+        )
+        self.db.add(escalation)
+        
+        self.log_decision(
+            issue_id=issue_id,
+            decision=f"Escalated from level {old_level} to {new_level}",
+            reasoning=reason
+        )
+        
+        event_record = IssueEvent(
+            issue_id=issue_id,
+            event_type="escalated",
+            agent_name=self.name,
+            event_data=json.dumps({
+                "from_level": old_level,
+                "to_level": new_level,
+                "reason": reason,
+                "notified": targets,
+            })
+        )
+        self.db.add(event_record)
+        await self.db.flush()
+        
+        esc_event = IssueEscalated(
+            issue_id=issue_id,
+            from_level=old_level,
+            to_level=new_level,
+            reason=reason,
+            hours_overdue=0,
+        )
+        await event_bus.publish(esc_event)
+        
+        return {
+            "escalated": True,
+            "from_level": old_level,
+            "to_level": new_level,
+            "reason": reason,
+            "notified": targets,
+        }
+    
+    async def check_all_pending(self) -> list[dict]:
+        query = (
+            select(Issue)
+            .where(Issue.state.in_(["assigned", "in_progress", "escalated"]))
+            .where(Issue.is_duplicate == False)
+            .where(Issue.sla_deadline.isnot(None))
+        )
+        result = await self.db.execute(query)
+        issues = result.scalars().all()
+        
+        results = []
+        for issue in issues:
+            result = await self.process_issue(issue.id)
+            if result.get("escalated"):
+                results.append(result)
+        
+        return results
+    
+    async def handle(self, event) -> None:
+        await self.process_issue(event.issue_id)

Backend/agents/geoDeduplicate/__init__.py

@@ -0,0 +1 @@
+from .agent import GeoDeduplicateAgent, IssueDeduplicated

Backend/agents/geoDeduplicate/agent.py

@@ -0,0 +1,225 @@
+import json
+from typing import Optional
+from uuid import UUID
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+import google.generativeai as genai
+
+from Backend.core.config import settings
+from Backend.core.events import event_bus, IssueClassified, Event
+from Backend.core.logging import get_logger
+from Backend.database.models import Issue, IssueEvent, Classification
+from Backend.utils.geo import haversine_distance, get_bounding_box
+from Backend.orchestration.base import BaseAgent
+
+logger = get_logger(__name__, agent_name="GeoDeduplicateAgent")
+
+if settings.gemini_api_key:
+    genai.configure(api_key=settings.gemini_api_key)
+
+
+class IssueDeduplicated(Event):
+    is_duplicate: bool
+    parent_issue_id: Optional[UUID] = None
+    cluster_id: Optional[str] = None
+    nearby_count: int = 0
+
+
+class GeoDeduplicateAgent(BaseAgent):
+    def __init__(self, db: AsyncSession):
+        super().__init__("GeoDeduplicateAgent")
+        self.db = db
+        self.radius_meters = settings.duplicate_radius_meters
+        if settings.gemini_api_key:
+            self.model = genai.GenerativeModel('gemma-3-27b-it')
+        else:
+            self.model = None
+    
+    async def semantic_similarity(self, desc1: str, desc2: str, cat1: str, cat2: str) -> float:
+        if not self.model:
+            return 0.5
+        
+        prompt = f"""Rate semantic similarity (0.0-1.0) between civic issue reports:
+
+Issue A:
+Category: {cat1}
+Description: {desc1[:200] if desc1 else 'N/A'}
+
+Issue B:
+Category: {cat2}
+Description: {desc2[:200] if desc2 else 'N/A'}
+
+Consider:
+- Same problem type?
+- Same physical location context?
+- Same infrastructure element?
+
+Return ONLY a decimal number between 0.0 and 1.0."""
+        
+        try:
+            response = self.model.generate_content(prompt)
+            score = float(response.text.strip())
+            return max(0.0, min(1.0, score))
+        except Exception as e:
+            logger.error(f"Gemini similarity failed: {e}")
+            return 0.5
+    
+    async def find_nearby_issues(
+        self,
+        latitude: float,
+        longitude: float,
+        exclude_id: UUID,
+        category: Optional[str] = None
+    ) -> list[tuple[Issue, float]]:
+        min_lat, max_lat, min_lon, max_lon = get_bounding_box(
+            latitude, longitude, self.radius_meters
+        )
+        
+        query = (
+            select(Issue)
+            .options(selectinload(Issue.classification))
+            .where(Issue.latitude >= min_lat)
+            .where(Issue.latitude <= max_lat)
+            .where(Issue.longitude >= min_lon)
+            .where(Issue.longitude <= max_lon)
+            .where(Issue.id != exclude_id)
+            .where(Issue.state.in_(["reported", "validated", "assigned", "in_progress"]))
+            .where(Issue.is_duplicate == False)
+        )
+        
+        result = await self.db.execute(query)
+        candidates = result.scalars().all()
+        
+        nearby = []
+        for issue in candidates:
+            distance = haversine_distance(
+                latitude, longitude,
+                issue.latitude, issue.longitude
+            )
+            if distance <= self.radius_meters:
+                if category and issue.classification:
+                    if issue.classification.primary_category == category:
+                        nearby.append((issue, distance))
+                else:
+                    nearby.append((issue, distance))
+        
+        return sorted(nearby, key=lambda x: x[1])
+    
+    async def check_duplicate(
+        self,
+        issue_id: UUID,
+        latitude: float,
+        longitude: float,
+        category: Optional[str] = None,
+        description: Optional[str] = None
+    ) -> tuple[bool, Optional[UUID], list[tuple[Issue, float]]]:
+        nearby = await self.find_nearby_issues(
+            latitude, longitude, issue_id, category
+        )
+        
+        if not nearby:
+            return False, None, []
+        
+        best_match = None
+        highest_score = 0.0
+        
+        for issue, distance in nearby:
+            if issue.classification and category:
+                cat1 = category
+                cat2 = issue.classification.primary_category
+                desc1 = description or ""
+                desc2 = issue.description or ""
+                
+                similarity = await self.semantic_similarity(desc1, desc2, cat1, cat2)
+                
+                if similarity > highest_score:
+                    highest_score = similarity
+                    best_match = issue
+        
+        if highest_score > 0.75 and best_match:
+            return True, best_match.id, nearby
+        
+        return False, None, nearby
+    
+    async def process_issue(self, issue_id: UUID) -> dict:
+        query = (
+            select(Issue)
+            .options(selectinload(Issue.classification))
+            .where(Issue.id == issue_id)
+        )
+        result = await self.db.execute(query)
+        issue = result.scalar_one_or_none()
+        if not issue:
+            return {"error": "Issue not found"}
+        
+        category = None
+        if issue.classification:
+            category = issue.classification.primary_category
+        
+        is_duplicate, parent_id, nearby = await self.check_duplicate(
+            issue.id,
+            issue.latitude,
+            issue.longitude,
+            category,
+            issue.description
+        )
+        
+        if is_duplicate and parent_id:
+            issue.is_duplicate = True
+            issue.parent_issue_id = parent_id
+            issue.geo_status = "duplicate"
+            issue.geo_cluster_id = str(parent_id)
+            
+            parent = await self.db.get(Issue, parent_id)
+            if parent and issue.priority and parent.priority:
+                if issue.priority < parent.priority:
+                    parent.priority = issue.priority
+            
+            self.log_decision(
+                issue_id=issue_id,
+                decision="Marked as duplicate",
+                reasoning=f"Found {len(nearby)} nearby issues within {self.radius_meters}m, linked to parent {parent_id}"
+            )
+        else:
+            issue.is_duplicate = False
+            issue.geo_status = "unique"
+            
+            self.log_decision(
+                issue_id=issue_id,
+                decision="Marked as unique",
+                reasoning=f"No similar issues found within {self.radius_meters}m radius"
+            )
+        
+        event_record = IssueEvent(
+            issue_id=issue_id,
+            event_type="geo_deduplicated",
+            agent_name=self.name,
+            event_data=json.dumps({
+                "is_duplicate": is_duplicate,
+                "parent_issue_id": str(parent_id) if parent_id else None,
+                "nearby_count": len(nearby),
+                "radius_meters": self.radius_meters,
+            })
+        )
+        self.db.add(event_record)
+        await self.db.flush()
+        
+        dedup_event = IssueDeduplicated(
+            issue_id=issue_id,
+            is_duplicate=is_duplicate,
+            parent_issue_id=parent_id,
+            cluster_id=str(parent_id) if parent_id else None,
+            nearby_count=len(nearby),
+        )
+        await event_bus.publish(dedup_event)
+        
+        return {
+            "is_duplicate": is_duplicate,
+            "parent_issue_id": str(parent_id) if parent_id else None,
+            "nearby_count": len(nearby),
+            "geo_status": issue.geo_status,
+        }
+    
+    async def handle(self, event: IssueClassified) -> None:
+        await self.process_issue(event.issue_id)

Backend/agents/notification/__init__.py

@@ -0,0 +1 @@
+from .agent import NotificationAgent, NotificationSent

Backend/agents/notification/agent.py

@@ -0,0 +1,333 @@
+import json
+from datetime import datetime
+from typing import Optional
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from Backend.core.events import Event, event_bus
+from Backend.core.logging import get_logger
+from Backend.core.config import settings
+from Backend.database.models import Classification, Issue, IssueEvent, Member
+from Backend.orchestration.base import BaseAgent
+from Backend.services.email import email_service
+
+logger = get_logger(__name__, agent_name="NotificationAgent")
+
+
+class NotificationSent(Event):
+    notification_type: str
+    recipients: list[str]
+    message: str
+
+
+class NotificationAgent(BaseAgent):
+    def __init__(self, db: AsyncSession):
+        super().__init__("NotificationAgent")
+        self.db = db
+        self.pending_notifications: list[dict] = []
+
+    async def get_issue_with_classification(self, issue_id: UUID) -> Optional[Issue]:
+        query = (
+            select(Issue)
+            .options(selectinload(Issue.classification))
+            .where(Issue.id == issue_id)
+        )
+        result = await self.db.execute(query)
+        return result.scalar_one_or_none()
+
+    def format_issue_summary(self, issue: Issue) -> str:
+        category = (
+            issue.classification.primary_category if issue.classification else "Unknown"
+        )
+        priority_map = {1: "CRITICAL", 2: "HIGH", 3: "MEDIUM", 4: "LOW"}
+        priority_str = priority_map.get(issue.priority, "UNKNOWN")
+
+        return (
+            f"Issue #{str(issue.id)[:8]}\n"
+            f"Category: {category}\n"
+            f"Priority: {priority_str}\n"
+            f"Location: ({issue.latitude:.4f}, {issue.longitude:.4f})\n"
+            f"Description: {issue.description or 'No description'}\n"
+            f"State: {issue.state}"
+        )
+
+    async def queue_notification(
+        self,
+        notification_type: str,
+        recipients: list[str],
+        subject: str,
+        message: str,
+        issue_id: Optional[UUID] = None,
+    ):
+        notification = {
+            "type": notification_type,
+            "recipients": recipients,
+            "subject": subject,
+            "message": message,
+            "issue_id": str(issue_id) if issue_id else None,
+            "queued_at": datetime.utcnow().isoformat(),
+        }
+        self.pending_notifications.append(notification)
+
+        logger.info(f"Notification queued: {notification_type} to {recipients}")
+
+        if issue_id:
+            event_record = IssueEvent(
+                issue_id=issue_id,
+                event_type="notification_queued",
+                agent_name=self.name,
+                event_data=json.dumps(notification),
+            )
+            self.db.add(event_record)
+            await self.db.flush()
+
+        return notification
+
+    async def notify_assignment(self, issue_id: UUID):
+        issue = await self.get_issue_with_classification(issue_id)
+        if not issue:
+            return
+
+        recipients = []
+        worker_name = "Worker"
+
+        if issue.assigned_member_id:
+            query = select(Member).where(Member.id == issue.assigned_member_id)
+            result = await self.db.execute(query)
+            member = result.scalar_one_or_none()
+            if member:
+                recipients.append(member.email)
+                worker_name = member.name
+                
+                category = (
+                    issue.classification.primary_category 
+                    if issue.classification 
+                    else "Unknown"
+                )
+                priority_map = {1: "CRITICAL", 2: "HIGH", 3: "MEDIUM", 4: "LOW"}
+                priority_str = priority_map.get(issue.priority, "UNKNOWN")
+                location = f"({issue.latitude:.4f}, {issue.longitude:.4f})"
+                
+                try:
+                    await email_service.send_assignment_email(
+                        worker_email=member.email,
+                        worker_name=worker_name,
+                        issue_id=str(issue.id)[:8],
+                        category=category,
+                        priority=priority_str,
+                        location=location,
+                        description=issue.description or "No description"
+                    )
+                    logger.info(f"Assignment email sent to {member.email}")
+                except Exception as e:
+                    logger.error(f"Failed to send assignment email: {e}")
+
+        if recipients:
+            summary = self.format_issue_summary(issue)
+            await self.queue_notification(
+                notification_type="assignment",
+                recipients=recipients,
+                subject=f"New Issue Assigned: #{str(issue.id)[:8]}",
+                message=f"You have been assigned a new issue:\n\n{summary}",
+                issue_id=issue_id,
+            )
+
+    async def notify_escalation(self, issue_id: UUID, reason: str, targets: list[str]):
+        issue = await self.get_issue_with_classification(issue_id)
+        if not issue:
+            return
+
+        category = (
+            issue.classification.primary_category 
+            if issue.classification 
+            else "Unknown"
+        )
+        priority_map = {1: "CRITICAL", 2: "HIGH", 3: "MEDIUM", 4: "LOW"}
+        priority_str = priority_map.get(issue.priority, "UNKNOWN")
+        
+        for target in targets:
+            try:
+                await email_service.send_escalation_email(
+                    admin_email=target,
+                    issue_id=str(issue.id)[:8],
+                    category=category,
+                    priority=priority_str,
+                    reason=reason,
+                    escalation_level=issue.escalation_level
+                )
+                logger.info(f"Escalation email sent to {target}")
+            except Exception as e:
+                logger.error(f"Failed to send escalation email: {e}")
+
+        summary = self.format_issue_summary(issue)
+        await self.queue_notification(
+            notification_type="escalation",
+            recipients=targets,
+            subject=f"ESCALATION: Issue #{str(issue.id)[:8]} - Level {issue.escalation_level}",
+            message=f"Issue has been escalated:\n\nReason: {reason}\n\n{summary}",
+            issue_id=issue_id,
+        )
+
+    async def notify_resolution(self, issue_id: UUID):
+        issue = await self.get_issue_with_classification(issue_id)
+        if not issue:
+            return
+        
+        category = (
+            issue.classification.primary_category 
+            if issue.classification 
+            else "Unknown"
+        )
+        location = f"({issue.latitude:.4f}, {issue.longitude:.4f})"
+        
+        if issue.user_id:
+            try:
+                await email_service.send_completion_email(
+                    user_email=issue.user_id,
+                    issue_id=str(issue.id)[:8],
+                    category=category,
+                    location=location,
+                    resolution_notes=issue.resolution_notes or "Issue resolved successfully"
+                )
+                logger.info(f"Resolution email sent to user {issue.user_id}")
+            except Exception as e:
+                logger.error(f"Failed to send resolution email: {e}")
+
+        await self.queue_notification(
+            notification_type="resolution",
+            recipients=[settings.admin_email],
+            subject=f"Issue Resolved: #{str(issue.id)[:8]}",
+            message=f"Issue has been marked as resolved.\n\n{self.format_issue_summary(issue)}",
+            issue_id=issue_id,
+        )
+    
+    async def notify_manual_review(self, issue_id: UUID, reason: str):
+        issue = await self.get_issue_with_classification(issue_id)
+        if not issue:
+            return
+        
+        category = (
+            issue.classification.primary_category 
+            if issue.classification 
+            else "Unknown"
+        )
+        location = f"({issue.latitude:.4f}, {issue.longitude:.4f})"
+        image_url = f"{settings.supabase_url}/storage/v1/object/public/{settings.supabase_bucket}/{issue.id}/original.jpg"
+        
+        try:
+            await email_service.send_manual_review_email(
+                issue_id=str(issue.id)[:8],
+                reason=reason,
+                category=category,
+                location=location,
+                image_url=image_url
+            )
+            logger.info(f"Manual review email sent to admin")
+        except Exception as e:
+            logger.error(f"Failed to send manual review email: {e}")
+        
+        await self.queue_notification(
+            notification_type="manual_review",
+            recipients=[settings.admin_email],
+            subject=f"Manual Review Required: #{str(issue.id)[:8]}",
+            message=f"Issue requires manual review.\n\nReason: {reason}\n\n{self.format_issue_summary(issue)}",
+            issue_id=issue_id,
+        )
+    
+    async def notify_user_confirmation(self, issue_id: UUID):
+        issue = await self.get_issue_with_classification(issue_id)
+        if not issue:
+            return
+        
+        category = (
+            issue.classification.primary_category 
+            if issue.classification 
+            else "Unknown"
+        )
+        confirmation_link = f"https://app.citytrack.city/confirm/{issue.id}"
+        
+        if issue.user_id:
+            try:
+                await email_service.send_confirmation_request_email(
+                    user_email=issue.user_id,
+                    issue_id=str(issue.id)[:8],
+                    category=category,
+                    confirmation_link=confirmation_link
+                )
+                logger.info(f"Confirmation request email sent to user {issue.user_id}")
+            except Exception as e:
+                logger.error(f"Failed to send confirmation email: {e}")
+        
+        await self.queue_notification(
+            notification_type="user_confirmation",
+            recipients=[issue.user_id] if issue.user_id else [],
+            subject=f"Please Confirm Resolution: #{str(issue.id)[:8]}",
+            message=f"Please confirm if this issue has been resolved.\n\n{self.format_issue_summary(issue)}",
+            issue_id=issue_id,
+        )
+    
+    async def notify_issue_accepted(self, issue_id: UUID, accepted_by: str = "automatic"):
+        issue = await self.get_issue_with_classification(issue_id)
+        if not issue:
+            return
+        
+        category = (
+            issue.classification.primary_category 
+            if issue.classification 
+            else "Unknown"
+        )
+        priority_map = {1: "CRITICAL", 2: "HIGH", 3: "MEDIUM", 4: "LOW"}
+        priority_str = priority_map.get(issue.priority, "UNKNOWN")
+        location = f"({issue.latitude:.4f}, {issue.longitude:.4f})"
+        tracking_url = f"https://app.citytrack.city/track/{issue.id}"
+        
+        if issue.user_id:
+            try:
+                await email_service.send_issue_accepted_email(
+                    user_email=issue.user_id,
+                    issue_id=str(issue.id)[:8],
+                    category=category,
+                    priority=priority_str,
+                    location=location,
+                    accepted_by=accepted_by,
+                    tracking_url=tracking_url
+                )
+                logger.info(f"Issue accepted email sent to user {issue.user_id} ({accepted_by})")
+            except Exception as e:
+                logger.error(f"Failed to send issue accepted email: {e}")
+        
+        await self.queue_notification(
+            notification_type="issue_accepted",
+            recipients=[issue.user_id] if issue.user_id else [],
+            subject=f"Issue Accepted: #{str(issue.id)[:8]}",
+            message=f"Your issue has been accepted {accepted_by}.\n\n{self.format_issue_summary(issue)}",
+            issue_id=issue_id,
+        )
+
+    async def process_issue(
+        self, issue_id: UUID, event_type: str = "assignment"
+    ) -> dict:
+        if event_type == "assignment":
+            await self.notify_assignment(issue_id)
+        elif event_type == "resolution":
+            await self.notify_resolution(issue_id)
+        elif event_type == "escalation":
+            await self.notify_escalation(issue_id, "SLA breach or priority escalation", [settings.admin_email])
+        elif event_type == "manual_review":
+            await self.notify_manual_review(issue_id, "Requires admin attention")
+        elif event_type == "user_confirmation":
+            await self.notify_user_confirmation(issue_id)
+        elif event_type == "issue_accepted":
+            accepted_by = "automatic"
+            await self.notify_issue_accepted(issue_id, accepted_by)
+        elif event_type == "issue_accepted_manual":
+            await self.notify_issue_accepted(issue_id, "manual")
+
+        return {"queued": len(self.pending_notifications)}
+
+    async def handle(self, event) -> None:
+        event_type = getattr(event, "notification_type", "assignment")
+        await self.process_issue(event.issue_id, event_type)

Backend/agents/priority/__init__.py

@@ -0,0 +1 @@
+from .agent import PriorityAgent, IssuePrioritized

Backend/agents/priority/agent.py

@@ -0,0 +1,144 @@
+import json
+from typing import Optional
+from uuid import UUID
+from sqlalchemy import select, func
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+import google.generativeai as genai
+
+from Backend.core.config import settings
+from Backend.core.events import event_bus, Event
+from Backend.core.logging import get_logger
+from Backend.database.models import Issue, IssueEvent, Classification
+from Backend.orchestration.base import BaseAgent
+
+logger = get_logger(__name__, agent_name="PriorityAgent")
+
+if settings.gemini_api_key:
+    genai.configure(api_key=settings.gemini_api_key)
+
+
+class IssuePrioritized(Event):
+    priority: int
+    reasoning: str
+
+
+class PriorityAgent(BaseAgent):
+    def __init__(self, db: AsyncSession):
+        super().__init__("PriorityAgent")
+        self.db = db
+        if settings.gemini_api_key:
+            self.model = genai.GenerativeModel('gemma-3-27b-it')
+        else:
+            self.model = None
+    
+    async def calculate_priority(
+        self,
+        category: Optional[str],
+        confidence: float,
+        is_duplicate: bool,
+        duplicate_count: int = 0,
+        description: Optional[str] = None,
+        city: Optional[str] = None
+    ) -> tuple[int, str]:
+        if not self.model:
+            return 3, "Gemini API not configured"
+        
+        prompt = f"""Assign priority for civic infrastructure issue:
+
+Category: {category or 'Unknown'}
+AI Confidence: {confidence:.1%}
+Duplicate Reports: {duplicate_count}
+Location: {city or 'Unknown'}
+Description: {description[:200] if description else 'N/A'}
+
+Priority Scale:
+1 = CRITICAL (Public safety, electrical hazards, major hazards)
+2 = HIGH (Potholes, road damage, fallen trees)
+3 = MEDIUM (Garbage, broken signs, minor structures)
+4 = LOW (Parking violations, minor vandalism)
+
+Consider safety impact, infrastructure criticality, and community accessibility.
+
+Return ONLY valid JSON:
+{{"priority": 1-4, "reasoning": "max 80 chars"}}"""
+        
+        try:
+            response = self.model.generate_content(prompt)
+            result = json.loads(response.text.replace("```json", "").replace("```", "").strip())
+            return result.get("priority", 3), result.get("reasoning", "Priority assigned")
+        except Exception as e:
+            logger.error(f"Gemini priority calculation failed: {e}")
+            return 3, "Analysis error"
+    
+    async def process_issue(self, issue_id: UUID) -> dict:
+        query = (
+            select(Issue)
+            .options(selectinload(Issue.classification))
+            .where(Issue.id == issue_id)
+        )
+        result = await self.db.execute(query)
+        issue = result.scalar_one_or_none()
+        if not issue:
+            return {"error": "Issue not found"}
+        
+        if issue.is_duplicate:
+            self.log_decision(
+                issue_id=issue_id,
+                decision="Skipped prioritization",
+                reasoning="Issue is a duplicate, priority inherited from parent"
+            )
+            return {"skipped": True, "reason": "duplicate"}
+        
+        category = None
+        confidence = 0.0
+        if issue.classification:
+            category = issue.classification.primary_category
+            confidence = issue.classification.primary_confidence
+        
+        dup_count_result = await self.db.execute(
+            select(func.count(Issue.id)).where(Issue.parent_issue_id == issue_id)
+        )
+        duplicate_count = dup_count_result.scalar() or 0
+        
+        priority, reasoning = await self.calculate_priority(
+            category, confidence, issue.is_duplicate, duplicate_count, issue.description, issue.city
+        )
+        
+        issue.priority = priority
+        issue.priority_reason = reasoning
+        
+        self.log_decision(
+            issue_id=issue_id,
+            decision=f"Priority set to {priority}",
+            reasoning=reasoning
+        )
+        
+        event_record = IssueEvent(
+            issue_id=issue_id,
+            event_type="prioritized",
+            agent_name=self.name,
+            event_data=json.dumps({
+                "priority": priority,
+                "reasoning": reasoning,
+                "category": category,
+                "confidence": confidence,
+            })
+        )
+        self.db.add(event_record)
+        await self.db.flush()
+        
+        priority_event = IssuePrioritized(
+            issue_id=issue_id,
+            priority=priority,
+            reasoning=reasoning,
+        )
+        await event_bus.publish(priority_event)
+        
+        return {
+            "priority": priority,
+            "reasoning": reasoning,
+        }
+    
+    async def handle(self, event) -> None:
+        await self.process_issue(event.issue_id)

Backend/agents/routing/__init__.py

@@ -0,0 +1 @@
+from .agent import RoutingAgent, IssueAssigned

Backend/agents/routing/agent.py

@@ -0,0 +1,222 @@
+import json
+from datetime import datetime, timedelta
+from typing import Optional
+from uuid import UUID
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+import google.generativeai as genai
+
+from Backend.core.config import settings
+from Backend.core.events import event_bus, Event
+from Backend.core.logging import get_logger
+from Backend.database.models import Issue, IssueEvent, Department, Member, Classification
+from Backend.orchestration.base import BaseAgent
+
+logger = get_logger(__name__, agent_name="RoutingAgent")
+
+if settings.gemini_api_key:
+    genai.configure(api_key=settings.gemini_api_key)
+
+PRIORITY_SLA_HOURS = {
+    1: 4,
+    2: 12,
+    3: 48,
+    4: 168,
+}
+
+
+class IssueAssigned(Event):
+    department_code: str
+    member_id: Optional[UUID] = None
+    member_name: Optional[str] = None
+    sla_deadline: datetime
+    sla_hours: int
+
+
+class RoutingAgent(BaseAgent):
+    def __init__(self, db: AsyncSession):
+        super().__init__("RoutingAgent")
+        self.db = db
+        if settings.gemini_api_key:
+            self.model = genai.GenerativeModel('gemma-3-27b-it')
+        else:
+            self.model = None
+    
+    async def find_department(self, category: Optional[str], description: Optional[str] = None) -> Optional[Department]:
+        query = select(Department).where(Department.is_active == True)
+        result = await self.db.execute(query)
+        departments = result.scalars().all()
+        
+        if not departments:
+            return None
+        
+        if not self.model or not category:
+            return departments[0]
+        
+        dept_info = "\n".join([f"- {d.code}: {d.name} ({d.categories})" for d in departments])
+        
+        prompt = f"""Route civic issue to correct department:
+
+Issue Category: {category}
+Description: {description[:150] if description else 'N/A'}
+
+Available Departments:
+{dept_info}
+
+Return ONLY the department CODE (e.g., PWD, TRAFFIC, SANITATION)"""
+        
+        try:
+            response = self.model.generate_content(prompt)
+            dept_code = response.text.strip().upper()
+            
+            for dept in departments:
+                if dept.code == dept_code:
+                    return dept
+        except Exception as e:
+            logger.error(f"Gemini routing failed: {e}")
+        
+        return departments[0]
+    
+    async def find_available_member(
+        self, 
+        department_id: UUID, 
+        city: Optional[str] = None,
+        locality: Optional[str] = None
+    ) -> Optional[Member]:
+        base_query = (
+            select(Member)
+            .where(Member.department_id == department_id)
+            .where(Member.is_active == True)
+            .where(Member.current_workload < Member.max_workload)
+        )
+        
+        if city:
+            city_query = base_query.where(Member.city.ilike(f"%{city}%"))
+            result = await self.db.execute(city_query.order_by(Member.current_workload.asc()))
+            member = result.scalars().first()
+            if member:
+                logger.info(f"Found member in city: {city}")
+                return member
+        
+        if locality:
+            locality_query = base_query.where(Member.locality.ilike(f"%{locality}%"))
+            result = await self.db.execute(locality_query.order_by(Member.current_workload.asc()))
+            member = result.scalars().first()
+            if member:
+                logger.info(f"Found member in locality: {locality}")
+                return member
+        
+        result = await self.db.execute(base_query.order_by(Member.current_workload.asc()))
+        member = result.scalars().first()
+        if member:
+            logger.info(f"Assigned to available member (no location match)")
+        return member
+    
+    def calculate_sla(self, priority: int, department: Optional[Department]) -> tuple[int, datetime]:
+        base_hours = PRIORITY_SLA_HOURS.get(priority, 48)
+        
+        if department and department.default_sla_hours:
+            base_hours = min(base_hours, department.default_sla_hours)
+        
+        deadline = datetime.utcnow() + timedelta(hours=base_hours)
+        return base_hours, deadline
+    
+    async def process_issue(self, issue_id: UUID) -> dict:
+        query = (
+            select(Issue)
+            .options(selectinload(Issue.classification))
+            .where(Issue.id == issue_id)
+        )
+        result = await self.db.execute(query)
+        issue = result.scalar_one_or_none()
+        if not issue:
+            return {"error": "Issue not found"}
+        
+        if issue.is_duplicate:
+            self.log_decision(
+                issue_id=issue_id,
+                decision="Skipped routing",
+                reasoning="Issue is a duplicate"
+            )
+            return {"skipped": True, "reason": "duplicate"}
+        
+        category = issue.classification.primary_category if issue.classification else None
+        priority = issue.priority or 3
+        
+        department = await self.find_department(category, issue.description)
+        
+        member = None
+        if department:
+            member = await self.find_available_member(
+                department.id, 
+                city=issue.city,
+                locality=issue.locality
+            )
+            if member:
+                member.current_workload += 1
+        
+        sla_hours, sla_deadline = self.calculate_sla(priority, department)
+        
+        issue.department_id = department.id if department else None
+        issue.assigned_member_id = member.id if member else None
+        issue.sla_hours = sla_hours
+        issue.sla_deadline = sla_deadline
+        issue.state = "assigned"
+        
+        dept_code = department.code if department else "UNASSIGNED"
+        member_name = member.name if member else "Unassigned"
+        member_city = member.city if member else "N/A"
+        
+        reasoning = f"Category '{category}' → {dept_code}"
+        if issue.city:
+            reasoning += f", Issue location: {issue.city}"
+        if member:
+            reasoning += f", Member location: {member_city}"
+        reasoning += f", SLA: {sla_hours}h"
+        
+        self.log_decision(
+            issue_id=issue_id,
+            decision=f"Routed to {dept_code} → {member_name}",
+            reasoning=reasoning
+        )
+        
+        event_record = IssueEvent(
+            issue_id=issue_id,
+            event_type="assigned",
+            agent_name=self.name,
+            event_data=json.dumps({
+                "department_code": dept_code,
+                "member_id": str(member.id) if member else None,
+                "member_name": member_name,
+                "issue_city": issue.city,
+                "issue_locality": issue.locality,
+                "member_city": member.city if member else None,
+                "sla_hours": sla_hours,
+                "sla_deadline": sla_deadline.isoformat(),
+            })
+        )
+        self.db.add(event_record)
+        await self.db.flush()
+        
+        assign_event = IssueAssigned(
+            issue_id=issue_id,
+            department_code=dept_code,
+            member_id=member.id if member else None,
+            member_name=member_name,
+            sla_deadline=sla_deadline,
+            sla_hours=sla_hours,
+        )
+        await event_bus.publish(assign_event)
+        
+        return {
+            "department": dept_code,
+            "member": member_name,
+            "issue_city": issue.city,
+            "issue_locality": issue.locality,
+            "sla_hours": sla_hours,
+            "sla_deadline": sla_deadline.isoformat(),
+        }
+    
+    async def handle(self, event) -> None:
+        await self.process_issue(event.issue_id)

Backend/agents/sla/agent.py

@@ -0,0 +1,157 @@
+import json
+from datetime import datetime
+from typing import Optional
+from uuid import UUID
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+import google.generativeai as genai
+
+from Backend.core.events import event_bus, Event
+from Backend.core.logging import get_logger
+from Backend.core.config import settings
+from Backend.database.models import Issue, IssueEvent, Member, Department
+from Backend.orchestration.base import BaseAgent
+
+logger = get_logger(__name__, agent_name="SLAAgent")
+
+if settings.gemini_api_key:
+    genai.configure(api_key=settings.gemini_api_key)
+
+
+class SLAWarning(Event):
+    hours_remaining: float
+    threshold_hours: float
+    warning_level: str  
+    assigned_email: Optional[str] = None
+
+
+class SLAAgent(BaseAgent):
+    def __init__(self, db: AsyncSession):
+        super().__init__("SLAAgent")
+        self.db = db
+        if settings.gemini_api_key:
+            self.model = genai.GenerativeModel('gemma-3-27b-it')
+        else:
+            self.model = None
+        
+    async def check_sla_status(self, issue: Issue) -> tuple[bool, str, Optional[str]]:
+        """
+        Checks if an issue needs an SLA warning.
+        Returns: (needs_warning, warning_type, reason)
+        """
+        if not issue.sla_deadline or issue.state in ["resolved", "verified", "closed", "escalated"]:
+            return False, "", None
+        
+        if not self.model:
+            now = datetime.utcnow()
+            hours_remaining = (issue.sla_deadline - now).total_seconds() / 3600
+            total_sla_hours = issue.sla_hours or 48
+            
+            if 0 < hours_remaining <= (total_sla_hours * 0.5) and hours_remaining > (total_sla_hours * 0.2):
+                 return True, "warning", f"50% SLA time remaining ({hours_remaining:.1f}h)"
+            elif 0 < hours_remaining <= (total_sla_hours * 0.2):
+                 return True, "critical", f"Critical: Less than 20% SLA time remaining ({hours_remaining:.1f}h)"
+            return False, "", None
+            
+        now = datetime.utcnow()
+        hours_remaining = (issue.sla_deadline - now).total_seconds() / 3600
+        total_sla_hours = issue.sla_hours or 48
+        hours_elapsed = total_sla_hours - hours_remaining
+        
+        prompt = f"""Assess SLA status for civic issue:
+
+Priority: {issue.priority} (1=Critical, 2=High, 3=Medium, 4=Low)
+State: {issue.state}
+Total SLA Hours: {total_sla_hours}
+Hours Elapsed: {hours_elapsed:.1f}
+Hours Remaining: {hours_remaining:.1f}
+Time Used: {(hours_elapsed/total_sla_hours*100):.1f}%
+
+Determine if warning is needed:
+- "none": No warning needed (>50% time remaining)
+- "warning": Warning level (20-50% time remaining)
+- "critical": Critical warning (<20% time remaining)
+
+Return ONLY valid JSON:
+{{"warning_level": "none/warning/critical", "reason": "max 60 chars"}}"""
+        
+        try:
+            response = self.model.generate_content(prompt)
+            result = json.loads(response.text.replace("```json", "").replace("```", "").strip())
+            level = result.get("warning_level", "none")
+            reason = result.get("reason", "SLA assessment completed")
+            
+            if level == "none":
+                return False, "", None
+            return True, level, reason
+        except Exception as e:
+            logger.error(f"Gemini SLA check failed: {e}")
+            if 0 < hours_remaining <= (total_sla_hours * 0.2):
+                return True, "critical", f"Less than 20% SLA time remaining"
+            elif 0 < hours_remaining <= (total_sla_hours * 0.5):
+                return True, "warning", f"50% SLA time remaining"
+            return False, "", None
+    
+    async def process_issue(self, issue_id: UUID) -> dict:
+        issue = await self.db.get(Issue, issue_id)
+        if not issue:
+            return {"error": "Issue not found"}
+            
+        needs_warning, level, reason = await self.check_sla_status(issue)
+        
+        if not needs_warning:
+            return {"status": "ok"}
+            
+        
+        assigned_email = None
+        if issue.assigned_member_id:
+            member = await self.db.get(Member, issue.assigned_member_id)
+            if member:
+                assigned_email = member.email
+        
+        
+        warning_event = SLAWarning(
+            issue_id=issue_id,
+            hours_remaining=(issue.sla_deadline - datetime.utcnow()).total_seconds() / 3600,
+            threshold_hours=0, 
+            warning_level=level,
+            assigned_email=assigned_email
+        )
+        await event_bus.publish(warning_event)
+        
+        
+        event_record = IssueEvent(
+            issue_id=issue_id,
+            event_type=f"sla_{level}",
+            agent_name=self.name,
+            event_data=json.dumps({
+                "hours_remaining": warning_event.hours_remaining,
+                "level": level,
+                "reason": reason
+            })
+        )
+        self.db.add(event_record)
+        await self.db.flush()
+        
+        return {"warning_sent": True, "level": level, "recipient": assigned_email}
+
+    async def check_all_active(self) -> list[dict]:
+        """Scans all active issues for SLA breaches."""
+        query = select(Issue).where(
+            Issue.state.in_(["assigned", "in_progress"]),
+            Issue.sla_deadline.isnot(None)
+        )
+        result = await self.db.execute(query)
+        issues = result.scalars().all()
+        
+        results = []
+        for issue in issues:
+            res = await self.process_issue(issue.id)
+            if res.get("warning_sent"):
+                results.append(res)
+        return results
+
+    async def handle(self, event) -> None:
+        
+        
+        pass

Backend/agents/vision/__init__.py

@@ -0,0 +1 @@
+from .agent import VisionAgent

Backend/agents/vision/agent.py

@@ -0,0 +1,296 @@
+import json
+import time
+import cv2
+import numpy as np
+import google.generativeai as genai
+from pathlib import Path
+from typing import Optional
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from Backend.core.config import settings
+from Backend.core.events import event_bus, IssueClassified, IssueCreated
+from Backend.core.logging import get_logger
+from Backend.core.schemas import ClassificationResult, DetectionBox, CLASS_ID_TO_CATEGORY, IssueCategory
+from Backend.database.models import Classification, Issue, IssueImage, IssueEvent
+from Backend.orchestration.base import BaseAgent
+from Backend.utils.fuzzy_match import auto_validate_issue
+from Backend.utils.storage import save_bytes, download_from_supabase, get_upload_url
+
+logger = get_logger(__name__, agent_name="VisionAgent")
+
+if settings.gemini_api_key:
+    genai.configure(api_key=settings.gemini_api_key)
+
+
+class VisionAgent(BaseAgent):
+    _model = None
+    
+    def __init__(self, db: Optional[AsyncSession] = None):
+        super().__init__("VisionAgent")
+        self.db = db
+        if settings.gemini_api_key:
+            self.gemini_model = genai.GenerativeModel('gemma-3-27b-it')
+        else:
+            self.gemini_model = None
+    
+    @classmethod
+    def load_model(cls):
+        if cls._model is None:
+            from ultralytics import YOLO
+            model_path = settings.model_path
+            if not model_path.exists():
+                raise FileNotFoundError(f"Model not found: {model_path}")
+            cls._model = YOLO(str(model_path))
+            logger.info(f"YOLO model loaded from {model_path}")
+        return cls._model
+    
+    @classmethod
+    def get_model(cls):
+        if cls._model is None:
+            cls.load_model()
+        return cls._model
+    
+    async def download_image(self, remote_path: str) -> bytes:
+        return await download_from_supabase(remote_path)
+    
+    async def save_annotated(self, results, original_path: str, subfolder: str) -> str:
+        im_array = results[0].plot()
+        
+        original_name = Path(original_path).stem
+        annotated_filename = f"annotated_{original_name}.jpg"
+        
+        _, buffer = cv2.imencode('.jpg', im_array, [cv2.IMWRITE_JPEG_QUALITY, 90])
+        image_bytes = buffer.tobytes()
+        
+        remote_path = await save_bytes(image_bytes, annotated_filename, subfolder=subfolder)
+        return remote_path
+    
+    async def run_inference(self, image_data: bytes) -> tuple[list, float]:
+        model = self.get_model()
+        
+        nparr = np.frombuffer(image_data, np.uint8)
+        img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
+        if img is None:
+            raise ValueError("Invalid image data")
+        
+        start_time = time.perf_counter()
+        results = model.predict(
+            source=img,
+            conf=settings.model_confidence_threshold,
+            imgsz=settings.model_input_size,
+            verbose=False,
+        )
+        inference_time = (time.perf_counter() - start_time) * 1000
+        
+        return results, inference_time
+
+    async def gemini_classify_image(
+        self,
+        image_data: bytes,
+        description: Optional[str] = None
+    ) -> tuple[Optional[IssueCategory], float, Optional[str]]:
+        if not self.gemini_model:
+            return None, 0.0, None
+
+        allowed = [
+            {"class_id": k, "class_name": v.value}
+            for k, v in CLASS_ID_TO_CATEGORY.items()
+        ]
+        prompt = (
+            "Classify the photo into exactly one of the allowed categories. "
+            "Return ONLY valid JSON with keys: class_id (int), confidence (0.0-1.0), reasoning (max 80 chars).\n\n"
+            f"Allowed categories: {json.dumps(allowed)}\n"
+            f"User description: {(description or '')[:200]}"
+        )
+
+        try:
+            response = self.gemini_model.generate_content(
+                [
+                    {"text": prompt},
+                    {
+                        "inline_data": {
+                            "mime_type": "image/jpeg",
+                            "data": image_data,
+                        }
+                    },
+                ]
+            )
+            text = (response.text or "").replace("```json", "").replace("```", "").strip()
+            data = json.loads(text)
+            class_id = data.get("class_id")
+            confidence = float(data.get("confidence", 0.0))
+            reasoning = data.get("reasoning")
+            if not isinstance(class_id, int):
+                return None, 0.0, None
+            category = CLASS_ID_TO_CATEGORY.get(class_id)
+            if not category:
+                return None, 0.0, None
+            confidence = max(0.0, min(1.0, confidence))
+            return category, confidence, reasoning
+        except Exception as e:
+            logger.error(f"Gemini vision classification failed: {e}")
+            return None, 0.0, None
+    
+    def extract_detections(self, results) -> list[DetectionBox]:
+        detections = []
+        for result in results:
+            boxes = result.boxes
+            if boxes is not None:
+                for i in range(len(boxes)):
+                    class_id = int(boxes.cls[i].item())
+                    confidence = float(boxes.conf[i].item())
+                    bbox = tuple(boxes.xyxy[i].tolist())
+                    
+                    category = CLASS_ID_TO_CATEGORY.get(class_id)
+                    if category:
+                        detections.append(DetectionBox(
+                            class_id=class_id,
+                            class_name=category.value,
+                            confidence=confidence,
+                            bbox=bbox,
+                        ))
+        return detections
+    
+    async def classify_image(
+        self,
+        image_path: str,
+        subfolder: str = "",
+        description: Optional[str] = None
+    ) -> tuple[list[DetectionBox], str, Optional[IssueCategory], float, Optional[str]]:
+        image_data = await self.download_image(image_path)
+        results, inference_time = await self.run_inference(image_data)
+        annotated_path = await self.save_annotated(results, image_path, subfolder)
+        detections = self.extract_detections(results)
+
+        gemini_category = None
+        gemini_confidence = 0.0
+        gemini_reasoning = None
+        if self.gemini_model and (not detections or max(d.confidence for d in detections) < 0.5):
+            gemini_category, gemini_confidence, gemini_reasoning = await self.gemini_classify_image(
+                image_data=image_data,
+                description=description
+            )
+        
+        logger.info(f"Inference completed in {inference_time:.2f}ms, {len(detections)} detections")
+        return detections, annotated_path, gemini_category, gemini_confidence, gemini_reasoning
+    
+    async def process_issue(
+        self,
+        issue_id: UUID,
+        image_paths: list[str],
+        description: Optional[str] = None
+    ) -> ClassificationResult:
+        all_detections = []
+        annotated_paths = []
+        total_time = 0.0
+        subfolder = str(issue_id)
+
+        gemini_best_category = None
+        gemini_best_confidence = 0.0
+        gemini_best_reasoning = None
+        
+        for path in image_paths:
+            start = time.perf_counter()
+            detections, annotated_path, gemini_category, gemini_confidence, gemini_reasoning = await self.classify_image(
+                path,
+                subfolder=subfolder,
+                description=description
+            )
+            total_time += (time.perf_counter() - start) * 1000
+            all_detections.extend(detections)
+            annotated_paths.append(annotated_path)
+
+            if gemini_category and gemini_confidence > gemini_best_confidence:
+                gemini_best_category = gemini_category
+                gemini_best_confidence = gemini_confidence
+                gemini_best_reasoning = gemini_reasoning
+            
+            if self.db:
+                query = select(IssueImage).where(IssueImage.file_path == path)
+                result = await self.db.execute(query)
+                image_record = result.scalar_one_or_none()
+                if image_record:
+                    image_record.annotated_path = annotated_path
+        
+        result = ClassificationResult(
+            issue_id=issue_id,
+            detections=all_detections,
+            annotated_urls=[get_upload_url(p) for p in annotated_paths],
+            inference_time_ms=total_time,
+        )
+
+        if gemini_best_category and (not result.primary_category or result.primary_confidence < 0.5):
+            result.primary_category = gemini_best_category
+            result.primary_confidence = gemini_best_confidence
+        
+        detected_categories = list(set(d.class_name for d in all_detections))
+        auto_validated, validation_reason = auto_validate_issue(description, detected_categories)
+        
+        validation_source = "auto" if auto_validated else "pending_manual"
+        new_state = "validated" if auto_validated else "reported"
+        
+        self.log_decision(
+            issue_id=issue_id,
+            decision=f"Validation: {validation_source}",
+            reasoning=validation_reason
+        )
+        
+        if self.db:
+            classification = Classification(
+                issue_id=issue_id,
+                primary_category=result.primary_category.value if result.primary_category else None,
+                primary_confidence=result.primary_confidence,
+                detections_json=json.dumps([d.model_dump() for d in all_detections]),
+                inference_time_ms=total_time,
+            )
+            self.db.add(classification)
+            
+            issue = await self.db.get(Issue, issue_id)
+            if issue:
+                issue.state = new_state
+                issue.validation_source = validation_source
+                issue.validation_reason = validation_reason
+            
+            event_record = IssueEvent(
+                issue_id=issue_id,
+                event_type="classified",
+                agent_name=self.name,
+                event_data=json.dumps({
+                    "category": result.primary_category.value if result.primary_category else None,
+                    "confidence": result.primary_confidence,
+                    "detections_count": len(all_detections),
+                    "validation_source": validation_source,
+                    "annotated_images": annotated_paths,
+                    "gemini_category": gemini_best_category.value if gemini_best_category else None,
+                    "gemini_confidence": gemini_best_confidence,
+                    "gemini_reasoning": gemini_best_reasoning,
+                })
+            )
+            self.db.add(event_record)
+            await self.db.flush()
+        
+        if result.primary_category:
+            event = IssueClassified(
+                issue_id=issue_id,
+                category=result.primary_category.value,
+                confidence=result.primary_confidence,
+                detections_count=len(all_detections),
+                metadata={
+                    "validation_source": validation_source,
+                    "validation_reason": validation_reason,
+                    "annotated_images": [get_upload_url(p) for p in annotated_paths],
+                }
+            )
+            await event_bus.publish(event)
+        
+        return result
+    
+    async def handle(self, event: IssueCreated) -> None:
+        await self.process_issue(
+            event.issue_id,
+            event.image_paths,
+            event.description
+        )

Backend/agents/vision/model.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:55bb189306a9882c84fb471b9cc81e2ba48363d1a4c49ccf914e9a08cde01c24
+size 22512426

Backend/api/routes/issues.py

@@ -0,0 +1,519 @@
+from typing import Optional
+from uuid import UUID
+from pydantic import BaseModel
+from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, UploadFile, status, BackgroundTasks
+from sqlalchemy import select, func
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from Backend.core.schemas import IssueCreate, IssueResponse, IssueListResponse, IssueState
+from Backend.core.flow_tracker import create_flow_tracker, remove_flow_tracker
+from Backend.database.connection import get_db, get_db_context
+from Backend.database.models import Issue, Classification
+from Backend.services.ingestion import IngestionService
+from Backend.agents import (
+    VisionAgent, 
+    GeoDeduplicateAgent, 
+    PriorityAgent, 
+    RoutingAgent,
+    NotificationAgent,
+)
+from Backend.utils.storage import get_upload_url
+from Backend.core.auth import get_user_id_from_form_token
+from Backend.core.logging import get_logger
+
+logger = get_logger(__name__)
+
+router = APIRouter()
+
+
+def issue_to_response(issue: Issue) -> IssueResponse:
+    image_urls = []
+    annotated_urls = []
+    for img in issue.images:
+        image_urls.append(get_upload_url(img.file_path))
+        if img.annotated_path:
+            annotated_urls.append(get_upload_url(img.annotated_path))
+
+    
+    return IssueResponse(
+        id=issue.id,
+        description=issue.description,
+        latitude=issue.latitude,
+        longitude=issue.longitude,
+        state=IssueState(issue.state),
+        priority=issue.priority,
+        category=issue.classification.primary_category if issue.classification else None,
+        confidence=issue.classification.primary_confidence if issue.classification else None,
+        image_urls=image_urls,
+        annotated_urls=annotated_urls,
+        validation_source=issue.validation_source,
+        is_duplicate=issue.is_duplicate,
+        parent_issue_id=issue.parent_issue_id,
+        city=issue.city,
+        locality=issue.locality,
+        full_address=issue.full_address,
+        geo_status="Duplicate" if issue.is_duplicate else "Clustered" if issue.geo_cluster_id else "Unique Location",
+        sla_hours=issue.sla_hours,
+        sla_deadline=issue.sla_deadline,
+        created_at=issue.created_at,
+        updated_at=issue.updated_at,
+    )
+
+
+async def get_issue_with_relations(db: AsyncSession, issue_id: UUID) -> Issue | None:
+    query = (
+        select(Issue)
+        .options(selectinload(Issue.images), selectinload(Issue.classification))
+        .where(Issue.id == issue_id)
+    )
+    result = await db.execute(query)
+    return result.scalar_one_or_none()
+
+
+async def run_agent_pipeline(db: AsyncSession, issue_id: UUID, image_paths: list[str], description: Optional[str]):
+    tracker = create_flow_tracker(issue_id)
+    
+    try:
+        await tracker.start_step("VisionAgent")
+        vision = VisionAgent(db)
+        vision_result = await vision.process_issue(issue_id, image_paths, description)
+        
+        detection_count = len(vision_result.detections)
+        
+        if detection_count == 0:
+            await tracker.complete_step(
+                "VisionAgent",
+                decision="No issues detected",
+                reasoning="0 detections - requires manual confirmation",
+                result={
+                    "detections": 0,
+                    "needs_confirmation": True,
+                    "annotated_urls": vision_result.annotated_urls,
+                }
+            )
+            
+            issue = await db.get(Issue, issue_id)
+            if issue:
+                issue.state = "pending_confirmation"
+                issue.validation_source = "pending_manual"
+                issue.validation_reason = "No issues detected by AI - awaiting user confirmation"
+                await db.flush()
+            
+            final_result = {
+                "issue_id": str(issue_id),
+                "state": "pending_confirmation",
+                "needs_confirmation": True,
+                "detections": 0,
+                "message": "No issues detected. Please confirm if you want to submit for manual review.",
+            }
+            await tracker.complete_flow(final_result)
+            return
+        
+        await tracker.complete_step(
+            "VisionAgent",
+            decision=f"Detected: {vision_result.primary_category.value if vision_result.primary_category else 'Unknown'}",
+            reasoning=f"Confidence: {vision_result.primary_confidence:.2%}, {detection_count} detections",
+            result=vision_result.model_dump(mode='json')
+        )
+        
+        await tracker.start_step("GeoDeduplicateAgent")
+        geo = GeoDeduplicateAgent(db)
+        geo_result = await geo.process_issue(issue_id)
+        await tracker.complete_step(
+            "GeoDeduplicateAgent",
+            decision=f"Status: {geo_result.get('geo_status', 'unknown')}",
+            reasoning=f"Nearby issues: {geo_result.get('nearby_count', 0)}",
+            result=geo_result
+        )
+        
+        if not geo_result.get("is_duplicate"):
+            await tracker.start_step("PriorityAgent")
+            priority = PriorityAgent(db)
+            priority_result = await priority.process_issue(issue_id)
+            await tracker.complete_step(
+                "PriorityAgent",
+                decision=f"Priority: {priority_result.get('priority', 'N/A')}",
+                reasoning=priority_result.get("reasoning", ""),
+                result=priority_result
+            )
+            
+            await tracker.start_step("RoutingAgent")
+            routing = RoutingAgent(db)
+            routing_result = await routing.process_issue(issue_id)
+            await tracker.complete_step(
+                "RoutingAgent",
+                decision=f"Routed to: {routing_result.get('department', 'N/A')}",
+                reasoning=f"Assigned: {routing_result.get('member', 'N/A')}, SLA: {routing_result.get('sla_hours', 0)}h",
+                result=routing_result
+            )
+            
+            await tracker.start_step("NotificationAgent")
+            notification = NotificationAgent(db)
+            await notification.notify_assignment(issue_id)
+            await tracker.complete_step(
+                "NotificationAgent",
+                decision="Notifications queued",
+                reasoning="Assignment notification sent to assigned member",
+                result={"queued": True}
+            )
+        else:
+            await tracker.complete_step(
+                "GeoDeduplicateAgent",
+                decision="Marked as duplicate",
+                reasoning=f"Linked to parent: {geo_result.get('parent_issue_id')}",
+                result=geo_result
+            )
+        
+        issue = await get_issue_with_relations(db, issue_id)
+        final_result = {
+            "issue_id": str(issue_id),
+            "state": issue.state if issue else "unknown",
+            "priority": issue.priority if issue else None,
+            "is_duplicate": issue.is_duplicate if issue else False,
+        }
+        await tracker.complete_flow(final_result)
+        
+    except Exception as e:
+        await tracker.error_flow(str(e))
+        raise
+    finally:
+        remove_flow_tracker(issue_id)
+
+
+
+async def run_agent_pipeline_background(issue_id: UUID, image_paths: list[str], description: Optional[str]):
+    async with get_db_context() as session:
+        await run_agent_pipeline(session, issue_id, image_paths, description)
+
+
+@router.post("", response_model=IssueResponse, status_code=status.HTTP_201_CREATED)
+
+async def create_issue(
+    background_tasks: BackgroundTasks,
+    images: list[UploadFile] = File(...),
+    description: Optional[str] = Form(None),
+    latitude: float = Form(...),
+    longitude: float = Form(...),
+    accuracy_meters: Optional[float] = Form(None),
+    platform: str = Form(...),
+    device_model: Optional[str] = Form(None),
+    authorization: Optional[str] = Form(None),
+    db: AsyncSession = Depends(get_db),
+):
+    user_id = get_user_id_from_form_token(authorization)
+    
+    data = IssueCreate(
+
+        description=description,
+        latitude=latitude,
+        longitude=longitude,
+        accuracy_meters=accuracy_meters,
+        platform=platform,
+        device_model=device_model,
+    )
+    
+    ingestion = IngestionService(db)
+    issue, image_paths = await ingestion.create_issue(data, images, user_id)
+    
+    
+    tracker = create_flow_tracker(issue.id)
+    
+    await tracker.start_step("LocationStep")
+    await tracker.complete_step(
+        "LocationStep", 
+        decision="Resolved", 
+        reasoning=f"{latitude:.4f}, {longitude:.4f}",
+        result={"city": "Mathura"} 
+    )
+    
+    await tracker.start_step("UploadStep")
+    await tracker.complete_step(
+       "UploadStep",
+       decision="Uploaded",
+       reasoning=f"{len(images)} images stored securely",
+       result={"count": len(images)}
+    )
+    
+    
+    background_tasks.add_task(run_agent_pipeline_background, issue.id, image_paths, data.description)
+
+    
+    issue = await get_issue_with_relations(db, issue.id)
+    issue = await get_issue_with_relations(db, issue.id)
+    return issue_to_response(issue)
+
+
+async def run_remaining_pipeline(db: AsyncSession, issue_id: UUID):
+    tracker = create_flow_tracker(issue_id)
+    try:
+        await tracker.start_step("GeoDeduplicateAgent")
+        geo = GeoDeduplicateAgent(db)
+        geo_result = await geo.process_issue(issue_id)
+        await tracker.complete_step(
+            "GeoDeduplicateAgent",
+            decision=f"Status: {geo_result.get('geo_status', 'unknown')}",
+            reasoning=f"Nearby issues: {geo_result.get('nearby_count', 0)}",
+            result=geo_result
+        )
+        
+        if not geo_result.get("is_duplicate"):
+            await tracker.start_step("PriorityAgent")
+            priority = PriorityAgent(db)
+            priority_result = await priority.process_issue(issue_id)
+            await tracker.complete_step(
+                "PriorityAgent",
+                decision=f"Priority: {priority_result.get('priority', 'N/A')}",
+                reasoning=priority_result.get("reasoning", ""),
+                result=priority_result
+            )
+            
+            
+            
+            
+            
+            
+            
+            await tracker.start_step("RoutingAgent")
+            await tracker.complete_step(
+                "RoutingAgent",
+                decision="Manual Review Requested",
+                reasoning="Skipped automatic routing due to 0 detections/manual confirmation. Sent to triage queue.",
+                result={"skipped": True, "queue": "manual_triage"}
+            )
+            
+            
+            
+        else:
+            await tracker.complete_step(
+                "GeoDeduplicateAgent",
+                decision="Marked as duplicate",
+                reasoning=f"Linked to parent: {geo_result.get('parent_issue_id')}",
+                result=geo_result
+            )
+        
+        issue = await get_issue_with_relations(db, issue_id)
+        final_result = {
+            "issue_id": str(issue_id),
+            "state": issue.state if issue else "unknown",
+            "priority": issue.priority if issue else None,
+            "is_duplicate": issue.is_duplicate if issue else False,
+        }
+        await tracker.complete_flow(final_result)
+        
+    except Exception as e:
+        await tracker.error_flow(str(e))
+        raise
+    finally:
+        remove_flow_tracker(issue_id)
+
+
+class ConfirmationBody(BaseModel):
+    confirmed: bool
+
+
+@router.post("/{issue_id}/confirm", response_model=IssueResponse)
+async def confirm_issue(
+    issue_id: UUID,
+    body: ConfirmationBody,
+    background_tasks: BackgroundTasks,
+    db: AsyncSession = Depends(get_db),
+):
+    issue = await get_issue_with_relations(db, issue_id)
+    if not issue:
+        raise HTTPException(status_code=404, detail="Issue not found")
+        
+    if body.confirmed:
+        issue.state = IssueState.REPORTED
+        issue.validation_reason = "Manual confirmation by user (0 detections)"
+        await db.flush()
+        
+        
+        issue = await get_issue_with_relations(db, issue_id)
+        
+        
+        background_tasks.add_task(pipeline_wrapper_resume, issue_id)
+        
+        return issue_to_response(issue)
+    else:
+        issue.state = IssueState.REJECTED
+        issue.validation_reason = "User rejected manual confirmation"
+        issue.resolution_notes = "User cancelled submission after 0 detections were found"
+        await db.flush()
+        
+        
+        issue = await get_issue_with_relations(db, issue_id)
+        
+        return issue_to_response(issue)
+
+
+from Backend.database.connection import get_db_context
+
+async def pipeline_wrapper(issue_id: UUID, image_paths: list[str], description: Optional[str]):
+    try:
+        async with get_db_context() as db:
+            await run_agent_pipeline(db, issue_id, image_paths, description)
+    except Exception:
+        pass
+
+async def pipeline_wrapper_resume(issue_id: UUID):
+    try:
+        async with get_db_context() as db:
+            await run_remaining_pipeline(db, issue_id)
+    except Exception:
+        pass
+
+@router.post("/stream", status_code=status.HTTP_201_CREATED)
+async def create_issue_with_stream(
+    background_tasks: BackgroundTasks,
+    images: list[UploadFile] = File(...),
+    description: Optional[str] = Form(None),
+    latitude: float = Form(...),
+    longitude: float = Form(...),
+    accuracy_meters: Optional[float] = Form(None),
+    platform: str = Form(...),
+    device_model: Optional[str] = Form(None),
+    authorization: Optional[str] = Form(None),
+    db: AsyncSession = Depends(get_db),
+):
+    user_id = get_user_id_from_form_token(authorization)
+    logger.info(f"[/stream] Creating issue - user_id: {user_id}, authorization_present: {bool(authorization)}")
+    
+    data = IssueCreate(
+        description=description,
+        latitude=latitude,
+        longitude=longitude,
+        accuracy_meters=accuracy_meters,
+        platform=platform,
+        device_model=device_model,
+    )
+    
+    ingestion = IngestionService(db)
+    issue, image_paths = await ingestion.create_issue(data, images, user_id)
+    logger.info(f"[/stream] Issue created: {issue.id} with user_id: {issue.user_id}")
+    
+    
+    await db.commit()
+    
+    
+    tracker = create_flow_tracker(issue.id)
+    
+    
+    background_tasks.add_task(pipeline_wrapper, issue.id, image_paths, data.description)
+    
+    return {
+        "issue_id": str(issue.id),
+        "stream_url": f"/flow/flow/{issue.id}",
+        "message": "Issue created. Pipeline started in background.",
+    }
+
+
+@router.post("/{issue_id}/process")
+async def process_issue_pipeline(
+    issue_id: UUID,
+    db: AsyncSession = Depends(get_db),
+):
+    issue = await get_issue_with_relations(db, issue_id)
+    if not issue:
+        raise HTTPException(status_code=404, detail="Issue not found")
+    
+    image_paths = [img.file_path for img in issue.images]
+    
+    await run_agent_pipeline(db, issue_id, image_paths, issue.description)
+    
+    issue = await get_issue_with_relations(db, issue_id)
+    return issue_to_response(issue)
+
+
+@router.get("/{issue_id}", response_model=IssueResponse)
+async def get_issue(
+    issue_id: UUID,
+    db: AsyncSession = Depends(get_db),
+):
+    issue = await get_issue_with_relations(db, issue_id)
+    if not issue:
+        raise HTTPException(status_code=404, detail="Issue not found")
+    return issue_to_response(issue)
+
+
+@router.patch("/{issue_id}/resolve")
+async def resolve_issue(
+    issue_id: UUID,
+    resolution_notes: Optional[str] = Form(None),
+    db: AsyncSession = Depends(get_db),
+):
+    issue = await db.get(Issue, issue_id)
+    if not issue:
+        raise HTTPException(status_code=404, detail="Issue not found")
+    
+    from datetime import datetime
+    issue.state = "resolved"
+    issue.resolved_at = datetime.utcnow()
+    issue.resolution_notes = resolution_notes
+    
+    if issue.assigned_member_id:
+        from Backend.database.models import Member
+        member = await db.get(Member, issue.assigned_member_id)
+        if member and member.current_workload > 0:
+            member.current_workload -= 1
+    
+    await db.flush()
+    
+    issue = await get_issue_with_relations(db, issue_id)
+    return issue_to_response(issue)
+
+
+@router.get("", response_model=IssueListResponse)
+async def list_issues(
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    state: Optional[IssueState] = None,
+    priority: Optional[int] = Query(None, ge=1, le=4),
+    department_id: Optional[UUID] = None,
+    is_duplicate: Optional[bool] = None,
+    user_id: Optional[str] = Query(None),
+    db: AsyncSession = Depends(get_db),
+):
+    query = (
+        select(Issue)
+        .options(selectinload(Issue.images), selectinload(Issue.classification))
+        .order_by(Issue.created_at.desc())
+    )
+    count_query = select(func.count(Issue.id))
+    
+    if state:
+        query = query.where(Issue.state == state.value)
+        count_query = count_query.where(Issue.state == state.value)
+    
+    if priority:
+        query = query.where(Issue.priority == priority)
+        count_query = count_query.where(Issue.priority == priority)
+    
+    if department_id:
+        query = query.where(Issue.department_id == department_id)
+        count_query = count_query.where(Issue.department_id == department_id)
+    
+    if is_duplicate is not None:
+        query = query.where(Issue.is_duplicate == is_duplicate)
+        count_query = count_query.where(Issue.is_duplicate == is_duplicate)
+    
+    if user_id:
+        query = query.where(Issue.user_id == user_id)
+        count_query = count_query.where(Issue.user_id == user_id)
+
+    
+    offset = (page - 1) * page_size
+    query = query.offset(offset).limit(page_size)
+    
+    result = await db.execute(query)
+    issues = result.scalars().all()
+    
+    count_result = await db.execute(count_query)
+    total = count_result.scalar() or 0
+    
+    return IssueListResponse(
+        items=[issue_to_response(issue) for issue in issues],
+        total=total,
+        page=page,
+        page_size=page_size,
+    )

Backend/core/config.py

@@ -51,8 +51,8 @@ class Settings(BaseSettings):
     gemini_api_key: Optional[str] = None
     google_client_secret: Optional[str] = None
     project_id: Optional[str] = None
-    sender_email: str = "noreply@urbanlens.city"
-    admin_email: str = "admin@urbanlens.city"
+    sender_email: str = "noreply@CityTrack.city"
+    admin_email: str = "admin@CityTrack.city"
 
     frontend_url: Optional[str] = None
     

Backend/services/__init__.py

@@ -0,0 +1,2 @@
+from .ingestion import IngestionService
+from .vision import VisionService

Backend/services/email.py

@@ -0,0 +1,273 @@
+import resend
+from typing import List
+from Backend.core.config import settings
+from Backend.core.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class EmailService:
+    def __init__(self):
+        self.sender_email = settings.sender_email
+        if settings.resend_api_key:
+            resend.api_key = settings.resend_api_key
+        else:
+            logger.warning("Resend API key not configured")
+    
+    async def send_email(
+        self,
+        to: List[str],
+        subject: str,
+        body: str,
+        html: bool = False
+    ) -> bool:
+        if not settings.resend_api_key:
+            logger.warning("Resend API key not configured. Email not sent.")
+            logger.info(f"Would send email to {to}: {subject}")
+            return False
+        
+        try:
+            for recipient in to:
+                try:
+                    params = {
+                        "from": self.sender_email,
+                        "to": [recipient],
+                        "subject": subject,
+                    }
+                    
+                    if html:
+                        params["html"] = body
+                    else:
+                        params["text"] = body
+                    
+                    resend.Emails.send(params)
+                    logger.info(f"Email sent successfully to {recipient}")
+                except Exception as e:
+                    logger.error(f"Failed to send email to {recipient}: {e}")
+                    return False
+            
+            return True
+            
+        except Exception as e:
+            logger.error(f"Email service error: {e}")
+            return False
+    
+    async def send_assignment_email(
+        self,
+        worker_email: str,
+        worker_name: str,
+        issue_id: str,
+        category: str,
+        priority: str,
+        location: str,
+        description: str
+    ):
+        subject = f"🔔 New Task Assigned: {category} [{priority}]"
+        
+        body = f"""
+Hello {worker_name},
+
+You have been assigned a new task in CityTrack.
+
+ISSUE DETAILS
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+Issue ID: {issue_id}
+Category: {category}
+Priority: {priority}
+Location: {location}
+Description: {description or 'No description provided'}
+
+NEXT STEPS
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+1. Review the issue details in your worker dashboard
+2. Navigate to the location
+3. Resolve the issue
+4. Upload proof of resolution
+
+Thank you for your service!
+
+CityTrack Team
+"Governance at the Speed of Software"
+"""
+        
+        return await self.send_email([worker_email], subject, body)
+    
+    async def send_manual_review_email(
+        self,
+        issue_id: str,
+        reason: str,
+        category: str,
+        location: str,
+        image_url: str
+    ):
+        subject = f"⚠️ Manual Review Required: {category}"
+        
+        body = f"""
+Admin Team,
+
+An issue requires manual review in CityTrack.
+
+ISSUE DETAILS
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+Issue ID: {issue_id}
+Category: {category}
+Location: {location}
+Reason: {reason}
+
+Image: {image_url}
+
+ACTION REQUIRED
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+Please review this issue in the admin dashboard and take appropriate action.
+
+CityTrack System
+"""
+        
+        return await self.send_email([settings.admin_email], subject, body)
+    
+    async def send_completion_email(
+        self,
+        user_email: str,
+        issue_id: str,
+        category: str,
+        location: str,
+        resolution_notes: str
+    ):
+        subject = f"✅ Your Report Has Been Resolved: {category}"
+        
+        body = f"""
+Dear Citizen,
+
+Great news! Your reported issue has been resolved.
+
+ISSUE DETAILS
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+Issue ID: {issue_id}
+Category: {category}
+Location: {location}
+Resolution: {resolution_notes or 'Issue has been successfully addressed'}
+
+FEEDBACK
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+We value your input! Please confirm if the issue has been resolved by checking the app.
+
+Thank you for making our city better!
+
+CityTrack Team
+"Governance at the Speed of Software"
+"""
+        
+        return await self.send_email([user_email], subject, body)
+    
+    async def send_escalation_email(
+        self,
+        admin_email: str,
+        issue_id: str,
+        category: str,
+        priority: str,
+        reason: str,
+        escalation_level: int
+    ):
+        subject = f"🚨 ESCALATION LEVEL {escalation_level}: {category}"
+        
+        body = f"""
+URGENT: Issue Escalation
+
+An issue has been escalated and requires immediate attention.
+
+ISSUE DETAILS
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+Issue ID: {issue_id}
+Category: {category}
+Priority: {priority}
+Escalation Level: {escalation_level}
+Reason: {reason}
+
+IMMEDIATE ACTION REQUIRED
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+Please review and address this issue immediately in the admin dashboard.
+
+CityTrack System
+"""
+        
+        return await self.send_email([admin_email], subject, body)
+    
+    async def send_confirmation_request_email(
+        self,
+        user_email: str,
+        issue_id: str,
+        category: str,
+        confirmation_link: str
+    ):
+        subject = f"🔍 Please Confirm: Is This Issue Resolved?"
+        
+        body = f"""
+Dear Citizen,
+
+Your reported issue has been marked as resolved by our team.
+
+ISSUE DETAILS
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+Issue ID: {issue_id}
+Category: {category}
+
+CONFIRMATION NEEDED
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+Please confirm if the issue has been properly resolved:
+{confirmation_link}
+
+Your feedback helps us improve our service quality.
+
+Thank you!
+
+CityTrack Team
+"""
+        
+        return await self.send_email([user_email], subject, body)
+    
+    async def send_issue_accepted_email(
+        self,
+        user_email: str,
+        issue_id: str,
+        category: str,
+        priority: str,
+        location: str,
+        accepted_by: str = "automatic",
+        tracking_url: str = None
+    ):
+        acceptance_type = "automatically" if accepted_by == "automatic" else "manually by our team"
+        subject = f"✓ Your Report Has Been Accepted: {category}"
+        
+        body = f"""
+Dear Citizen,
+
+Thank you for reporting an issue! Your report has been accepted {acceptance_type}.
+
+ISSUE DETAILS
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+Issue ID: {issue_id}
+Category: {category}
+Priority: {priority}
+Location: {location}
+
+WHAT HAPPENS NEXT
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+1. Your issue has been assigned to the appropriate department
+2. A field worker will be dispatched to address it
+3. You will receive updates on the progress
+4. Once resolved, you'll get a confirmation notification
+
+TRACK YOUR REPORT
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+{tracking_url or 'Check the CityTrack app for real-time updates'}
+
+Thank you for helping make our city better!
+
+CityTrack Team
+"Governance at the Speed of Software"
+"""
+        
+        return await self.send_email([user_email], subject, body)
+
+
+email_service = EmailService()

Backend/services/geocoding.py

@@ -0,0 +1,100 @@
+import aiohttp
+from typing import Optional
+from dataclasses import dataclass
+
+from Backend.core.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+@dataclass
+class LocationInfo:
+    city: Optional[str] = None
+    locality: Optional[str] = None
+    district: Optional[str] = None
+    state: Optional[str] = None
+    country: Optional[str] = None
+    pincode: Optional[str] = None
+    full_address: Optional[str] = None
+
+
+class GeocodingService:
+    NOMINATIM_URL = "https://nominatim.openstreetmap.org/reverse"
+    
+    async def reverse_geocode(self, latitude: float, longitude: float) -> LocationInfo:
+        params = {
+            "lat": latitude,
+            "lon": longitude,
+            "format": "json",
+            "addressdetails": 1,
+            "zoom": 18,
+        }
+        
+        headers = {
+            "User-Agent": "CityIssueResolutionAgent/1.0"
+        }
+        
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.get(
+                    self.NOMINATIM_URL,
+                    params=params,
+                    headers=headers,
+                    timeout=aiohttp.ClientTimeout(total=10)
+                ) as response:
+                    if response.status == 200:
+                        data = await response.json()
+                        return self._parse_response(data)
+                    else:
+                        logger.warning(f"Geocoding failed: {response.status}")
+                        return LocationInfo()
+        except Exception as e:
+            logger.error(f"Geocoding error: {e}")
+            return LocationInfo()
+    
+    def _parse_response(self, data: dict) -> LocationInfo:
+        address = data.get("address", {})
+        
+        city = (
+            address.get("city") or
+            address.get("town") or
+            address.get("municipality") or
+            address.get("village") or
+            address.get("suburb")
+        )
+        
+        locality = (
+            address.get("suburb") or
+            address.get("neighbourhood") or
+            address.get("quarter") or
+            address.get("borough")
+        )
+        
+        district = (
+            address.get("county") or
+            address.get("district") or
+            address.get("state_district")
+        )
+        
+        state = address.get("state")
+        country = address.get("country")
+        pincode = address.get("postcode")
+        
+        full_address = data.get("display_name")
+        
+        return LocationInfo(
+            city=city,
+            locality=locality,
+            district=district,
+            state=state,
+            country=country,
+            pincode=pincode,
+            full_address=full_address,
+        )
+    
+    async def get_city_from_coordinates(self, latitude: float, longitude: float) -> Optional[str]:
+        location = await self.reverse_geocode(latitude, longitude)
+        return location.city or location.locality or location.district
+
+
+geocoding_service = GeocodingService()

Backend/services/ingestion.py

@@ -0,0 +1,85 @@
+from uuid import UUID
+from fastapi import UploadFile
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from Backend.core.events import event_bus, IssueCreated
+from Backend.core.logging import get_logger
+from Backend.core.schemas import IssueCreate, IssueState
+from Backend.database.models import Issue, IssueImage
+from Backend.services.geocoding import geocoding_service
+from Backend.utils.storage import save_upload, get_upload_url, validate_file_extension, validate_file_size
+
+logger = get_logger(__name__)
+
+
+class IngestionService:
+    def __init__(self, db: AsyncSession):
+        self.db = db
+    
+    async def create_issue(
+        self,
+        data: IssueCreate,
+        images: list[UploadFile],
+        user_id: str | None = None
+    ) -> tuple[Issue, list[str]]:
+        if not images:
+            raise ValueError("At least one image is required")
+        
+        for image in images:
+            if not validate_file_extension(image.filename or ""):
+                raise ValueError(f"Invalid file extension: {image.filename}")
+        
+        location_info = await geocoding_service.reverse_geocode(
+            data.latitude, data.longitude
+        )
+        
+        logger.info(f"Location resolved: {location_info.city}, {location_info.locality}")
+
+        final_description = data.description or "Issue reported"
+        
+        issue = Issue(
+            user_id=user_id,
+            description=final_description,
+            latitude=data.latitude,
+            longitude=data.longitude,
+            accuracy_meters=data.accuracy_meters,
+            platform=data.platform,
+            device_model=data.device_model,
+            state=IssueState.REPORTED,
+            city=location_info.city,
+            locality=location_info.locality,
+            full_address=location_info.full_address,
+        )
+
+        self.db.add(issue)
+        await self.db.flush()
+        
+        image_paths = []
+        for image in images:
+            file_path = await save_upload(image, subfolder=str(issue.id))
+            
+            issue_image = IssueImage(
+                issue_id=issue.id,
+                file_path=file_path,
+                original_filename=image.filename,
+            )
+            self.db.add(issue_image)
+            image_paths.append(file_path)
+        
+        await self.db.flush()
+        
+        event = IssueCreated(
+            issue_id=issue.id,
+            image_paths=image_paths,
+            latitude=issue.latitude,
+            longitude=issue.longitude,
+            description=issue.description,
+        )
+        await event_bus.publish(event)
+        
+        logger.info(f"Issue created: {issue.id} in {issue.city}")
+        
+        return issue, image_paths
+    
+    async def get_issue(self, issue_id: UUID) -> Issue | None:
+        return await self.db.get(Issue, issue_id)

Backend/services/vision.py

@@ -0,0 +1,3 @@
+from Backend.agents.vision import VisionAgent
+
+VisionService = VisionAgent

Backend/utils/__init__.py

@@ -0,0 +1,3 @@
+from .geo import haversine_distance, is_within_radius, find_nearby_issues
+from .storage import save_upload, generate_filename, get_upload_url, save_bytes, download_from_supabase
+from .fuzzy_match import auto_validate_issue, match_description_to_category

Backend/utils/fuzzy_match.py

@@ -0,0 +1,113 @@
+from difflib import SequenceMatcher
+from typing import Optional
+
+CATEGORY_KEYWORDS: dict[str, list[str]] = {
+    "Damaged Road Issues": [
+        "road", "damage", "damaged", "broken", "crack", "cracked", "pavement",
+        "asphalt", "street", "highway", "lane", "surface", "rough", "uneven"
+    ],
+    "Pothole Issues": [
+        "pothole", "hole", "pit", "crater", "dip", "depression", "bump",
+        "cavity", "hollow", "gap"
+    ],
+    "Illegal Parking Issues": [
+        "parking", "parked", "car", "vehicle", "illegal", "wrong", "blocking",
+        "obstruction", "no parking", "double park", "sidewalk"
+    ],
+    "Broken Road Sign Issues": [
+        "sign", "signboard", "traffic sign", "road sign", "broken sign",
+        "fallen sign", "damaged sign", "missing sign", "bent"
+    ],
+    "Fallen Trees": [
+        "tree", "fallen", "branch", "trunk", "uprooted", "collapsed",
+        "blocking", "storm", "wind", "timber"
+    ],
+    "Littering/Garbage on Public Places": [
+        "garbage", "trash", "litter", "waste", "rubbish", "dump", "dirty",
+        "filth", "debris", "plastic", "pile", "mess", "junk", "disposal"
+    ],
+    "Vandalism Issues": [
+        "vandal", "graffiti", "spray", "paint", "defaced", "broken",
+        "smashed", "destroyed", "damaged property", "torn"
+    ],
+    "Dead Animal Pollution": [
+        "dead", "animal", "carcass", "body", "corpse", "rotting", "smell",
+        "stink", "dog", "cat", "bird", "cow", "roadkill"
+    ],
+    "Damaged Concrete Structures": [
+        "concrete", "structure", "wall", "pillar", "bridge", "flyover",
+        "footpath", "sidewalk", "curb", "crack", "broken"
+    ],
+    "Damaged Electric Wires and Poles": [
+        "electric", "wire", "pole", "cable", "power", "electricity",
+        "hanging", "exposed", "sparking", "transformer", "light pole"
+    ],
+}
+
+
+def normalize_text(text: str) -> str:
+    return text.lower().strip()
+
+
+def calculate_similarity(s1: str, s2: str) -> float:
+    return SequenceMatcher(None, s1.lower(), s2.lower()).ratio()
+
+
+def fuzzy_match_word(word: str, keywords: list[str], threshold: float = 0.7) -> bool:
+    word = normalize_text(word)
+    for keyword in keywords:
+        if word == keyword:
+            return True
+        if len(word) >= 4 and calculate_similarity(word, keyword) >= threshold:
+            return True
+        if keyword in word or word in keyword:
+            return True
+    return False
+
+
+def match_description_to_category(
+    description: Optional[str],
+    detected_category: str,
+    threshold: float = 0.6
+) -> tuple[bool, float, list[str]]:
+    if not description:
+        return False, 0.0, []
+    
+    keywords = CATEGORY_KEYWORDS.get(detected_category, [])
+    if not keywords:
+        return False, 0.0, []
+    
+    words = normalize_text(description).replace(",", " ").replace(".", " ").split()
+    
+    matched_words = []
+    for word in words:
+        if len(word) < 3:
+            continue
+        if fuzzy_match_word(word, keywords):
+            matched_words.append(word)
+    
+    if not words:
+        return False, 0.0, []
+    
+    match_score = len(matched_words) / max(len(words), 1)
+    is_match = len(matched_words) >= 1 or match_score >= threshold
+    
+    return is_match, match_score, matched_words
+
+
+def auto_validate_issue(
+    description: Optional[str],
+    detected_categories: list[str],
+    confidence_threshold: float = 0.5
+) -> tuple[bool, str]:
+    if not description or not detected_categories:
+        return False, "No description or no detections for auto-validation"
+    
+    for category in detected_categories:
+        is_match, score, matched_words = match_description_to_category(
+            description, category
+        )
+        if is_match:
+            return True, f"Auto-validated: '{category}' matched with keywords: {matched_words}"
+    
+    return False, f"Manual verification required: no match between description and detected categories {detected_categories}"

Backend/utils/geo.py

@@ -0,0 +1,52 @@
+from math import radians, cos, sin, asin, sqrt
+from typing import Sequence
+from uuid import UUID
+
+
+def haversine_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
+    R = 6371000
+    
+    lat1, lon1, lat2, lon2 = map(radians, [lat1, lon1, lat2, lon2])
+    
+    dlat = lat2 - lat1
+    dlon = lon2 - lon1
+    
+    a = sin(dlat / 2) ** 2 + cos(lat1) * cos(lat2) * sin(dlon / 2) ** 2
+    c = 2 * asin(sqrt(a))
+    
+    return R * c
+
+
+def is_within_radius(
+    lat1: float, lon1: float,
+    lat2: float, lon2: float,
+    radius_meters: float
+) -> bool:
+    return haversine_distance(lat1, lon1, lat2, lon2) <= radius_meters
+
+
+def find_nearby_issues(
+    target_lat: float,
+    target_lon: float,
+    issues: Sequence[tuple[UUID, float, float]],
+    radius_meters: float
+) -> list[tuple[UUID, float]]:
+    nearby = []
+    for issue_id, lat, lon in issues:
+        distance = haversine_distance(target_lat, target_lon, lat, lon)
+        if distance <= radius_meters:
+            nearby.append((issue_id, distance))
+    return sorted(nearby, key=lambda x: x[1])
+
+
+def get_bounding_box(lat: float, lon: float, radius_meters: float) -> tuple[float, float, float, float]:
+    R = 6371000
+    lat_delta = (radius_meters / R) * (180 / 3.14159265359)
+    lon_delta = lat_delta / cos(radians(lat))
+    
+    return (
+        lat - lat_delta,
+        lat + lat_delta,
+        lon - lon_delta,
+        lon + lon_delta
+    )

main.py

@@ -0,0 +1,14 @@
+import uvicorn
+from Backend.core.config import settings
+
+def main():
+    uvicorn.run(
+        "Backend.api:app",
+        host=settings.api_host,
+        port=settings.api_port,
+        reload=True,
+        workers=1,
+    )
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,19 @@
+fastapi>=0.109.0
+uvicorn[standard]>=0.27.0
+pydantic>=2.5.0
+pydantic-settings>=2.1.0
+email-validator>=2.0.0
+sqlalchemy[asyncio]>=2.0.25
+asyncpg>=0.29.0
+aiofiles>=23.2.1
+aiohttp>=3.9.0
+python-multipart>=0.0.6
+PyJWT>=2.8.0
+bcrypt>=4.1.0
+numpy<2
+opencv-python-headless==4.11.0.86
+torch>=2.1.0,<3
+torchvision>=0.16.0,<1
+ultralytics>=8.1.0
+resend>=2.0.0
+google-generativeai>=0.8.3

static/flow.html

@@ -4,7 +4,7 @@
 <head>
     <meta charset="UTF-8">
     <meta name="viewport" content="width=device-width, initial-scale=1.0">
-    <title>Agent Mission Control | UrbanLens</title>
+    <title>Agent Mission Control | CityTrack</title>
     <!-- Tailwind CSS -->
     <script src="https://cdn.tailwindcss.com"></script>
     <!-- Fonts -->
@@ -124,7 +124,7 @@
                 <i data-lucide="cpu" class="w-5 h-5 text-white"></i>
             </div>
             <div>
-                <h1 class="font-bold text-lg leading-tight tracking-tight">UrbanLens <span
+                <h1 class="font-bold text-lg leading-tight tracking-tight">CityTrack <span
                         class="text-blue-500">Core</span></h1>
                 <p class="text-[10px] text-slate-400 font-mono tracking-wider uppercase">Agent Orchestration Node</p>
             </div>
@@ -362,7 +362,7 @@
         let currentAnnotatedImage = null; // URI
 
         // --- IndexedDB Setup ---
-        const DB_NAME = 'UrbanLensDB';
+        const DB_NAME = 'CityTrackDB';
         const DB_VERSION = 1;
         let db;