fix(handoff): bound escalation assessment latency

Co-Authored-By: Codex <noreply@openai.com>

backend/app/core/config.py

@@ -111,6 +111,7 @@ class Settings(BaseSettings):
     GOOGLE_AI_API_KEY: Optional[str] = None
     AI_MODEL_GEMINI: str = "gemini-2.5-flash"
     AI_MODEL_ANTHROPIC: str = "claude-sonnet-4-6"
+    ESCALATION_AI_ASSESSMENT_TIMEOUT_SECONDS: int = 5
 
     # Model tier routing — maps action types to model tiers
     AI_MODEL_TIERS: dict[str, str] = {

backend/app/services/handoff_manager.py

@@ -57,7 +57,9 @@ class HandoffManager:
         ai_assessment = None
         ai_assessment_data = None
         if intent == "escalate":
-            ai_assessment, ai_assessment_data = await self._generate_ai_assessment(session)
+            ai_assessment, ai_assessment_data = (
+                await self._generate_ai_assessment_with_timeout(session)
+            )
 
         handoff = SessionHandoff(
             session_id=session_id,
@@ -311,6 +313,24 @@ class HandoffManager:
             logger.exception("Failed to generate AI assessment")
             return None, None
 
+    async def _generate_ai_assessment_with_timeout(
+        self, session: AISession
+    ) -> tuple[str | None, dict[str, Any] | None]:
+        """Generate optional escalation assessment within the click-path budget."""
+        timeout = settings.ESCALATION_AI_ASSESSMENT_TIMEOUT_SECONDS
+        try:
+            return await asyncio.wait_for(
+                self._generate_ai_assessment(session),
+                timeout=timeout,
+            )
+        except asyncio.TimeoutError:
+            logger.warning(
+                "Escalation AI assessment timed out after %ss for session %s",
+                timeout,
+                session.id,
+            )
+            return None, None
+
     async def generate_briefing(
         self, handoff_id: UUID, claiming_user_id: UUID
     ) -> str:

backend/tests/test_handoff_manager.py

@@ -1,4 +1,5 @@
 """Integration tests for HandoffManager service."""
+import asyncio
 from unittest.mock import AsyncMock, patch
 
 import pytest
@@ -101,6 +102,55 @@ async def test_create_escalate_handoff(client: AsyncClient, test_user, auth_head
     assert "branch_map" in session.escalation_package or "snapshot" in session.escalation_package
 
 
+@pytest.mark.asyncio
+async def test_create_escalate_handoff_does_not_wait_on_slow_ai_assessment(
+    client: AsyncClient, test_user, auth_headers, test_db, monkeypatch
+):
+    """Escalate should commit a handoff even when optional AI assessment is slow."""
+    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.flush()
+
+    async def slow_assessment(self, session):
+        await asyncio.sleep(0.2)
+        return "too slow", {"confidence": "medium"}
+
+    monkeypatch.setattr(
+        "app.services.handoff_manager.settings."
+        "ESCALATION_AI_ASSESSMENT_TIMEOUT_SECONDS",
+        0.01,
+    )
+    with patch.object(
+        HandoffManager,
+        "_generate_ai_assessment",
+        new=slow_assessment,
+    ):
+        manager = HandoffManager(test_db)
+        handoff = await manager.create_handoff(
+            session_id=session.id,
+            intent="escalate",
+            engineer_notes="Need senior help",
+            user_id=test_user["user_data"]["id"],
+        )
+
+    assert handoff.intent == "escalate"
+    assert handoff.ai_assessment is None
+    assert handoff.ai_assessment_data is None
+
+    await test_db.refresh(session)
+    assert session.status == "escalated"
+    assert session.handoff_count == 1
+
+
 @pytest.mark.asyncio
 async def test_claim_session(client: AsyncClient, test_user, test_admin, auth_headers, test_db):
     """Claiming a handoff sets claimed_by and reactivates session."""