Merge pull request 'fix(security): add unsafe-eval to CSP script-src (Alpine.js runtime requires it)' (#14) from fm/fix-denya-csp-unsafe-eval-20260909 into main

Alpine 3.17.2's CDN build compiles every x-data/x-show/x-text expression
with new Function(), which the strict P0 CSP (script-src 'self'
'unsafe-inline') blocked. Every Alpine directive threw "Evaluating a
string as JavaScript violates ... 'unsafe-eval' is not an allowed
source", Alpine never initialized, and the loading overlay
(x-show="loading" in base.html) stayed visible forever on /login and
every Alpine-driven page.

Add 'unsafe-eval' to script-src (Alpine's documented CSP requirement for
its runtime); everything else in the header is unchanged. Regression test
asserts the /login CSP header carries 'unsafe-eval' inside script-src.

Verified live: headless chromium (playwright build 1243) shows zero
CSP/eval console errors after the fix, with Alpine applying
style="display:none" to the loading overlay; the pre-fix header produces
the Alpine Expression Error spam and leaves the overlay visible.

AGENTS.md

@@ -66,6 +66,13 @@ sign-up UI; users are created/managed by admins only (P0 hardening batch).
 | 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.
+
 ### Pages (Sprint 3) — Jinja2 templates at `app/templates/`
 | Method | Path | Auth | Description |
 |--------|------|------|-------------|
@@ -77,7 +84,29 @@ sign-up UI; users are created/managed by admins only (P0 hardening batch).
 | 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`.
+
+### 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`.
 
 ### Tickets (Sprint 2)
 | Method | Path | Auth | Description |
@@ -107,13 +136,17 @@ Frontend: Alpine.js (CDN) + Tailwind CSS (CDN). Auth state in localStorage. Role
   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, and
+  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)
 

HARDENING.md

@@ -16,7 +16,7 @@
 - **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`.
 - **Demo-only posture resolved (PR #10 + P0 batch):** `SECRET_KEY` now fails

app/core/roles.py

@@ -12,8 +12,9 @@ uses (PRD §4, ``app/services/seed.py``, the frontend nav in base.html):
 
 Legacy databases created under the pre-P0 open-registration builds can carry
 lowercase/nickname role strings (``technician``, ``cs``, ``fm``, ``ceo``,
-``admin``, ``superadmin`` …).  ``ROLE_ALIASES`` maps the *unambiguous*
-nicknames onto a canonical role so startup normalization (see
+``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
@@ -52,10 +53,13 @@ ROLE_ALIASES: dict[str, str] = {
     "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",
 }

app/main.py

@@ -52,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' "
@@ -100,8 +141,18 @@ class SecurityHeadersMiddleware:
 
     * ``X-Frame-Options: DENY`` and ``X-Content-Type-Options: nosniff`` on
       all responses;
-    * CSP on HTML pages (login + app pages; the Alpine.js/Tailwind CDNs need
-      the CDN hosts + inline script/style for this demo);
+    * 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).
     """
@@ -109,8 +160,8 @@ class SecurityHeadersMiddleware:
     HSTS = "max-age=31536000; includeSubDomains"
     CSP = (
         "default-src 'self'; "
-        "script-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net https://cdn.tailwindcss.com; "
-        "style-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net https://cdn.tailwindcss.com; "
+        "script-src 'self' 'unsafe-inline' 'unsafe-eval'; "
+        "style-src 'self' 'unsafe-inline'; "
         "img-src 'self' data: blob:; "
         "font-src 'self' data:; "
         "connect-src 'self'; "
@@ -143,6 +194,9 @@ class SecurityHeadersMiddleware:
                 )
                 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:
@@ -171,11 +225,32 @@ app.add_middleware(
     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/templates/base.html

@@ -4,8 +4,9 @@
     <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>
-    <script src="https://cdn.tailwindcss.com"></script>
+    <!-- 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: {

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

@@ -308,6 +308,43 @@ async def test_normalize_does_not_map_ambiguous_admin_alias(client: AsyncClient)
     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

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