Sprint 2: Ticket Engine, SLA Engine, Photo Uploads, Category System

- Ticket CRUD with full 15-status lifecycle enforcement
- Auto-generated ticket numbers (PAV-YYYY-NNNNN)
- Status transitions validated and auto-logged to timeline
- Reopen with 7-day window + auto-escalation to Ama
- SLA engine: priority-based deadlines and breach detection
- Photo uploads to uploads/ with multipart support
- Category system seeded from PRD §7 taxonomy (20 top-level)
- Flat and tree category listing endpoints
- SLA status check endpoint (/api/tickets/{id}/sla)
- Paginated ticket listing with filters

.gitignore

@@ -4,6 +4,8 @@ __pycache__/
 *.sqlite3
 .env
 .venv/
+uploads/
+test_*.py
 venv/
 *.egg-info/
 dist/

AGENTS.md

@@ -19,9 +19,10 @@ app/
 ├── core/          # config, database, security (JWT + bcrypt + RBAC)
 ├── models/        # SQLAlchemy ORM models
 ├── schemas/       # Pydantic request/response schemas
-├── services/      # Business logic (auth, seed)
+├── services/      # Business logic (auth, seed, ticket, sla)
 └── routers/       # FastAPI route handlers
 alembic/           # Database migrations
+uploads/           # Photo uploads (created at runtime)
 ```
 
 ## Commands
@@ -31,14 +32,16 @@ alembic/           # Database migrations
 
 ## Seed data
 
-Users and units are auto-seeded on first startup via lifespan hook:
+Users, units, and categories are auto-seeded on first startup via lifespan hook:
 - 17 users covering all roles (Admin/Jerome, Admin/Wahab, CS Rep, CS Manager, FM Dispatcher, Tech, CEO, Director)
 - 120 apartment units (East/West, 10 floors × 6 apts per wing)
 - Default password for all seed users: `denya123`
 - Units load from `apartment_mapping.json` if present, else built-in fallback
+- Categories: 20 top-level (10 Maintenance, 6 CS, 4 Emergency) with sub-categories, seeded from `app/services/seed.py::SEED_CATEGORIES_DATA`
 
 ## Key API endpoints
 
+### Auth & Health
 | Method | Path | Auth | Description |
 |--------|------|------|-------------|
 | GET | `/health` | No | Health check |
@@ -50,6 +53,20 @@ Users and units are auto-seeded on first startup via lifespan hook:
 | POST | `/api/whatsapp/mock` | No | Mock WhatsApp |
 | GET | `/api/whatsapp/mock-log` | No | Recent mock submissions |
 
+### Tickets (Sprint 2)
+| Method | Path | Auth | Description |
+|--------|------|------|-------------|
+| POST | `/api/tickets` | Bearer | Create ticket (auto-number PAV-YYYY-NNNNN) |
+| GET | `/api/tickets` | No | List tickets (filter: status, priority, property, category_id, assigned_to, date_from, date_to) |
+| GET | `/api/tickets/{id}` | No | Get ticket detail with timeline, photos, SLA status |
+| PATCH | `/api/tickets/{id}` | Bearer | Update ticket (validates status transitions) |
+| POST | `/api/tickets/{id}/status` | Bearer | Change status with note |
+| GET | `/api/tickets/{id}/sla` | No | Check SLA breach status |
+| POST | `/api/tickets/{id}/photos` | Bearer | Upload photos (multipart, is_before param) |
+| GET | `/api/tickets/{id}/photos` | No | List photos |
+| GET | `/api/tickets/categories` | No | Category tree (optional `?type=` filter) |
+| GET | `/api/tickets/categories/flat` | No | Flat category list |
+
 ## Auth
 
 - JWT access (30min) + refresh (7d) tokens
@@ -57,6 +74,28 @@ Users and units are auto-seeded on first startup via lifespan hook:
 - Use `require_roles("Admin/Jerome", "Admin/Wahab")` dependency for RBAC
 - `sub` claim holds string user ID
 
+## Ticket System (Sprint 2)
+
+### Status Lifecycle (15 statuses)
+New → Logged → Triage → Assigned → Accepted → Travelling → On Site → In Progress → Waiting Parts → Escalated → Completed → On-Field Verification → Wahab Review → Closed → Reopened
+
+Valid transitions defined in `app/services/ticket.py::VALID_TRANSITIONS`. Invalid transitions return 400.
+
+### SLA Engine
+Defined in `app/services/sla.py`. Priority-based targets:
+- Urgent: respond 15min, resolve 4h
+- High: respond 30min, resolve 24h
+- Medium: respond 4h, resolve 72h (3d)
+- Low: respond 24h, resolve 168h (7d)
+
+`sla_deadline` auto-calculated on ticket creation. SLA status check at `GET /api/tickets/{id}/sla`.
+
+### Ticket Number Format
+`PAV-YYYY-NNNNN` — sequential per year (e.g., PAV-2026-00001).
+
+### Photo Uploads
+Stored under `uploads/` with UUID filenames. Static-files mounted at `/uploads/`. Multipart POST with `is_before` query param.
+
 ## Database
 
 SQLite via aiosqlite with async SQLAlchemy 2.0. Alembic for migrations.

app/main.py

@@ -4,16 +4,16 @@ from __future__ import annotations
 
 import logging
 from contextlib import asynccontextmanager
+from pathlib import Path
 
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
-from sqlalchemy import text
-from sqlalchemy.ext.asyncio import AsyncSession
+from fastapi.staticfiles import StaticFiles
 
 from app.core.config import settings
 from app.core.database import Base, async_session_factory, engine
-from app.routers import auth, health, whatsapp
-from app.services.seed import seed_units, seed_users
+from app.routers import auth, health, tickets, whatsapp
+from app.services.seed import seed_categories, seed_units, seed_users
 
 logger = logging.getLogger(__name__)
 
@@ -29,6 +29,8 @@ async def lifespan(app: FastAPI):
         await session.commit()
         await seed_units(session, json_path=str(settings.BASE_DIR / "apartment_mapping.json"))
         await session.commit()
+        await seed_categories(session)
+        await session.commit()
     yield
     await engine.dispose()
     logger.info("Denya OneCare stopped.")
@@ -49,7 +51,13 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
+# ── Static files (uploads) ───────────────────────────────────────────
+uploads_dir = Path(settings.BASE_DIR / "uploads")
+uploads_dir.mkdir(parents=True, exist_ok=True)
+app.mount("/uploads", StaticFiles(directory=str(uploads_dir)), name="uploads")
+
 # ── Routers ──────────────────────────────────────────────────────────
 app.include_router(health.router)
 app.include_router(auth.router)
 app.include_router(whatsapp.router)
+app.include_router(tickets.router)

app/routers/tickets.py

@@ -0,0 +1,297 @@
+"""Ticket CRUD endpoints, photo uploads, SLA checks, and category listing."""
+
+from __future__ import annotations
+
+import os
+import uuid
+from datetime import datetime
+from pathlib import Path
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from app.core.config import settings
+from app.core.database import get_db
+from app.core.security import get_current_user
+from app.models.category import Category
+from app.models.ticket import Ticket, TicketPhoto
+from app.models.user import User
+from app.schemas.ticket import (
+    CategoryOut,
+    CategoryTreeOut,
+    SLAStatusOut,
+    TicketBrief,
+    TicketCreate,
+    TicketListResponse,
+    TicketOut,
+    TicketPhotoOut,
+    TicketUpdate,
+)
+from app.services import ticket as ticket_service
+from app.services.sla import get_sla_status
+
+router = APIRouter(prefix="/api/tickets", tags=["tickets"])
+
+# Ensure uploads directory exists
+UPLOADS_DIR = settings.BASE_DIR / "uploads"
+UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
+
+
+# ── Categories ──────────────────────────────────────────────────────
+async def _build_category_tree(db: AsyncSession, parent_id: int | None = None) -> list[CategoryTreeOut]:
+    """Build a nested category tree."""
+    result = await db.execute(
+        select(Category)
+        .where(Category.parent_id == parent_id)
+        .order_by(Category.name)
+    )
+    categories = result.scalars().all()
+    tree = []
+    for cat in categories:
+        children = await _build_category_tree(db, cat.id)
+        tree.append(CategoryTreeOut(
+            id=cat.id,
+            type=cat.type,
+            name=cat.name,
+            parent_id=cat.parent_id,
+            sla_urgency=cat.sla_urgency,
+            children=children,
+        ))
+    return tree
+
+
+@router.get("/categories", response_model=list[CategoryTreeOut])
+async def list_categories(
+    db: Annotated[AsyncSession, Depends(get_db)],
+    type_filter: str | None = Query(None, alias="type"),
+) -> list[CategoryTreeOut]:
+    """Return the full category tree, optionally filtered by type (maintenance, cs, emergency)."""
+    if type_filter:
+        # Return flat list of top-level categories of the given type
+        result = await db.execute(
+            select(Category)
+            .where(Category.parent_id.is_(None), Category.type == type_filter)
+            .order_by(Category.name)
+        )
+        parents = result.scalars().all()
+        tree = []
+        for parent in parents:
+            children_result = await db.execute(
+                select(Category)
+                .where(Category.parent_id == parent.id)
+                .order_by(Category.name)
+            )
+            children = children_result.scalars().all()
+            tree.append(CategoryTreeOut(
+                id=parent.id,
+                type=parent.type,
+                name=parent.name,
+                parent_id=parent.parent_id,
+                sla_urgency=parent.sla_urgency,
+                children=[CategoryOut.model_validate(c) for c in children],
+            ))
+        return tree
+    return await _build_category_tree(db)
+
+
+@router.get("/categories/flat", response_model=list[CategoryOut])
+async def list_categories_flat(
+    db: Annotated[AsyncSession, Depends(get_db)],
+    type_filter: str | None = Query(None, alias="type"),
+) -> list[CategoryOut]:
+    """Return a flat list of all categories (no nesting), optionally filtered by type."""
+    query = select(Category).order_by(Category.type, Category.name)
+    if type_filter:
+        query = query.where(Category.type == type_filter)
+    result = await db.execute(query)
+    categories = result.scalars().all()
+    return [CategoryOut.model_validate(c) for c in categories]
+
+
+# ── Ticket CRUD ──────────────────────────────────────────────────────
+@router.post("", response_model=TicketOut, status_code=status.HTTP_201_CREATED)
+async def create_ticket(
+    body: TicketCreate,
+    db: Annotated[AsyncSession, Depends(get_db)],
+    current_user: Annotated[User, Depends(get_current_user)],
+) -> Ticket:
+    """Create a new ticket. Auto-generates ticket number PAV-YYYY-NNNNN."""
+    ticket = await ticket_service.create_ticket(db, body.model_dump(exclude_none=True), user=current_user)
+    # Reload with relationships
+    ticket = await ticket_service.get_ticket(db, ticket.id)
+    sla = await get_sla_status(ticket)
+    ticket.sla_status = sla  # type: ignore[attr-defined]
+    return ticket
+
+
+@router.get("", response_model=TicketListResponse)
+async def list_tickets(
+    db: Annotated[AsyncSession, Depends(get_db)],
+    page: int = Query(1, ge=1),
+    page_size: int = Query(50, ge=1, le=200),
+    status: str | None = Query(None),
+    priority: str | None = Query(None),
+    property: str | None = Query(None),
+    category_id: int | None = Query(None),
+    assigned_to: int | None = Query(None),
+    date_from: datetime | None = Query(None),
+    date_to: datetime | None = Query(None),
+) -> TicketListResponse:
+    """List tickets with optional filtering and pagination."""
+    tickets, total = await ticket_service.list_tickets(
+        db,
+        status_filter=status,
+        priority_filter=priority,
+        property_filter=property,
+        category_id=category_id,
+        assigned_to=assigned_to,
+        date_from=date_from,
+        date_to=date_to,
+        page=page,
+        page_size=page_size,
+    )
+    items = [TicketBrief.model_validate(t) for t in tickets]
+    return TicketListResponse(items=items, total=total, page=page, page_size=page_size)
+
+
+@router.get("/{ticket_id}", response_model=TicketOut)
+async def get_ticket(
+    ticket_id: int,
+    db: Annotated[AsyncSession, Depends(get_db)],
+) -> Ticket:
+    """Get a single ticket by ID with full details, timeline, photos, and SLA status."""
+    ticket = await ticket_service.get_ticket(db, ticket_id)
+    sla = await get_sla_status(ticket)
+    ticket.sla_status = sla  # type: ignore[attr-defined]
+    return ticket
+
+
+@router.patch("/{ticket_id}", response_model=TicketOut)
+async def update_ticket(
+    ticket_id: int,
+    body: TicketUpdate,
+    db: Annotated[AsyncSession, Depends(get_db)],
+    current_user: Annotated[User, Depends(get_current_user)],
+) -> Ticket:
+    """Update a ticket. Validates status transitions and logs timeline."""
+    ticket = await ticket_service.update_ticket(db, ticket_id, body.model_dump(exclude_none=True), user=current_user)
+    sla = await get_sla_status(ticket)
+    ticket.sla_status = sla  # type: ignore[attr-defined]
+    return ticket
+
+
+# ── Status Transitions (convenience endpoints) ───────────────────────
+@router.post("/{ticket_id}/status", response_model=TicketOut)
+async def change_ticket_status(
+    ticket_id: int,
+    body: dict,
+    db: Annotated[AsyncSession, Depends(get_db)],
+    current_user: Annotated[User, Depends(get_current_user)],
+) -> Ticket:
+    """Change a ticket's status with optional note.
+
+    Request body:
+    ```json
+    {"status": "Logged", "note": "Verified with customer"}
+    ```
+    """
+    status_val = body.get("status")
+    if not status_val:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="status is required")
+    update_data = {"status": status_val}
+    if "note" in body:
+        update_data["note"] = body["note"]
+
+    ticket = await ticket_service.update_ticket(db, ticket_id, update_data, user=current_user)
+    sla = await get_sla_status(ticket)
+    ticket.sla_status = sla  # type: ignore[attr-defined]
+    return ticket
+
+
+# ── SLA ──────────────────────────────────────────────────────────────
+@router.get("/{ticket_id}/sla", response_model=SLAStatusOut)
+async def check_sla(
+    ticket_id: int,
+    db: Annotated[AsyncSession, Depends(get_db)],
+) -> dict:
+    """Check SLA status for a ticket — response deadline, resolution deadline, breach flags."""
+    ticket = await ticket_service.get_ticket(db, ticket_id)
+    return await get_sla_status(ticket)
+
+
+# ── Photo Uploads ────────────────────────────────────────────────────
+ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
+ALLOWED_MIME_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"}
+
+
+@router.post("/{ticket_id}/photos", response_model=list[TicketPhotoOut], status_code=status.HTTP_201_CREATED)
+async def upload_photos(
+    ticket_id: int,
+    db: Annotated[AsyncSession, Depends(get_db)],
+    current_user: Annotated[User, Depends(get_current_user)],
+    files: list[UploadFile],
+    is_before: bool = True,
+) -> list[TicketPhoto]:
+    """Upload photos for a ticket (before/after). Stores files under uploads/."""
+    # Verify ticket exists
+    await ticket_service.get_ticket(db, ticket_id)
+
+    # Validate all files before any write operations
+    for file in files:
+        if file.content_type not in ALLOWED_MIME_TYPES:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Unsupported content type '{file.content_type}'. Allowed: image/jpeg, image/png, image/gif, image/webp",
+            )
+        ext = Path(file.filename or "photo.jpg").suffix.lower() if file.filename else ".jpg"
+        if ext not in ALLOWED_EXTENSIONS:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Unsupported file extension '{ext}'. Allowed: .jpg, .jpeg, .png, .gif, .webp",
+            )
+
+    created_photos: list[TicketPhoto] = []
+    file_data: list[tuple[Path, bytes]] = []
+    for file in files:
+        # Generate unique filename
+        ext = Path(file.filename or "photo.jpg").suffix.lower() if file.filename else ".jpg"
+        unique_name = f"{uuid.uuid4().hex}{ext}"
+        file_path = UPLOADS_DIR / unique_name
+
+        content = await file.read()
+        file_data.append((file_path, content))
+
+        # Database record (no file write yet)
+        photo = TicketPhoto(
+            ticket_id=ticket_id,
+            photo_url=f"/uploads/{unique_name}",
+            is_before=is_before,
+        )
+        db.add(photo)
+        created_photos.append(photo)
+
+    await db.flush()
+    for p in created_photos:
+        await db.refresh(p)
+
+    # Write files to disk only after successful DB flush
+    for file_path, content in file_data:
+        file_path.write_bytes(content)
+
+    return created_photos
+
+
+@router.get("/{ticket_id}/photos", response_model=list[TicketPhotoOut])
+async def list_photos(
+    ticket_id: int,
+    db: Annotated[AsyncSession, Depends(get_db)],
+) -> list[TicketPhoto]:
+    """List all photos for a ticket."""
+    await ticket_service.get_ticket(db, ticket_id)
+    result = await db.execute(
+        select(TicketPhoto)
+        .where(TicketPhoto.ticket_id == ticket_id)
+        .order_by(TicketPhoto.id)
+    )
+    return list(result.scalars().all())

app/schemas/ticket.py

@@ -0,0 +1,117 @@
+"""Pydantic schemas for ticket CRUD, timeline, photos, and SLA."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from decimal import Decimal
+
+from pydantic import BaseModel, Field
+
+
+# ── Category ─────────────────────────────────────────────────────────
+class CategoryOut(BaseModel):
+    id: int
+    type: str
+    name: str
+    parent_id: int | None = None
+    sla_urgency: str | None = None
+
+    model_config = {"from_attributes": True}
+
+
+class CategoryTreeOut(CategoryOut):
+    children: list[CategoryTreeOut] = []
+
+
+# ── Ticket ───────────────────────────────────────────────────────────
+class TicketCreate(BaseModel):
+    unit_id: int | None = None
+    category_id: int | None = None
+    priority: str | None = None  # urgent, high, medium, low
+    reporter: str | None = None
+    reported_via: str | None = None  # whatsapp, phone, walk-in, qr, agent
+    description: str | None = None
+    assigned_to: int | None = None
+
+
+class TicketUpdate(BaseModel):
+    status: str | None = None
+    unit_id: int | None = None
+    category_id: int | None = None
+    priority: str | None = None
+    reporter: str | None = None
+    reported_via: str | None = None
+    description: str | None = None
+    assigned_to: int | None = None
+    eta: datetime | None = None
+    cost: Decimal | None = None
+    parts_used: str | None = None
+
+
+class TicketTimelineOut(BaseModel):
+    id: int
+    ticket_id: int
+    from_status: str | None = None
+    to_status: str | None = None
+    note: str | None = None
+    user_id: int | None = None
+    created_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class TicketPhotoOut(BaseModel):
+    id: int
+    ticket_id: int
+    photo_url: str
+    is_before: bool
+
+    model_config = {"from_attributes": True}
+
+
+class TicketBrief(BaseModel):
+    """Summary view for list endpoints."""
+    id: int
+    ticket_number: str
+    status: str
+    priority: str | None = None
+    unit_id: int | None = None
+    category_id: int | None = None
+    assigned_to: int | None = None
+    reporter: str | None = None
+    description: str | None = None
+    sla_deadline: datetime | None = None
+    reopen_count: int = 0
+    created_at: datetime
+    updated_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class TicketOut(TicketBrief):
+    """Full ticket detail with relations."""
+    closed_at: datetime | None = None
+    eta: datetime | None = None
+    cost: Decimal | None = None
+    parts_used: str | None = None
+    customer_rating: int | None = None
+    timeline: list[TicketTimelineOut] = []
+    photos: list[TicketPhotoOut] = []
+    sla_status: dict | None = None
+
+
+class TicketListResponse(BaseModel):
+    items: list[TicketBrief]
+    total: int
+    page: int
+    page_size: int
+
+
+# ── SLA ──────────────────────────────────────────────────────────────
+class SLAStatusOut(BaseModel):
+    priority: str | None = None
+    response_deadline: str | None = None
+    resolution_deadline: str | None = None
+    response_breached: bool = False
+    resolution_breached: bool = False
+    current_status: str | None = None

app/services/seed.py

@@ -1,6 +1,6 @@
-"""Seed script — users and units.
+"""Seed script — users, units, and categories.
 
-Called on first startup or via :func:`seed_all`.
+Called on first startup via lifespan hook.
 """
 
 from __future__ import annotations
@@ -15,6 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.core.security import hash_password
 from app.models.unit import Unit
 from app.models.user import User
+from app.models.category import Category
 
 logger = logging.getLogger(__name__)
 
@@ -118,3 +119,87 @@ async def seed_units(db: AsyncSession, json_path: str | Path | None = None) -> l
     else:
         logger.info("All units already exist")
     return created
+
+
+# ── Category taxonomy (PRD §7) ───────────────────────────────────────
+# Top-level categories with sub-categories
+SEED_CATEGORIES_DATA: list[dict] = [
+    # ── Maintenance ──────────────────────────────────────────────
+    {"type": "maintenance", "name": "Plumbing", "subs": ["WC not flushing", "Leaky tap", "Pipe burst", "Blocked drain", "Water heater"]},
+    {"type": "maintenance", "name": "Electrical", "subs": ["No power", "Tripped breaker", "Light fitting", "Socket repair", "Voltage issues"]},
+    {"type": "maintenance", "name": "HVAC", "subs": ["AC not cooling", "AC leaking", "Thermostat", "Fan not working"]},
+    {"type": "maintenance", "name": "Furniture", "subs": ["Broken bed", "Broken chair", "Broken table", "Wardrobe issue"]},
+    {"type": "maintenance", "name": "Appliances", "subs": ["Fridge", "Washing machine", "Microwave", "TV", "Kettle"]},
+    {"type": "maintenance", "name": "Cleaning", "subs": ["Deep clean requested", "Post-checkout turnover", "Common area cleaning"]},
+    {"type": "maintenance", "name": "Pest Control", "subs": ["Insects", "Rodents", "Fumigation"]},
+    {"type": "maintenance", "name": "Security", "subs": ["Lock broken", "Door not closing", "Window latch", "CCTV issue"]},
+    {"type": "maintenance", "name": "Internet", "subs": ["WiFi down", "Slow speed", "Router reset"]},
+    {"type": "maintenance", "name": "Structural", "subs": ["Wall crack", "Ceiling leak", "Floor tile", "Paint touch-up"]},
+    # ── Customer Service ─────────────────────────────────────────
+    {"type": "cs", "name": "Check-in", "subs": ["Early check-in", "Key handover", "Welcome instructions"]},
+    {"type": "cs", "name": "Check-out", "subs": ["Late checkout", "Key return", "Inspection"]},
+    {"type": "cs", "name": "Housekeeping", "subs": ["Mid-stay cleaning", "Linen change", "Restocking"]},
+    {"type": "cs", "name": "Lost Property", "subs": ["Guest left items behind"]},
+    {"type": "cs", "name": "Billing", "subs": ["Invoice question", "Payment issue", "Deposit query"]},
+    {"type": "cs", "name": "Staff Behaviour", "subs": ["Staff conduct feedback"]},
+    # ── Emergency ────────────────────────────────────────────────
+    {"type": "emergency", "name": "Fire", "subs": ["Smoke detected", "Fire alarm", "Sprinkler issue"], "sla_urgency": "urgent"},
+    {"type": "emergency", "name": "Flood", "subs": ["Major water leak", "Burst pipe", "Overflowing"], "sla_urgency": "urgent"},
+    {"type": "emergency", "name": "Gas Leak", "subs": ["Gas smell", "Suspected leak"], "sla_urgency": "urgent"},
+    {"type": "emergency", "name": "Electrical Hazard", "subs": ["Sparking", "Exposed wires", "Power outage multiple units"], "sla_urgency": "urgent"},
+]
+
+
+async def seed_categories(db: AsyncSession) -> list[Category]:
+    """Seed the categories table with the full PRD §7 taxonomy."""
+    created: list[Category] = []
+
+    for group in SEED_CATEGORIES_DATA:
+        # Check if parent category exists
+        result = await db.execute(
+            select(Category).where(
+                Category.type == group["type"],
+                Category.name == group["name"],
+                Category.parent_id.is_(None),
+            )
+        )
+        parent = result.scalar_one_or_none()
+        if parent is None:
+            parent = Category(
+                type=group["type"],
+                name=group["name"],
+                parent_id=None,
+                sla_urgency=group.get("sla_urgency"),
+            )
+            db.add(parent)
+            await db.flush()
+            created.append(parent)
+
+        # Seed sub-categories
+        for sub_name in group["subs"]:
+            result = await db.execute(
+                select(Category).where(
+                    Category.type == group["type"],
+                    Category.name == sub_name,
+                    Category.parent_id == parent.id,
+                )
+            )
+            existing = result.scalar_one_or_none()
+            if existing is None:
+                sub = Category(
+                    type=group["type"],
+                    name=sub_name,
+                    parent_id=parent.id,
+                    sla_urgency=None,
+                )
+                db.add(sub)
+                created.append(sub)
+
+    if created:
+        await db.flush()
+        for c in created:
+            await db.refresh(c)
+        logger.info("Seeded %d categories", len(created))
+    else:
+        logger.info("All seed categories already exist")
+    return created

app/services/sla.py

@@ -0,0 +1,100 @@
+"""SLA engine — priority-based deadlines, escalation triggers, and status checks."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+from app.models.ticket import Ticket
+
+# ── Priority SLA Targets (from PRD §11) ──────────────────────────────
+# Each entry: (response_window_minutes, resolution_window_hours)
+SLA_TARGETS: dict[str, tuple[int, int]] = {
+    "urgent": (15, 4),       # respond 10-15min, resolve 2-4h → use upper bound
+    "high": (30, 24),        # respond 30min, resolve 8-24h
+    "medium": (240, 72),     # respond 4h (240min), resolve 2-3d (72h)
+    "low": (1440, 168),      # respond 1d (1440min), resolve 5-7d (168h)
+}
+
+
+def compute_sla_deadline(priority: str, created_at: datetime | None = None) -> datetime:
+    """Return the resolution deadline for a given priority.
+
+    Uses the upper-bound resolution target.
+    """
+    now = created_at or datetime.now(timezone.utc)
+    _response_min, resolution_hours = SLA_TARGETS.get(priority, (240, 72))
+    return now + timedelta(hours=resolution_hours)
+
+
+def get_sla_response_window(priority: str) -> timedelta:
+    """Return the response-window timedelta for a given priority."""
+    minutes, _hours = SLA_TARGETS.get(priority, (240, 72))
+    return timedelta(minutes=minutes)
+
+
+def get_sla_resolution_window(priority: str) -> timedelta:
+    """Return the resolution-window timedelta for a given priority."""
+    _minutes, hours = SLA_TARGETS.get(priority, (240, 72))
+    return timedelta(hours=hours)
+
+
+# ── Escalation detection ─────────────────────────────────────────────
+def should_escalate_on_response(ticket: Ticket) -> bool:
+    """Return True if the ticket's response SLA has been breached.
+
+    A ticket is in breach if its status is still in the pre-acknowledgement
+    states (New, Logged, Triage, Assigned) beyond the response window.
+    """
+    if not ticket.priority or ticket.created_at is None:
+        return False
+    pre_ack_states = {"New", "Logged", "Triage", "Assigned"}
+    if ticket.status not in pre_ack_states:
+        return False
+    response_window = get_sla_response_window(ticket.priority)
+    deadline = ticket.created_at.replace(tzinfo=timezone.utc) + response_window
+    return datetime.now(timezone.utc) > deadline
+
+
+def is_sla_breached(ticket: Ticket) -> bool:
+    """Return True if the ticket's resolution SLA deadline has passed."""
+    if ticket.sla_deadline is None or ticket.status in ("Closed", "Completed"):
+        return False
+    deadline = ticket.sla_deadline
+    if deadline.tzinfo is None:
+        deadline = deadline.replace(tzinfo=timezone.utc)
+    return datetime.now(timezone.utc) > deadline
+
+
+# ── API helper ───────────────────────────────────────────────────────
+async def get_sla_status(ticket: Ticket) -> dict:
+    """Return a structured SLA status dict for a ticket."""
+    now = datetime.now(timezone.utc)
+    created = ticket.created_at
+    if created and created.tzinfo is None:
+        created = created.replace(tzinfo=timezone.utc)
+
+    response_deadline = None
+    resolution_deadline = ticket.sla_deadline
+
+    if ticket.priority and created:
+        response_deadline = created + get_sla_response_window(ticket.priority)
+
+    response_breached = False
+    if response_deadline and ticket.status in {"New", "Logged", "Triage", "Assigned"}:
+        response_breached = now > response_deadline
+
+    resolution_breached = False
+    if resolution_deadline:
+        rd = resolution_deadline
+        if rd.tzinfo is None:
+            rd = rd.replace(tzinfo=timezone.utc)
+        resolution_breached = now > rd if ticket.status not in ("Closed", "Completed") else False
+
+    return {
+        "priority": ticket.priority,
+        "response_deadline": response_deadline.isoformat() if response_deadline else None,
+        "resolution_deadline": resolution_deadline.isoformat() if resolution_deadline else None,
+        "response_breached": response_breached,
+        "resolution_breached": resolution_breached,
+        "current_status": ticket.status,
+    }

app/services/ticket.py

@@ -0,0 +1,292 @@
+"""Ticket business logic — CRUD, status transitions, SLA enforcement."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+from fastapi import HTTPException, status
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from app.models.ticket import Escalation, Ticket, TicketTimeline
+from app.models.unit import Unit
+from app.models.user import User
+from app.services.sla import compute_sla_deadline
+
+# ── Status Transition Map ────────────────────────────────────────────
+# Keys: current status → list of valid next statuses
+VALID_TRANSITIONS: dict[str, list[str]] = {
+    "New": ["Logged"],
+    "Logged": ["Triage", "Closed"],
+    "Triage": ["Assigned", "Escalated"],
+    "Assigned": ["Accepted", "Triage"],
+    "Accepted": ["Travelling", "Triage"],
+    "Travelling": ["On Site", "Triage"],
+    "On Site": ["In Progress", "Triage"],
+    "In Progress": ["Waiting Parts", "Escalated", "Completed"],
+    "Waiting Parts": ["In Progress", "Escalated"],
+    "Escalated": ["Triage", "In Progress", "Completed", "Closed"],
+    "Completed": ["On-Field Verification", "In Progress"],
+    "On-Field Verification": ["Wahab Review", "Completed", "Closed"],
+    "Wahab Review": ["Closed", "On-Field Verification"],
+    "Closed": ["Reopened"],
+    "Reopened": ["Triage", "Logged"],
+}
+
+REOPEN_WINDOW_DAYS = 7
+SLA_ACK_USER = "Ama"
+
+
+# ── Helpers ──────────────────────────────────────────────────────────
+async def _generate_ticket_number(db: AsyncSession) -> str:
+    """Generate the next ticket number in PAV-YYYY-NNNNN format."""
+    year = datetime.now(timezone.utc).year
+    prefix = f"PAV-{year}-"
+    # Count existing tickets this year
+    result = await db.execute(
+        select(func.count(Ticket.id)).where(Ticket.ticket_number.like(f"{prefix}%"))
+    )
+    count = result.scalar() or 0
+    return f"{prefix}{count + 1:05d}"
+
+
+async def _log_status_change(
+    db: AsyncSession,
+    ticket_id: int,
+    from_status: str | None,
+    to_status: str | None,
+    note: str | None = None,
+    user_id: int | None = None,
+) -> TicketTimeline:
+    """Create a timeline entry for a status change."""
+    entry = TicketTimeline(
+        ticket_id=ticket_id,
+        from_status=from_status,
+        to_status=to_status,
+        note=note,
+        user_id=user_id,
+    )
+    db.add(entry)
+    await db.flush()
+    return entry
+
+
+async def _get_ticket_or_404(db: AsyncSession, ticket_id: int) -> Ticket:
+    """Fetch a ticket by ID or raise 404."""
+    result = await db.execute(
+        select(Ticket)
+        .options(
+            selectinload(Ticket.timeline),
+            selectinload(Ticket.photos),
+            selectinload(Ticket.assigned_technician),
+            selectinload(Ticket.unit),
+            selectinload(Ticket.category),
+        )
+        .where(Ticket.id == ticket_id)
+    )
+    ticket = result.scalar_one_or_none()
+    if ticket is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Ticket not found")
+    return ticket
+
+
+# ── CRUD ─────────────────────────────────────────────────────────────
+async def create_ticket(
+    db: AsyncSession,
+    data: dict[str, Any],
+    user: User | None = None,
+) -> Ticket:
+    """Create a new ticket with auto-numbering and SLA deadline."""
+    ticket_number = await _generate_ticket_number(db)
+    priority = data.get("priority")
+    sla_deadline = compute_sla_deadline(priority) if priority else None
+
+    ticket = Ticket(
+        ticket_number=ticket_number,
+        status="New",
+        unit_id=data.get("unit_id"),
+        category_id=data.get("category_id"),
+        priority=priority,
+        reporter=data.get("reporter"),
+        reported_via=data.get("reported_via"),
+        description=data.get("description"),
+        assigned_to=data.get("assigned_to"),
+        sla_deadline=sla_deadline,
+    )
+    db.add(ticket)
+    await db.flush()
+    await db.refresh(ticket)
+
+    # Log initial creation
+    await _log_status_change(
+        db,
+        ticket.id,
+        from_status=None,
+        to_status="New",
+        note="Ticket created",
+        user_id=user.id if user else None,
+    )
+
+    # Auto-advance New → Logged (creation is also the logging step)
+    ticket.status = "Logged"
+    await _log_status_change(
+        db,
+        ticket.id,
+        from_status="New",
+        to_status="Logged",
+        note="Ticket logged and numbered",
+        user_id=user.id if user else None,
+    )
+
+    await db.flush()
+    await db.refresh(ticket)
+    return ticket
+
+
+async def get_ticket(db: AsyncSession, ticket_id: int) -> Ticket:
+    """Get a single ticket with full details."""
+    return await _get_ticket_or_404(db, ticket_id)
+
+
+async def list_tickets(
+    db: AsyncSession,
+    *,
+    status_filter: str | None = None,
+    priority_filter: str | None = None,
+    property_filter: str | None = None,
+    category_id: int | None = None,
+    assigned_to: int | None = None,
+    date_from: datetime | None = None,
+    date_to: datetime | None = None,
+    page: int = 1,
+    page_size: int = 50,
+) -> tuple[list[Ticket], int]:
+    """List tickets with optional filters. Returns (tickets, total_count)."""
+    query = select(Ticket)
+    count_query = select(func.count(Ticket.id))
+
+    if status_filter:
+        query = query.where(Ticket.status == status_filter)
+        count_query = count_query.where(Ticket.status == status_filter)
+    if priority_filter:
+        query = query.where(Ticket.priority == priority_filter)
+        count_query = count_query.where(Ticket.priority == priority_filter)
+    if property_filter:
+        # Join unit to filter by property
+        query = query.join(Ticket.unit).where(Unit.property == property_filter)
+        count_query = count_query.join(Ticket.unit).where(Unit.property == property_filter)
+    if category_id:
+        query = query.where(Ticket.category_id == category_id)
+        count_query = count_query.where(Ticket.category_id == category_id)
+    if assigned_to:
+        query = query.where(Ticket.assigned_to == assigned_to)
+        count_query = count_query.where(Ticket.assigned_to == assigned_to)
+    if date_from:
+        query = query.where(Ticket.created_at >= date_from)
+        count_query = count_query.where(Ticket.created_at >= date_from)
+    if date_to:
+        query = query.where(Ticket.created_at <= date_to)
+        count_query = count_query.where(Ticket.created_at <= date_to)
+
+    # Count total
+    total_result = await db.execute(count_query)
+    total = total_result.scalar() or 0
+
+    # Paginate
+    offset = (page - 1) * page_size
+    query = query.order_by(Ticket.created_at.desc()).offset(offset).limit(page_size)
+
+    result = await db.execute(query)
+    tickets = list(result.scalars().all())
+    return tickets, total
+
+
+async def update_ticket(
+    db: AsyncSession,
+    ticket_id: int,
+    data: dict[str, Any],
+    user: User | None = None,
+) -> Ticket:
+    """Update a ticket. Status changes are validated and logged."""
+    ticket = await _get_ticket_or_404(db, ticket_id)
+
+    # Handle status transitions separately
+    new_status = data.get("status")
+    if new_status is not None:
+        old_status = ticket.status
+        if old_status != new_status:
+            valid_targets = VALID_TRANSITIONS.get(old_status, [])
+            if new_status not in valid_targets:
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail=f"Invalid status transition: {old_status} → {new_status}. "
+                    f"Valid targets: {valid_targets}",
+                )
+
+            # Handle special logic for reopening
+            if new_status == "Reopened":
+                if old_status != "Closed":
+                    raise HTTPException(
+                        status_code=status.HTTP_400_BAD_REQUEST,
+                        detail="Can only reopen a Closed ticket",
+                    )
+                if ticket.closed_at:
+                    closed_at = ticket.closed_at
+                    if closed_at.tzinfo is None:
+                        closed_at = closed_at.replace(tzinfo=timezone.utc)
+                    days_since_close = (datetime.now(timezone.utc) - closed_at).days
+                    if days_since_close > REOPEN_WINDOW_DAYS:
+                        raise HTTPException(
+                            status_code=status.HTTP_400_BAD_REQUEST,
+                            detail=f"Cannot reopen: ticket closed {days_since_close} days ago "
+                            f"(max {REOPEN_WINDOW_DAYS} days)",
+                        )
+
+                # Auto-escalate to Ama
+                ama_result = await db.execute(
+                    select(User).where(User.full_name == SLA_ACK_USER)
+                )
+                ama_user = ama_result.scalar_one_or_none()
+                ticket.reopen_count += 1
+                ticket.closed_at = None
+                if ama_user:
+                    escalation = Escalation(
+                        ticket_id=ticket.id,
+                        escalated_to=ama_user.id,
+                        reason=f"Auto-escalation: ticket reopened (reopen #{ticket.reopen_count})",
+                    )
+                    db.add(escalation)
+
+            # Handle closing
+            if new_status == "Closed":
+                ticket.closed_at = datetime.now(timezone.utc)
+
+            # Log the transition
+            await _log_status_change(
+                db,
+                ticket.id,
+                from_status=old_status,
+                to_status=new_status,
+                note=data.get("note"),
+                user_id=user.id if user else None,
+            )
+
+            ticket.status = new_status
+
+    # Update other fields
+    for field in ("unit_id", "category_id", "priority", "reporter", "reported_via",
+                  "description", "assigned_to", "eta", "cost", "parts_used"):
+        if field in data:
+            setattr(ticket, field, data[field])
+
+    # Recompute SLA if priority changed
+    if data.get("priority"):
+        ticket.sla_deadline = compute_sla_deadline(data["priority"])
+
+    await db.flush()
+    await db.refresh(ticket)
+    return ticket
+
+