fix(personas): H-2/H-3/M-2/M-5 — audit table + admin guards + rate limit

Per #229 review:

H-2: append-only persona_assignment_audit table. New migration 0023
chains from 0022. CREATED / CHANGED / DISABLED rows written in the
same transaction as the upsert/disable so the audit and state never
desync. HEAD_REVISION bumped to 0023_persona_assignment_audit;
test_head_revision_is_current also updated (was stale at 0021
before this change).

H-3: last-admin guard. POST .../disable refuses with HTTP 409 +
code=LAST_ADMIN when the target is persona='admin' and active admin
count <= 1. New store method count_active_admins() does the count.

M-2: PATCH no longer silently re-enables a disabled assignment.
Returns 409 + code=USER_DISABLED pointing the admin at the
yet-to-land POST .../enable endpoint.

M-5: per-admin invite rate limit. POST /admin/personas rejects with
HTTP 429 + code=RATE_LIMIT when the calling admin has issued >= 20
assignments in the trailing hour. New store method
count_invites_by_admin().

Tests added:
* test_persona_changes_create_audit_rows
* test_disabling_last_admin_returns_409_LAST_ADMIN
* test_patch_disabled_user_returns_409_USER_DISABLED
* test_admin_cannot_send_more_than_20_invites_per_hour

Review: #229 (comment)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: dwinter3

server/backend/alembic/versions/0023_persona_assignment_audit.py

@@ -0,0 +1,77 @@
+"""AS-1 follow-up: persona_assignment_audit — append-only history table.
+
+Revision ID: 0023_persona_assignment_audit
+Revises: 0022_persona_assignments
+Create Date: 2026-05-12
+
+The base 0022 table overwrites a single row per Human, which destroys the
+``assigned_by`` history every time an admin changes a persona. The audit
+table records every CREATED/CHANGED/DISABLED/ENABLED transition so the
+operator surface can answer "who set Carol to admin and when?"
+
+# Chain note
+
+After this migration lands, HEAD_REVISION in migrations.py must be
+``0023_persona_assignment_audit``.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0023_persona_assignment_audit"
+down_revision: str | Sequence[str] | None = "0022_persona_assignments"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _table_exists(bind: sa.engine.Connection, table_name: str) -> bool:
+    inspector = sa.inspect(bind)
+    return table_name in inspector.get_table_names()
+
+
+def upgrade() -> None:
+    """Create the ``persona_assignment_audit`` table."""
+    bind = op.get_bind()
+
+    if _table_exists(bind, "persona_assignment_audit"):
+        return
+
+    op.create_table(
+        "persona_assignment_audit",
+        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column("username", sa.Text(), nullable=False),
+        # old_persona is NULL on the CREATED row (no prior state).
+        sa.Column("old_persona", sa.Text(), nullable=True),
+        # new_persona is NULL on the DISABLED row (no live persona after).
+        sa.Column("new_persona", sa.Text(), nullable=True),
+        sa.Column("changed_by", sa.Text(), nullable=False),
+        sa.Column(
+            "changed_at",
+            sa.DateTime(),
+            nullable=False,
+            server_default=sa.func.current_timestamp(),
+        ),
+        sa.Column("action", sa.Text(), nullable=False),
+        sa.CheckConstraint(
+            "action IN ('CREATED', 'CHANGED', 'DISABLED', 'ENABLED')",
+            name="ck_audit_action",
+        ),
+    )
+    op.create_index(
+        "idx_audit_username",
+        "persona_assignment_audit",
+        ["username", sa.text("changed_at DESC")],
+    )
+
+
+def downgrade() -> None:
+    """Drop the audit table."""
+    bind = op.get_bind()
+    if not _table_exists(bind, "persona_assignment_audit"):
+        return
+    op.drop_index("idx_audit_username", table_name="persona_assignment_audit")
+    op.drop_table("persona_assignment_audit")

server/backend/src/cq_server/migrations.py

@@ -36,7 +36,7 @@
 # Phase 2 (task #100) — chain head after porting fork-delta tables to
 # Alembic. Update this string when adding a new migration so test
 # assertions and ops scripts stay in sync with the actual chain head.
-HEAD_REVISION = "0022_persona_assignments"
+HEAD_REVISION = "0023_persona_assignment_audit"
 
 
 def _find_alembic_ini() -> Path:

server/backend/src/cq_server/persona_routes.py

@@ -19,7 +19,7 @@
 
 import logging
 import uuid
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from typing import Literal
 
 from fastapi import APIRouter, Depends, HTTPException, Query
@@ -180,6 +180,21 @@ async def create_persona(
             detail=f"persona assignment already exists for username={req.username!r}",
         )
 
+    # M-5: per-admin invite rate-limit. Count persona assignments this
+    # admin has issued in the trailing hour; cap at 20.
+    since = (datetime.now(UTC) - timedelta(hours=1)).isoformat()
+    recent_count = await store.count_invites_by_admin(
+        admin_username=admin, since=since
+    )
+    if recent_count >= 20:
+        raise HTTPException(
+            status_code=429,
+            detail={
+                "error": "Invite rate limit exceeded (20/hour).",
+                "code": "RATE_LIMIT",
+            },
+        )
+
     # Ensure the user row exists (or create it).  We create a stub user
     # without a password hash — the invite claim flow will set the password.
     from .auth import hash_password
@@ -198,6 +213,8 @@ async def create_persona(
         persona=req.persona,
         assigned_at=now,
         assigned_by=admin,
+        audit_action="CREATED",
+        audit_old_persona=None,
     )
 
     # Fire invite email (best-effort — same pattern as invite_routes).
@@ -255,12 +272,29 @@ async def patch_persona(
     if existing is None:
         raise HTTPException(status_code=404, detail=f"no persona assignment for {username!r}")
 
+    # M-2: don't silently re-enable a disabled user. PATCH is not the
+    # enable path; admins should call POST /admin/personas/{username}/enable
+    # (follow-up issue) once that lands.
+    if existing.get("disabled_at") is not None:
+        raise HTTPException(
+            status_code=409,
+            detail={
+                "error": (
+                    "user is disabled — re-enable first via "
+                    "POST /admin/personas/{username}/enable"
+                ),
+                "code": "USER_DISABLED",
+            },
+        )
+
     now = datetime.now(UTC).isoformat()
     await store.upsert_persona_assignment(
         username=username,
         persona=req.persona,
         assigned_at=now,
         assigned_by=admin,
+        audit_action="CHANGED",
+        audit_old_persona=existing.get("persona"),
     )
     return PatchPersonaResponse(
         username=username,
@@ -273,23 +307,43 @@ async def patch_persona(
 @router.post("/{username}/disable", response_model=DisableResponse)
 async def disable_persona(
     username: str,
-    admin: str = Depends(require_admin),  # noqa: ARG001 — auth gate only
+    admin: str = Depends(require_admin),
     store: SqliteStore = Depends(get_store),
 ) -> DisableResponse:
     """Soft-disable a Human's persona assignment.
 
-    Sets ``disabled_at`` to now. The users row is NOT deleted — audit trail
-    is preserved. Re-enabling is done via PATCH (which clears disabled_at).
+    Sets ``disabled_at`` to now. The users row is NOT deleted — audit
+    trail is preserved. PATCH no longer silently re-enables (see M-2);
+    a future POST .../enable endpoint owns the re-enable path.
 
     404 when no assignment row exists.
     409 when the assignment is already disabled.
+    409 + code=LAST_ADMIN when disabling would leave zero active admins.
     """
     existing = await store.get_persona_assignment(username)
     if existing is None:
         raise HTTPException(status_code=404, detail=f"no persona assignment for {username!r}")
     if existing.get("disabled_at") is not None:
         raise HTTPException(status_code=409, detail=f"{username!r} is already disabled")
 
+    # H-3: last-admin guard. Refuse to disable the only remaining admin
+    # so the L2 surface never loses its escape hatch.
+    if existing.get("persona") == "admin":
+        active_admins = await store.count_active_admins()
+        if active_admins <= 1:
+            raise HTTPException(
+                status_code=409,
+                detail={
+                    "error": "Cannot disable the last admin.",
+                    "code": "LAST_ADMIN",
+                },
+            )
+
     now = datetime.now(UTC).isoformat()
-    await store.disable_persona_assignment(username=username, disabled_at=now)
+    await store.disable_persona_assignment(
+        username=username,
+        disabled_at=now,
+        changed_by=admin,
+        old_persona=existing.get("persona"),
+    )
     return DisableResponse(username=username, disabled_at=now)