feat: add stream_ticket_notes generator for SSE doc streaming

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

backend/app/services/flowpilot_engine.py

@@ -8,6 +8,7 @@ import json
 import logging
 import uuid
 from datetime import datetime, timezone
+from collections.abc import AsyncIterator
 from typing import Any, Optional
 from uuid import UUID
 
@@ -1011,6 +1012,104 @@ async def generate_status_update(
     )
 
 
+async def stream_ticket_notes(
+    session_id: UUID,
+    user_id: UUID,
+    db: AsyncSession,
+) -> AsyncIterator[str]:
+    """Stream AI-generated structured ticket notes for a resolved session.
+
+    Yields text chunks suitable for SSE streaming.
+    """
+    session = await _load_session(session_id, user_id, db)
+
+    # Build conversation summary from messages (chat sessions)
+    # or steps (guided sessions)
+    messages = session.conversation_messages or []
+    if messages:
+        recent = messages[-20:]  # Last 20 messages for richer context
+        convo_text = "\n".join(
+            f"{'Engineer' if m['role'] == 'user' else 'AI Assistant'}: {m['content'][:500]}"
+            for m in recent
+            if isinstance(m, dict) and "role" in m and "content" in m
+        )
+    else:
+        # Fall back to steps for guided sessions
+        steps_summary = []
+        for step in sorted(session.steps, key=lambda s: s.step_order):
+            content = step.content or {}
+            text = content.get("text", "")
+            response = step.free_text_input or step.selected_option or ("Skipped" if step.was_skipped else None)
+            entry = f"Step {step.step_order + 1}: {text}"
+            if response:
+                entry += f"\n  Engineer response: {response}"
+            steps_summary.append(entry)
+        convo_text = "\n".join(steps_summary) if steps_summary else "No session data."
+
+    # Calculate time spent
+    now = datetime.now(timezone.utc)
+    ref_time = session.resolved_at or now
+    delta = ref_time - session.created_at
+    total_minutes = int(delta.total_seconds() / 60)
+    time_display = f"{total_minutes} minutes" if total_minutes < 60 else f"{total_minutes // 60}h {total_minutes % 60}m"
+
+    system_prompt = """You are generating internal ticket notes for an MSP engineer's PSA system.
+
+Generate EXACTLY these four markdown sections, in this order:
+
+## Problem Summary
+Summarize what the engineer reported and the initial symptoms. 1-3 sentences.
+
+## Steps Taken
+List the key diagnostic steps, commands run, checks performed, and findings. Use bullet points.
+
+## Resolution
+What fixed the issue or what the final action was. Be specific and technical.
+
+## Next Steps
+Any follow-up items, monitoring to watch, or preventive measures. Write "None" if not applicable.
+
+Rules:
+- Be technical, concise, and factual
+- Use markdown formatting (headers, bullet lists, bold for emphasis)
+- Include specific technical details (commands, settings, error messages) where available
+- Do NOT include greetings, sign-offs, or pleasantries
+- Do NOT wrap output in code fences
+- Output ONLY the four sections above, nothing else"""
+
+    user_message_parts = [
+        f"Session status: {session.status}",
+        f"Time spent: {time_display}",
+        f"Problem summary: {session.problem_summary or 'Not specified'}",
+    ]
+    if session.problem_domain:
+        user_message_parts.append(f"Problem domain: {session.problem_domain}")
+    if session.resolution_summary:
+        user_message_parts.append(f"Resolution notes: {session.resolution_summary}")
+    user_message_parts.append(f"\nSession conversation:\n{convo_text}")
+
+    user_message = "\n".join(user_message_parts)
+
+    provider = get_ai_provider(settings.get_model_for_action("quick_action"))
+
+    # Use streaming if provider supports it (Anthropic), otherwise fall back
+    try:
+        async for chunk in provider.generate_text_stream(
+            system_prompt=system_prompt,
+            messages=[{"role": "user", "content": user_message}],
+            max_tokens=1500,
+        ):
+            yield chunk
+    except NotImplementedError:
+        # Fallback for non-streaming providers (Gemini)
+        text, _, _ = await provider.generate_text(
+            system_prompt=system_prompt,
+            messages=[{"role": "user", "content": user_message}],
+            max_tokens=1500,
+        )
+        yield text
+
+
 def _build_status_update_prompt(
     audience: str,
     length: str,