Merge pull request 'feat(dashboard): real technician names + technician performance dashboard (relay #748 v3)' (#16) from fm/denya-wahab-tech-perf-20260909 into main

Wahab customer request (relay #748 v3 next-wave).

3a. Fix 'Tech #N' display:
- FM dashboard's "Technician Workload" card was built from
  `Tech #${t.assigned_to}`; it now reads Ticket.assigned_technician_name
  (falling back to 'Unassigned', matching /tickets).
- Template sweep confirms no other `Tech #` placeholder exists.

3b. Technician performance dashboard:
- New bearer-gated GET /api/tickets/tech-performance aggregating existing
  ticket columns only (no schema change): per-technician total assigned,
  completed, in progress, escalated, cancelled, pending, open tasks,
  completion rate, and created_at -> closed_at resolution time with explicit
  counts for finished tasks lacking a timestamp. Rows key on user id so
  same-name technicians stay separate; labels are real names. Fleet totals
  and an unassigned-ticket count are included.
- New /dashboard/tech-performance page (FM + CEO nav and links) with sortable
  table, completion bars and empty states; same-origin vendored assets only.
- Bucket map, service, schemas and tests in app/services/ticket.py,
  app/schemas/ticket.py, tests/test_tech_performance.py.

Tests: 111 passed (101 existing + 10 new).

.env.example

@@ -0,0 +1,33 @@
+# Denya OneCare — runtime environment template (HARDENING.md P0.1/P1.1)
+# Copy to .env and fill in real values. NEVER commit .env.
+# Generate the secret with: openssl rand -hex 32
+
+# ── Required ─────────────────────────────────────────────
+SECRET_KEY=
+DATABASE_URL=sqlite+aiosqlite:///./data/denya_onecare.db
+# Explicit origin allow-list — "*" is rejected at startup (P0.2)
+CORS_ORIGINS=http://localhost:8000
+
+# ── Optional (WhatsApp; needed before wiring Meta) ───────
+WHATSAPP_PHONE_NUMBER_ID=
+WHATSAPP_ACCESS_TOKEN=
+WHATSAPP_VERIFY_TOKEN=
+META_GRAPH_BASE=https://graph.facebook.com/v18.0
+
+# ── Webhook auth (P0) ──────────────────────────────────
+# Shared secret for inbound WhatsApp webhook POSTs (X-Webhook-Secret header).
+# FAIL-CLOSED: when unset/empty, every webhook message is rejected (403).
+# Generate with: openssl rand -hex 32
+WHATSAPP_WEBHOOK_SECRET=
+
+# ── WhatsApp demo path (optional) ───────────────────────
+# Expected sender/recipient number (E.164) for the WhatsApp demo round trip
+# (webhook -> ticket -> mock-log). Set the real number ONLY on the deploy
+# host's .env — keep this template an empty placeholder, never a live number.
+# Empty (default): the demo payload builder fails closed (no fabricated sender).
+WHATSAPP_DEMO_TO=
+
+# ── Login rate limiting (P0) ────────────────────────────
+# ~5 failed login attempts per 15 minutes per IP+email → HTTP 429
+LOGIN_RATE_LIMIT_MAX_ATTEMPTS=5
+LOGIN_RATE_LIMIT_WINDOW_SECONDS=900

AGENTS.md

@@ -47,14 +47,40 @@ Users, units, and categories are auto-seeded on first startup via lifespan hook:
 | Method | Path | Auth | Description |
 |--------|------|------|-------------|
 | GET | `/health` | No | Health check |
-| POST | `/api/auth/register` | No | Create user |
+| POST | `/api/auth/login` | No | Get JWT tokens (rate-limited ~5 fails/15min/IP+email → 429) |
-| POST | `/api/auth/login` | No | Get JWT tokens |
 | POST | `/api/auth/refresh` | Token | Refresh tokens |
 | GET | `/api/auth/me` | Bearer | Current user |
 | GET | `/api/auth/admin-only` | Admin/Jerome, Admin/Wahab | RBAC demo endpoint |
 | GET | `/api/auth/users` | Bearer | List users (id, name, role) for the assign-technician picker |
-| POST | `/api/whatsapp/mock` | No | Mock WhatsApp |
+| POST | `/api/auth/users` | Admin/Jerome, Admin/Wahab | Admin creates a user (forced canonical role; unknown roles → 422) |
-| GET | `/api/whatsapp/mock-log` | No | Recent mock submissions |
+| PATCH | `/api/auth/users/{id}` | Admin/Jerome, Admin/Wahab | Role change / deactivate (self-modification → 400) |
+| DELETE | `/api/auth/users/{id}` | Admin/Jerome, Admin/Wahab | Delete user (409 if referenced by tickets/timeline/escalations) |
+
+**Self-registration is removed** — `POST /api/auth/register` 404s and there is no
+sign-up UI; users are created/managed by admins only (P0 hardening batch).
+
+### WhatsApp
+| Method | Path | Auth | Description |
+|--------|------|------|-------------|
+| GET | `/api/whatsapp/webhook` | No | Meta handshake (`hub.verify_token`, constant-time; mismatch → 403) |
+| POST | `/api/whatsapp/webhook` | `X-Webhook-Secret` header | Inbound message → ticket + log. Fail-closed: 403 when `WHATSAPP_WEBHOOK_SECRET` is unset or the header doesn't match |
+| GET | `/api/whatsapp/mock-log` | Bearer | Recent webhook submissions (debug; auth required) |
+
+`whatsapp_log` predates the real Meta webhook (the model gained
+`message_text`/`wa_message_id`/`ticket_number` and dropped `command` without a
+migration), so legacy tables keep the old `(command, …)` shape and every ORM
+read/write 500s. `ensure_legacy_schema` (app/main.py) adds the missing columns
+and backfills+drops the obsolete NOT NULL `command` column idempotently at
+startup — do not hand-edit legacy DBs, ship a self-heal there instead.
+
+Demo WhatsApp round trip: `WHATSAPP_DEMO_TO` (E.164, .env-only — never commit
+a real number; `.env.example` keeps an empty placeholder) is the expected
+sender/recipient for the demo path.
+`app/routers/whatsapp.py::build_demo_webhook_payload` builds a Meta webhook
+payload from it (fails closed when unset), so posting it to
+`POST /api/whatsapp/webhook` with the secret logs `from_number` = demo number
+(visible via `GET /api/whatsapp/mock-log`) and the auto-reply targets the same
+number. Covered by `tests/test_whatsapp_demo_number.py`.
 
 ### Pages (Sprint 3) — Jinja2 templates at `app/templates/`
 | Method | Path | Auth | Description |
@@ -63,17 +89,65 @@ Users, units, and categories are auto-seeded on first startup via lifespan hook:
 | GET | `/dashboard/cs` | Client | CS dashboard |
 | GET | `/dashboard/fm` | Client | FM dashboard |
 | GET | `/dashboard/ceo` | Client | CEO dashboard |
+| GET | `/dashboard/tech-performance` | Client | Technician performance report (FM + CEO nav) |
 | GET | `/tickets` | Client | All Issues filterable table |
 | GET | `/tickets/new` | Client | Create Issue form |
 | GET | `/tickets/{id}` | Client | Issue detail with timeline |
 
-Frontend: Alpine.js (CDN) + Tailwind CSS (CDN). Auth state in localStorage. Role-based nav routing in `base.html`.
+Frontend: Alpine.js + Tailwind CSS vendored same-origin (no CDN) — see
+"Frontend assets" below. Auth state in localStorage. Role-based nav routing in `base.html`.
+
+### Technician performance (Wahab request)
+`GET /api/tickets/tech-performance` (Bearer) aggregates per-technician workload/outcomes by
+**real name** using only existing ticket columns (`assigned_to`/`status`/`created_at`/`closed_at`
+→ `users.full_name`) — no schema change. Aggregation lives in
+`app/services/ticket.py::get_technician_performance`; bucket map `_TECH_STATUS_BUCKETS` defines
+`completed` (Completed/Closed), `in_progress`, `escalated`, `cancelled`, and `pending` (remainder),
+so the buckets always sum to `total_assigned`. `completion_rate = completed/total_assigned`;
+`avg_resolution_hours` averages `created_at → closed_at` only where `closed_at` is set, with
+`resolved_without_timestamps` counting finished tasks that lack one. Rows key on user id (same-name
+techs stay separate), sorted completed desc → workload → name. UI:
+`app/templates/dashboard/tech-performance.html`, linked from FM + CEO nav and the FM
+"Technician Workload" card — that card reads `Ticket.assigned_technician_name`, never an id
+placeholder. Regression: `tests/test_tech_performance.py`.
+
+### Frontend assets (vendored, LAN-safe)
+- Alpine.js 3.17.2 + Tailwind Play 3.4.17 are committed under `app/static/vendor/`
+  and served at `/static/vendor/…` (mounted in `app/main.py`, versioned
+  filenames → immutable cache `public, max-age=31536000, immutable`). Templates
+  must never reference a CDN; update `app/templates/base.html` when upgrading:
+  download `alpinejs@<ver>/dist/cdn.min.js` (jsDelivr) and the tailwind play
+  script (`cdn.tailwindcss.com/<ver>`), save them under `app/static/vendor/`
+  mirroring the committed names (Alpine keeps `.min.js`, e.g.
+  `alpine-3.17.2.min.js`; the tailwind play file does not, e.g.
+  `tailwind-3.4.17.js`), then bump the `<script src>` + the
+  `VENDORED_SCRIPTS` tuple in the regression file `tests/test_frontend_vendoring.py`.
+- HTML pages ship `Cache-Control: no-cache` and CSP allows no external host
+  (`script-src 'self' 'unsafe-inline' 'unsafe-eval'`, `style-src 'self'
+  'unsafe-inline'`, `connect-src 'self'`); no CDN host is allowed in CSP
+  (`app/main.py::SecurityHeadersMiddleware`). `'unsafe-eval'` is required by
+  the Alpine 3.17.2 CDN build: its evaluator compiles every `x-*` expression
+  with `new Function()`, and without it CSP blocks Alpine entirely (stuck
+  loading overlay on every page) — covered by
+  `test_csp_script_src_allows_unsafe_eval_for_alpine` in
+  `tests/test_frontend_vendoring.py`.
+
+### Branding (logo)
+Official Denya Developers logo derivatives are committed under
+`app/static/branding/` (Gitea release `logo-assets-v1`, sha256-verified
+monochrome forest-green lockup; sourced from the Gitea release, never from
+kagentz). Placement: login header uses `denya-logo-h96.png` (full lockup); the
+logged-in topbar (base.html nav) uses `denya-logo-h48.png` on a light chip
+(logo ink is only ~2:1 against the dark `#0d2b18` nav — keep a light chip
+there); favicons 32x32+16x16 declared in `base.html <head>`.
+`img-src 'self' data: blob:` already covers `/static/branding/*` — no CSP
+change. Regression coverage: `tests/test_branding_assets.py`.
 
 ### 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, building, unit_id, category_id, assigned_to, date_from, date_to) |
+| GET | `/api/tickets` | No | List tickets (filter: status, priority, property, building, unit_id, category_id, assigned_to, date_from, date_to; paginate with `page`/`page_size` or `limit` alias — hard cap 200, both given → 422) |
 | GET | `/api/tickets/{id}` | No | Get ticket detail with timeline, photos, SLA status, nested unit/category, phone |
 | GET | `/api/tickets/{id}/transitions` | No | Valid next statuses for the ticket's current status (drives the detail-page status picker) |
 | PATCH | `/api/tickets/{id}` | Bearer | Update ticket (validates status transitions; assigning a technician auto-advances New/Logged/Triage to Assigned) |
@@ -88,9 +162,26 @@ Frontend: Alpine.js (CDN) + Tailwind CSS (CDN). Auth state in localStorage. Role
 ## Auth
 
 - JWT access (30min) + refresh (7d) tokens
-- Roles: CS Rep, CS Manager, FM Dispatcher, Admin/Jerome, Admin/Wahab, Tech, CEO, Director
+- **Unified role model** lives in `app/core/roles.py` (`CANONICAL_ROLES`,
-- Use `require_roles("Admin/Jerome", "Admin/Wahab")` dependency for RBAC
+  `ADMIN_ROLES` = Admin/Jerome + Admin/Wahab, `ROLE_ALIASES` for legacy
-- `sub` claim holds string user ID
+  nickname roles like `technician`/`cs`/`fm`/`ceo`). Canonical stored roles:
+  CS Rep, CS Manager, FM Dispatcher, Admin/Jerome, Admin/Wahab, Tech, CEO, Director.
+- Role checks, admin user creation, login, and JWT validation all derive from the
+  role module; unknown/junk roles (e.g. lowercase `admin`/`superadmin` from the
+  old open register) fail closed at login/JWT and can never be recreated via the
+  API (422). Startup self-heals (lifespan in `app/main.py`, helpers in
+  `app/services/seed.py`) converge legacy rows: `normalize_legacy_user_roles`
+  maps unambiguous alias nicknames onto canonical roles (space and underscore
+  forms alike — `cs rep`/`cs_rep` → `CS Rep`, `fm dispatcher`/`fm_dispatcher`,
+  `cs manager`/`cs_manager`), and
+  `normalize_legacy_user_emails` lowercases stored emails — login and the admin
+  create-user duplicate check both compare on the lowercased form, so pre-P0
+  mixed-case emails are never silently locked out.
+- Use `require_roles(*ADMIN_ROLES)` for admin gates; `sub` claim holds string user ID
+- Security headers middleware in `app/main.py`: X-Frame-Options DENY +
+  nosniff on everything, CSP on HTML pages, HSTS when `X-Forwarded-Proto: https`
+  (CSP allows no external host — frontend libs are vendored, see "Frontend
+  assets"; `script-src` carries `'unsafe-eval'` for the Alpine runtime)
 
 ## Ticket System (Sprint 2)
 

CLAUDE.md

@@ -1 +0,0 @@
-AGENTS.md

CLAUDE.md

@@ -0,0 +1,2 @@
+<!-- Points Claude at AGENTS.md via import; edit AGENTS.md, not this file. -->
+@AGENTS.md

HARDENING.md

@@ -16,18 +16,21 @@
 - **Working & verified:** auth (JWT 30m/7d, bcrypt, RBAC via `require_roles`), ticket
   CRUD with 16-status `VALID_TRANSITIONS` state machine, SLA engine, photo uploads,
   category/unit hierarchy, 3 role dashboards (CS/FM/CEO), Alembic migrations with
-  legacy-schema self-heal. **32 pytest tests pass.**
+  legacy-schema self-heal. **pytest suite passes.**
 - **Live:** container `denya-onecare` on LXC `scottdenya` (192.168.68.75:8000),
   image built 2026-08-02, `restart: unless-stopped`.
-- **Known demo-only posture (must change):** `SECRET_KEY=change-me-in-production`,
+- **Demo-only posture resolved (PR #10 + P0 batch):** `SECRET_KEY` now fails
-  `CORS_ORIGINS=*`, open `/api/auth/register`, SQLite backend, WhatsApp webhook
+  closed without a real key, CORS is an explicit origin allow-list, and open
+  self-registration was removed — `POST /api/auth/register` → 404, users are
+  admin-managed only (see AGENTS.md).
+- **Known demo-only posture (must change):** SQLite backend; WhatsApp webhook
   code is present but **no real credentials wired**.
 
 ---
 
 ## 1. P0 — Security blockers (do these FIRST, before any real data)
 
-### P0.1 Hardcode-safe secrets; never ship the default key
+### P0.1 Hardcode-safe secrets; never ship the default key — **DONE (PR #10)**
 - **Files:** `docker-compose.yml`, `app/core/config.py`
 - Replace the hardcoded `SECRET_KEY=change-me-in-production` default with a
   fail-closed default: if `SECRET_KEY` is unset/empty or still the well-known
@@ -37,7 +40,7 @@
 - **Acceptance:** starting the app without a real key fails loudly; container env
   contains a strong random key (≥32 bytes, e.g. `openssl rand -hex 32`).
 
-### P0.2 Lock down CORS
+### P0.2 Lock down CORS — **DONE (PR #10)**
 - **Files:** `app/main.py`, `docker-compose.yml`
 - `CORS_ORIGINS=*` + `allow_credentials=True` is an invalid/unsafe combo
   (browsers reject `*` with credentials anyway). Replace with an explicit
@@ -47,18 +50,21 @@
 - **Acceptance:** `settings.CORS_ORIGINS` is a comma-separated explicit list; the
   middleware builds an allow-list, not `["*"]`.
 
-### P0.3 Gate user registration
+### P0.3 Gate user registration — **DONE (P0 batch, 2026-09)**
-- **File:** `app/routers/auth.py` (`POST /api/auth/register`)
+- Open `POST /api/auth/register` was removed entirely (404) — there is no sign-up UI.
-- Today anyone on the network can self-register. Decide the model:
+- Users are admin-managed: `POST /api/auth/users` (Admin/Jerome + Admin/Wahab only)
-  - **Recommended:** require an admin-issued invitation token, or restrict
+  creates users with a **forced canonical role** (unknown roles → 422);
-    registration to a seed/allowed list, or remove the open route and create
+  `PATCH /api/auth/users/{id}` changes role / deactivates (self-modification → 400);
-    users only via seed/admin.
+  `DELETE /api/auth/users/{id}` is guarded (users referenced by
-  - If a public self-service resident/tenant signup is genuinely required
+  tickets/timeline/escalations → 409); duplicate email → 409.
-    (Phase 2 QR/self-service), it must be a SEPARATE endpoint with a **role
+- Emails are normalized (strip + lowercase) on every write path; a startup
-    default of the least-privilege role** and rate-limiting — never able to mint
+  self-heal lowercases legacy mixed-case rows so pre-P0 accounts can't be locked
-    admin/FM roles.
+  out of login. Unified role model lives in `app/core/roles.py`.
-- **Acceptance:** a raw, unauthenticated register call can no longer mint an
+- **Regression tests:** `tests/test_p0_auth_admin_batch.py`.
-  `Admin/*` or `Director` account.
+- If a public self-service resident/tenant signup is genuinely required later
+  (Phase 2 QR/self-service), it must be a SEPARATE endpoint with a **role
+  default of the least-privilege role** and rate-limiting — never able to mint
+  admin/FM roles.
 
 ### P0.4 Reconsider SQLite for the final product
 - **Files:** `docker-compose.yml`, `app/core/database.py`, `app/core/config.py`, PRD §16
@@ -71,21 +77,21 @@
 - **Acceptance:** `pytest` green against Postgres (tests param via conftest),
   Alembic applies cleanly on a fresh Postgres DB.
 
-### P0.5 WhatsApp webhook auth + hardening (finish wiring, then lock it)
+### P0.5 WhatsApp webhook auth + hardening — **PARTIAL (P0 batch, 2026-09)**
 - **File:** `app/routers/whatsapp.py`
-- The handler exists but no credentials are set. When wiring this week:
+- **Done:** the `GET` handshake validates `hub.verify_token` with a constant-time
-  - Verify the `hub.verify_token` check is constant-time (compare with
+  compare and returns **403 on mismatch**; inbound `POST`s are gated by the
-    `secrets.compare_digest`). **The GET verification path currently returns
+  `X-Webhook-Secret` header matching `WHATSAPP_WEBHOOK_SECRET` (fail-closed 403
-    `{"error": ...}` with HTTP 200** — flip to `403` on token mismatch.
+  when the env var is unset — see `.env.example`); the debug
-  - Validate **inbound messages only from Meta** — the webhook MUST authenticate
+  `GET /api/whatsapp/mock-log` now requires Bearer auth.
-    Meta's request signature (X-Hub-Signature-256 HMAC over the raw body with your
+- **Still open:** Meta request-signature validation (`X-Hub-Signature-256` HMAC
-    app secret) before processing, otherwise anyone who discovers the endpoint can
+  over the raw body with the app secret — the shared-secret header above is the
-    forge tickets. This is the single most important WhatsApp hardening item.
+  interim gate); per-sender rate limiting / dedupe idempotency keyed on
-  - Add per-sender rate limiting / dedupe on `wa_message_id` (webhook retries can
+  `wa_message_id` (retries can still double-create tickets); redacting the raw
-    double-create tickets). Create an idempotency guard keyed on `wa_message_id`.
+  access token in `send_whatsapp_reply` error paths.
-  - Never log the raw access token; redact in `send_whatsapp_reply` error paths.
+- **Acceptance (open items):** a forged POST without the Meta signature is
-- **Acceptance:** a forged POST without the Meta signature is rejected; duplicate
+  rejected; duplicate `wa_message_id` does not create a second ticket;
-  `wa_message_id` does not create a second ticket; verify-token mismatch returns 403.
+  verify-token mismatch returns 403 (done).
 
 ---
 
@@ -130,10 +136,13 @@
   `X-Content-Type-Options: nosniff`.
 
 ### P1.6 API hardening & rate limiting
-- Add rate limiting on `POST /api/auth/login` (brute-force) — per-IP/IP+account.
+- **Done (P0 batch):** `POST /api/auth/login` is rate-limited in-process —
-- Consider rate limits on ticket creation (spam / mass-creation).
+  ~5 failures / 15 min per IP+email → 429 (env-tunable `LOGIN_RATE_LIMIT_*`,
-- Normalize/validate `page_size` (already capped `le=200`) and pagination
+  a successful login resets the window). **Open:** ticket-creation rate limiting
-  tie-breaker (`id DESC` present — good).
+  (spam / mass-creation).
+- **Done (P0 batch):** ticket-list pagination — `page`/`page_size` (default 50,
+  cap 200), `limit` alias for `page_size` also capped; passing both with
+  different values → 422. Tie-breaker `id DESC` present.
 
 ---
 
@@ -145,9 +154,9 @@ Assumes Denya provides: **phone number ID, access token, verify token, app secre
    `WHATSAPP_VERIFY_TOKEN`, `WHATSAPP_APP_SECRET`, `META_GRAPH_BASE`) to `.env`
    (git-ignored) and inject at runtime. Never commit.
 2. **Webhook handshake:** in Meta dashboard point the webhook URL at
-   `<domain>/api/whatsapp/webhook`. The GET verify path currently echoes
+   `<domain>/api/whatsapp/webhook`. The GET verify path echoes `hub.challenge`
-   `hub.challenge` when the verify token matches — confirm this works, then apply
+   when the verify token matches (constant-time compare; mismatch → 403). The
-   P0.5 (403 on mismatch, HMAC signature validation).
+   still-open P0.5 work is the Meta signature validation in step 3.
 3. **Verify incoming signature** (P0.5) — use `X-Hub-Signature-256` = HMAC-SHA256
    of the raw body with your app secret, compared with `compare_digest`.
 4. **Reply flow:** confirm `send_whatsapp_reply` posts correctly to
@@ -194,14 +203,18 @@ Assumes Denya provides: **phone number ID, access token, verify token, app secre
 
 ## 5. Definition of Done (production-ready)
 
-- [ ] No default `SECRET_KEY`; app fails closed without a real key
+- [x] No default `SECRET_KEY`; app fails closed without a real key (PR #10)
-- [ ] CORS is an explicit origin allow-list
+- [x] CORS is an explicit origin allow-list (PR #10)
-- [ ] Self-registration cannot mint privileged roles (or is admin-gated/removed)
+- [x] Self-registration removed (register → 404); users are admin-managed only
+      with forced canonical roles (P0 batch)
 - [ ] Postgres backend; Alembic applies cleanly on fresh DB; nightly backups
 - [ ] TLS-terminated reverse proxy with real domain; no raw :8000 on WAN
-- [ ] WhatsApp webhook: Meta signature validated, verify-token mismatch → 403,
+- [x] Webhook POST gated by `X-Webhook-Secret` (fail-closed); verify-token
-      idempotent on `wa_message_id`, real credentials injected at runtime
+      mismatch → 403; `mock-log` requires auth
-- [ ] Login/ticket rate limiting in place
+- [ ] WhatsApp webhook: Meta `X-Hub-Signature-256` HMAC validated; idempotent on
+      `wa_message_id`; real credentials injected at runtime
+- [x] Login rate limiting in place (~5 fails / 15 min per IP+email → 429)
+- [ ] Ticket-creation rate limiting in place
 - [ ] Photo uploads size-limited and content-sniffed
 - [ ] RBAC audited per-route; phone data access controlled
 - [ ] Expanded test suite (auth, WhatsApp, SLA boundary, uploads) — all green

app/core/config.py

@@ -23,7 +23,7 @@ class Settings(BaseSettings):
     DATABASE_URL: str = "sqlite+aiosqlite:///./denya_onecare.db"
 
     # ── Auth ─────────────────────────────────────────────────────────
-    SECRET_KEY: str = "change-me-in-production-use-a-real-secret"
+    SECRET_KEY: str = ""
     ALGORITHM: str = "HS256"
     ACCESS_TOKEN_EXPIRE_MINUTES: int = 60  # Phase 1: raised 30 -> 60 for fewer re-logins
     REFRESH_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 7  # 7 days
@@ -36,9 +36,42 @@ class Settings(BaseSettings):
     WHATSAPP_ACCESS_TOKEN: str = ""
     WHATSAPP_VERIFY_TOKEN: str = ""
     META_GRAPH_BASE: str = "https://graph.facebook.com/v18.0"
+    # Shared secret for inbound webhook POSTs (header ``X-Webhook-Secret``).
+    # Fail-closed: when unset/empty the webhook rejects every message.
+    WHATSAPP_WEBHOOK_SECRET: str = ""
+    # Expected sender/recipient number (E.164) for the WhatsApp demo round
+    # trip (webhook -> ticket -> mock-log). Set per deployment in .env only —
+    # never commit a real number. When set, the demo payload builder
+    # (``app/routers/whatsapp.py::build_demo_webhook_payload``) originates
+    # messages from it, the auto-reply targets it, and ``/api/whatsapp/mock-log``
+    # surfaces it. Empty (default) means the demo payload cannot be built:
+    # the helper fails closed rather than fabricating a sender.
+    WHATSAPP_DEMO_TO: str = ""
+
+    # ── Login rate limiting ──────────────────────────────────────────
+    LOGIN_RATE_LIMIT_MAX_ATTEMPTS: int = 5
+    LOGIN_RATE_LIMIT_WINDOW_SECONDS: int = 15 * 60
 
     # ── Paths ────────────────────────────────────────────────────────
     BASE_DIR: Path = Path(__file__).resolve().parent.parent.parent
 
 
 settings = Settings()
+
+# ── Fail-closed secret validation (HARDENING.md P0.1) ─────────────────
+# Refuse to boot without a real SECRET_KEY. Devs must create a local .env
+# (see .env.example); production injects it via docker-compose env_file.
+_KNOWN_PLACEHOLDER_SECRETS = {
+    "",
+    "change-me-in-production",
+    "change-me-in-production-use-a-real-secret",
+    "changeme",
+    "secret",
+}
+
+if settings.SECRET_KEY in _KNOWN_PLACEHOLDER_SECRETS or len(settings.SECRET_KEY) < 32:
+    raise RuntimeError(
+        "SECRET_KEY is missing, a known placeholder, or shorter than 32 chars. "
+        "Generate one with: openssl rand -hex 32 — and set it in .env "
+        "(dev) or the runtime environment (prod). Refusing to start."
+    )

app/core/ratelimit.py

@@ -0,0 +1,82 @@
+"""Dependency-light login rate limiter.
+
+Brute-force protection for ``POST /api/auth/login``: a sliding window of
+failed attempts keyed by ``ip|email``.  Defaults to ~5 failures / 15 minutes
+(env-tunable via ``LOGIN_RATE_LIMIT_MAX_ATTEMPTS`` /
+``LOGIN_RATE_LIMIT_WINDOW_SECONDS``).
+
+In-process storage is intentional: the app currently runs a single uvicorn
+worker, and keeping the limiter dependency-light avoids pulling slowapi in
+for one endpoint.  ``_now`` is a module-level hook so tests can fast-forward
+the clock.
+"""
+
+from __future__ import annotations
+
+import time
+from collections import defaultdict, deque
+
+from fastapi import HTTPException, status
+
+from app.core.config import settings
+
+
+def _now() -> float:
+    """Wall-clock epoch seconds; overridable in tests via monkeypatch."""
+    return time.time()
+
+
+class LoginRateLimiter:
+    """Sliding-window failure limiter keyed by ``ip|email``."""
+
+    def __init__(self, max_attempts: int = 5, window_seconds: int = 15 * 60) -> None:
+        self.max_attempts = max(max_attempts, 1)
+        self.window_seconds = max(window_seconds, 1)
+        self._failures: defaultdict[str, deque[float]] = defaultdict(deque)
+
+    def key(self, ip: str, email: str) -> str:
+        return f"{ip}|{email.strip().lower()}"
+
+    def _prune(self, key: str, now: float | None = None) -> None:
+        now = now if now is not None else _now()
+        window_start = now - self.window_seconds
+        bucket = self._failures.get(key)
+        if bucket is None:
+            return
+        while bucket and bucket[0] <= window_start:
+            bucket.popleft()
+        if not bucket:
+            self._failures.pop(key, None)
+
+    def failure_count(self, key: str) -> int:
+        self._prune(key)
+        return len(self._failures.get(key, ()))
+
+    def is_blocked(self, key: str) -> bool:
+        return self.failure_count(key) >= self.max_attempts
+
+    def record_failure(self, key: str) -> None:
+        self._failures[key].append(_now())
+        self._prune(key)
+
+    def clear(self, key: str) -> None:
+        self._failures.pop(key, None)
+
+    def reset(self) -> None:
+        self._failures.clear()
+
+    def check_or_raise(self, key: str) -> None:
+        """Raise HTTP 429 when the key has exhausted its attempts."""
+        if self.is_blocked(key):
+            raise HTTPException(
+                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+                detail="Too many failed login attempts. Try again later.",
+            )
+
+
+# Shared instance — module import is safe because the app fails closed at
+# boot (app/core/config.py) before any request can reach the login route.
+login_rate_limiter = LoginRateLimiter(
+    max_attempts=settings.LOGIN_RATE_LIMIT_MAX_ATTEMPTS,
+    window_seconds=settings.LOGIN_RATE_LIMIT_WINDOW_SECONDS,
+)

app/core/roles.py

@@ -0,0 +1,103 @@
+"""Unified role model — single source of truth for user roles.
+
+HARDENING (P0 batch): every role string used by seeds, RBAC checks, admin
+user management, login, and JWT validation derives from this module so the
+system can never silently drift between role vocabularies.
+
+Canonical roles are the human-readable taxonomy the whole product already
+uses (PRD §4, ``app/services/seed.py``, the frontend nav in base.html):
+
+    Admin/Jerome, Admin/Wahab, CS Rep, CS Manager, FM Dispatcher,
+    Tech, CEO, Director
+
+Legacy databases created under the pre-P0 open-registration builds can carry
+lowercase/nickname role strings (``technician``, ``cs``, ``fm``, ``ceo``,
+``admin``, ``superadmin`` …) and underscore variants of the space-separated
+ones (``cs_rep``, ``cs_manager``, ``fm_dispatcher``).  ``ROLE_ALIASES`` maps
+the *unambiguous* nicknames onto a canonical role so startup normalization (see
+``app/services/seed.py::normalize_legacy_user_roles``) can converge the data.
+
+``admin`` / ``superadmin`` are deliberately NOT aliased: they are
+identity-ambiguous (they cannot be attributed to Jerome or Wahab) and were
+mintable by anyone during the open-registration window, so they are treated
+as unknown and fail closed — the operator must remediate those rows manually
+(role cleanup on the live DB is owned by the deployer).
+"""
+
+from __future__ import annotations
+
+# ── Canonical taxonomy ────────────────────────────────────────────────
+# Order is cosmetic; membership is what matters.
+CANONICAL_ROLES: tuple[str, ...] = (
+    "Admin/Jerome",
+    "Admin/Wahab",
+    "CS Rep",
+    "CS Manager",
+    "FM Dispatcher",
+    "Tech",
+    "CEO",
+    "Director",
+)
+
+# Roles that pass admin gates (ticket DELETE, user management, …).
+ADMIN_ROLES: tuple[str, ...] = ("Admin/Jerome", "Admin/Wahab")
+
+# Roles shown to the frontend nav/assignment helpers as "technician" pool.
+TECHNICIAN_ROLE = "Tech"
+
+# ── Legacy alias → canonical mapping (case-insensitive) ───────────────
+# Keys are lowercased. Unambiguous nicknames from legacy/early seeds and the
+# brief's role model ("technician/cs/fm/ceo") converge onto canonical roles.
+ROLE_ALIASES: dict[str, str] = {
+    "technician": TECHNICIAN_ROLE,
+    "tech": TECHNICIAN_ROLE,
+    "cs": "CS Rep",
+    "cs rep": "CS Rep",
+    "cs_rep": "CS Rep",
+    "cs representative": "CS Rep",
+    "cs manager": "CS Manager",
+    "cs_manager": "CS Manager",
+    "fm": "FM Dispatcher",
+    "fm dispatcher": "FM Dispatcher",
+    "fm_dispatcher": "FM Dispatcher",
+    "ceo": "CEO",
+    "director": "Director",
+}
+
+# Aliases that are explicitly NOT auto-mapped (identity-ambiguous and/or
+# mintable by the old open register). They stay unknown → denied at login
+# and JWT validation until an operator remediates the row.
+_BLOCKED_LEGACY_ROLES = frozenset({"admin", "superadmin", "administrator"})
+
+
+def normalize_role(role: str | None) -> str | None:
+    """Return the canonical role for *role*, or ``None`` when unrecognised.
+
+    ``Admin/Jerome`` → ``Admin/Jerome``; ``technician`` → ``Tech``;
+    ``superadmin`` → ``None`` (unknown; caller must fail closed).
+    """
+    if not role:
+        return None
+    stripped = role.strip()
+    if stripped in CANONICAL_ROLES:
+        return stripped
+    return ROLE_ALIASES.get(stripped.lower())
+
+
+def is_known_role(role: str | None) -> bool:
+    """True when *role* is canonical or maps to a canonical role."""
+    return normalize_role(role) is not None
+
+
+def is_admin_role(role: str | None) -> bool:
+    """True when *role* is one of the canonical administrator roles."""
+    return normalize_role(role) in ADMIN_ROLES
+
+
+def is_blocked_legacy_role(role: str | None) -> bool:
+    """True for legacy ``admin``/``superadmin`` rows that need remediation.
+
+    Such rows are not canonical, are not auto-mapped, and must not pass any
+    authorization gate; an operator should reassign or remove them.
+    """
+    return bool(role) and role.strip().lower() in _BLOCKED_LEGACY_ROLES

app/core/security.py

@@ -15,6 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.config import settings
 from app.core.database import get_db
+from app.core.roles import is_known_role
 from app.models.user import User
 
 bearer_scheme = HTTPBearer(auto_error=False)
@@ -90,6 +91,14 @@ async def get_current_user(
     user = result.scalar_one_or_none()
     if user is None or not user.active:
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found or inactive")
+    # Unified role model: reject rows whose stored role is not canonical or a
+    # known legacy alias. Legacy junk roles (e.g. lowercase ``admin``) must
+    # fail closed here so they can never ride an access token into the app.
+    if not is_known_role(user.role):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Account role is not recognised; contact an administrator",
+        )
     return user
 
 

app/main.py

@@ -14,7 +14,13 @@ from sqlalchemy import text
 from app.core.config import settings
 from app.core.database import Base, async_session_factory, engine
 from app.routers import auth, health, pages, tickets, whatsapp
-from app.services.seed import seed_categories, seed_units, seed_users
+from app.services.seed import (
+    normalize_legacy_user_emails,
+    normalize_legacy_user_roles,
+    seed_categories,
+    seed_units,
+    seed_users,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -46,6 +52,47 @@ async def ensure_legacy_schema(conn) -> None:
                 text("UPDATE tickets SET reported_at = created_at WHERE reported_at IS NULL")
             )
             logger.info("Added missing tickets.reported_at column (legacy database)")
+
+    # whatsapp_log predates the real Meta webhook (the model gained
+    # message_text/wa_message_id/ticket_number and dropped `command` in commit
+    # 4afdc36 with no migration), so legacy DBs still carry the old shape and
+    # every read/write through the ORM 500s (no such column: message_text).
+    result = await conn.execute(
+        text("SELECT name FROM sqlite_master WHERE type='table' AND name='whatsapp_log'")
+    )
+    if result.scalar():
+        result = await conn.execute(text("PRAGMA table_info(whatsapp_log)"))
+        log_columns = {row[1] for row in result}
+        if "message_text" not in log_columns:
+            await conn.execute(
+                text("ALTER TABLE whatsapp_log ADD COLUMN message_text TEXT NOT NULL DEFAULT ''")
+            )
+            logger.info("Added missing whatsapp_log.message_text column (legacy database)")
+        if "wa_message_id" not in log_columns:
+            await conn.execute(
+                text("ALTER TABLE whatsapp_log ADD COLUMN wa_message_id VARCHAR(100)")
+            )
+            logger.info("Added missing whatsapp_log.wa_message_id column (legacy database)")
+        if "ticket_number" not in log_columns:
+            await conn.execute(
+                text("ALTER TABLE whatsapp_log ADD COLUMN ticket_number VARCHAR(30)")
+            )
+            logger.info("Added missing whatsapp_log.ticket_number column (legacy database)")
+        if "command" in log_columns:
+            # Legacy rows stored the message body in `command`, which the model
+            # no longer defines (NOT NULL, no default): any ORM insert omitting
+            # it would violate NOT NULL.  Preserve the old bodies in
+            # message_text, then drop the obsolete column to match the model.
+            await conn.execute(
+                text(
+                    "UPDATE whatsapp_log SET message_text = command "
+                    "WHERE (message_text IS NULL OR message_text = '') "
+                    "AND command IS NOT NULL AND command != ''"
+                )
+            )
+            await conn.execute(text("ALTER TABLE whatsapp_log DROP COLUMN command"))
+            logger.info("Dropped obsolete whatsapp_log.command column (legacy database)")
+
     result = await conn.execute(
         text(
             "UPDATE categories SET name = 'Missing Item' "
@@ -67,6 +114,10 @@ async def lifespan(app: FastAPI):
         await ensure_legacy_schema(conn)
     async with async_session_factory() as session:
         await seed_users(session)
+        # P0 role-model unification: converge legacy nickname roles (e.g.
+        # ``technician``/``cs``/``fm``) onto the canonical taxonomy at startup.
+        await normalize_legacy_user_roles(session)
+        await normalize_legacy_user_emails(session)
         await session.commit()
         await seed_units(session, json_path=str(settings.BASE_DIR / "apartment_mapping.json"))
         await session.commit()
@@ -83,20 +134,123 @@ app = FastAPI(
     lifespan=lifespan,
 )
 
-# ── CORS ─────────────────────────────────────────────────────────────
+
+# ── Security headers (P0 batch) ──────────────────────────────────────
+class SecurityHeadersMiddleware:
+    """Set hardening headers on every HTTP response.
+
+    * ``X-Frame-Options: DENY`` and ``X-Content-Type-Options: nosniff`` on
+      all responses;
+    * CSP on HTML pages (login + app pages). Alpine.js and Tailwind are
+      vendored same-origin (``/static/vendor/``), so no external hosts are
+      allowed and the page is fully self-contained — safe on LAN-only demo
+      clients. ``script-src`` keeps ``'unsafe-inline'`` for the inline
+      ``tailwind.config``/``app()`` blocks and ``'unsafe-eval'`` because
+      Alpine 3.17.2's expression evaluator compiles ``x-data``/``x-show``/
+      ``x-text`` etc. with ``new Function()`` — without ``'unsafe-eval'`` CSP
+      blocks every Alpine expression and the loading overlay never clears;
+      style-src keeps ``'unsafe-inline'`` for the Tailwind runtime;
+    * ``Cache-Control: no-cache`` on HTML pages so templates always
+      revalidate (the vendored assets themselves are cached immutably via
+      versioned filenames);
+    * ``Strict-Transport-Security`` only when TLS terminates (https scheme
+      or ``X-Forwarded-Proto: https`` from the reverse proxy).
+    """
+
+    HSTS = "max-age=31536000; includeSubDomains"
+    CSP = (
+        "default-src 'self'; "
+        "script-src 'self' 'unsafe-inline' 'unsafe-eval'; "
+        "style-src 'self' 'unsafe-inline'; "
+        "img-src 'self' data: blob:; "
+        "font-src 'self' data:; "
+        "connect-src 'self'; "
+        "frame-ancestors 'none'; "
+        "base-uri 'self'; "
+        "form-action 'self'; "
+        "object-src 'none'"
+    )
+
+    def __init__(self, app):
+        self.app = app
+
+    async def __call__(self, scope, receive, send):
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        is_tls = scope.get("scheme") == "https"
+        for name, value in scope.get("headers") or []:
+            if name.lower() == b"x-forwarded-proto":
+                first = value.decode("latin-1").split(",", 1)[0].strip().lower()
+                if first == "https":
+                    is_tls = True
+
+        async def send_wrapper(message):
+            if message["type"] == "http.response.start":
+                headers = list(message.get("headers") or [])
+                content_type = next(
+                    (v for k, v in headers if k.lower() == b"content-type"), b""
+                )
+                if content_type.startswith(b"text/html"):
+                    headers.append((b"content-security-policy", self.CSP.encode()))
+                    # Templates must always revalidate: never serve a stale
+                    # page that still points at old vendored filenames.
+                    headers.append((b"cache-control", b"no-cache"))
+                headers.append((b"x-frame-options", b"DENY"))
+                headers.append((b"x-content-type-options", b"nosniff"))
+                if is_tls:
+                    headers.append((b"strict-transport-security", self.HSTS.encode()))
+                message["headers"] = headers
+            await send(message)
+
+        await self.app(scope, receive, send_wrapper)
+
+
+app.add_middleware(SecurityHeadersMiddleware)
+
+# ── CORS (HARDENING.md P0.2 — explicit origin allow-list, never "*") ──
+_origins = [o.strip() for o in settings.CORS_ORIGINS.split(",") if o.strip()]
+if "*" in _origins or not _origins:
+    raise RuntimeError(
+        "CORS_ORIGINS must be an explicit comma-separated origin allow-list "
+        "(e.g. 'https://denya.sysloggh.net,http://localhost:8000'). "
+        "'*' with allow_credentials=True is invalid and unsafe. Refusing to start."
+    )
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=settings.CORS_ORIGINS.split(",") if settings.CORS_ORIGINS != "*" else ["*"],
+    allow_origins=_origins,
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
 )
 
-# ── Static files (uploads) ───────────────────────────────────────────
+# ── Static files (uploads + vendored frontend assets) ────────────────
+class ImmutableStaticFiles(StaticFiles):
+    """StaticFiles that serves long-lived immutable cache headers.
+
+    Used for the vendored frontend libraries under ``app/static/vendor/``
+    (Alpine.js + Tailwind Play). Their URLs embed the version, so upgrading
+    later just bumps the filename and clients fetch the new artifact instead
+    of a stale immutable copy.
+    """
+
+    def file_response(self, full_path, stat_result, scope, status_code=200):
+        response = super().file_response(full_path, stat_result, scope, status_code)
+        response.headers["cache-control"] = "public, max-age=31536000, immutable"
+        return response
+
+
 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")
 
+# Alpine.js/Tailwind are vendored same-origin so LAN-only demo clients render
+# the login/dashboards with no external network (see app/templates/base.html).
+static_dir = Path(__file__).resolve().parent / "static"
+static_dir.mkdir(parents=True, exist_ok=True)
+app.mount("/static", ImmutableStaticFiles(directory=str(static_dir)), name="static")
+
 # ── Routers ──────────────────────────────────────────────────────────
 app.include_router(health.router)
 app.include_router(auth.router)

app/routers/auth.py

@@ -1,20 +1,27 @@
-"""Authentication router — register, login, refresh, me."""
+"""Authentication router — login, refresh, me, and admin user management.
+
+Self-registration was removed (P0 hardening): users are created/managed by
+admins only via ``POST/PATCH/DELETE /api/auth/users``.
+"""
 
 from __future__ import annotations
 
 from typing import Annotated
 
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, HTTPException, Request, status
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.database import get_db
+from app.core.ratelimit import login_rate_limiter
+from app.core.roles import ADMIN_ROLES
 from app.core.security import get_current_user, require_roles
 from app.models.user import User
 from app.schemas.auth import (
+    AdminCreateUserRequest,
+    AdminUpdateUserRequest,
     LoginRequest,
     RefreshRequest,
-    RegisterRequest,
     TokenResponse,
     UserOut,
 )
@@ -22,36 +29,29 @@ from app.services import auth as auth_service
 
 router = APIRouter(prefix="/api/auth", tags=["auth"])
 
-
+_require_admin = require_roles(*ADMIN_ROLES)
-@router.get("/users", response_model=list[UserOut])
-async def list_users(
-    db: Annotated[AsyncSession, Depends(get_db)],
-    current_user: Annotated[User, Depends(get_current_user)],
-) -> list[User]:
-    """List users (id, name, role) for assignment pickers.
-
-    Previously the frontend hard-coded technician ids/names in detail.html;
-    this endpoint makes the assign dropdown data-driven so a seed change never
-    silently breaks technician assignment.
-    """
-    result = await db.execute(select(User).order_by(User.full_name))
-    return list(result.scalars().all())
-
-
-@router.post("/register", response_model=UserOut, status_code=201)
-async def register(
-    body: RegisterRequest,
-    db: Annotated[AsyncSession, Depends(get_db)],
-) -> User:
-    return await auth_service.register(db, body)
 
 
+# ── Public authN ─────────────────────────────────────────────────────
 @router.post("/login", response_model=TokenResponse)
 async def login(
+    request: Request,
     body: LoginRequest,
     db: Annotated[AsyncSession, Depends(get_db)],
 ) -> TokenResponse:
-    access, refresh, _user = await auth_service.login(db, body.email, body.password)
+    """Log in. Brute-force limited to ~5 failures / 15 min per IP+email (429)."""
+    client_ip = request.client.host if request.client else "unknown"
+    key = login_rate_limiter.key(client_ip, body.email)
+    login_rate_limiter.check_or_raise(key)
+
+    try:
+        access, refresh, _user = await auth_service.login(db, body.email, body.password)
+    except HTTPException as exc:
+        # Count only real auth failures toward the limit; success resets it.
+        if exc.status_code == status.HTTP_401_UNAUTHORIZED:
+            login_rate_limiter.record_failure(key)
+        raise
+    login_rate_limiter.clear(key)  # successful login resets the failure window
     return TokenResponse(access_token=access, refresh_token=refresh)
 
 
@@ -71,7 +71,54 @@ async def me(current_user: Annotated[User, Depends(get_current_user)]) -> User:
 
 @router.get("/admin-only", response_model=UserOut)
 async def admin_only(
-    current_user: Annotated[User, Depends(require_roles("Admin/Jerome", "Admin/Wahab"))],
+    current_user: Annotated[User, Depends(_require_admin)],
 ) -> User:
-    """Example RBAC-protected endpoint — only Admins can access."""
+    """Example RBAC-protected endpoint — only canonical Admins can access."""
     return current_user
+
+
+# ── User directory ───────────────────────────────────────────────────
+@router.get("/users", response_model=list[UserOut])
+async def list_users(
+    db: Annotated[AsyncSession, Depends(get_db)],
+    current_user: Annotated[User, Depends(get_current_user)],
+) -> list[User]:
+    """List users (id, name, role) for authenticated assignment pickers."""
+    result = await db.execute(select(User).order_by(User.full_name))
+    return list(result.scalars().all())
+
+
+# ── Admin user management ────────────────────────────────────────────
+@router.post("/users", response_model=UserOut, status_code=status.HTTP_201_CREATED)
+async def create_user(
+    body: AdminCreateUserRequest,
+    db: Annotated[AsyncSession, Depends(get_db)],
+    current_user: Annotated[User, Depends(_require_admin)],
+) -> User:
+    """Admin-only: create a user with a forced canonical role.
+
+    The client cannot self-register or pick an arbitrary role — unknown roles
+    (e.g. ``admin``, ``superadmin``) are rejected with 422.
+    """
+    return await auth_service.create_user(db, body)
+
+
+@router.patch("/users/{user_id}", response_model=UserOut)
+async def update_user(
+    user_id: int,
+    body: AdminUpdateUserRequest,
+    db: Annotated[AsyncSession, Depends(get_db)],
+    current_user: Annotated[User, Depends(_require_admin)],
+) -> User:
+    """Admin-only: change a user's role and/or deactivate the account."""
+    return await auth_service.update_user(db, current_user, user_id, body)
+
+
+@router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_user(
+    user_id: int,
+    db: Annotated[AsyncSession, Depends(get_db)],
+    current_user: Annotated[User, Depends(_require_admin)],
+) -> None:
+    """Admin-only: delete a user account (guarded; see auth_service)."""
+    await auth_service.delete_user(db, current_user, user_id)

app/routers/pages.py

@@ -33,6 +33,14 @@ async def ceo_dashboard(request: Request):
     return templates.TemplateResponse(request, "dashboard/ceo.html")
 
 
+@router.get("/dashboard/tech-performance", response_class=HTMLResponse)
+async def tech_performance_dashboard(request: Request):
+    """Technician performance report (FM + CEO). Data comes from
+    ``GET /api/tickets/tech-performance``; the page itself follows the same
+    client-side auth pattern as the other dashboards."""
+    return templates.TemplateResponse(request, "dashboard/tech-performance.html")
+
+
 @router.get("/tickets", response_class=HTMLResponse)
 async def ticket_list(request: Request):
     return templates.TemplateResponse(request, "tickets/list.html")

app/routers/tickets.py

@@ -13,6 +13,7 @@ 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.roles import ADMIN_ROLES
 from app.core.security import get_current_user, require_roles
 from app.models.category import Category
 from app.models.ticket import Ticket, TicketPhoto
@@ -22,6 +23,7 @@ from app.schemas.ticket import (
     CategoryOut,
     CategoryTreeOut,
     SLAStatusOut,
+    TechnicianPerformanceReportOut,
     TicketBrief,
     TicketCreate,
     TicketListResponse,
@@ -37,6 +39,13 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/api/tickets", tags=["tickets"])
 
+_require_admin = require_roles(*ADMIN_ROLES)
+
+# Pagination contract: sane defaults and a hard page-size cap so list
+# responses never balloon into truncation territory (P0 batch).
+DEFAULT_PAGE_SIZE = 50
+MAX_PAGE_SIZE = 200
+
 # Ensure uploads directory exists
 UPLOADS_DIR = settings.BASE_DIR / "uploads"
 UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
@@ -197,7 +206,10 @@ async def create_ticket(
 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),
+    page_size: int | None = Query(None, ge=1, le=MAX_PAGE_SIZE),
+    limit: int | None = Query(
+        None, ge=1, le=MAX_PAGE_SIZE, description="Alias for page_size (also capped)"
+    ),
     status: str | None = Query(None),
     priority: str | None = Query(None),
     property: str | None = Query(None),
@@ -208,7 +220,17 @@ async def list_tickets(
     date_from: datetime | None = Query(None),
     date_to: datetime | None = Query(None),
 ) -> TicketListResponse:
-    """List tickets with optional filtering and pagination."""
+    """List tickets with optional filtering and pagination.
+
+    Both ``page_size`` and its alias ``limit`` are capped at MAX_PAGE_SIZE;
+    supplying both is an error. Neither defaults to DEFAULT_PAGE_SIZE.
+    """
+    if page_size is not None and limit is not None and page_size != limit:
+        raise HTTPException(
+            status_code=422,  # ``status`` query param shadows fastapi.status in this scope
+            detail="Provide either 'page_size' or 'limit', not both",
+        )
+    effective_page_size = page_size if page_size is not None else (limit or DEFAULT_PAGE_SIZE)
     tickets, total = await ticket_service.list_tickets(
         db,
         status_filter=status,
@@ -221,10 +243,26 @@ async def list_tickets(
         date_from=date_from,
         date_to=date_to,
         page=page,
-        page_size=page_size,
+        page_size=effective_page_size,
     )
     items = [TicketBrief.model_validate(t) for t in tickets]
-    return TicketListResponse(items=items, total=total, page=page, page_size=page_size)
+    return TicketListResponse(items=items, total=total, page=page, page_size=effective_page_size)
+
+
+@router.get("/tech-performance", response_model=TechnicianPerformanceReportOut)
+async def technician_performance(
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _current_user: Annotated[User, Depends(get_current_user)],
+) -> dict:
+    """Per-technician performance aggregate for the FM/CEO dashboards.
+
+    Technician names come from ``users.full_name`` (never ``Tech #<id>``);
+    counts and resolution times are derived from existing ticket columns, so no
+    schema change is involved. Registered before ``/{ticket_id}`` so the literal
+    path is not swallowed by the int-typed ticket-id route. Requires a bearer
+    token, matching every other endpoint that powers a logged-in dashboard.
+    """
+    return await ticket_service.get_technician_performance(db)
 
 
 @router.get("/{ticket_id}/transitions")
@@ -270,7 +308,7 @@ async def update_ticket(
 async def delete_ticket(
     ticket_id: int,
     db: Annotated[AsyncSession, Depends(get_db)],
-    current_user: Annotated[User, Depends(require_roles("Admin/Jerome", "Admin/Wahab"))],
+    current_user: Annotated[User, Depends(_require_admin)],
 ) -> None:
     """Delete a ticket and its children (timeline, photos, escalations).
 

app/routers/whatsapp.py

@@ -1,18 +1,31 @@
-"""WhatsApp webhook handler — Meta Graph API integration."""
+"""WhatsApp webhook handler — Meta Graph API integration.
+
+P0 hardening:
+* ``POST /api/whatsapp/webhook`` requires the ``X-Webhook-Secret`` header to
+  match ``WHATSAPP_WEBHOOK_SECRET``.  Fail-closed: when the env var is unset
+  every message is rejected (same posture as the SECRET_KEY guard).
+* ``GET /api/whatsapp/webhook`` (Meta handshake) validates ``hub.verify_token``
+  with a constant-time compare and returns 403 on mismatch.
+* ``GET /api/whatsapp/mock-log`` now requires authentication (was a public
+  debug endpoint that could 500 and leak stack traces without auth).
+"""
 
 from __future__ import annotations
 
 import logging
+import secrets
 from datetime import datetime, timezone
 from typing import Annotated
 
 import httpx
-from fastapi import APIRouter, Depends, Query
+from fastapi import APIRouter, Depends, Header, HTTPException, Query, 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.user import User
 from app.models.whatsapp_log import WhatsAppLog
 from app.schemas.whatsapp import (
     MetaWebhookRequest,
@@ -33,6 +46,47 @@ REPLY_TEMPLATE = (
 )
 
 
+# ── WhatsApp demo round trip (WHATSAPP_DEMO_TO) ─────────────────────
+def build_demo_webhook_payload(
+    text: str = "Demo message — Denya OneCare WhatsApp round trip",
+    wa_message_id: str = "wamid.demo.000001",
+) -> dict:
+    """Build a Meta webhook payload for the WhatsApp demo round trip.
+
+    The message ``from`` is ``settings.WHATSAPP_DEMO_TO`` (E.164), so the demo
+    surfaces the expected number end-to-end: the webhook logs it in
+    ``whatsapp_log`` (visible via ``GET /api/whatsapp/mock-log``) and the
+    auto-reply is sent back to the same number.  The demo number is configured
+    per deployment in .env (never committed); when it is unset this raises
+    rather than fabricating a sender (same fail-closed posture as the webhook
+    secret).
+    """
+    demo_to = (settings.WHATSAPP_DEMO_TO or "").strip()
+    if not demo_to:
+        raise RuntimeError(
+            "WHATSAPP_DEMO_TO is not configured — set the demo sender number "
+            "in .env to run the WhatsApp demo round trip."
+        )
+    return {
+        "object": "whatsapp_business_account",
+        "entry": [
+            {
+                "id": "1",
+                "changes": [
+                    {
+                        "id": wa_message_id,
+                        "message": {
+                            "from": demo_to,
+                            "id": wa_message_id,
+                            "text": {"text": text},
+                        },
+                    }
+                ],
+            }
+        ],
+    }
+
+
 # ── Meta Graph API helpers ──────────────────────────────────────────
 async def send_whatsapp_reply(
     to_phone: str,
@@ -62,24 +116,55 @@ async def send_whatsapp_reply(
         return WhatsAppReplyResponse(success=False, message=str(exc))
 
 
+# ── Webhook auth (fail-closed) ──────────────────────────────────────
+async def require_webhook_secret(
+    x_webhook_secret: Annotated[str | None, Header(alias="X-Webhook-Secret")] = None,
+) -> None:
+    """Reject webhook messages unless X-Webhook-Secret matches the env secret.
+
+    Reads ``settings.WHATSAPP_WEBHOOK_SECRET`` at request time so the value
+    can be injected per-deployment.  Empty/unset env ⇒ reject everything.
+    """
+    expected = settings.WHATSAPP_WEBHOOK_SECRET
+    if not expected:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Webhook disabled: WHATSAPP_WEBHOOK_SECRET is not configured",
+        )
+    if not x_webhook_secret or not secrets.compare_digest(x_webhook_secret, expected):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Invalid webhook secret",
+        )
+
+
 # ── Webhook endpoint ────────────────────────────────────────────────
-@router.post("/webhook")
+@router.get("/webhook")
-async def whatsapp_webhook(
+async def whatsapp_webhook_verify(
+    mode: str | None = Query(None, alias="hub.mode"),
+    verify_token: str | None = Query(None, alias="hub.verify_token"),
+    challenge: str | None = Query(None, alias="hub.challenge"),
+) -> WebhookVerificationResponse | dict:
+    """Meta webhook handshake (GET): echo the challenge when the token matches."""
+    if mode != "subscribe" or not challenge:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Missing hub.mode / hub.challenge",
+        )
+    expected = settings.WHATSAPP_VERIFY_TOKEN
+    if not expected or not secrets.compare_digest(verify_token or "", expected):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Verify token mismatch",
+        )
+    return WebhookVerificationResponse(challenge=challenge)
+
+
+async def _process_entries(
     body: MetaWebhookRequest,
-    db: Annotated[AsyncSession, Depends(get_db)],
+    db: AsyncSession,
-    hub_verify_token: str | None = Query(None, alias="hub.verify_token"),
-    mode: str | None = Query(None),
-    hub_challenge: str | None = Query(None),
 ) -> dict:
-    """Handle incoming WhatsApp webhook from Meta."""
+    """Create tickets + logs for inbound messages. Shared by the POST handler."""
-
-    # ── Verification GET request (Meta sends this on webhook setup) ──
-    if mode and hub_challenge:
-        if hub_verify_token != settings.WHATSAPP_VERIFY_TOKEN:
-            return {"error": "Verify token mismatch"}
-        return WebhookVerificationResponse(challenge=hub_challenge).model_dump()
-
-    # ── Process inbound messages ────────────────────────────────────
     if not body.entry:
         return {"status": "no entry"}
 
@@ -153,13 +238,24 @@ async def whatsapp_webhook(
     return {"status": "no messages"}
 
 
-# ── Legacy debug endpoint ──────────────────────────────────────────
+@router.post("/webhook")
+async def whatsapp_webhook(
+    body: MetaWebhookRequest,
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _auth: None = Depends(require_webhook_secret),
+) -> dict:
+    """Process an inbound WhatsApp message (Meta POST). Requires webhook secret."""
+    return await _process_entries(body, db)
+
+
+# ── Debug endpoint (auth required) ──────────────────────────────────
 @router.get("/mock-log", response_model=list[MockWhatsAppLogEntry])
 async def mock_whatsapp_log(
     db: Annotated[AsyncSession, Depends(get_db)],
-    limit: int = 50,
+    current_user: Annotated[User, Depends(get_current_user)],
+    limit: int = Query(50, ge=1, le=200),
 ) -> list[MockWhatsAppLogEntry]:
-    """Return recent WhatsApp webhook submissions for debugging."""
+    """Return recent WhatsApp webhook submissions (authenticated only)."""
     result = await db.execute(
         select(WhatsAppLog).order_by(WhatsAppLog.received_at.desc()).limit(limit)
     )

app/schemas/auth.py

@@ -2,15 +2,27 @@
 
 from __future__ import annotations
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field, field_validator, model_validator
+
+from app.core.roles import CANONICAL_ROLES
 
 
-class RegisterRequest(BaseModel):
+def _validate_canonical_role(value: str | None) -> str | None:
-    email: str
+    """Reject any role that is not part of the unified canonical taxonomy.
-    password: str
+
-    full_name: str
+    The role vocabulary is closed: admin user-management must only ever mint
-    phone: str | None = None
+    canonical roles (see app/core/roles.py). Legacy/unknown strings
-    role: str = "CS Rep"
+    (``admin``, ``superadmin``, ``technician``, …) are rejected here so junk
+    roles can never be (re)created through the API.
+    """
+    if value is None:
+        return None
+    role = value.strip()
+    if role not in CANONICAL_ROLES:
+        raise ValueError(
+            f"Unknown role '{value}'. Allowed roles: {', '.join(CANONICAL_ROLES)}"
+        )
+    return role
 
 
 class LoginRequest(BaseModel):
@@ -37,3 +49,50 @@ class UserOut(BaseModel):
     active: bool
 
     model_config = {"from_attributes": True}
+
+
+# ── Admin user management (self-registration is removed) ──────────────
+class AdminCreateUserRequest(BaseModel):
+    """Admin-created user. The role is mandatory and must be canonical.
+
+    ``role`` is deliberately NOT optional and has no default — an admin must
+    state the intended role explicitly; the server never infers one.
+    """
+
+    email: str = Field(min_length=1)
+    password: str = Field(min_length=8, description="Minimum 8 characters")
+    full_name: str = Field(min_length=1)
+    phone: str | None = None
+    role: str
+
+    @field_validator("role")
+    @classmethod
+    def _role_canonical(cls, value: str) -> str:
+        return _validate_canonical_role(value)  # type: ignore[return-value]
+
+    @field_validator("email")
+    @classmethod
+    def _lower_email(cls, value: str) -> str:
+        return value.strip().lower()
+
+
+class AdminUpdateUserRequest(BaseModel):
+    """Admin edits to an existing user: role change and/or deactivation.
+
+    At least one field must be present. ``active=False`` deactivates the
+    account (login and token refresh then fail closed).
+    """
+
+    role: str | None = None
+    active: bool | None = None
+
+    @field_validator("role")
+    @classmethod
+    def _role_canonical(cls, value: str | None) -> str | None:
+        return _validate_canonical_role(value)
+
+    @model_validator(mode="after")
+    def _at_least_one_field(self) -> AdminUpdateUserRequest:
+        if self.role is None and self.active is None:
+            raise ValueError("Provide at least one of 'role' or 'active'")
+        return self

app/schemas/ticket.py

@@ -129,6 +129,60 @@ class TicketListResponse(BaseModel):
     page_size: int
 
 
+# ── Technician performance ───────────────────────────────────────────
+class TechnicianPerformanceOut(BaseModel):
+    """Per-technician workload and outcome aggregate (by real name).
+
+    ``open_tickets`` is ``total_assigned - completed - cancelled``;
+    ``pending`` is the remainder bucket (statuses such as New/Logged/Triage/
+    Assigned plus verification stages) and keeps the named buckets summing to
+    ``total_assigned``. ``avg_resolution_hours`` averages ``created_at ->
+    closed_at`` and is ``null`` when no finished task carries a timestamp —
+    ``resolved_without_timestamps`` then says how many those are.
+    """
+    technician_id: int
+    name: str
+    total_assigned: int
+    completed: int
+    closed: int
+    in_progress: int
+    escalated: int
+    cancelled: int
+    pending: int
+    open_tickets: int
+    completion_rate: float
+    avg_resolution_hours: float | None = None
+    resolved_with_timestamps: int
+    resolved_without_timestamps: int
+    status_breakdown: dict[str, int] = {}
+
+
+class TechnicianPerformanceTotalsOut(BaseModel):
+    """Fleet-wide roll-up of :class:`TechnicianPerformanceOut`."""
+    technicians: int
+    total_tickets: int
+    total_assigned: int
+    completed: int
+    closed: int
+    in_progress: int
+    escalated: int
+    cancelled: int
+    pending: int
+    open_tickets: int
+    completion_rate: float
+    avg_resolution_hours: float | None = None
+    resolved_with_timestamps: int
+    resolved_without_timestamps: int
+    unassigned_tickets: int
+
+
+class TechnicianPerformanceReportOut(BaseModel):
+    """Response for ``GET /api/tickets/tech-performance``."""
+    generated_at: datetime
+    technicians: list[TechnicianPerformanceOut]
+    totals: TechnicianPerformanceTotalsOut
+
+
 # ── SLA ──────────────────────────────────────────────────────────────
 class SLAStatusOut(BaseModel):
     priority: str | None = None

app/services/auth.py

@@ -1,11 +1,16 @@
-"""Authentication service — register, login, refresh."""
+"""Authentication service — login, refresh, and admin user management.
+
+Self-registration was removed (HARDENING/P0 batch): users are created and
+managed exclusively by admins through the admin user-management endpoints.
+"""
 
 from __future__ import annotations
 
 from fastapi import HTTPException, status
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.core.roles import is_known_role, normalize_role
 from app.core.security import (
     create_access_token,
     create_refresh_token,
@@ -13,38 +18,35 @@ from app.core.security import (
     hash_password,
     verify_password,
 )
+from app.models.ticket import Escalation, Ticket, TicketTimeline
 from app.models.user import User
-from app.schemas.auth import RegisterRequest
+from app.schemas.auth import AdminCreateUserRequest, AdminUpdateUserRequest
 
-
+_UNAUTHORIZED = status.HTTP_401_UNAUTHORIZED
-async def register(db: AsyncSession, body: RegisterRequest) -> User:
-    """Create a new user.  Raises 409 if email already exists."""
-    result = await db.execute(select(User).where(User.email == body.email))
-    if result.scalar_one_or_none():
-        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email already registered")
-
-    user = User(
-        email=body.email,
-        password_hash=hash_password(body.password),
-        full_name=body.full_name,
-        phone=body.phone,
-        role=body.role,
-    )
-    db.add(user)
-    await db.flush()
-    await db.refresh(user)
-    return user
 
 
+# ── AuthN ────────────────────────────────────────────────────────────
 async def login(db: AsyncSession, email: str, password: str) -> tuple[str, str, User]:
-    """Authenticate and return (access_token, refresh_token, user)."""
+    """Authenticate and return (access_token, refresh_token, user).
-    result = await db.execute(select(User).where(User.email == email))
-    user = result.scalar_one_or_none()
-    if user is None or not verify_password(password, user.password_hash):
-        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid email or password")
-    if not user.active:
-        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Account is inactive")
 
+    Fails closed (401) for bad credentials, inactive accounts, and any user
+    whose stored role is not part of the unified role model.
+    """
+    result = await db.execute(
+        select(User)
+        .where(func.lower(User.email) == email.strip().lower())
+        .order_by(User.id)
+    )
+    user = result.scalars().first()
+    if user is None or not verify_password(password, user.password_hash):
+        raise HTTPException(status_code=_UNAUTHORIZED, detail="Invalid email or password")
+    if not user.active:
+        raise HTTPException(status_code=_UNAUTHORIZED, detail="Account is inactive")
+    if not is_known_role(user.role):
+        raise HTTPException(
+            status_code=_UNAUTHORIZED,
+            detail="Account role is not recognised; contact an administrator",
+        )
     access_token = create_access_token({"sub": str(user.id)})
     refresh_token = create_refresh_token({"sub": str(user.id)})
     return access_token, refresh_token, user
@@ -55,18 +57,131 @@ async def refresh_access_token(db: AsyncSession, token: str) -> tuple[str, str]:
     try:
         payload = decode_token(token)
         if payload.get("type") != "refresh":
-            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token type")
+            raise HTTPException(status_code=_UNAUTHORIZED, detail="Invalid token type")
     except HTTPException:
         raise
     except Exception:
-        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid refresh token")
+        raise HTTPException(status_code=_UNAUTHORIZED, detail="Invalid refresh token")
 
     user_id: int = int(payload["sub"])
     result = await db.execute(select(User).where(User.id == user_id))
     user = result.scalar_one_or_none()
     if user is None or not user.active:
-        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found or inactive")
+        raise HTTPException(status_code=_UNAUTHORIZED, detail="User not found or inactive")
+    if not is_known_role(user.role):
+        raise HTTPException(
+            status_code=_UNAUTHORIZED,
+            detail="Account role is not recognised; contact an administrator",
+        )
 
     new_access = create_access_token({"sub": str(user.id)})
     new_refresh = create_refresh_token({"sub": str(user.id)})
     return new_access, new_refresh
+
+
+# ── Admin user management ────────────────────────────────────────────
+async def _get_user_or_404(db: AsyncSession, user_id: int) -> User:
+    result = await db.execute(select(User).where(User.id == user_id))
+    user = result.scalar_one_or_none()
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+    return user
+
+
+async def create_user(db: AsyncSession, body: AdminCreateUserRequest) -> User:
+    """Admin-created user with an explicit, canonical role. 409 on duplicate email."""
+    email = body.email.strip().lower()
+    result = await db.execute(select(User).where(func.lower(User.email) == email))
+    if result.scalar_one_or_none():
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email already registered")
+
+    # Defense in depth: schema already guarantees a canonical role.
+    role = normalize_role(body.role)
+    if role is None:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Unknown role")
+
+    user = User(
+        email=email,
+        password_hash=hash_password(body.password),
+        full_name=body.full_name.strip(),
+        phone=body.phone,
+        role=role,
+    )
+    db.add(user)
+    await db.flush()
+    await db.refresh(user)
+    return user
+
+
+async def update_user(
+    db: AsyncSession,
+    actor: User,
+    user_id: int,
+    body: AdminUpdateUserRequest,
+) -> User:
+    """Admin role-change / deactivation for an existing user.
+
+    Guards:
+    * an admin cannot modify their own account through the API (self-lockout);
+    * role changes are limited to the canonical taxonomy.
+
+    The ≥1-active-admin invariant holds structurally: only admins can demote
+    admins, and no admin can demote/deactivate themselves, so at least one
+    canonical admin always remains.
+    """
+    user = await _get_user_or_404(db, user_id)
+
+    if actor.id == user.id:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Admins cannot change their own role or active state through the API",
+        )
+
+    new_role = normalize_role(body.role) if body.role is not None else None
+    new_active = body.active
+
+    if new_role is not None:
+        user.role = new_role
+    if new_active is not None:
+        user.active = new_active
+    await db.flush()
+    await db.refresh(user)
+    return user
+
+
+async def delete_user(db: AsyncSession, actor: User, user_id: int) -> None:
+    """Admin deletes a user account (hard delete).
+
+    Guards:
+    * an admin cannot delete their own account (self-guard also keeps the
+      ≥1-active-admin invariant: admins can never remove themselves);
+    * users referenced by tickets / timeline / escalations are kept (409) so
+      historical data never dangles — reassign or deactivate instead.
+    """
+    user = await _get_user_or_404(db, user_id)
+
+    if actor.id == user.id:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Admins cannot delete their own account through the API",
+        )
+
+    referenced = False
+    for clause in (
+        select(func.count(Ticket.id)).where(Ticket.assigned_to == user_id),
+        select(func.count(TicketTimeline.id)).where(TicketTimeline.user_id == user_id),
+        select(func.count(Escalation.id)).where(Escalation.escalated_to == user_id),
+    ):
+        count = (await db.execute(clause)).scalar() or 0
+        if count:
+            referenced = True
+            break
+    if referenced:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="User has related tickets, timeline entries, or escalations; "
+            "reassign or deactivate instead of deleting",
+        )
+
+    await db.delete(user)
+    await db.flush()

app/services/seed.py

@@ -13,6 +13,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.security import hash_password
+from app.core.roles import is_blocked_legacy_role, normalize_role
 from app.models.unit import Unit
 from app.models.user import User
 from app.models.category import Category
@@ -41,6 +42,88 @@ SEED_USERS_DATA = [
 ]
 
 
+async def normalize_legacy_user_roles(db: AsyncSession) -> int:
+    """Converge legacy role strings onto the unified canonical taxonomy.
+
+    Databases built before the P0 role-model batch can hold nickname roles
+    (``technician``, ``cs``, ``fm``, ``ceo`` …) minted by the old open
+    self-registration.  Unambiguous aliases are rewritten to their canonical
+    role so RBAC keeps working.  Ambiguous/unknown roles (e.g. lowercase
+    ``admin``/``superadmin``) are NOT auto-mapped — they fail closed at login
+    and JWT validation until an operator remediates the row.
+
+    Returns the number of rows rewritten. Idempotent.
+    """
+    result = await db.execute(select(User))
+    changed = 0
+    for user in result.scalars().all():
+        canonical = normalize_role(user.role)
+        if canonical and canonical != user.role:
+            logger.info("Normalizing legacy role %r → %r for %s", user.role, canonical, user.email)
+            user.role = canonical
+            changed += 1
+        elif canonical is None and not is_blocked_legacy_role(user.role):
+            logger.warning(
+                "User %s has unrecognized role %r; login will be denied until fixed",
+                user.email,
+                user.role,
+            )
+    if changed:
+        await db.flush()
+    return changed
+
+
+async def normalize_legacy_user_emails(db: AsyncSession) -> int:
+    """Lowercase stored user emails to match the normalized login lookup.
+
+    Databases built before the P0 batch can hold mixed-case emails (the old
+    open self-registration stored them verbatim) while login now compares on
+    the lowercase form, so such rows would otherwise be silently locked out.
+    Rewrites each stored email to its stripped/lowercase form.  When two rows
+    share an email that differs only in case, only the lowest-id row becomes
+    canonical (any others keep their stored value and are logged as a warning)
+    so the unique constraint is never violated.  Idempotent; returns the
+    number of rows rewritten.
+    """
+    result = await db.execute(select(User).order_by(User.id))
+    users = list(result.scalars().all())
+    groups: dict[str, list[User]] = {}
+    for user in users:
+        groups.setdefault(user.email.strip().lower(), []).append(user)
+
+    changed = 0
+    for normalized, members in groups.items():
+        if len(members) == 1:
+            user = members[0]
+            if user.email != normalized:
+                logger.info(
+                    "Normalizing legacy email %r → %r for user %d", user.email, normalized, user.id
+                )
+                user.email = normalized
+                changed += 1
+            continue
+        if any(user.email == normalized for user in members):
+            losers = [u for u in members if u.email != normalized]
+        else:
+            winner = members[0]
+            logger.info(
+                "Normalizing legacy email %r → %r for user %d", winner.email, normalized, winner.id
+            )
+            winner.email = normalized
+            changed += 1
+            losers = members[1:]
+        for loser in losers:
+            logger.warning(
+                "Cannot normalize email %r for user %d: another account already holds "
+                "that normalized email; keeping the stored value",
+                normalized,
+                loser.id,
+            )
+    if changed:
+        await db.flush()
+    return changed
+
+
 async def seed_users(db: AsyncSession, default_password: str = "denya123") -> list[User]:
     """Insert seed users if they don't already exist."""
     hashed = hash_password(default_password)

app/services/ticket.py

@@ -345,6 +345,169 @@ async def update_ticket(
     return ticket
 
 
+# ── Technician performance ───────────────────────────────────────────
+# Buckets for the technician-performance dashboard (Wahab request). The map is
+# intentionally not exhaustive: any status missing from it is counted as
+# "pending" so the named buckets always sum to ``total_assigned`` and no
+# assigned ticket is silently dropped from the report.
+_TECH_STATUS_BUCKETS: dict[str, str] = {
+    "Completed": "completed",
+    "Closed": "completed",
+    "Accepted": "in_progress",
+    "Travelling": "in_progress",
+    "On Site": "in_progress",
+    "In Progress": "in_progress",
+    "Waiting Parts": "in_progress",
+    "Escalated": "escalated",
+    "Cancelled": "cancelled",
+}
+
+
+def _bucket_for_status(status: str) -> str:
+    """Map a ticket status onto its technician-performance bucket."""
+    return _TECH_STATUS_BUCKETS.get(status, "pending")
+
+
+def _empty_tech_entry(technician_id: int, name: str) -> dict[str, Any]:
+    return {
+        "technician_id": technician_id,
+        "name": name,
+        "total_assigned": 0,
+        "completed": 0,
+        "closed": 0,
+        "in_progress": 0,
+        "escalated": 0,
+        "cancelled": 0,
+        "pending": 0,
+        "open_tickets": 0,
+        "completion_rate": 0.0,
+        "avg_resolution_hours": None,
+        "resolved_with_timestamps": 0,
+        "resolved_without_timestamps": 0,
+        "status_breakdown": {},
+        "_hours_sum": 0.0,
+    }
+
+
+async def get_technician_performance(db: AsyncSession) -> dict[str, Any]:
+    """Aggregate per-technician workload/outcome stats for the dashboard.
+
+    Derived entirely from existing ``tickets`` columns (``assigned_to``,
+    ``status``, ``created_at``, ``closed_at``) joined to ``users.full_name`` —
+    no schema change. Technician identity is keyed on the user id so two people
+    sharing a display name stay separate rows, while the reported label is the
+    real name (never ``"Tech #<id>"``).
+
+    Completion rate is ``completed / total_assigned`` (Completed + Closed count
+    as completed). Resolution time averages ``created_at -> closed_at`` only for
+    rows where ``closed_at`` is set; the number of finished tasks lacking that
+    timestamp is reported separately so an absent average is never mistaken for
+    missing work.
+    """
+    rows = (
+        await db.execute(
+            select(
+                Ticket.assigned_to,
+                User.full_name,
+                Ticket.status,
+                Ticket.created_at,
+                Ticket.closed_at,
+            )
+            .join(User, User.id == Ticket.assigned_to)
+            .where(Ticket.assigned_to.is_not(None))
+        )
+    ).all()
+
+    unassigned_result = await db.execute(
+        select(func.count(Ticket.id)).where(Ticket.assigned_to.is_(None))
+    )
+    unassigned_tickets = unassigned_result.scalar() or 0
+
+    by_tech: dict[int, dict[str, Any]] = {}
+    for assigned_to, full_name, status, created_at, closed_at in rows:
+        entry = by_tech.get(assigned_to)
+        if entry is None:
+            entry = _empty_tech_entry(assigned_to, full_name)
+            by_tech[assigned_to] = entry
+
+        entry["total_assigned"] += 1
+        bucket = _bucket_for_status(status)
+        if bucket == "completed":
+            entry["completed"] += 1
+            if status == "Closed":
+                entry["closed"] += 1
+            if closed_at is not None and created_at is not None:
+                hours = (closed_at - created_at).total_seconds() / 3600
+                entry["_hours_sum"] += hours
+                entry["resolved_with_timestamps"] += 1
+            else:
+                entry["resolved_without_timestamps"] += 1
+        else:
+            entry[bucket] += 1
+
+        breakdown = entry["status_breakdown"]
+        breakdown[status] = breakdown.get(status, 0) + 1
+
+    technicians: list[dict[str, Any]] = []
+    total_assigned = total_completed = total_closed = 0
+    total_in_progress = total_escalated = total_cancelled = total_pending = 0
+    total_hours = 0.0
+    total_with_timestamps = total_without_timestamps = 0
+
+    for entry in by_tech.values():
+        assigned = entry["total_assigned"]
+        entry["open_tickets"] = assigned - entry["completed"] - entry["cancelled"]
+        entry["completion_rate"] = round(entry["completed"] / assigned * 100, 1) if assigned else 0.0
+        if entry["resolved_with_timestamps"]:
+            entry["avg_resolution_hours"] = round(
+                entry["_hours_sum"] / entry["resolved_with_timestamps"], 1
+            )
+        entry["status_breakdown"] = dict(
+            sorted(entry["status_breakdown"].items(), key=lambda kv: (-kv[1], kv[0]))
+        )
+
+        total_assigned += assigned
+        total_completed += entry["completed"]
+        total_closed += entry["closed"]
+        total_in_progress += entry["in_progress"]
+        total_escalated += entry["escalated"]
+        total_cancelled += entry["cancelled"]
+        total_pending += entry["pending"]
+        total_hours += entry["_hours_sum"]
+        total_with_timestamps += entry["resolved_with_timestamps"]
+        total_without_timestamps += entry["resolved_without_timestamps"]
+
+        del entry["_hours_sum"]
+        technicians.append(entry)
+
+    # Busiest/most productive first; ties broken by workload then real name.
+    technicians.sort(key=lambda e: (-e["completed"], -e["total_assigned"], e["name"].lower()))
+
+    totals = {
+        "technicians": len(technicians),
+        "total_assigned": total_assigned,
+        "completed": total_completed,
+        "closed": total_closed,
+        "in_progress": total_in_progress,
+        "escalated": total_escalated,
+        "cancelled": total_cancelled,
+        "pending": total_pending,
+        "open_tickets": total_assigned - total_completed - total_cancelled,
+        "completion_rate": round(total_completed / total_assigned * 100, 1) if total_assigned else 0.0,
+        "avg_resolution_hours": round(total_hours / total_with_timestamps, 1) if total_with_timestamps else None,
+        "resolved_with_timestamps": total_with_timestamps,
+        "resolved_without_timestamps": total_without_timestamps,
+        "unassigned_tickets": unassigned_tickets,
+        "total_tickets": total_assigned + unassigned_tickets,
+    }
+
+    return {
+        "generated_at": datetime.now(timezone.utc),
+        "technicians": technicians,
+        "totals": totals,
+    }
+
+
 async def delete_ticket(db: AsyncSession, ticket_id: int) -> Ticket:
     """Delete a ticket and all dependent rows (timeline, photos, escalations).
 

app/templates/base.html

@@ -4,8 +4,12 @@
     <meta charset="UTF-8">
     <meta name="viewport" content="width=device-width, initial-scale=1.0">
     <title>Denya OneCare</title>
-    <script src="https://cdn.jsdelivr.net/npm/alpinejs@3.x.x/dist/cdn.min.js" defer></script>
+    <!-- Favicons (same-origin app/static/branding) -->
-    <script src="https://cdn.tailwindcss.com"></script>
+    <link rel="icon" type="image/png" sizes="32x32" href="/static/branding/denya-logo-32x32.png">
+    <link rel="icon" type="image/png" sizes="16x16" href="/static/branding/denya-logo-16x16.png">
+    <!-- Vendored same-origin (no CDN): app/static/vendor/ — LAN-safe demo -->
+    <script src="/static/vendor/alpine-3.17.2.min.js" defer></script>
+    <script src="/static/vendor/tailwind-3.4.17.js"></script>
     <script>
         tailwind.config = {
             theme: {
@@ -67,15 +71,11 @@
                 <!-- Left side -->
                 <div class="flex items-center space-x-4">
                     <a href="/dashboard/cs" class="flex items-center space-x-3">
-                        <!-- Denya Developers Logo Mark -->
+                        <!-- Denya Developers logo (same-origin app/static/branding) -->
-                        <div class="w-9 h-9 bg-gold rounded-lg flex items-center justify-center shadow-sm">
+                        <img src="/static/branding/denya-logo-h48.png" alt="Denya Developers"
-                            <svg class="w-5 h-5 text-[#0d2b18]" fill="none" stroke="currentColor" stroke-width="2.5" viewBox="0 0 24 24">
+                             class="h-9 w-auto bg-white rounded-md px-1.5 py-1 shadow-sm">
-                                <path stroke-linecap="round" stroke-linejoin="round" d="M3 12l2-2m0 0l7-7 7 7M5 10v10a1 1 0 001 1h3m10-11l2 2m-2-2v10a1 1 0 01-1 1h-3m-6 0a1 1 0 001-1v-4a1 1 0 011-1h2a1 1 0 011 1v4a1 1 0 001 1m-6 0h6"/>
-                            </svg>
-                        </div>
                         <div class="flex flex-col">
-                            <span class="text-white font-bold text-base leading-tight">Denya Developers</span>
+                            <span class="text-gold text-sm leading-tight font-bold">OneCare</span>
-                            <span class="text-gold text-xs leading-tight font-medium">OneCare</span>
                         </div>
                     </a>
                     <!-- Nav Links -->
@@ -89,6 +89,7 @@
                     <template x-if="isFM">
                         <div class="hidden md:flex space-x-1 ml-6">
                             <a href="/dashboard/fm" class="px-3 py-2 rounded-md text-sm font-medium transition-colors" :class="currentPath === '/dashboard/fm' ? 'bg-denya-800 text-gold' : 'text-gray-300 hover:text-white hover:bg-denya-800/50'">Dashboard</a>
+                            <a href="/dashboard/tech-performance" class="px-3 py-2 rounded-md text-sm font-medium transition-colors" :class="currentPath === '/dashboard/tech-performance' ? 'bg-denya-800 text-gold' : 'text-gray-300 hover:text-white hover:bg-denya-800/50'">Tech Performance</a>
                             <a href="/tickets" class="px-3 py-2 rounded-md text-sm font-medium transition-colors" :class="currentPath.startsWith('/tickets') && !currentPath.endsWith('/new') ? 'bg-denya-800 text-gold' : 'text-gray-300 hover:text-white hover:bg-denya-800/50'">All Issues</a>
                             <a href="/tickets/new" class="px-3 py-2 rounded-md text-sm font-medium text-gray-300 hover:text-white hover:bg-denya-800/50 transition-colors">Create Issue</a>
                         </div>
@@ -96,6 +97,7 @@
                     <template x-if="isExecutive">
                         <div class="hidden md:flex space-x-1 ml-6">
                             <a href="/dashboard/ceo" class="px-3 py-2 rounded-md text-sm font-medium transition-colors" :class="currentPath === '/dashboard/ceo' ? 'bg-denya-800 text-gold' : 'text-gray-300 hover:text-white hover:bg-denya-800/50'">Dashboard</a>
+                            <a href="/dashboard/tech-performance" class="px-3 py-2 rounded-md text-sm font-medium transition-colors" :class="currentPath === '/dashboard/tech-performance' ? 'bg-denya-800 text-gold' : 'text-gray-300 hover:text-white hover:bg-denya-800/50'">Tech Performance</a>
                             <a href="/tickets" class="px-3 py-2 rounded-md text-sm font-medium transition-colors" :class="currentPath.startsWith('/tickets') ? 'bg-denya-800 text-gold' : 'text-gray-300 hover:text-white hover:bg-denya-800/50'">Issues</a>
                         </div>
                     </template>
@@ -113,16 +115,25 @@
 
     <!-- Mobile Nav -->
     <div class="md:hidden border-b bg-[#0d2b18] border-denya-800" x-show="isLoggedIn" x-cloak>
-        <template x-if="isCS || isFM">
+        <template x-if="isCS">
             <div class="flex overflow-x-auto px-4 py-2 space-x-2">
                 <a href="/dashboard/cs" class="px-3 py-1.5 rounded text-sm font-medium whitespace-nowrap" :class="currentPath === '/dashboard/cs' ? 'bg-denya-800 text-gold' : 'text-gray-300'">Dashboard</a>
                 <a href="/tickets" class="px-3 py-1.5 rounded text-sm font-medium whitespace-nowrap" :class="currentPath.startsWith('/tickets') && !currentPath.endsWith('/new') ? 'bg-denya-800 text-gold' : 'text-gray-300'">Issues</a>
                 <a href="/tickets/new" class="px-3 py-1.5 rounded text-sm font-medium whitespace-nowrap text-gray-300">New Issue</a>
             </div>
         </template>
+        <template x-if="isFM">
+            <div class="flex overflow-x-auto px-4 py-2 space-x-2">
+                <a href="/dashboard/fm" class="px-3 py-1.5 rounded text-sm font-medium whitespace-nowrap" :class="currentPath === '/dashboard/fm' ? 'bg-denya-800 text-gold' : 'text-gray-300'">Dashboard</a>
+                <a href="/dashboard/tech-performance" class="px-3 py-1.5 rounded text-sm font-medium whitespace-nowrap" :class="currentPath === '/dashboard/tech-performance' ? 'bg-denya-800 text-gold' : 'text-gray-300'">Tech Performance</a>
+                <a href="/tickets" class="px-3 py-1.5 rounded text-sm font-medium whitespace-nowrap" :class="currentPath.startsWith('/tickets') && !currentPath.endsWith('/new') ? 'bg-denya-800 text-gold' : 'text-gray-300'">Issues</a>
+                <a href="/tickets/new" class="px-3 py-1.5 rounded text-sm font-medium whitespace-nowrap text-gray-300">New Issue</a>
+            </div>
+        </template>
         <template x-if="isExecutive">
             <div class="flex overflow-x-auto px-4 py-2 space-x-2">
                 <a href="/dashboard/ceo" class="px-3 py-1.5 rounded text-sm font-medium whitespace-nowrap" :class="currentPath === '/dashboard/ceo' ? 'bg-denya-800 text-gold' : 'text-gray-300'">CEO Dashboard</a>
+                <a href="/dashboard/tech-performance" class="px-3 py-1.5 rounded text-sm font-medium whitespace-nowrap" :class="currentPath === '/dashboard/tech-performance' ? 'bg-denya-800 text-gold' : 'text-gray-300'">Tech Performance</a>
                 <a href="/tickets" class="px-3 py-1.5 rounded text-sm font-medium whitespace-nowrap text-gray-300">Issues</a>
             </div>
         </template>

app/templates/dashboard/ceo.html

@@ -1,9 +1,12 @@
 {% extends "base.html" %}
 {% block content %}
 <div x-data="ceoDashboard()" x-init="init()">
-    <div class="mb-6">
+    <div class="mb-6 flex flex-wrap items-start justify-between gap-3">
-        <h1 class="text-2xl font-bold text-gray-900">Executive Dashboard</h1>
+        <div>
-        <p class="text-gray-500 mt-1">Read-only strategic overview</p>
+            <h1 class="text-2xl font-bold text-gray-900">Executive Dashboard</h1>
+            <p class="text-gray-500 mt-1">Read-only strategic overview</p>
+        </div>
+        <a href="/dashboard/tech-performance" class="px-3 py-2 text-sm border border-gray-300 rounded-lg hover:bg-gray-50 text-gray-700">Technician Performance →</a>
     </div>
 
     <!-- Executive KPI Cards -->

app/templates/dashboard/fm.html

@@ -80,7 +80,10 @@
     <div class="grid grid-cols-1 lg:grid-cols-2 gap-6 mb-8">
         <!-- Tech Workload -->
         <div class="bg-white rounded-xl shadow-sm border border-gray-200 p-5">
-            <h3 class="text-sm font-semibold text-gray-700 uppercase tracking-wide mb-4">Technician Workload</h3>
+            <div class="flex items-center justify-between mb-4">
+                <h3 class="text-sm font-semibold text-gray-700 uppercase tracking-wide">Technician Workload</h3>
+                <a href="/dashboard/tech-performance" class="text-xs text-denya-600 hover:text-denya-800">Performance →</a>
+            </div>
             <div class="space-y-3">
                 <template x-for="tech in techWorkload" :key="tech.id">
                     <div class="flex items-center justify-between p-2 hover:bg-gray-50 rounded">
@@ -251,11 +254,13 @@
                         this.kpi.westJobs = active.filter(t => t.unit_id && unitMap[t.unit_id] === 'West').length;
                     } catch (e) { console.error('Property stats error', e); }
 
-                    // Tech workload (simulated from assigned_to counts)
+                    // Tech workload — real technician names (Ticket.assigned_technician_name),
+                    // never an id placeholder; keyed on user id so two people sharing a
+                    // display name stay separate rows.
                     const techMap = {};
                     active.forEach(t => {
                         if (t.assigned_to) {
-                            if (!techMap[t.assigned_to]) techMap[t.assigned_to] = { id: t.assigned_to, name: `Tech #${t.assigned_to}`, activeJobs: 0 };
+                            if (!techMap[t.assigned_to]) techMap[t.assigned_to] = { id: t.assigned_to, name: t.assigned_technician_name || 'Unassigned', activeJobs: 0 };
                             techMap[t.assigned_to].activeJobs++;
                         }
                     });

app/templates/dashboard/tech-performance.html

@@ -0,0 +1,239 @@
+{% extends "base.html" %}
+{% block content %}
+<div x-data="techPerformance()" x-init="init()">
+    <div class="mb-6 flex flex-wrap items-start justify-between gap-3">
+        <div>
+            <h1 class="text-2xl font-bold text-gray-900">Technician Performance</h1>
+            <p class="text-gray-500 mt-1">Tasks completed per technician, by name — FM &amp; CEO view</p>
+        </div>
+        <div class="flex items-center space-x-2">
+            <a x-show="isFM" href="/dashboard/fm" class="px-3 py-2 text-sm border border-gray-300 rounded-lg hover:bg-gray-50 text-gray-700">FM Dashboard</a>
+            <a x-show="isExecutive" href="/dashboard/ceo" class="px-3 py-2 text-sm border border-gray-300 rounded-lg hover:bg-gray-50 text-gray-700">CEO Dashboard</a>
+            <button @click="loadData()" class="px-3 py-2 text-sm bg-denya-700 text-white rounded-lg hover:bg-denya-800">Refresh</button>
+        </div>
+    </div>
+
+    <!-- KPI Cards -->
+    <div class="grid grid-cols-2 md:grid-cols-3 lg:grid-cols-6 gap-4 mb-8">
+        <div class="bg-white rounded-xl shadow-sm border border-gray-200 p-4">
+            <p class="text-xs text-gray-500 font-medium uppercase tracking-wide">Technicians</p>
+            <p class="text-3xl font-bold text-gray-900 mt-1" x-text="totals.technicians || 0"></p>
+        </div>
+        <div class="bg-white rounded-xl shadow-sm border border-gray-200 p-4">
+            <p class="text-xs text-gray-500 font-medium uppercase tracking-wide">Assigned</p>
+            <p class="text-3xl font-bold text-gray-900 mt-1" x-text="totals.total_assigned || 0"></p>
+        </div>
+        <div class="bg-white rounded-xl shadow-sm border border-gray-200 p-4">
+            <p class="text-xs text-gray-500 font-medium uppercase tracking-wide">Completed</p>
+            <p class="text-3xl font-bold text-green-600 mt-1" x-text="totals.completed || 0"></p>
+        </div>
+        <div class="bg-white rounded-xl shadow-sm border border-gray-200 p-4">
+            <p class="text-xs text-gray-500 font-medium uppercase tracking-wide">Completion Rate</p>
+            <p class="text-3xl font-bold mt-1" :class="rateClass(totals.completion_rate)" x-text="formatRate(totals.completion_rate)"></p>
+        </div>
+        <div class="bg-white rounded-xl shadow-sm border border-gray-200 p-4">
+            <p class="text-xs text-gray-500 font-medium uppercase tracking-wide">Avg Resolution</p>
+            <p class="text-3xl font-bold text-gray-900 mt-1" x-text="formatHours(totals.avg_resolution_hours)"></p>
+        </div>
+        <div class="bg-white rounded-xl shadow-sm border border-gray-200 p-4">
+            <p class="text-xs text-gray-500 font-medium uppercase tracking-wide">Open Tasks</p>
+            <p class="text-3xl font-bold text-orange-600 mt-1" x-text="totals.open_tickets || 0"></p>
+        </div>
+    </div>
+
+    <!-- Status strip -->
+    <div class="grid grid-cols-2 md:grid-cols-4 gap-4 mb-8">
+        <div class="bg-white rounded-xl shadow-sm border border-gray-200 p-4 flex items-center justify-between">
+            <span class="text-sm font-medium text-sky-700">In Progress</span>
+            <span class="text-2xl font-bold text-sky-700" x-text="totals.in_progress || 0"></span>
+        </div>
+        <div class="bg-white rounded-xl shadow-sm border border-gray-200 p-4 flex items-center justify-between">
+            <span class="text-sm font-medium text-red-700">Escalated</span>
+            <span class="text-2xl font-bold text-red-700" x-text="totals.escalated || 0"></span>
+        </div>
+        <div class="bg-white rounded-xl shadow-sm border border-gray-200 p-4 flex items-center justify-between">
+            <span class="text-sm font-medium text-gray-600">Pending / In review</span>
+            <span class="text-2xl font-bold text-gray-700" x-text="totals.pending || 0"></span>
+        </div>
+        <div class="bg-white rounded-xl shadow-sm border border-gray-200 p-4 flex items-center justify-between">
+            <span class="text-sm font-medium text-gray-500">Cancelled</span>
+            <span class="text-2xl font-bold text-gray-500" x-text="totals.cancelled || 0"></span>
+        </div>
+    </div>
+
+    <!-- Error state -->
+    <div x-show="error" class="mb-6 p-4 bg-red-50 border border-red-200 rounded-xl text-sm text-red-700">
+        <span class="font-semibold">Could not load technician performance.</span>
+        <span x-text="error"></span>
+    </div>
+
+    <!-- Per-technician table -->
+    <div class="bg-white rounded-xl shadow-sm border border-gray-200">
+        <div class="px-5 py-4 border-b border-gray-200 flex flex-wrap items-center justify-between gap-3">
+            <h3 class="text-sm font-semibold text-gray-700 uppercase tracking-wide">By Technician</h3>
+            <div class="flex items-center space-x-2 text-sm">
+                <label for="tech-sort" class="text-gray-500">Sort by</label>
+                <select id="tech-sort" x-model="sortKey" class="border border-gray-300 rounded-lg px-2 py-1.5 text-sm text-gray-700 bg-white">
+                    <option value="completed">Completed</option>
+                    <option value="total_assigned">Workload (assigned)</option>
+                    <option value="completion_rate">Completion rate</option>
+                    <option value="avg_resolution_hours">Avg resolution</option>
+                    <option value="name">Name</option>
+                </select>
+                <button @click="dir = dir === 'desc' ? 'asc' : 'desc'" class="px-2 py-1.5 text-xs border border-gray-300 rounded-lg hover:bg-gray-50 text-gray-600" x-text="dir === 'desc' ? 'Desc ↓' : 'Asc ↑'"></button>
+            </div>
+        </div>
+
+        <div class="overflow-x-auto" x-show="technicians.length">
+            <table class="w-full text-sm">
+                <thead class="bg-gray-50 text-gray-600 text-xs uppercase tracking-wider">
+                    <tr>
+                        <th class="px-5 py-3 text-left">Technician</th>
+                        <th class="px-4 py-3 text-right">Assigned</th>
+                        <th class="px-4 py-3 text-right">Completed</th>
+                        <th class="px-4 py-3 text-right">In Progress</th>
+                        <th class="px-4 py-3 text-right">Escalated</th>
+                        <th class="px-4 py-3 text-right">Pending</th>
+                        <th class="px-4 py-3 text-right">Cancelled</th>
+                        <th class="px-5 py-3 text-left min-w-[160px]">Completion Rate</th>
+                        <th class="px-4 py-3 text-right">Avg Resolution</th>
+                    </tr>
+                </thead>
+                <tbody class="divide-y divide-gray-100">
+                    <template x-for="tech in sortedTechnicians" :key="tech.technician_id">
+                        <tr class="hover:bg-gray-50 transition" :title="breakdownTitle(tech)">
+                            <td class="px-5 py-3">
+                                <div class="flex items-center space-x-3">
+                                    <div class="w-8 h-8 bg-denya-100 rounded-full flex items-center justify-center text-sm font-medium text-denya-700" x-text="initial(tech.name)"></div>
+                                    <span class="font-medium text-gray-800" x-text="tech.name"></span>
+                                </div>
+                            </td>
+                            <td class="px-4 py-3 text-right font-medium text-gray-700" x-text="tech.total_assigned"></td>
+                            <td class="px-4 py-3 text-right font-semibold text-green-700" x-text="tech.completed"></td>
+                            <td class="px-4 py-3 text-right text-sky-700" x-text="tech.in_progress"></td>
+                            <td class="px-4 py-3 text-right" :class="tech.escalated ? 'text-red-600 font-medium' : 'text-gray-400'" x-text="tech.escalated"></td>
+                            <td class="px-4 py-3 text-right text-gray-600" x-text="tech.pending"></td>
+                            <td class="px-4 py-3 text-right" :class="tech.cancelled ? 'text-gray-500' : 'text-gray-300'" x-text="tech.cancelled"></td>
+                            <td class="px-5 py-3">
+                                <div class="flex items-center space-x-2">
+                                    <div class="flex-1 bg-gray-100 rounded-full h-2.5 overflow-hidden">
+                                        <div class="h-full rounded-full transition-all" :class="barClass(tech.completion_rate)" :style="'width: ' + Math.min(tech.completion_rate, 100) + '%'"></div>
+                                    </div>
+                                    <span class="text-xs font-medium text-gray-600 w-12 text-right" x-text="formatRate(tech.completion_rate)"></span>
+                                </div>
+                            </td>
+                            <td class="px-4 py-3 text-right text-gray-700">
+                                <span x-text="formatHours(tech.avg_resolution_hours)"></span>
+                                <span x-show="tech.resolved_without_timestamps > 0" class="block text-[11px] text-gray-400"
+                                      x-text="tech.resolved_without_timestamps + ' task' + (tech.resolved_without_timestamps === 1 ? '' : 's') + ' without timestamps'"></span>
+                            </td>
+                        </tr>
+                    </template>
+                </tbody>
+            </table>
+        </div>
+
+        <!-- Empty state -->
+        <div x-show="!technicians.length && !error" class="px-5 py-16 text-center">
+            <p class="text-gray-500 font-medium">No technician assignments yet</p>
+            <p class="text-gray-400 text-sm mt-1">Once tickets are assigned to a technician they will appear here with their completion rate and resolution time.</p>
+        </div>
+
+        <div x-show="technicians.length" class="px-5 py-3 border-t border-gray-100 text-xs text-gray-400 flex flex-wrap items-center justify-between gap-2">
+            <span>Completion rate = completed (Completed + Closed) ÷ assigned. Avg resolution spans created → closed where the close timestamp exists.</span>
+            <span><span x-text="totals.unassigned_tickets || 0"></span> ticket(s) have no technician assigned and are excluded from the rows above.</span>
+        </div>
+    </div>
+</div>
+
+<script>
+    function techPerformance() {
+        return {
+            technicians: [],
+            totals: {},
+            sortKey: 'completed',
+            dir: 'desc',
+            error: '',
+
+            get isFM() { return app().isFM },
+            get isExecutive() { return app().isExecutive },
+
+            get sortedTechnicians() {
+                const list = [...this.technicians];
+                const key = this.sortKey;
+                const flip = this.dir === 'asc' ? 1 : -1;
+                list.sort((a, b) => {
+                    if (key === 'name') {
+                        return flip * a.name.localeCompare(b.name);
+                    }
+                    let av = a[key];
+                    let bv = b[key];
+                    // Missing resolution times sort last in either direction;
+                    // two missing values fall through to the name tiebreak.
+                    if (key === 'avg_resolution_hours') {
+                        const aMissing = av === null || av === undefined;
+                        const bMissing = bv === null || bv === undefined;
+                        if (aMissing || bMissing) {
+                            if (aMissing && bMissing) return a.name.localeCompare(b.name);
+                            return aMissing ? 1 : -1;
+                        }
+                    }
+                    av = av || 0;
+                    bv = bv || 0;
+                    if (av === bv) return a.name.localeCompare(b.name);
+                    return flip * (av - bv);
+                });
+                return list;
+            },
+
+            async init() {
+                await this.loadData();
+            },
+
+            async loadData() {
+                this.error = '';
+                try {
+                    const data = await app().apiGet('/api/tickets/tech-performance');
+                    this.technicians = data?.technicians || [];
+                    this.totals = data?.totals || {};
+                } catch (e) {
+                    this.error = e.message || 'Unknown error';
+                }
+            },
+
+            formatRate(rate) {
+                if (rate === null || rate === undefined) return '—';
+                return `${rate}%`;
+            },
+
+            formatHours(hours) {
+                if (hours === null || hours === undefined) return '—';
+                if (hours < 1) return `${Math.round(hours * 60)}m`;
+                return `${hours}h`;
+            },
+
+            initial(name) {
+                return (name || '?').charAt(0).toUpperCase();
+            },
+
+            rateClass(rate) {
+                if (!rate) return 'text-gray-400';
+                if (rate >= 70) return 'text-green-600';
+                if (rate >= 40) return 'text-yellow-600';
+                return 'text-red-600';
+            },
+
+            barClass(rate) {
+                if (rate >= 70) return 'bg-green-500';
+                if (rate >= 40) return 'bg-yellow-500';
+                return 'bg-red-500';
+            },
+
+            breakdownTitle(tech) {
+                const parts = Object.entries(tech.status_breakdown || {}).map(([s, n]) => `${s}: ${n}`);
+                return parts.length ? parts.join(' · ') : '';
+            }
+        }
+    }
+</script>
+{% endblock %}

app/templates/login.html

@@ -4,11 +4,9 @@
     <div class="w-full max-w-md" x-data="loginForm()">
         <div class="bg-white rounded-2xl shadow-lg p-8">
             <div class="text-center mb-8">
-                <div class="mx-auto w-16 h-16 bg-denya-100 rounded-full flex items-center justify-center mb-4">
+                <!-- Denya Developers full lockup (same-origin app/static/branding) -->
-                    <svg class="w-8 h-8 text-denya-600" fill="none" stroke="currentColor" viewBox="0 0 24 24">
+                <img src="/static/branding/denya-logo-h96.png" alt="Denya Developers"
-                        <path stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M19 21V5a2 2 0 00-2-2H7a2 2 0 00-2 2v16m14 0h2m-2 0h-5m-9 0H3m2 0h5M9 7h1m-1 4h1m4-4h1m-1 4h1m-5 10v-5a1 1 0 011-1h2a1 1 0 011 1v5m-4 0h4"/>
+                     class="mx-auto mb-4 h-24 w-auto">
-                    </svg>
-                </div>
                 <h1 class="text-2xl font-bold text-gray-900">Denya OneCare</h1>
                 <p class="text-gray-500 mt-1">Sign in to your dashboard</p>
             </div>

docker-compose.yml

@@ -4,10 +4,9 @@ services:
     container_name: denya-onecare
     ports:
       - "8000:8000"
+    env_file:
+      - .env            # git-ignored; see .env.example for required keys
     environment:
-      - DATABASE_URL=sqlite+aiosqlite:///./data/denya_onecare.db
-      - SECRET_KEY=change-me-in-production
-      - CORS_ORIGINS=*
       - DEBUG=false
     volumes:
       - app-data:/app/data

tests/conftest.py

@@ -12,18 +12,32 @@ import tempfile
 
 _TMP_DIR = tempfile.mkdtemp(prefix="denya-test-")
 os.environ["DATABASE_URL"] = f"sqlite+aiosqlite:///{_TMP_DIR}/test.db"
+# HARDENING.md P0.1/P0.2: the app now fails closed without a real SECRET_KEY
+# and an explicit CORS allow-list — tests must satisfy both.
+os.environ.setdefault("SECRET_KEY", "test-secret-key-not-for-production-0123456789abcdef")
+os.environ.setdefault("CORS_ORIGINS", "http://test")
 
+import pytest  # noqa: E402
 import pytest_asyncio  # noqa: E402  (DATABASE_URL must be set before app imports)
 from httpx import ASGITransport, AsyncClient  # noqa: E402
 
 from app.core.database import Base, async_session_factory, engine  # noqa: E402
+from app.core.ratelimit import login_rate_limiter  # noqa: E402
 from app.main import app  # noqa: E402
 from app.models.ticket import Ticket  # noqa: E402
 from app.services.seed import seed_categories, seed_units, seed_users  # noqa: E402
 
 
+@pytest.fixture
+def _reset_login_rate_limiter():
+    """Isolate login rate-limit state between tests (shared in-process store)."""
+    login_rate_limiter.reset()
+    yield
+    login_rate_limiter.reset()
+
+
 @pytest_asyncio.fixture
-async def client():
+async def client(_reset_login_rate_limiter):
     """Async test client with a fresh, seeded database per test."""
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)

tests/test_branding_assets.py

@@ -0,0 +1,80 @@
+"""Denya logo branding — repo-local assets under app/static/branding/.
+
+The official Denya Developers logo derivatives (Gitea release
+``logo-assets-v1``, sha256-verified) are committed same-origin under
+``app/static/branding/`` like the vendored frontend libraries — no CDN, no CSP
+change needed (``img-src 'self' data: blob:`` already covers them).
+
+Placement contract:
+* Login page header uses the h96 full lockup (``denya-logo-h96.png``).
+* Dashboard topbar (base.html nav) uses the h48 full lockup (compact spot:
+  the DEVELOPERS subtext is unreadable below ~48px, so the mark reads as
+  symbol+DENYA — the intended compact treatment).
+* Favicon: 32x32 declared first, 16x16 declared, both in <head>.
+"""
+
+from __future__ import annotations
+
+import re
+
+import pytest
+from httpx import AsyncClient
+
+pytestmark = pytest.mark.asyncio
+
+BRANDING_ASSETS = (
+    "denya-logo.png",
+    "denya-logo-trimmed.png",
+    "denya-logo-h48.png",
+    "denya-logo-h96.png",
+    "denya-logo-64x64.png",
+    "denya-logo-32x32.png",
+    "denya-logo-16x16.png",
+)
+
+
+def _directive_sources(csp: str, directive: str) -> list[str]:
+    """Return the source list of one CSP directive (e.g. ``img-src``)."""
+    for part in csp.split(";"):
+        tokens = part.split()
+        if tokens and tokens[0].strip() == directive:
+            return [t.strip() for t in tokens[1:]]
+    return []
+
+
+async def test_login_page_header_uses_h96_logo(client: AsyncClient):
+    """/login must carry the h96 full-lockup logo (same-origin URL)."""
+    resp = await client.get("/login")
+    assert resp.status_code == 200, resp.text
+    assert "/static/branding/denya-logo-h96.png" in resp.text
+
+
+async def test_topbar_and_favicon_on_dashboard_pages(client: AsyncClient):
+    """Dashboard chrome (base.html) carries h48 topbar logo + both favicons."""
+    resp = await client.get("/dashboard/fm")
+    assert resp.status_code == 200, resp.text
+    assert "/static/branding/denya-logo-h48.png" in resp.text
+    # Favicon 32x32 with 16x16 declared in <head> (link rel="icon").
+    assert 'rel="icon" type="image/png" sizes="32x32" href="/static/branding/denya-logo-32x32.png"' in resp.text
+    assert 'rel="icon" type="image/png" sizes="16x16" href="/static/branding/denya-logo-16x16.png"' in resp.text
+
+
+async def test_branding_assets_served_same_origin(client: AsyncClient):
+    """Every committed branding asset must resolve locally as a PNG."""
+    for name in BRANDING_ASSETS:
+        url = f"/static/branding/{name}"
+        resp = await client.get(url)
+        assert resp.status_code == 200, f"{url} -> {resp.status_code}"
+        assert resp.headers.get("content-type", "").startswith("image/png"), f"{url}: {resp.headers.get('content-type')!r}"
+        assert len(resp.content) > 100, f"{url} looks empty"
+
+
+async def test_csp_serves_branding_without_changes(client: AsyncClient):
+    """img-src already allows same-origin PNGs — no external host needed."""
+    resp = await client.get("/login")
+    assert resp.status_code == 200
+    csp = resp.headers["content-security-policy"]
+    img_sources = _directive_sources(csp, "img-src")
+    assert img_sources, f"no img-src directive in CSP: {csp}"
+    assert "'self'" in img_sources and "data:" in img_sources and "blob:" in img_sources
+    assert "cdn." not in csp  # fully self-contained, like the vendored scripts

tests/test_frontend_vendoring.py

@@ -0,0 +1,104 @@
+"""Frontend must be fully self-contained — no CDN (LAN page-freeze regression).
+
+The P0 batch's templates loaded Alpine.js from ``cdn.jsdelivr.net`` and Tailwind
+from ``cdn.tailwindcss.com``, so any demo client that cannot reach those CDNs
+(LAN-only devices, filtered networks) got a login page whose JS never engaged
+(a stuck form). Both libraries are now vendored under ``app/static/vendor/``
+and served same-origin with no external ``script src`` in the HTML. HTML pages
+always revalidate (``Cache-Control: no-cache``); the vendored assets carry
+long-lived immutable caching (their URLs embed the version).
+"""
+
+from __future__ import annotations
+
+import re
+
+import pytest
+from httpx import AsyncClient
+
+pytestmark = pytest.mark.asyncio
+
+# Any <script … src="//host/…"> or src="https?://host/…"> — i.e. NOT same-origin.
+_EXTERNAL_SRC = re.compile(r"""<script\b[^>]*\bsrc\s*=\s*["'](?:https?:)?//[^"']+["']""")
+
+VENDORED_SCRIPTS = (
+    "/static/vendor/alpine-3.17.2.min.js",
+    "/static/vendor/tailwind-3.4.17.js",
+)
+
+
+async def test_login_page_has_no_external_script_srcs(client: AsyncClient):
+    """/login must reference only same-origin scripts — regex over the body."""
+    resp = await client.get("/login")
+    assert resp.status_code == 200, resp.text
+    body = resp.text
+    external = _EXTERNAL_SRC.findall(body)
+    assert not external, f"external script srcs found: {external}"
+    for src in VENDORED_SCRIPTS:
+        assert src in body, f"missing vendored script {src} in /login HTML"
+
+
+async def test_vendor_assets_served_same_origin(client: AsyncClient):
+    """Both vendored libraries must resolve locally with real JS content."""
+    for src in VENDORED_SCRIPTS:
+        resp = await client.get(src)
+        assert resp.status_code == 200, f"{src} -> {resp.status_code}"
+        assert len(resp.content) > 1000, f"{src} looks empty ({len(resp.content)} bytes)"
+
+
+async def test_html_pages_are_not_cached(client: AsyncClient):
+    """HTML page responses must always revalidate (Cache-Control: no-cache)."""
+    resp = await client.get("/login")
+    assert resp.status_code == 200
+    assert resp.headers["cache-control"] == "no-cache"
+
+
+async def test_vendor_assets_cached_immutable(client: AsyncClient):
+    """Versioned vendor assets must carry long-lived immutable caching."""
+    for src in VENDORED_SCRIPTS:
+        resp = await client.get(src)
+        assert resp.status_code == 200
+        cc = resp.headers.get("cache-control", "")
+        assert "max-age=31536000" in cc and "immutable" in cc, f"{src}: {cc!r}"
+
+
+async def test_csp_no_longer_allows_cdn_hosts(client: AsyncClient):
+    """CSP must be 'self'-only for scripts/styles; connect-src stays 'self'."""
+    resp = await client.get("/login")
+    assert resp.status_code == 200
+    csp = resp.headers["content-security-policy"]
+    for host in ("cdn.jsdelivr.net", "cdn.tailwindcss.com"):
+        assert host not in csp, f"CSP still allows {host}"
+    assert "script-src 'self' 'unsafe-inline'" in csp
+    assert "style-src 'self' 'unsafe-inline'" in csp
+    assert "connect-src 'self'" in csp
+
+
+def _directive_sources(csp: str, directive: str) -> list[str]:
+    """Return the source list of one CSP directive (e.g. ``script-src``)."""
+    for part in csp.split(";"):
+        tokens = part.split()
+        if tokens and tokens[0].strip() == directive:
+            return [t.strip() for t in tokens[1:]]
+    return []
+
+
+async def test_csp_script_src_allows_unsafe_eval_for_alpine(client: AsyncClient):
+    """/login CSP must permit 'unsafe-eval' in script-src (Alpine 3.17.2 runtime).
+
+    Alpine's expression evaluator compiles every ``x-data``/``x-show``/``x-text``
+    expression with ``new Function()``. A strict CSP without ``'unsafe-eval'``
+    blocks each evaluation ("Refused to evaluate a string as JavaScript ..."),
+    Alpine never initializes, and the loading overlay (``x-show="loading"`` in
+    base.html) stays visible forever — regression shipped with the P0 CSP.
+    """
+    resp = await client.get("/login")
+    assert resp.status_code == 200
+    csp = resp.headers["content-security-policy"]
+    script_sources = _directive_sources(csp, "script-src")
+    assert script_sources, f"no script-src directive in CSP: {csp}"
+    assert "'unsafe-eval'" in script_sources, f"script-src missing 'unsafe-eval': {csp}"
+    # Everything else stays as hardened: still 'self'-only apart from the two
+    # Alpine-required relaxations, and connect-src remains 'self'.
+    assert "'self'" in script_sources
+    assert "connect-src 'self'" in csp

tests/test_p0_auth_admin_batch.py

@@ -0,0 +1,647 @@
+"""P0 security batch — admin user management, unified role model, login rate
+limiting, WhatsApp webhook secret, mock-log auth, security headers, pagination.
+
+Each item in the P0 hardening batch has an executable behavioral test here
+(plus the register-removal tests living in test_p0_hardening.py).
+"""
+
+from __future__ import annotations
+
+import pytest
+from httpx import AsyncClient
+
+from app.core.config import settings
+from app.core.database import async_session_factory
+from app.core.security import hash_password
+from app.models.user import User
+from app.services.seed import normalize_legacy_user_emails, normalize_legacy_user_roles
+
+pytestmark = pytest.mark.asyncio
+
+
+# ── helpers ───────────────────────────────────────────────────────────
+async def _login(client, email="wahab@denya.com", password="denya123") -> str:
+    resp = await client.post("/api/auth/login", json={"email": email, "password": password})
+    assert resp.status_code == 200, resp.text
+    return resp.json()["access_token"]
+
+
+def _auth(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}
+
+
+async def _insert_user(email: str, role: str, *, active: bool = True) -> int:
+    """Insert a user row directly (bypasses API role validation) — used to
+    simulate legacy bootstrap/registration rows in the DB."""
+    async with async_session_factory() as session:
+        user = User(
+            email=email,
+            password_hash=hash_password("denya123"),
+            full_name=f"Legacy {email}",
+            role=role,
+            active=active,
+        )
+        session.add(user)
+        await session.commit()
+        return user.id
+
+
+async def _normalize_roles() -> None:
+    async with async_session_factory() as session:
+        await normalize_legacy_user_roles(session)
+        await session.commit()
+
+
+async def _normalize_emails() -> None:
+    async with async_session_factory() as session:
+        await normalize_legacy_user_emails(session)
+        await session.commit()
+
+
+async def _create_ticket(client, token: str, **overrides) -> dict:
+    payload = {
+        "unit_id": 2,
+        "category_id": 3,
+        "priority": "medium",
+        "reporter": "P0 Batch Test",
+        "reported_via": "walk-in",
+        "description": "p0 batch ticket",
+        **overrides,
+    }
+    resp = await client.post("/api/tickets", json=payload, headers=_auth(token))
+    assert resp.status_code == 201, resp.text
+    return resp.json()
+
+
+# ── Item 2/3: admin user management ───────────────────────────────────
+async def test_create_user_requires_admin(client: AsyncClient):
+    token = await _login(client, email="bella@denya.com")  # CS Rep
+    resp = await client.post(
+        "/api/auth/users",
+        json={"email": "x@example.com", "password": "password1", "full_name": "X", "role": "CS Rep"},
+        headers=_auth(token),
+    )
+    assert resp.status_code == 403
+
+
+async def test_create_user_requires_auth(client: AsyncClient):
+    resp = await client.post(
+        "/api/auth/users",
+        json={"email": "x@example.com", "password": "password1", "full_name": "X", "role": "CS Rep"},
+    )
+    assert resp.status_code == 401
+
+
+async def test_admin_creates_user_with_forced_role(client: AsyncClient):
+    token = await _login(client)  # Admin/Wahab
+    resp = await client.post(
+        "/api/auth/users",
+        json={"email": "New.Tech@Example.com", "password": "password1", "full_name": "New Tech", "role": "Tech"},
+        headers=_auth(token),
+    )
+    assert resp.status_code == 201, resp.text
+    created = resp.json()
+    assert created["role"] == "Tech"
+    assert created["active"] is True
+    assert created["email"] == "new.tech@example.com"  # normalized lower-case
+
+    # The new user can actually log in with their forced role.
+    login = await client.post(
+        "/api/auth/login", json={"email": "new.tech@example.com", "password": "password1"}
+    )
+    assert login.status_code == 200
+    me = await client.get("/api/auth/me", headers=_auth(login.json()["access_token"]))
+    assert me.json()["role"] == "Tech"
+
+
+async def test_admin_create_user_rejects_unknown_roles(client: AsyncClient):
+    """Unified role model: junk/legacy roles cannot be minted at creation."""
+    token = await _login(client)
+    for role in ("admin", "superadmin", "technician", "root"):
+        resp = await client.post(
+            "/api/auth/users",
+            json={"email": f"{role.strip().lower()}@example.com", "password": "password1", "full_name": "X", "role": role},
+            headers=_auth(token),
+        )
+        assert resp.status_code == 422, (role, resp.text)
+
+
+async def test_admin_create_user_duplicate_email_conflict(client: AsyncClient):
+    token = await _login(client)
+    payload = {"email": "dupe2@example.com", "password": "password1", "full_name": "D", "role": "CS Rep"}
+    assert (await client.post("/api/auth/users", json=payload, headers=_auth(token))).status_code == 201
+    resp = await client.post("/api/auth/users", json=payload, headers=_auth(token))
+    assert resp.status_code == 409
+
+
+async def test_patch_user_role_change(client: AsyncClient):
+    token = await _login(client)
+    users = (await client.get("/api/auth/users", headers=_auth(token))).json()
+    tech = next(u for u in users if u["role"] == "Tech")
+    resp = await client.patch(
+        f"/api/auth/users/{tech['id']}",
+        json={"role": "CS Rep"},
+        headers=_auth(token),
+    )
+    assert resp.status_code == 200, resp.text
+    assert resp.json()["role"] == "CS Rep"
+    assert resp.json()["active"] is True
+
+
+async def test_patch_user_rejects_unknown_role(client: AsyncClient):
+    token = await _login(client)
+    users = (await client.get("/api/auth/users", headers=_auth(token))).json()
+    tech = next(u for u in users if u["role"] == "Tech")
+    resp = await client.patch(
+        f"/api/auth/users/{tech['id']}",
+        json={"role": "superadmin"},
+        headers=_auth(token),
+    )
+    assert resp.status_code == 422, resp.text
+
+
+async def test_patch_deactivate_blocks_login(client: AsyncClient):
+    token = await _login(client)
+    users = (await client.get("/api/auth/users", headers=_auth(token))).json()
+    tech = next(u for u in users if u["role"] == "Tech")
+    resp = await client.patch(
+        f"/api/auth/users/{tech['id']}",
+        json={"active": False},
+        headers=_auth(token),
+    )
+    assert resp.status_code == 200
+    assert resp.json()["active"] is False
+
+    login = await client.post(
+        "/api/auth/login", json={"email": tech["email"], "password": "denya123"}
+    )
+    assert login.status_code == 401
+
+
+async def test_admin_cannot_modify_own_account(client: AsyncClient):
+    token = await _login(client)  # wahab
+    me = (await client.get("/api/auth/me", headers=_auth(token))).json()
+    resp = await client.patch(
+        f"/api/auth/users/{me['id']}", json={"active": False}, headers=_auth(token)
+    )
+    assert resp.status_code == 400
+
+
+async def test_admin_can_demote_other_admin_but_not_self(client: AsyncClient):
+    """An admin can manage the other admin seat, but self-removal stays blocked,
+    so at least one canonical admin always remains (structural invariant)."""
+    token = await _login(client)  # wahab
+    users = (await client.get("/api/auth/users", headers=_auth(token))).json()
+    jerome = next(u for u in users if u["role"] == "Admin/Jerome")
+    wahab = next(u for u in users if u["role"] == "Admin/Wahab")
+
+    # Demote the OTHER admin → allowed, wahab is still the acting admin.
+    resp = await client.patch(
+        f"/api/auth/users/{jerome['id']}", json={"role": "CEO"}, headers=_auth(token)
+    )
+    assert resp.status_code == 200, resp.text
+
+    # Demoting/deactivating yourself is always rejected.
+    resp = await client.patch(
+        f"/api/auth/users/{wahab['id']}", json={"active": False}, headers=_auth(token)
+    )
+    assert resp.status_code == 400
+
+
+async def test_delete_user_admin_only_and_works(client: AsyncClient):
+    admin_token = await _login(client)
+    users = (await client.get("/api/auth/users", headers=_auth(admin_token))).json()
+    tech = next(u for u in users if u["role"] == "Tech")
+
+    # Non-admin cannot delete.
+    cs_token = await _login(client, email="bella@denya.com")
+    resp = await client.delete(f"/api/auth/users/{tech['id']}", headers=_auth(cs_token))
+    assert resp.status_code == 403
+
+    # Admin deletes → 204, user gone, login fails.
+    resp = await client.delete(f"/api/auth/users/{tech['id']}", headers=_auth(admin_token))
+    assert resp.status_code == 204
+    login = await client.post(
+        "/api/auth/login", json={"email": tech["email"], "password": "denya123"}
+    )
+    assert login.status_code == 401
+
+
+async def test_delete_user_referenced_by_ticket_is_409(client: AsyncClient):
+    token = await _login(client)
+    users = (await client.get("/api/auth/users", headers=_auth(token))).json()
+    tech = next(u for u in users if u["role"] == "Tech")
+    ticket = await _create_ticket(client, token)
+    resp = await client.patch(
+        f"/api/tickets/{ticket['id']}",
+        json={"assigned_to": tech["id"]},
+        headers=_auth(token),
+    )
+    assert resp.status_code == 200, resp.text
+
+    resp = await client.delete(f"/api/auth/users/{tech['id']}", headers=_auth(token))
+    assert resp.status_code == 409
+    assert "related" in resp.json()["detail"].lower()
+
+
+async def test_admin_cannot_delete_self(client: AsyncClient):
+    token = await _login(client)
+    me = (await client.get("/api/auth/me", headers=_auth(token))).json()
+    resp = await client.delete(f"/api/auth/users/{me['id']}", headers=_auth(token))
+    assert resp.status_code == 400
+
+
+# ── Item 3: unified role model — legacy rows & JWT validation ─────────
+async def test_legacy_alias_role_normalized_at_startup(client: AsyncClient):
+    """A legacy 'technician' row is mapped onto canonical 'Tech' at startup."""
+    await _insert_user("legacy-tech@example.com", "technician")
+    await _normalize_roles()  # what lifespan does each boot
+
+    login = await client.post(
+        "/api/auth/login", json={"email": "legacy-tech@example.com", "password": "denya123"}
+    )
+    assert login.status_code == 200, login.text
+    me = await client.get("/api/auth/me", headers=_auth(login.json()["access_token"]))
+    assert me.json()["role"] == "Tech"
+
+
+async def test_login_rejects_unknown_legacy_role_fail_closed(client: AsyncClient):
+    """Lowercase 'superadmin' rows (mintable by the old open register) cannot
+    log in — fail closed, never granted admin powers."""
+    await _insert_user("legacy-admin@example.com", "superadmin")
+    # NOTE: no normalize call — the row is exactly what the live DB holds today.
+
+    login = await client.post(
+        "/api/auth/login", json={"email": "legacy-admin@example.com", "password": "denya123"}
+    )
+    assert login.status_code == 401
+    assert "role" in login.json()["detail"].lower()
+
+
+async def test_jwt_validation_rejects_unknown_role(client: AsyncClient):
+    """A token for a user whose row later becomes junk-role must fail closed."""
+    token = await _login(client)  # wahab is a canonical admin at token time
+    me = (await client.get("/api/auth/me", headers=_auth(token))).json()
+
+    # Simulate a legacy DB row flip to a non-canonical role.
+    async with async_session_factory() as session:
+        from sqlalchemy import select
+        user = (await session.execute(select(User).where(User.id == me["id"]))).scalar_one()
+        user.role = "admin"
+        await session.commit()
+
+    resp = await client.get("/api/auth/me", headers=_auth(token))
+    assert resp.status_code == 401
+
+
+async def test_normalize_does_not_map_ambiguous_admin_alias(client: AsyncClient):
+    """normalize_legacy_user_roles leaves identity-ambiguous 'admin' rows for
+    operator remediation instead of guessing a canonical admin."""
+    await _insert_user("legacy-admin2@example.com", "admin")
+    await _normalize_roles()
+
+    async with async_session_factory() as session:
+        from sqlalchemy import select
+        user = (
+            await session.execute(select(User).where(User.email == "legacy-admin2@example.com"))
+        ).scalar_one()
+    assert user.role == "admin"
+
+
+async def test_underscore_legacy_aliases_normalize_to_canonical():
+    """Open-register rows can carry underscore role forms ('cs_rep',
+    'cs_manager', 'fm_dispatcher') — the exact gap Mumuni relay #747 found on
+    user 18 (test@denya.com).  Each must map onto its canonical role so startup
+    normalization can converge the row instead of stranding it on /tickets."""
+    from app.core.roles import normalize_role
+
+    assert normalize_role("cs_rep") == "CS Rep"
+    assert normalize_role("cs_manager") == "CS Manager"
+    assert normalize_role("fm_dispatcher") == "FM Dispatcher"
+    # Matching is case/whitespace tolerant, like the space-form aliases.
+    assert normalize_role(" CS_REP ") == "CS Rep"
+    assert normalize_role("cs_rep") is not None  # known, not fail-closed
+
+
+async def test_legacy_cs_rep_row_converges_and_authenticates(client: AsyncClient):
+    """A 'cs_rep' row (user 18 test@denya.com shape) is converged to canonical
+    'CS Rep' by the startup self-heal and can log in again (no fail-closed
+    denial, and the frontend CS nav sees the canonical role)."""
+    await _insert_user("cs-rep-legacy@example.com", "cs_rep")
+    await _normalize_roles()  # what lifespan does each boot
+
+    async with async_session_factory() as session:
+        from sqlalchemy import select
+        user = (
+            await session.execute(select(User).where(User.email == "cs-rep-legacy@example.com"))
+        ).scalar_one()
+    assert user.role == "CS Rep"
+
+    login = await client.post(
+        "/api/auth/login", json={"email": "cs-rep-legacy@example.com", "password": "denya123"}
+    )
+    assert login.status_code == 200, login.text
+    me = await client.get("/api/auth/me", headers=_auth(login.json()["access_token"]))
+    assert me.json()["role"] == "CS Rep"
+
+
+# ── P0 email normalization (legacy mixed-case rows) ───────────────────
+async def test_legacy_mixed_case_email_migrated_and_authenticates(client: AsyncClient):
+    """A legacy row whose email was stored verbatim in mixed case (the old open
+    register) is lowercased by the startup self-heal and still authenticates."""
+    await _insert_user("DemoUser@Example.com", "Tech")
+    await _normalize_emails()  # what lifespan does each boot
+
+    async with async_session_factory() as session:
+        from sqlalchemy import select
+        user = (
+            await session.execute(select(User).where(User.email == "demouser@example.com"))
+        ).scalar_one()
+    assert user.email == "demouser@example.com"
+
+    for variant in ("demouser@example.com", "DemoUser@Example.com"):
+        resp = await client.post(
+            "/api/auth/login", json={"email": variant, "password": "denya123"}
+        )
+        assert resp.status_code == 200, resp.text
+
+
+async def test_login_matches_legacy_mixed_case_email_before_migration(client: AsyncClient):
+    """Login compares on the normalized form, so an un-migrated mixed-case row
+    is still matched by its lowercase login (no hard dependency on the
+    self-heal having run)."""
+    await _insert_user("DemoUser@Example.com", "Tech")
+    resp = await client.post(
+        "/api/auth/login", json={"email": "demouser@example.com", "password": "denya123"}
+    )
+    assert resp.status_code == 200, resp.text
+
+
+async def test_legacy_email_normalization_is_idempotent(client: AsyncClient):
+    """The startup self-heal rewrites once and no-ops on subsequent boots."""
+    await _insert_user("DemoUser@Example.com", "Tech")
+    async with async_session_factory() as session:
+        first = await normalize_legacy_user_emails(session)
+        await session.commit()
+    async with async_session_factory() as session:
+        second = await normalize_legacy_user_emails(session)
+        await session.commit()
+    assert first == 1
+    assert second == 0
+
+
+async def test_create_user_rejects_case_variant_of_legacy_email(client: AsyncClient):
+    """The admin create-user duplicate check compares on the normalized form:
+    creating a case-variant of a legacy mixed-case row returns 409, not 201."""
+    await _insert_user("DemoUser@Example.com", "Tech")
+    token = await _login(client)
+    resp = await client.post(
+        "/api/auth/users",
+        json={
+            "email": "demouser@example.com",
+            "password": "password1",
+            "full_name": "X",
+            "role": "Tech",
+        },
+        headers=_auth(token),
+    )
+    assert resp.status_code == 409, resp.text
+
+
+async def test_admin_only_rbac_gate(client: AsyncClient):
+    """Canonical admins pass /api/auth/admin-only; everyone else 403."""
+    wahab = await _login(client)
+    assert (await client.get("/api/auth/admin-only", headers=_auth(wahab))).status_code == 200
+
+    bella = await _login(client, email="bella@denya.com")
+    assert (await client.get("/api/auth/admin-only", headers=_auth(bella))).status_code == 403
+
+
+# ── Item 4: login rate limiting ───────────────────────────────────────
+async def test_login_rate_limited_after_five_failures(client: AsyncClient):
+    email, password = "rate-limited@example.com", "denya123"
+    # Make sure the account exists with a valid password.
+    token = await _login(client)
+    await client.post(
+        "/api/auth/users",
+        json={"email": email, "password": password, "full_name": "Rate", "role": "CS Rep"},
+        headers=_auth(token),
+    )
+
+    for _ in range(5):
+        resp = await client.post(
+            "/api/auth/login", json={"email": email, "password": "wrong-password"}
+        )
+        assert resp.status_code == 401
+
+    # 6th attempt — even with the CORRECT password — is throttled.
+    resp = await client.post(
+        "/api/auth/login", json={"email": email, "password": password}
+    )
+    assert resp.status_code == 429, resp.text
+
+
+async def test_rate_limit_is_per_email(client: AsyncClient):
+    """Failures for one account never lock out another account."""
+    token = await _login(client)
+    for email in ("victim@example.com", "other@example.com"):
+        await client.post(
+            "/api/auth/users",
+            json={"email": email, "password": "password1", "full_name": "U", "role": "CS Rep"},
+            headers=_auth(token),
+        )
+
+    for _ in range(6):
+        resp = await client.post(
+            "/api/auth/login", json={"email": "victim@example.com", "password": "bad"}
+        )
+        assert resp.status_code in (401, 429)
+
+    # Unaffected account still logs in fine.
+    resp = await client.post(
+        "/api/auth/login", json={"email": "other@example.com", "password": "password1"}
+    )
+    assert resp.status_code == 200, resp.text
+
+
+async def test_rate_limit_window_expires(client: AsyncClient, monkeypatch):
+    """After the 15-minute window passes, the account can log in again."""
+    import time as _time
+
+    import app.core.ratelimit as ratelimit_mod
+
+    email, password = "window@example.com", "password1"
+    token = await _login(client)
+    await client.post(
+        "/api/auth/users",
+        json={"email": email, "password": password, "full_name": "W", "role": "CS Rep"},
+        headers=_auth(token),
+    )
+
+    # Pin the limiter clock so the window can be fast-forwarded deterministically.
+    clock = {"now": _time.time()}
+    monkeypatch.setattr(ratelimit_mod, "_now", lambda: clock["now"])
+    for _ in range(5):
+        await client.post("/api/auth/login", json={"email": email, "password": "bad"})
+
+    blocked = await client.post("/api/auth/login", json={"email": email, "password": password})
+    assert blocked.status_code == 429, blocked.text
+
+    clock["now"] += settings.LOGIN_RATE_LIMIT_WINDOW_SECONDS + 1
+    resp = await client.post("/api/auth/login", json={"email": email, "password": password})
+    assert resp.status_code == 200, resp.text
+
+
+# ── Item 5: WhatsApp webhook secret (fail closed) ─────────────────────
+WEBHOOK_BODY = {
+    "object": "whatsapp_business_account",
+    "entry": [
+        {
+            "id": "1",
+            "changes": [
+                {
+                    "id": "wamid.1",
+                    "message": {"from": "+233000000000", "id": "wamid.1", "text": {"text": "AC leaking"}},
+                }
+            ],
+        }
+    ],
+}
+
+
+async def test_webhook_fail_closed_when_env_unset(client: AsyncClient):
+    """WHATSAPP_WEBHOOK_SECRET unset ⇒ every message rejected (403)."""
+    assert settings.WHATSAPP_WEBHOOK_SECRET == ""  # test env default is unset
+    resp = await client.post("/api/whatsapp/webhook", json=WEBHOOK_BODY)
+    assert resp.status_code == 403
+    assert "not configured" in resp.json()["detail"].lower()
+
+
+async def test_webhook_rejects_missing_or_wrong_secret(client: AsyncClient, monkeypatch):
+    monkeypatch.setattr(settings, "WHATSAPP_WEBHOOK_SECRET", "test-webhook-secret")
+    resp = await client.post("/api/whatsapp/webhook", json=WEBHOOK_BODY)
+    assert resp.status_code == 403
+
+    resp = await client.post(
+        "/api/whatsapp/webhook", json=WEBHOOK_BODY, headers={"X-Webhook-Secret": "wrong"}
+    )
+    assert resp.status_code == 403
+
+
+async def test_webhook_accepts_valid_secret_and_creates_ticket(client: AsyncClient, monkeypatch):
+    from app.routers import whatsapp as whatsapp_router
+
+    async def _fake_reply(to_phone: str, text: str):
+        from app.schemas.whatsapp import WhatsAppReplyResponse
+        return WhatsAppReplyResponse(success=True, message="sent")
+
+    monkeypatch.setattr(settings, "WHATSAPP_WEBHOOK_SECRET", "test-webhook-secret")
+    monkeypatch.setattr(whatsapp_router, "send_whatsapp_reply", _fake_reply)
+
+    resp = await client.post(
+        "/api/whatsapp/webhook",
+        json=WEBHOOK_BODY,
+        headers={"X-Webhook-Secret": "test-webhook-secret"},
+    )
+    assert resp.status_code == 200, resp.text
+    data = resp.json()
+    assert data["status"] == "processed"
+    assert data["ticket_number"].startswith("PAV-")
+
+    # The message is logged and visible to an authenticated mock-log caller.
+    token = await _login(client)
+    log = await client.get("/api/whatsapp/mock-log", headers=_auth(token))
+    assert log.status_code == 200
+    assert len(log.json()) == 1
+    assert log.json()[0]["ticket_number"] == data["ticket_number"]
+
+
+async def test_webhook_verify_token_mismatch_403(client: AsyncClient, monkeypatch):
+    monkeypatch.setattr(settings, "WHATSAPP_VERIFY_TOKEN", "verify-me")
+    resp = await client.get(
+        "/api/whatsapp/webhook",
+        params={"hub.mode": "subscribe", "hub.verify_token": "nope", "hub.challenge": "1234"},
+    )
+    assert resp.status_code == 403
+
+
+async def test_webhook_verify_token_match_returns_challenge(client: AsyncClient, monkeypatch):
+    monkeypatch.setattr(settings, "WHATSAPP_VERIFY_TOKEN", "verify-me")
+    resp = await client.get(
+        "/api/whatsapp/webhook",
+        params={"hub.mode": "subscribe", "hub.verify_token": "verify-me", "hub.challenge": "1234"},
+    )
+    assert resp.status_code == 200
+    assert resp.json() == {"challenge": "1234"}
+
+
+# ── Item 6: mock-log requires auth ────────────────────────────────────
+async def test_mock_log_requires_auth(client: AsyncClient):
+    resp = await client.get("/api/whatsapp/mock-log")
+    assert resp.status_code == 401, resp.text
+
+    token = await _login(client)
+    resp = await client.get("/api/whatsapp/mock-log", headers=_auth(token))
+    assert resp.status_code == 200
+    assert resp.json() == []
+
+
+# ── Item 7: security headers ──────────────────────────────────────────
+async def test_security_headers_on_html_page(client: AsyncClient):
+    resp = await client.get("/login")
+    assert resp.status_code == 200
+    assert resp.headers["x-frame-options"] == "DENY"
+    assert resp.headers["x-content-type-options"] == "nosniff"
+    assert "content-security-policy" in resp.headers
+    assert "default-src 'self'" in resp.headers["content-security-policy"]
+
+
+async def test_csp_only_on_html_not_json_api(client: AsyncClient):
+    resp = await client.get("/api/tickets")
+    assert resp.status_code == 200
+    assert "content-type" in resp.headers and resp.headers["content-type"].startswith("application/json")
+    assert "content-security-policy" not in resp.headers
+    # Frame/type hardening headers apply everywhere.
+    assert resp.headers["x-frame-options"] == "DENY"
+    assert resp.headers["x-content-type-options"] == "nosniff"
+
+
+async def test_hsts_only_when_tls_terminates(client: AsyncClient):
+    plain = await client.get("/login")
+    assert "strict-transport-security" not in plain.headers
+
+    tls = await client.get("/login", headers={"X-Forwarded-Proto": "https"})
+    assert tls.headers["strict-transport-security"] == "max-age=31536000; includeSubDomains"
+
+
+# ── Item 8: pagination (page/limit) and sane max page size ────────────
+async def test_limit_alias_over_cap_rejected(client: AsyncClient, seed_tickets):
+    await seed_tickets(10)
+    resp = await client.get("/api/tickets", params={"limit": 500})
+    assert resp.status_code == 422
+
+
+async def test_limit_alias_paginates(client: AsyncClient, seed_tickets):
+    await seed_tickets(14)
+    resp = await client.get("/api/tickets", params={"page": 2, "limit": 5})
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["total"] == 14
+    assert len(data["items"]) == 5
+    assert data["page_size"] == 5
+    assert data["page"] == 2
+
+
+async def test_page_beyond_last_returns_empty_with_total(client: AsyncClient, seed_tickets):
+    await seed_tickets(7)
+    resp = await client.get("/api/tickets", params={"page": 999, "page_size": 10})
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["items"] == []
+    assert data["total"] == 7
+
+
+async def test_page_size_and_limit_conflict_is_422(client: AsyncClient, seed_tickets):
+    await seed_tickets(3)
+    resp = await client.get("/api/tickets", params={"page_size": 10, "limit": 20})
+    assert resp.status_code == 422

tests/test_p0_hardening.py

@@ -0,0 +1,132 @@
+"""P0 hardening regression tests (HARDENING.md P0.1 / P0.2 / P0.3).
+
+Covers:
+- P0.3: self-registration is REMOVED — POST /api/auth/register returns 404 and
+  no public path can mint a user at all (users are admin-created only).
+- P0.1: app fails to import/boot with placeholder or missing SECRET_KEY
+- P0.2: app fails to boot with CORS_ORIGINS="*"
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+from httpx import AsyncClient
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+
+pytestmark = pytest.mark.asyncio
+
+
+# ── P0.3: self-registration is removed entirely ───────────────────────
+
+async def test_register_endpoint_is_removed(client: AsyncClient):
+    """A raw unauthenticated register call must 404 — no public signup path."""
+    resp = await client.post(
+        "/api/auth/register",
+        json={
+            "email": "attacker@example.com",
+            "password": "Sup3rSecret!",
+            "full_name": "Attacker",
+            "role": "Admin/Jerome",
+        },
+    )
+    assert resp.status_code == 404, resp.text
+
+
+async def test_register_endpoint_removed_regardless_of_role(client: AsyncClient):
+    """Attempts to mint privileged (or any) roles via register all 404."""
+    for role in ("Admin/Jerome", "Admin/Wahab", "admin", "superadmin", "CS Rep"):
+        resp = await client.post(
+            "/api/auth/register",
+            json={
+                "email": f"attacker-{role.lower().replace('/', '-')}@example.com",
+                "password": "Sup3rSecret!",
+                "full_name": "Attacker",
+                "role": role,
+            },
+        )
+        assert resp.status_code == 404, (role, resp.text)
+
+
+async def test_register_does_not_create_user(client: AsyncClient):
+    """No user row is ever created through the removed register endpoint."""
+    await client.post(
+        "/api/auth/register",
+        json={
+            "email": "ghost@example.com",
+            "password": "Sup3rSecret!",
+            "full_name": "Ghost",
+        },
+    )
+    # The ghost account must not be able to log in.
+    resp = await client.post(
+        "/api/auth/login",
+        json={"email": "ghost@example.com", "password": "Sup3rSecret!"},
+    )
+    assert resp.status_code == 401, resp.text
+
+
+def _boot_with_env(env_overrides: dict[str, str]) -> subprocess.CompletedProcess:
+    """Try importing app.main in a subprocess with the given env; the import
+    must fail (non-zero) when fail-closed validation trips."""
+    env = os.environ.copy()
+    env["DATABASE_URL"] = "sqlite+aiosqlite:///:memory:"
+    env.pop("SECRET_KEY", None)
+    env.pop("CORS_ORIGINS", None)
+    env.update(env_overrides)
+    script = (
+        "import sys; sys.path.insert(0, ''); "
+        "import app.main"  # noqa
+    )
+    return subprocess.run(
+        [sys.executable, "-c", script],
+        cwd=str(REPO_ROOT),
+        env=env,
+        capture_output=True,
+        text=True,
+        timeout=60,
+    )
+
+
+def test_boot_fails_with_placeholder_secret():
+    result = _boot_with_env({"SECRET_KEY": "change-me-in-production"})
+    assert result.returncode != 0, "app booted with placeholder SECRET_KEY!"
+    assert "SECRET_KEY" in result.stderr
+
+
+def test_boot_fails_with_short_secret():
+    result = _boot_with_env({"SECRET_KEY": "tooshort"})
+    assert result.returncode != 0, "app booted with a <32-char SECRET_KEY!"
+    assert "SECRET_KEY" in result.stderr
+
+
+def test_boot_fails_without_secret():
+    result = _boot_with_env({"SECRET_KEY": ""})
+    assert result.returncode != 0, "app booted without a SECRET_KEY!"
+    assert "SECRET_KEY" in result.stderr
+
+
+def test_boot_fails_with_wildcard_cors():
+    result = _boot_with_env(
+        {
+            "SECRET_KEY": "test-secret-key-not-for-production-0123456789abcdef",
+            "CORS_ORIGINS": "*",
+        }
+    )
+    assert result.returncode != 0, "app booted with CORS_ORIGINS=* !"
+    assert "CORS_ORIGINS" in result.stderr
+
+
+def test_boot_succeeds_with_valid_env():
+    result = _boot_with_env(
+        {
+            "SECRET_KEY": "test-secret-key-not-for-production-0123456789abcdef",
+            "CORS_ORIGINS": "http://test",
+        }
+    )
+    assert result.returncode == 0, result.stderr

tests/test_tech_performance.py

@@ -0,0 +1,287 @@
+"""Tests for the technician-performance dashboard and the ``Tech #N`` fix.
+
+Anchors two Wahab-facing defects/requests:
+
+1. The FM dashboard built its "Technician Workload" card from
+   ``Tech #<user id>`` instead of the technician's real name. The data sources
+   the card consumes must expose the technician's real name via
+   ``assigned_technician_name``, never an id placeholder.
+2. ``GET /api/tickets/tech-performance`` reports per-technician workload and
+   outcomes by real name: totals, per-status buckets, completion rate and
+   ``created_at -> closed_at`` resolution times. Everything derives from
+   existing ticket columns — no schema change.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+import pytest
+from sqlalchemy import select
+
+from app.core.database import async_session_factory
+from app.models.ticket import Ticket
+from app.models.user import User
+
+pytestmark = pytest.mark.asyncio
+
+
+async def _login(client, email: str = "wahab@denya.com", password: str = "denya123") -> str:
+    resp = await client.post("/api/auth/login", json={"email": email, "password": password})
+    assert resp.status_code == 200, resp.text
+    return resp.json()["access_token"]
+
+
+def _auth(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}
+
+
+async def _user_id(email: str) -> int:
+    async with async_session_factory() as session:
+        user = (
+            await session.execute(select(User).where(User.email == email))
+        ).scalar_one()
+        return user.id
+
+
+async def _seed_tickets(specs: list[dict]) -> None:
+    """Insert tickets directly so tests control status/timestamps exactly."""
+    async with async_session_factory() as session:
+        for i, spec in enumerate(specs):
+            session.add(Ticket(ticket_number=f"PAV-PERF-{i:05d}", **spec))
+        await session.commit()
+
+
+async def _make_user(email: str, full_name: str, role: str = "Tech") -> int:
+    """Create a user directly (no API) so tests can build same-name technicians."""
+    async with async_session_factory() as session:
+        user = User(email=email, password_hash="not-a-real-hash", full_name=full_name, role=role)
+        session.add(user)
+        await session.commit()
+        await session.refresh(user)
+        return user.id
+
+
+def _ticket(status: str, assigned_to: int | None, *, created_at: datetime | None = None,
+            closed_at: datetime | None = None) -> dict:
+    spec: dict = {
+        "status": status,
+        "priority": "medium",
+        "description": f"{status} ticket",
+        "assigned_to": assigned_to,
+    }
+    if created_at is not None:
+        spec["created_at"] = created_at
+    if closed_at is not None:
+        spec["closed_at"] = closed_at
+    return spec
+
+
+# ── 3a. Real technician names on the workload data paths ─────────────
+async def test_workload_sources_report_real_technician_names(client):
+    """The FM workload card reads ``/api/tickets?page_size=200`` and the
+    performance report reads ``/api/tickets/tech-performance``. For an assigned
+    ticket both must expose the technician's real full name, never a
+    ``Tech #<id>`` placeholder.
+    """
+    prosper = await _user_id("prosper@denya.com")
+    await _seed_tickets([_ticket("In Progress", prosper), _ticket("Escalated", prosper)])
+
+    token = await _login(client)
+    list_resp = await client.get("/api/tickets?page_size=200", headers=_auth(token))
+    assert list_resp.status_code == 200
+    active = [
+        t for t in list_resp.json()["items"]
+        if t["status"] not in ("Closed", "Completed", "Cancelled")
+    ]
+
+    # Same mapping the FM workload card builds from active assigned tickets.
+    workload: dict[int, str] = {}
+    for t in active:
+        if t["assigned_to"]:
+            workload.setdefault(t["assigned_to"], t["assigned_technician_name"] or "Unassigned")
+    assert workload == {prosper: "Prosper"}
+
+    perf_resp = await client.get("/api/tickets/tech-performance", headers=_auth(token))
+    assert perf_resp.status_code == 200
+    rows = {t["technician_id"]: t for t in perf_resp.json()["technicians"]}
+    assert rows[prosper]["name"] == "Prosper"
+    assert all(not row["name"].startswith("Tech #") for row in rows.values())
+
+
+# ── Page + route wiring ──────────────────────────────────────────────
+async def test_tech_performance_page_is_same_origin(client):
+    """The new dashboard is reachable and uses only same-origin assets."""
+    resp = await client.get("/dashboard/tech-performance")
+    assert resp.status_code == 200
+    body = resp.text
+    assert "/api/tickets/tech-performance" in body
+    assert "https://" not in body
+    assert "cdn." not in body
+
+
+async def test_base_nav_links_fm_and_ceo_to_tech_performance(client):
+    """FM and CEO navigation exposes the report."""
+    for path in ("/dashboard/fm", "/dashboard/ceo"):
+        body = (await client.get(path)).text
+        assert "/dashboard/tech-performance" in body
+
+
+async def test_tech_performance_requires_auth(client):
+    """The aggregate endpoint is bearer-gated, like other dashboard data paths."""
+    resp = await client.get("/api/tickets/tech-performance")
+    assert resp.status_code == 401
+
+
+async def test_tech_performance_route_not_shadowed_by_ticket_id(client):
+    """`/tech-performance` is a literal route, not an int ticket id (no 422)."""
+    token = await _login(client)
+    resp = await client.get("/api/tickets/tech-performance", headers=_auth(token))
+    assert resp.status_code == 200
+
+
+# ── 3b. Aggregation ──────────────────────────────────────────────────
+async def test_tech_performance_empty_state(client):
+    """No assigned tickets → empty rows and a fully zeroed roll-up."""
+    token = await _login(client)
+    resp = await client.get("/api/tickets/tech-performance", headers=_auth(token))
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["technicians"] == []
+    assert data["totals"]["technicians"] == 0
+    assert data["totals"]["total_assigned"] == 0
+    assert data["totals"]["completion_rate"] == 0.0
+    assert data["totals"]["avg_resolution_hours"] is None
+
+
+async def test_tech_performance_aggregates_by_real_name(client):
+    """Counts, completion rate, aging, sorting and totals per technician."""
+    prosper = await _user_id("prosper@denya.com")
+    sam = await _user_id("sam@denya.com")
+
+    await _seed_tickets(
+        [
+            # Prosper: 2 completed (one timestamped), 1 escalated, 1 in progress, 1 cancelled
+            _ticket("Completed", prosper, created_at=datetime(2026, 1, 1, 0, 0),
+                    closed_at=datetime(2026, 1, 1, 10, 0)),
+            _ticket("Completed", prosper),
+            _ticket("Escalated", prosper),
+            _ticket("In Progress", prosper),
+            _ticket("Cancelled", prosper),
+            # Sam: 3 closed (timestamped), 1 assigned, 1 awaiting verification
+            _ticket("Closed", sam, created_at=datetime(2026, 1, 1, 0, 0),
+                    closed_at=datetime(2026, 1, 2, 0, 0)),
+            _ticket("Closed", sam, created_at=datetime(2026, 1, 1, 0, 0),
+                    closed_at=datetime(2026, 1, 3, 0, 0)),
+            _ticket("Closed", sam, created_at=datetime(2026, 1, 1, 0, 0),
+                    closed_at=datetime(2026, 1, 4, 0, 0)),
+            _ticket("Assigned", sam),
+            _ticket("On-Field Verification", sam),
+            # Unassigned tickets are reported separately, never as a fake technician.
+            _ticket("Logged", None),
+            _ticket("New", None),
+        ]
+    )
+
+    token = await _login(client)
+    resp = await client.get("/api/tickets/tech-performance", headers=_auth(token))
+    assert resp.status_code == 200
+    data = resp.json()
+
+    by_name = {t["name"]: t for t in data["technicians"]}
+    assert set(by_name) == {"Prosper", "Sam"}
+    assert all("Tech #" not in name for name in by_name)
+
+    prosper_row = by_name["Prosper"]
+    assert prosper_row["technician_id"] == prosper
+    assert prosper_row["total_assigned"] == 5
+    assert prosper_row["completed"] == 2
+    assert prosper_row["closed"] == 0
+    assert prosper_row["in_progress"] == 1
+    assert prosper_row["escalated"] == 1
+    assert prosper_row["cancelled"] == 1
+    assert prosper_row["pending"] == 0
+    # Buckets always reconcile with the assigned total.
+    assert (
+        prosper_row["completed"] + prosper_row["in_progress"] + prosper_row["escalated"]
+        + prosper_row["cancelled"] + prosper_row["pending"]
+    ) == prosper_row["total_assigned"]
+    assert prosper_row["open_tickets"] == 2
+    assert prosper_row["completion_rate"] == 40.0
+    assert prosper_row["avg_resolution_hours"] == 10.0
+    assert prosper_row["resolved_with_timestamps"] == 1
+    assert prosper_row["resolved_without_timestamps"] == 1
+    assert prosper_row["status_breakdown"] == {"Completed": 2, "Cancelled": 1, "Escalated": 1, "In Progress": 1}
+
+    sam_row = by_name["Sam"]
+    assert sam_row["total_assigned"] == 5
+    assert sam_row["completed"] == 3
+    assert sam_row["closed"] == 3
+    assert sam_row["in_progress"] == 0
+    assert sam_row["pending"] == 2
+    assert sam_row["open_tickets"] == 2
+    assert sam_row["completion_rate"] == 60.0
+    assert sam_row["avg_resolution_hours"] == 48.0
+    assert sam_row["resolved_without_timestamps"] == 0
+    assert sam_row["status_breakdown"] == {"Closed": 3, "Assigned": 1, "On-Field Verification": 1}
+
+    # Sorted by completed desc → Sam first.
+    assert [t["name"] for t in data["technicians"]] == ["Sam", "Prosper"]
+
+    totals = data["totals"]
+    assert totals["technicians"] == 2
+    assert totals["total_assigned"] == 10
+    assert totals["completed"] == 5
+    assert totals["closed"] == 3
+    assert totals["in_progress"] == 1
+    assert totals["escalated"] == 1
+    assert totals["cancelled"] == 1
+    assert totals["pending"] == 2
+    assert totals["open_tickets"] == 4
+    assert totals["completion_rate"] == 50.0
+    # Fleet average is weighted over tickets, not an average of per-tech averages.
+    assert totals["avg_resolution_hours"] == 38.5
+    assert totals["resolved_with_timestamps"] == 4
+    assert totals["resolved_without_timestamps"] == 1
+    assert totals["unassigned_tickets"] == 2
+    assert totals["total_tickets"] == 12
+
+
+async def test_tech_performance_reports_counts_when_timestamps_absent(client):
+    """Finished tasks without ``closed_at`` yield no average but a count."""
+    afful = await _user_id("afful@denya.com")
+    await _seed_tickets([
+        _ticket("Completed", afful),
+        _ticket("Closed", afful),
+        _ticket("Completed", afful),
+    ])
+
+    token = await _login(client)
+    resp = await client.get("/api/tickets/tech-performance", headers=_auth(token))
+    row = next(t for t in resp.json()["technicians"] if t["name"] == "Afful")
+    assert row["completed"] == 3
+    assert row["closed"] == 1
+    assert row["avg_resolution_hours"] is None
+    assert row["resolved_with_timestamps"] == 0
+    assert row["resolved_without_timestamps"] == 3
+
+
+async def test_tech_performance_groups_same_name_by_user_id(client):
+    """Two technicians sharing a display name stay separate rows."""
+    first = await _make_user("kwame.one@denya.com", "Kwame Mensah")
+    second = await _make_user("kwame.two@denya.com", "Kwame Mensah")
+    await _seed_tickets([
+        _ticket("Completed", first),
+        _ticket("Completed", second),
+        _ticket("Escalated", second),
+    ])
+
+    token = await _login(client)
+    resp = await client.get("/api/tickets/tech-performance", headers=_auth(token))
+    rows = [t for t in resp.json()["technicians"] if t["name"] == "Kwame Mensah"]
+    assert len(rows) == 2
+    assert {r["technician_id"] for r in rows} == {first, second}
+    by_id = {r["technician_id"]: r for r in rows}
+    assert by_id[first]["total_assigned"] == 1
+    assert by_id[second]["total_assigned"] == 2
+    assert by_id[second]["escalated"] == 1

tests/test_whatsapp_demo_number.py

@@ -0,0 +1,164 @@
+"""WhatsApp demo-number wiring (WHATSAPP_DEMO_TO).
+
+The demo round trip (webhook POST -> ticket -> auto-reply -> mock-log) surfaces
+the expected sender/recipient number. That number is **never committed**: it
+lives only in the deploy host's .env and reaches the app through the
+``WHATSAPP_DEMO_TO`` setting (same fail-closed env pattern as the webhook
+secret). ``build_demo_webhook_payload()`` builds the demo payload from the
+setting so mock-log and the auto-reply show the configured number; with the
+setting unset it raises instead of fabricating a sender.
+
+Anchors:
+* ``settings.WHATSAPP_DEMO_TO`` defaults to ``""`` and no tracked file assigns
+  it a value (real numbers stay out of git history).
+* ``build_demo_webhook_payload`` uses the configured number as the message
+  ``from`` and fails closed when unset.
+* A full round trip with the secret + demo number logs the number and surfaces
+  it in ``/api/whatsapp/mock-log``; the auto-reply targets the same number.
+* Webhook stays 403 without the secret and mock-log stays 401 without auth.
+"""
+
+from __future__ import annotations
+
+import re
+import subprocess
+from pathlib import Path
+
+import pytest
+
+pytestmark = pytest.mark.asyncio
+
+from app.core.config import settings  # noqa: E402
+
+# Clearly-fake test number — never use a real contact number in source.
+_FAKE_DEMO_TO = "+233559999999"
+
+
+async def _login(client, email="wahab@denya.com", password="denya123") -> str:
+    resp = await client.post(
+        "/api/auth/login",
+        json={"email": email, "password": password},
+    )
+    assert resp.status_code == 200, resp.text
+    return resp.json()["access_token"]
+
+
+def _auth(token: str) -> dict:
+    return {"Authorization": f"Bearer {token}"}
+
+
+async def _capture_reply(monkeypatch, calls: list):
+    """Swap the Meta client for a recorder; returns the fake."""
+    from app.routers import whatsapp as whatsapp_router
+    from app.schemas.whatsapp import WhatsAppReplyResponse
+
+    async def _fake_reply(to_phone: str, text: str):
+        calls.append((to_phone, text))
+        return WhatsAppReplyResponse(success=True, message="sent")
+
+    monkeypatch.setattr(whatsapp_router, "send_whatsapp_reply", _fake_reply)
+    return whatsapp_router
+
+
+# ── Config plumbing ───────────────────────────────────────────────────
+def test_demo_number_defaults_empty_and_never_committed():
+    """WHATSAPP_DEMO_TO must default empty; no tracked file may set a value.
+
+    Real WhatsApp numbers are deploy-host .env secrets — a committed value
+    (even in tests or .env.example) would leak into git history.
+    """
+    assert settings.WHATSAPP_DEMO_TO == ""
+
+    root = Path.cwd()
+    listed = subprocess.run(
+        ["git", "ls-files", "-z"], cwd=root, capture_output=True, text=True
+    )
+    assert listed.returncode == 0, "git ls-files failed inside the test repo"
+    tracked = [p for p in listed.stdout.split("\0") if p]
+
+    assignment = re.compile(r"^WHATSAPP_DEMO_TO[ \t]*=[ \t]*(\S*)$")
+    offenders = []
+    for rel in tracked:
+        path = root / rel
+        if path.suffix.lower() in {".png", ".jpg", ".jpeg", ".db", ".pyc", ".ico", ".woff", ".woff2", ".gz"}:
+            continue  # binaries cannot carry a text assignment
+        try:
+            text = path.read_text(encoding="utf-8", errors="ignore")
+        except OSError:
+            continue
+        for lineno, line in enumerate(text.splitlines(), start=1):
+            match = assignment.match(line)
+            if match and match.group(1):
+                offenders.append(f"{rel}:{lineno}: WHATSAPP_DEMO_TO={match.group(1)!r}")
+    assert not offenders, (
+        "WHATSAPP_DEMO_TO must stay unset in tracked files (set it in the "
+        f"deploy host .env only); found: {offenders}"
+    )
+
+
+# ── Demo payload builder ─────────────────────────────────────────────
+def test_demo_payload_builder_fails_closed_when_unset(monkeypatch):
+    """Without WHATSAPP_DEMO_TO the builder raises rather than fabricating."""
+    from app.routers.whatsapp import build_demo_webhook_payload
+
+    monkeypatch.setattr(settings, "WHATSAPP_DEMO_TO", "")
+    with pytest.raises(RuntimeError, match="WHATSAPP_DEMO_TO"):
+        build_demo_webhook_payload()
+
+
+def test_demo_payload_builder_uses_configured_number(monkeypatch):
+    """The demo payload's message ``from`` is the configured demo number."""
+    from app.routers.whatsapp import build_demo_webhook_payload
+
+    monkeypatch.setattr(settings, "WHATSAPP_DEMO_TO", _FAKE_DEMO_TO)
+    payload = build_demo_webhook_payload(text="Leaking tap", wa_message_id="wamid.demo.42")
+    entry = payload["entry"][0]["changes"][0]
+    assert entry["message"]["from"] == _FAKE_DEMO_TO
+    assert entry["message"]["id"] == "wamid.demo.42"
+    assert entry["message"]["text"]["text"] == "Leaking tap"
+
+
+# ── Demo round trip ──────────────────────────────────────────────────
+async def test_demo_round_trip_surfaces_number_in_mock_log(client, monkeypatch):
+    """Webhook demo payload -> ticket + log; mock-log shows the demo number."""
+    from app.routers.whatsapp import build_demo_webhook_payload
+
+    replies: list[tuple[str, str]] = []
+    await _capture_reply(monkeypatch, replies)
+
+    monkeypatch.setattr(settings, "WHATSAPP_WEBHOOK_SECRET", "test-webhook-secret")
+    monkeypatch.setattr(settings, "WHATSAPP_DEMO_TO", _FAKE_DEMO_TO)
+
+    resp = await client.post(
+        "/api/whatsapp/webhook",
+        json=build_demo_webhook_payload(text="Demo leak"),
+        headers={"X-Webhook-Secret": "test-webhook-secret"},
+    )
+    assert resp.status_code == 200, resp.text
+    data = resp.json()
+    assert data["status"] == "processed"
+    assert data["ticket_number"].startswith("PAV-")
+
+    # Auto-reply went back to the demo number.
+    assert replies and replies[0][0] == _FAKE_DEMO_TO, replies
+
+    token = await _login(client)
+    log = await client.get("/api/whatsapp/mock-log", headers=_auth(token))
+    assert log.status_code == 200, log.text
+    entries = log.json()
+    assert any(e["from_number"] == _FAKE_DEMO_TO for e in entries)
+    assert any(e["ticket_number"] == data["ticket_number"] for e in entries)
+
+
+async def test_webhook_still_403_without_secret_even_with_demo_number(client, monkeypatch):
+    """The webhook secret gate is independent of the demo number."""
+    monkeypatch.setattr(settings, "WHATSAPP_WEBHOOK_SECRET", "test-webhook-secret")
+    monkeypatch.setattr(settings, "WHATSAPP_DEMO_TO", _FAKE_DEMO_TO)
+    resp = await client.post("/api/whatsapp/webhook", json={"object": "whatsapp_business_account"})
+    assert resp.status_code == 403
+
+
+async def test_mock_log_still_401_gated(client):
+    """mock-log stays authenticated-only (no token -> 401)."""
+    resp = await client.get("/api/whatsapp/mock-log")
+    assert resp.status_code == 401, resp.text

tests/test_whatsapp_log_legacy_heal.py

@@ -0,0 +1,144 @@
+"""Regression: legacy ``whatsapp_log`` tables self-heal at startup.
+
+Producer/consumer for the CT115 defect (Mumuni relay #747): the live demo DB's
+``whatsapp_log`` table still has the pre-webhook shape — ``command`` instead of
+``message_text`` and no ``wa_message_id``/``ticket_number`` — so
+``GET /api/whatsapp/mock-log`` 500'd with ``OperationalError: no such column:
+whatsapp_log.message_text`` even for a valid admin token.
+
+Producer: build the legacy-shaped table (as the live DB holds it) and seed it
+with ``command`` rows.
+Consumer: boot the app's startup self-heal (``ensure_legacy_schema``), then read
+back through the authenticated mock-log endpoint, insert through the ORM write
+path, and re-run the self-heal to prove idempotency.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+import pytest
+from sqlalchemy import text
+
+from app.core.database import async_session_factory, engine
+from app.main import ensure_legacy_schema
+
+pytestmark = pytest.mark.asyncio
+
+# Pre-webhook shape (commit 4afdc36 changed the model but no migration shipped):
+# id, command (NOT NULL), from_number, ticket_id, received_at.
+LEGACY_WHATSAPP_LOG_DDL = (
+    "CREATE TABLE whatsapp_log ("
+    "id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, "
+    "command TEXT NOT NULL, "
+    "from_number VARCHAR(50), "
+    "ticket_id INTEGER, "
+    "received_at DATETIME NOT NULL)"
+)
+
+EXPECTED_MODEL_COLUMNS = {
+    "id",
+    "from_number",
+    "message_text",
+    "wa_message_id",
+    "ticket_id",
+    "ticket_number",
+    "received_at",
+}
+
+
+async def _login(client, email="wahab@denya.com", password="denya123") -> str:
+    resp = await client.post("/api/auth/login", json={"email": email, "password": password})
+    assert resp.status_code == 200, resp.text
+    return resp.json()["access_token"]
+
+
+def _auth(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}
+
+
+async def _replace_with_legacy_whatsapp_log() -> None:
+    """Drop the model-shaped table and recreate the legacy shape with rows."""
+    async with engine.begin() as conn:
+        await conn.execute(text("DROP TABLE IF EXISTS whatsapp_log"))
+        await conn.execute(text(LEGACY_WHATSAPP_LOG_DDL))
+        await conn.execute(
+            text(
+                "INSERT INTO whatsapp_log (command, from_number, ticket_id, received_at) VALUES "
+                "('legacy older message', '+233200000001', 1, '2026-09-08 08:00:00'), "
+                "('legacy newest message', '+233200000002', 2, '2026-09-09 09:30:00')"
+            )
+        )
+
+
+async def _columns() -> set[str]:
+    async with engine.begin() as conn:
+        result = await conn.execute(text("PRAGMA table_info(whatsapp_log)"))
+        return {row[1] for row in result}
+
+
+async def test_legacy_whatsapp_log_self_heals_and_mock_log_200(client):
+    """Producer: legacy whatsapp_log (command shape).  Consumer: startup self-heal
+    then authenticated mock-log — the exact 500 from the live instance."""
+    token = await _login(client)
+    await _replace_with_legacy_whatsapp_log()
+
+    # Pre-fix reproduction: on the legacy table this endpoint fails exactly as
+    # reported (production: HTTP 500; under ASGITransport the app never returns
+    # a response so the sqlalchemy OperationalError propagates).
+    import sqlalchemy.exc
+
+    with pytest.raises(sqlalchemy.exc.OperationalError):
+        await client.get("/api/whatsapp/mock-log", headers=_auth(token))
+
+    # ── Boot the startup self-heal (what lifespan does each boot) ──
+    async with engine.begin() as conn:
+        await ensure_legacy_schema(conn)
+
+    columns = await _columns()
+    assert EXPECTED_MODEL_COLUMNS <= columns
+    assert "command" not in columns  # obsolete model-dropped column is gone
+
+    # Authenticated mock-log returns 200 and the backfilled rows are readable.
+    resp = await client.get("/api/whatsapp/mock-log", headers=_auth(token))
+    assert resp.status_code == 200, resp.text
+    entries = resp.json()
+    assert [e["message_text"] for e in entries] == [
+        "legacy newest message",
+        "legacy older message",
+    ]  # order: received_at desc; bodies preserved from legacy `command`
+
+    # ORM write path works on the healed table (the current app inserts
+    # message_text/wa_message_id and never writes `command`).
+    from app.models.whatsapp_log import WhatsAppLog
+
+    async with async_session_factory() as session:
+        session.add(
+            WhatsAppLog(
+                from_number="+233200000003",
+                message_text="inbound after self-heal",
+                wa_message_id="wamid.healed.1",
+                ticket_id=None,
+                ticket_number=None,
+                received_at=datetime(2026, 9, 10, 10, 0, 0),
+            )
+        )
+        await session.commit()
+
+    resp = await client.get("/api/whatsapp/mock-log", headers=_auth(token))
+    assert resp.status_code == 200, resp.text
+    entries = resp.json()
+    assert entries[0]["message_text"] == "inbound after self-heal"
+    assert entries[0]["from_number"] == "+233200000003"
+    assert len(entries) == 3
+
+    # ── Idempotent on the next boot ──
+    async with engine.begin() as conn:
+        await ensure_legacy_schema(conn)
+    columns = await _columns()
+    assert EXPECTED_MODEL_COLUMNS <= columns
+    assert "command" not in columns
+
+    resp = await client.get("/api/whatsapp/mock-log", headers=_auth(token))
+    assert resp.status_code == 200, resp.text
+    assert len(resp.json()) == 3