feat(api): role-gate handoff claim to engineer-or-admin

POST /ai-sessions/{id}/handoffs/{hid}/claim previously required only an
authenticated user, so a viewer-role account user could claim escalations.
Codex review flagged this as wedge-relevant: the Escalation Mode race-
condition story (two seniors clicking Pick Up simultaneously) depends on
auth gating for audit integrity. Originally captured as a deferred TODO
during /plan-eng-review, then moved in-scope by /codex review.

Swap the dep to require_engineer_or_admin. One-line change. Two new tests:
- viewer_role gets 403 with "Engineer or admin access required"
- engineer/owner role still succeeds and claimed_at + claimed_by populate

Existing handoff create + queue tests unaffected.

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

backend/app/api/endpoints/session_handoffs.py

@@ -13,7 +13,7 @@ from fastapi import APIRouter, Depends, HTTPException, status
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.api.deps import get_current_active_user, get_db
+from app.api.deps import get_current_active_user, get_db, require_engineer_or_admin
 from app.models.user import User
 from app.models.ai_session import AISession
 from app.models.session_handoff import SessionHandoff
@@ -86,10 +86,16 @@ async def list_handoffs(
 async def claim_handoff(
     session_id: UUID,
     handoff_id: UUID,
-    current_user: Annotated[User, Depends(get_current_active_user)],
+    current_user: Annotated[User, Depends(require_engineer_or_admin)],
     db: Annotated[AsyncSession, Depends(get_db)],
 ) -> HandoffResponse:
-    """Claim a handed-off session."""
+    """Claim a handed-off session.
+
+    Role-gated to engineer/admin/owner — viewers cannot claim. The race-condition
+    story (two seniors clicking Pick Up simultaneously) depends on auth gating
+    for audit integrity. Codex review flagged this as wedge-relevant; locked
+    in-scope for Escalation Mode v1.
+    """
     manager = HandoffManager(db)
     try:
         handoff = await manager.claim_session(

backend/tests/test_session_handoffs_api.py

@@ -1,8 +1,12 @@
 """API endpoint tests for session handoffs."""
+from uuid import UUID as PyUUID
+
 import pytest
 from httpx import AsyncClient
+from sqlalchemy import select
 
 from app.models.ai_session import AISession
+from app.models.user import User
 
 
 @pytest.mark.asyncio
@@ -58,3 +62,88 @@ async def test_get_queue(client: AsyncClient, test_user, auth_headers, test_db):
     assert resp.status_code == 200
     data = resp.json()
     assert len(data) >= 1
+
+
+@pytest.mark.asyncio
+async def test_claim_blocked_for_viewer_role(
+    client: AsyncClient, test_user, auth_headers, test_db
+):
+    """POST /handoffs/{id}/claim must 403 for viewer-role users.
+
+    Codex review flagged the missing role gate as wedge-relevant: the
+    race-condition story (two seniors clicking Pick Up simultaneously)
+    requires auth gating for audit integrity. Viewers must not be able
+    to claim escalations.
+    """
+    # Create a session + handoff as the engineer-role test_user (default = owner).
+    session = AISession(
+        user_id=test_user["user_data"]["id"],
+        account_id=test_user["user_data"]["account_id"],
+        session_type="guided",
+        intake_type="free_text",
+        intake_content={"text": "test"},
+        status="active",
+        confidence_tier="discovery",
+        conversation_messages=[],
+    )
+    test_db.add(session)
+    await test_db.commit()
+
+    create_resp = await client.post(
+        f"/api/v1/ai-sessions/{session.id}/handoff",
+        headers=auth_headers,
+        json={"intent": "escalate", "engineer_notes": "Need help"},
+    )
+    assert create_resp.status_code == 201
+    handoff_id = create_resp.json()["id"]
+
+    # Downgrade the user to viewer.
+    user_id = PyUUID(test_user["user_data"]["id"])
+    user = (
+        await test_db.execute(select(User).where(User.id == user_id))
+    ).scalar_one()
+    user.account_role = "viewer"
+    await test_db.commit()
+
+    claim_resp = await client.post(
+        f"/api/v1/ai-sessions/{session.id}/handoffs/{handoff_id}/claim",
+        headers=auth_headers,
+    )
+    assert claim_resp.status_code == 403
+    assert "engineer" in claim_resp.json()["detail"].lower()
+
+
+@pytest.mark.asyncio
+async def test_claim_allowed_for_engineer_role(
+    client: AsyncClient, test_user, auth_headers, test_db
+):
+    """POST /handoffs/{id}/claim succeeds for engineer-or-admin roles."""
+    session = AISession(
+        user_id=test_user["user_data"]["id"],
+        account_id=test_user["user_data"]["account_id"],
+        session_type="guided",
+        intake_type="free_text",
+        intake_content={"text": "test"},
+        status="active",
+        confidence_tier="discovery",
+        conversation_messages=[],
+    )
+    test_db.add(session)
+    await test_db.commit()
+
+    create_resp = await client.post(
+        f"/api/v1/ai-sessions/{session.id}/handoff",
+        headers=auth_headers,
+        json={"intent": "escalate", "engineer_notes": "Need help"},
+    )
+    assert create_resp.status_code == 201
+    handoff_id = create_resp.json()["id"]
+
+    # Default test_user role is "owner", which passes engineer-or-admin.
+    claim_resp = await client.post(
+        f"/api/v1/ai-sessions/{session.id}/handoffs/{handoff_id}/claim",
+        headers=auth_headers,
+    )
+    assert claim_resp.status_code == 200
+    assert claim_resp.json()["claimed_by"] == test_user["user_data"]["id"]
+    assert claim_resp.json()["claimed_at"] is not None