feat(escalations): magic-moment 3-option CTA + claim 500 fix

- HandoffContextScreen: 3-option layout (Continue/AI analysis/Own thing)
  with hasTaskLane, activeOptionKey, spinner/disabled states
- AssistantChatPage: wire up handleContinue, handleAIAnalysis, handleOwnThing
  handlers; chip detail expansion inline with copy-button fix; post-escalation
  redirect to dashboard on ConcludeSessionModal close
- TaskLane: fix async copy button (await + execCommand fallback + copiedKey
  visual feedback); whitespace-pre-wrap on command blocks
- Fix 500 on claim: Pydantic v2 model_validate() + model_copy(update={})
  (was passing update= kwarg directly which v2 rejects)
- HandoffResponse schema: handed_off_by_name field

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

backend/app/services/flowpilot_engine.py

@@ -913,6 +913,41 @@ async def generate_status_update(
     """Generate a status update for ticket notes, client communication, or email draft."""
     session = await _load_session(session_id, user_id, db)
 
+    # For escalation/ticket_notes, return the pre-generated handoff prose immediately
+    # if enrich_escalation_async has already populated it. This eliminates the
+    # redundant Sonnet re-summarization on every "Ticket Notes" click.
+    if request.context == "escalation" and request.audience == "ticket_notes":
+        from app.models.session_handoff import SessionHandoff
+
+        handoff_q = await db.execute(
+            select(SessionHandoff)
+            .where(
+                SessionHandoff.session_id == session_id,
+                SessionHandoff.intent == "escalate",
+            )
+            .order_by(SessionHandoff.created_at.desc())
+            .limit(1)
+        )
+        escalation_handoff = handoff_q.scalar_one_or_none()
+        saved_data = (
+            escalation_handoff.ai_assessment_data or {}
+        ) if escalation_handoff else {}
+        prose = saved_data.get("summary_prose") or (
+            escalation_handoff.ai_assessment if escalation_handoff else None
+        )
+        if prose:
+            return StatusUpdateResponse(
+                content=prose,
+                audience=request.audience,
+                length=request.length,
+                context=request.context,
+                session_status=session.status,
+                steps_completed=session.step_count or 0,
+                time_spent_display=None,
+                client_name=None,
+                generated_at=datetime.now(timezone.utc),
+            )
+
     # Build conversation summary from session steps
     steps_summary = []
     for step in sorted(session.steps, key=lambda s: s.step_order):

backend/app/services/handoff_manager.py

@@ -14,6 +14,7 @@ on top of per-user emails. The `/escalate` endpoint is now a thin shim
 calling these in sequence.
 """
 import asyncio
+import json
 import logging
 from datetime import datetime, timezone
 from typing import Any
@@ -23,6 +24,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
+from app.core.ai_provider import get_ai_provider
 from app.core.config import settings
 from app.core.email import EmailService
 from app.core.escalation_bus import bus as escalation_bus
@@ -432,7 +434,10 @@ class HandoffManager:
         """
         result = await self.db.execute(
             select(SessionHandoff)
-            .options(selectinload(SessionHandoff.claimed_by_user))
+            .options(
+                selectinload(SessionHandoff.claimed_by_user),
+                selectinload(SessionHandoff.handed_off_by_user),
+            )
             .where(SessionHandoff.id == handoff_id)
         )
         handoff = result.scalar_one_or_none()
@@ -463,61 +468,111 @@ class HandoffManager:
         await self.db.flush()
         return handoff
 
-    async def _generate_ai_assessment(
+    async def _generate_handoff_summary(
         self, session: AISession
-    ) -> tuple[str | None, dict[str, Any] | None]:
-        """Generate AI diagnostic assessment for escalation handoffs."""
-        try:
-            from app.services.assistant_chat_service import _call_ai
+    ) -> dict[str, Any] | None:
+        """Single structured AI call for the escalation magic-moment screen.
 
-            context = f"Problem: {session.problem_summary or 'Unknown'}\nDomain: {session.problem_domain or 'Unknown'}"
-            msgs = session.conversation_messages or []
-            # Include last 10 messages for context
-            recent = "\n".join(
-                f"[{m.get('role', '?')}]: {m.get('content', '')[:200]}"
-                for m in msgs[-10:]
-            )
-
-            assessment_text, _, _ = await _call_ai(
-                system_base="You are a diagnostic assessment generator for MSP escalations.",
-                rag_context="",
-                history=[],
-                new_message=(
-                    f"Generate a brief diagnostic assessment for this escalation.\n"
-                    f"{context}\n\nRecent conversation:\n{recent}\n\n"
-                    f"Return: 1) Most likely cause, 2) Suggested next steps, 3) Confidence (low/medium/high)"
-                ),
-                max_tokens=500,
-            )
-
-            assessment_data = {
-                "likely_cause": "See assessment text",
-                "suggested_steps": [],
-                "confidence": "medium",
-            }
-
-            return assessment_text, assessment_data
-        except Exception:
-            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."""
+        Returns a dict with summary_prose, what_we_know, likely_cause,
+        suggested_steps, and confidence. Returns None on timeout or error.
+        Replaces the old _generate_ai_assessment + _generate_ai_assessment_with_timeout
+        pair, which returned freeform prose with no usable structured fields.
+        """
         timeout = settings.ESCALATION_AI_ASSESSMENT_TIMEOUT_SECONDS
         try:
             return await asyncio.wait_for(
-                self._generate_ai_assessment(session),
+                self._generate_handoff_summary_inner(session),
                 timeout=timeout,
             )
         except asyncio.TimeoutError:
             logger.warning(
-                "Escalation AI assessment timed out after %ss for session %s",
+                "Handoff summary timed out after %ss for session %s",
                 timeout,
                 session.id,
             )
-            return None, None
+            return None
+        except Exception:
+            logger.exception(
+                "Handoff summary failed for session %s", session.id
+            )
+            return None
+
+    async def _generate_handoff_summary_inner(
+        self, session: AISession
+    ) -> dict[str, Any]:
+        steps = session.steps or []
+        steps_tried = []
+        for step in sorted(steps, key=lambda s: s.step_order):
+            content = step.content or {}
+            text = content.get("text", "").strip()
+            if not text:
+                continue
+            entry = text
+            if step.selected_option:
+                entry += f" → {step.selected_option}"
+            elif step.free_text_input:
+                entry += f" → {step.free_text_input[:100]}"
+            elif step.was_skipped:
+                entry += " (skipped)"
+            steps_tried.append(entry)
+        steps_text = (
+            "\n".join(f"- {s}" for s in steps_tried[:15])
+            or "No diagnostic steps recorded."
+        )
+
+        msgs = session.conversation_messages or []
+        recent_msgs = "\n".join(
+            f"[{m.get('role', '?')}]: {m.get('content', '')[:200]}"
+            for m in msgs[-10:]
+        )
+
+        prompt = (
+            "Generate a structured escalation handoff summary.\n\n"
+            f"Problem: {session.problem_summary or 'Unknown'}\n"
+            f"Domain: {session.problem_domain or 'Unknown'}\n"
+            f"Escalation reason: {session.escalation_reason or 'Not provided'}\n\n"
+            f"Diagnostic steps taken:\n{steps_text}\n\n"
+            f"Recent conversation:\n{recent_msgs}\n\n"
+            "Respond with ONLY a valid JSON object matching this schema exactly:\n"
+            '{"summary_prose": "<2-3 sentences suitable for PSA ticket notes>",\n'
+            ' "what_we_know": ["<confirmed fact 1>", "<confirmed fact 2>"],\n'
+            ' "likely_cause": "<one sentence root cause hypothesis>",\n'
+            ' "suggested_steps": ["<next step 1>", "<next step 2>"],\n'
+            ' "confidence": "<low or medium or high>"}'
+        )
+
+        provider = get_ai_provider(settings.get_model_for_action("escalation_package"))
+        raw, _, _ = await provider.generate_json(
+            system_prompt=(
+                "You are a diagnostic assessment generator for MSP tech support escalations. "
+                "Always respond with valid JSON and nothing else. "
+                "Be concise and factual."
+            ),
+            messages=[{"role": "user", "content": prompt}],
+            max_tokens=700,
+        )
+
+        cleaned = raw.strip()
+        if cleaned.startswith("```"):
+            lines = cleaned.split("\n", 1)
+            cleaned = lines[1] if len(lines) > 1 else cleaned
+            if cleaned.endswith("```"):
+                cleaned = cleaned[:-3].rstrip()
+
+        result = json.loads(cleaned)
+
+        if not isinstance(result.get("suggested_steps"), list):
+            result["suggested_steps"] = []
+        if not isinstance(result.get("what_we_know"), list):
+            result["what_we_know"] = []
+        if result.get("confidence") not in ("low", "medium", "high"):
+            result["confidence"] = "medium"
+        if not isinstance(result.get("summary_prose"), str) or not result.get("summary_prose"):
+            result["summary_prose"] = result.get("likely_cause", "Assessment generated.")
+        if not isinstance(result.get("likely_cause"), str):
+            result["likely_cause"] = ""
+
+        return result
 
     async def generate_briefing(
         self, handoff_id: UUID, claiming_user_id: UUID
@@ -671,37 +726,29 @@ async def enrich_escalation_async(handoff_id: UUID, user_id: UUID) -> None:
 
             manager = HandoffManager(db)
 
-            # Build the enhanced package (Sonnet). Don't fail the whole
-            # task if it errors — the assessment is independently useful.
+            # Single consolidated AI call — replaces the old
+            # _generate_ai_assessment + _build_enhanced_escalation_package pair.
             try:
-                enhanced_pkg = await manager._build_enhanced_escalation_package(
-                    session, user_id
-                )
-                if enhanced_pkg:
-                    enhanced_pkg["intent"] = "escalate"
-                    enhanced_pkg["engineer_notes"] = handoff.engineer_notes
-                    enhanced_pkg["handoff_id"] = str(handoff.id)
-                    if isinstance(session.escalation_package, dict):
-                        enhanced_pkg.setdefault(
-                            "snapshot", session.escalation_package.get("snapshot")
-                        )
-                    session.escalation_package = enhanced_pkg
+                summary = await manager._generate_handoff_summary(session)
+                if summary:
+                    # ai_assessment (text) holds the PSA prose for backward compat
+                    # (push_to_psa reads it; generate_status_update falls back to it).
+                    handoff.ai_assessment = summary.get("summary_prose")
+                    handoff.ai_assessment_data = summary
+                    # Keep suggested_next_steps in escalation_package so
+                    # psa_documentation_service can read it without a handoff join.
+                    existing_pkg = (
+                        session.escalation_package
+                        if isinstance(session.escalation_package, dict)
+                        else {}
+                    )
+                    session.escalation_package = {
+                        **existing_pkg,
+                        "suggested_next_steps": summary.get("suggested_steps", []),
+                    }
             except Exception:
                 logger.exception(
-                    "enrich_escalation_async: enhanced package build failed for handoff %s",
-                    handoff_id,
-                )
-
-            # Generate the diagnostic AI assessment.
-            try:
-                ai_assessment, ai_assessment_data = (
-                    await manager._generate_ai_assessment_with_timeout(session)
-                )
-                handoff.ai_assessment = ai_assessment
-                handoff.ai_assessment_data = ai_assessment_data
-            except Exception:
-                logger.exception(
-                    "enrich_escalation_async: assessment generation failed for handoff %s",
+                    "enrich_escalation_async: summary generation failed for handoff %s",
                     handoff_id,
                 )
 
@@ -714,7 +761,7 @@ async def enrich_escalation_async(handoff_id: UUID, user_id: UUID) -> None:
                         "type": "handoff_assessment_ready",
                         "handoff_id": str(handoff.id),
                         "session_id": str(handoff.session_id),
-                        "has_assessment": handoff.ai_assessment is not None,
+                        "has_assessment": handoff.ai_assessment_data is not None,
                     },
                 )
             except Exception: