feat: add HandoffManager service with dual-write and integration tests

Unified park/escalate handoff management with snapshot generation,
AI diagnostic assessment for escalations (via _call_ai), claim workflow
that reactivates sessions, PSA push via existing psa_documentation_service,
and team queue query. Dual-writes to ai_sessions.escalation_package for
backward compatibility.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

backend/tests/test_handoff_manager.py

@@ -0,0 +1,115 @@
+"""Integration tests for HandoffManager service."""
+import pytest
+from httpx import AsyncClient
+
+from app.models.ai_session import AISession
+
+
+@pytest.mark.asyncio
+async def test_create_park_handoff(client: AsyncClient, test_user, auth_headers, test_db):
+    """Parking a session creates a handoff with snapshot."""
+    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=[{"role": "user", "content": "help me"}],
+    )
+    test_db.add(session)
+    await test_db.flush()
+
+    from app.services.handoff_manager import HandoffManager
+    manager = HandoffManager(test_db)
+
+    handoff = await manager.create_handoff(
+        session_id=session.id,
+        intent="park",
+        engineer_notes="Waiting for client to provide logs",
+        user_id=test_user["user_data"]["id"],
+    )
+
+    assert handoff is not None
+    assert handoff.intent == "park"
+    assert handoff.engineer_notes == "Waiting for client to provide logs"
+    assert handoff.snapshot is not None
+
+    # Session should be paused
+    await test_db.refresh(session)
+    assert session.status == "paused"
+    assert session.handoff_count == 1
+
+
+@pytest.mark.asyncio
+async def test_create_escalate_handoff(client: AsyncClient, test_user, auth_headers, test_db):
+    """Escalating creates handoff + dual-writes to escalation_package."""
+    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()
+
+    from app.services.handoff_manager import HandoffManager
+    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"
+
+    # Dual-write check
+    await test_db.refresh(session)
+    assert session.status == "escalated"
+    assert session.escalation_package is not None
+    assert "branch_map" in session.escalation_package or "snapshot" in session.escalation_package
+
+
+@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."""
+    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()
+
+    from app.services.handoff_manager import HandoffManager
+    manager = HandoffManager(test_db)
+
+    handoff = await manager.create_handoff(
+        session_id=session.id,
+        intent="escalate",
+        engineer_notes="Need help",
+        user_id=test_user["user_data"]["id"],
+    )
+
+    claimed = await manager.claim_session(
+        handoff_id=handoff.id,
+        claiming_user_id=test_admin["user_data"]["id"],
+    )
+
+    assert claimed.claimed_by == test_admin["user_data"]["id"]
+    assert claimed.claimed_at is not None
+
+    await test_db.refresh(session)
+    assert session.status == "active"