From 7a5b853b3b1cd3be830e3b72461f0508d34d1c92 Mon Sep 17 00:00:00 2001 From: Michael Chihlas Date: Mon, 27 Apr 2026 15:46:59 -0400 Subject: [PATCH] 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 --- backend/app/api/endpoints/session_handoffs.py | 12 ++- backend/tests/test_session_handoffs_api.py | 89 +++++++++++++++++++ 2 files changed, 98 insertions(+), 3 deletions(-) diff --git a/backend/app/api/endpoints/session_handoffs.py b/backend/app/api/endpoints/session_handoffs.py index 513eefc6..2e3ec65f 100644 --- a/backend/app/api/endpoints/session_handoffs.py +++ b/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( diff --git a/backend/tests/test_session_handoffs_api.py b/backend/tests/test_session_handoffs_api.py index 26a47988..6edaac1e 100644 --- a/backend/tests/test_session_handoffs_api.py +++ b/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