feat(escalations): close out plan-locked wedge polish

Four items from the design-plan audit, all flagged as locked-design or
Codex corrections, shipped together so the GTM demo path covers them
end-to-end before bug bash.

1. Live AI assessment refresh on the magic-moment screen. Backend already
   publishes handoff_assessment_ready when enrich_escalation_async commits;
   wire the frontend listener so the senior sees the assessment populate
   without a manual reopen. New event type + onAssessmentReady handler on
   streamEscalations; AssistantChatPage opens a scoped SSE subscription
   whenever it tracks a handoff missing its assessment, refetches on match,
   and replaces magicHandoff / overlayHandoff in place. Closes the loop on
   the async-assessment commit e8ba74e.

2. Suggested-step chips below the chat input. Locked design from the plan
   (Codex correction). Chip strip renders above the composer post-claim
   when ai_assessment_data.suggested_steps[] is non-empty. Click prefills
   the input and focuses; first send or explicit X hides for the session.

3. Unread 6px dot on EscalationQueue cards. localStorage-persisted seen
   set (rf-escalation-seen, capped 200). Dot top-right when not seen.
   Cleared on open (card click) or claim (Pick Up) — NOT on hover, per
   Codex correction. Pick Up stops propagation so it doesn't double-fire.

4. Race-condition toast on claim conflict. The /claim endpoint previously
   silently overwrote claimed_by — both seniors thought they owned the
   session. New HandoffAlreadyClaimedError carries the winner's id/name/
   timestamp; claim_session rejects different-user re-claims (same-user is
   idempotent for double-click safety); endpoint returns 409 with
   structured detail. AssistantChatPage.handleStartHere extracts and
   surfaces "Already claimed by {name} {time_ago}." via toast, drops
   ?pickup=true, dismisses magic-moment so the loser flows back to queue.

Tests: 2 new unit tests in test_handoff_manager.py (conflict raises,
same-user idempotent). Full handoff + escalation suite (34 tests) green.
Frontend tsc -b clean.

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

backend/app/api/endpoints/session_handoffs.py

@@ -22,7 +22,7 @@ from app.core.escalation_bus import bus as escalation_bus
 from app.models.user import User
 from app.models.ai_session import AISession
 from app.models.session_handoff import SessionHandoff
-from app.services.handoff_manager import HandoffManager
+from app.services.handoff_manager import HandoffAlreadyClaimedError, HandoffManager
 from app.schemas.session_handoff import (
     HandoffCreateRequest,
     HandoffResponse,
@@ -129,6 +129,19 @@ async def claim_handoff(
             handoff_id=handoff_id,
             claiming_user_id=current_user.id,
         )
+    except HandoffAlreadyClaimedError as e:
+        # Loser of the race — the API surfaces structured detail so the
+        # client can render "Already claimed by {name} {time_ago}" without
+        # a follow-up fetch.
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail={
+                "error": "already_claimed",
+                "claimed_by_id": str(e.claimed_by_id),
+                "claimed_by_name": e.claimed_by_name,
+                "claimed_at": e.claimed_at.isoformat(),
+            },
+        )
     except ValueError as e:
         raise HTTPException(status_code=404, detail=str(e))
 

backend/app/services/handoff_manager.py

@@ -36,6 +36,30 @@ from app.services.notification_service import notify
 logger = logging.getLogger(__name__)
 
 
+class HandoffAlreadyClaimedError(Exception):
+    """Raised when a senior tries to claim a handoff another senior already won.
+
+    Carries the winning claimer's id, display name, and claim timestamp so the
+    API layer can surface a "Already claimed by {name} {time_ago}" toast on
+    the losing client. The race story is the locked design — without this
+    exception the endpoint would silently overwrite `claimed_by` and both
+    seniors would think they own the session.
+    """
+
+    def __init__(
+        self,
+        claimed_by_id: UUID,
+        claimed_by_name: str,
+        claimed_at: datetime,
+    ) -> None:
+        super().__init__(
+            f"Handoff already claimed by {claimed_by_name} at {claimed_at.isoformat()}"
+        )
+        self.claimed_by_id = claimed_by_id
+        self.claimed_by_name = claimed_by_name
+        self.claimed_at = claimed_at
+
+
 class HandoffManager:
     """Unified park/escalate handoff management."""
 
@@ -398,14 +422,31 @@ class HandoffManager:
         handoff_id: UUID,
         claiming_user_id: UUID,
     ) -> SessionHandoff:
-        """Claim a handed-off session."""
+        """Claim a handed-off session.
+
+        If the handoff was already claimed by a *different* user (the race
+        story: two seniors clicking Pick Up simultaneously), raise
+        `HandoffAlreadyClaimedError` with the winning claimer's details so
+        the API can return 409 with the data the loser's toast needs. A
+        re-claim by the same user is idempotent.
+        """
         result = await self.db.execute(
-            select(SessionHandoff).where(SessionHandoff.id == handoff_id)
+            select(SessionHandoff)
+            .options(selectinload(SessionHandoff.claimed_by_user))
+            .where(SessionHandoff.id == handoff_id)
         )
         handoff = result.scalar_one_or_none()
         if not handoff:
             raise ValueError(f"Handoff {handoff_id} not found")
 
+        if handoff.claimed_by is not None and handoff.claimed_by != claiming_user_id:
+            claimer = handoff.claimed_by_user
+            raise HandoffAlreadyClaimedError(
+                claimed_by_id=handoff.claimed_by,
+                claimed_by_name=claimer.name if claimer else "another engineer",
+                claimed_at=handoff.claimed_at or datetime.now(timezone.utc),
+            )
+
         handoff.claimed_by = claiming_user_id
         handoff.claimed_at = datetime.now(timezone.utc)
 

backend/tests/test_handoff_manager.py

@@ -189,6 +189,99 @@ async def test_claim_session(client: AsyncClient, test_user, test_admin, auth_he
     assert session.status == "active"
 
 
+@pytest.mark.asyncio
+async def test_claim_session_conflict_raises_already_claimed(
+    client: AsyncClient, test_user, test_admin, auth_headers, test_db
+):
+    """Two seniors claiming simultaneously: the second raises the typed
+    HandoffAlreadyClaimedError carrying the winner's identity. Without this
+    guard both calls would silently overwrite claimed_by — the locked
+    race-condition story depends on a real conflict response."""
+    from app.services.handoff_manager import (
+        HandoffAlreadyClaimedError,
+        HandoffManager,
+    )
+
+    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()
+
+    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"],
+    )
+
+    # First claim — admin wins.
+    await manager.claim_session(
+        handoff_id=handoff.id,
+        claiming_user_id=test_admin["user_data"]["id"],
+    )
+
+    # Second claim by a different user — owner of the original session,
+    # standing in for "the other senior who lost the race."
+    with pytest.raises(HandoffAlreadyClaimedError) as exc_info:
+        await manager.claim_session(
+            handoff_id=handoff.id,
+            claiming_user_id=test_user["user_data"]["id"],
+        )
+
+    err = exc_info.value
+    assert err.claimed_by_id == test_admin["user_data"]["id"]
+    assert err.claimed_by_name  # populated from User.name
+    assert err.claimed_at is not None
+
+
+@pytest.mark.asyncio
+async def test_claim_session_idempotent_for_same_user(
+    client: AsyncClient, test_user, test_admin, auth_headers, test_db
+):
+    """A re-claim by the user who already won is a no-op, not a conflict.
+    Defends against double-clicks / network retries on the loser-side toast."""
+    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()
+
+    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"],
+    )
+
+    first = await manager.claim_session(
+        handoff_id=handoff.id,
+        claiming_user_id=test_admin["user_data"]["id"],
+    )
+    second = await manager.claim_session(
+        handoff_id=handoff.id,
+        claiming_user_id=test_admin["user_data"]["id"],
+    )
+
+    assert first.claimed_by == second.claimed_by == test_admin["user_data"]["id"]
+
+
 # ─── Notification dispatch ────────────────────────────────────────────────────