feat: command palette, PSA ticket context, session-to-flow converter (#108)

* feat: add paletteIntent utility for command palette query classification

Detects query intent ('question' | 'keyword' | 'page' | 'empty') to drive
smart result ordering in the enhanced command palette.

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

* feat: add recentFlows localStorage utility for command palette empty state

Tracks recently visited flows (capped at 10) with deduplication by id,
surfaced in command palette when query is empty.

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

* feat: rewrite CommandPalette with categorized results and smart ranking

- Adds FlowPilot AI result (always present when query is non-empty)
- Intent-aware ordering: question → FlowPilot prominent; page → pages first;
  keyword → FlowPilot at top with flows/sessions/tags below
- Pages section with admin-gated items (uses useAuthStore)
- Tags extracted from flow search results with ?tag= navigation
- Quick Actions for create/import/scripts
- Empty state shows recent flows + quick actions
- Grouped rendering with section labels per design system
- Keyboard nav flattened across groups

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

* feat: add FlowPilot prefill handoff from command palette to AssistantChatPage

When navigated to /assistant with location.state.prefill, automatically
creates a new chat and sends the prefill message without user interaction.
Clears location state after handling to prevent re-trigger on back navigation.

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

* feat: track recently visited flows for command palette empty state

Calls addRecentFlow after tree data loads in both TreeNavigationPage and
ProceduralNavigationPage so the command palette can surface recent flows
when the query is empty.

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

* refactor: use useMemo instead of useCallback for groups builder in CommandPalette

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

* feat: add PSA ticket context Pydantic schemas (Task 6)

Add TicketDetails, CompanyInfo, ContactInfo, ConfigItem, TicketNote,
RelatedTicket, and TicketContext models in schemas/psa_context.py for
structured ticket context enrichment used by AI prompt injection.

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

* feat: add ticket context prompt formatter (Task 7)

format_ticket_context_for_prompt() in services/psa/ticket_context.py
serializes TicketContext into structured text for AI system prompts,
with 10-note limit, 200-char text previews, and human-readable timestamps.

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

* feat: add get_ticket_context() to ConnectWise provider (Task 8)

Fetches ticket details, company, contact, configurations, notes, and
related open tickets in parallel via asyncio.gather with partial failure
tolerance. Results are cached with a 5-minute TTL per ticket/connection.

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

* feat: add GET /integrations/psa/tickets/{id}/context endpoint (Task 9)

Returns rich TicketContext for a ticket ID. Handles PSA auth failures
(returns structured error), ticket-not-found (404), and general PSA
errors (502). Requires active PSA connection for the user's account.

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

* feat: inject PSA ticket context into copilot system prompt (Task 10)

When a copilot conversation has an associated session with a linked PSA
ticket, fetch the ticket context and append it to the system prompt.
Failure is non-critical — errors are logged and the copilot proceeds
without context rather than failing the request.

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

* feat: add PSA context API client with TypeScript interfaces

Defines TicketDetails, CompanyInfo, ContactInfo, ConfigItemInfo,
TicketNote, RelatedTicket, and TicketContext interfaces matching backend
psa_context.py schemas. Exports psaContextApi with getTicketContext().

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

* feat: add useTicketContext hook for PSA ticket context fetching

Accepts psaTicketId and psaConnectionId, fetches context on mount
when both IDs are present, and exposes refresh() for manual re-fetch.

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

* feat: add TicketContextPanel component with accordion sections

Glass-card panel showing ticket summary, status/priority/SLA, and
accordion sections for Client, Contact, Devices, Notes, and Related
tickets. Matches design system with font-label labels and ice-cyan accents.

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

* feat: mount TicketContextPanel in session runners when ticket is linked

ProceduralNavigationPage renders panel in left sidebar below step checklist.
TreeNavigationPage renders panel above breadcrumb trail. Both use
useTicketContext hook and show panel only when psa_ticket_id is set.

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

* feat: add fallback_steps to TypeScript types (Task 15)

Add optional fallback_steps field to ProceduralStep interface.
Add FallbackStepRecord interface and fallback_decisions field to Session.

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

* feat: add backend validation for fallback steps (Task 16)

Validate fallback_steps in procedural flow validation: required fields,
no nested fallback_steps, no duplicate IDs. Add FallbackStepRecord schema
and fallback_decisions field to SessionResponse.

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

* feat: create FallbackSteps UI component (Task 17)

Collapsible component supporting edit and execute modes. Edit mode
provides title/description inputs with add/remove controls. Execute
mode shows "This worked" / "Didn't help" action buttons with emerald/
rose styling. Amber accent styling throughout.

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

* feat: integrate FallbackSteps into editor and session runner (Task 18)

Wire FallbackSteps edit mode into StepEditor for procedure_step type
with add/remove/update handlers using crypto.randomUUID(). Add execute
mode rendering in ProceduralNavigationPage with fallbackDecisions state
tracking per parent step.

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

* feat: add session-to-flow request/response schemas (Task 19)

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

* feat: add session-to-flow AI generation service (Task 20)

Converts completed troubleshooting sessions into reusable procedural flows
with fallback branches. Includes PSA ticket context integration and
AI-generated step validation.

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

* feat: add POST /ai/session-to-flow endpoint (Task 21)

Converts a completed session into a reusable procedural flow using AI.
Includes quota checking, usage recording, and proper error handling.

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

* feat: add Create Flow from Session button to session detail page (Task 22)

Adds sessionToFlow API client, exports from api/index.ts, and integrates
a prominent "Create Flow from Session" button on SessionDetailPage for
completed sessions. Generates a procedural flow via AI then navigates
to the procedural editor.

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

* fix: cast tree_type to TreeType in session-to-flow creation

Fixes build error where string was not assignable to TreeType.

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

* fix: update Playwright test selectors to match actual UI

- Use Control+k instead of Meta+k (Linux/CI compatibility)
- Use 'AI Assistant' group label instead of 'FlowPilot AI'
- Match actual FlowPilot chat page elements (Start a Conversation, New Chat, textarea)

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

* fix: update Playwright test selectors to match actual UI

- Use specific command palette placeholder to avoid ambiguous matches
- Fix 'Quick Actions' scoping (two elements with same text)
- Fix 'Resolved' exact match on session detail page
- Fix tree editor to use getByText instead of getByDisplayValue
- Fix 'Add Step' strict mode by using .first()
- Fix fallback description placeholder text
- Update playwright.config.ts to use port 5433 and resolutionflow DB
- Update FlowPilot chat selectors to match actual page layout

11/17 new tests now passing. Remaining 6 need procedural session
navigation investigation.

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

* fix: resolve all Playwright test failures — 16/16 passing

- Fix procedural session tests: sessions auto-start, no Start button
- Fix strict mode violations: use getByRole('heading') for step titles
- Fix FlowPilot chat: use button role selector for New Chat
- Fix command palette page nav: scope Analytics click to palette modal
- Fix fallback runner: remove non-existent Start button click
- Update playwright.config to port 5433 for local Docker

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

---------

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

backend/app/api/endpoints/ai_builder.py

@@ -6,6 +6,9 @@
   POST /ai/branch-detail — Stage 3: AI generates detail for one branch
   POST /ai/assemble      — Stage 4: assemble branches into tree (no AI)
   GET  /ai/quota         — quota status
+
+Session conversion:
+  POST /ai/session-to-flow — Convert a completed session into a procedural flow
 """
 import logging
 from typing import Annotated
@@ -40,6 +43,8 @@ from app.schemas.ai_builder import (
     AIAssembleResponse,
     AIQuotaStatusResponse,
 )
+from app.schemas.session_to_flow import SessionToFlowRequest, SessionToFlowResponse
+from app.services.session_to_flow_service import generate_flow_from_session
 
 logger = logging.getLogger(__name__)
 
@@ -437,3 +442,97 @@ async def assemble(
         summary=stats,
         status="completed",
     )
+
+
+@router.post("/session-to-flow", response_model=SessionToFlowResponse)
+@limiter.limit("5/minute")
+async def session_to_flow(
+    request: Request,
+    data: SessionToFlowRequest,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+):
+    """Convert a completed troubleshooting session into a reusable procedural flow."""
+    _require_ai_enabled()
+
+    # Check AI quota
+    allowed, quota_status = await check_ai_quota(
+        user_id=current_user.id,
+        account_id=current_user.account_id,
+        db=db,
+        billing_anchor=current_user.ai_billing_cycle_anchor_at,
+        is_super_admin=current_user.is_super_admin,
+    )
+    if not allowed:
+        reset_key = (
+            "daily_reset_at"
+            if quota_status.get("deny_reason") == "daily"
+            else "monthly_reset_at"
+        )
+        raise HTTPException(
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            detail={
+                "message": f"AI build limit exceeded ({quota_status['deny_reason']})",
+                "reset_at": quota_status.get(reset_key),
+                "quota": quota_status,
+            },
+        )
+
+    plan = await get_user_plan(current_user.account_id, db)
+
+    try:
+        result = await generate_flow_from_session(
+            session_id=data.session_id,
+            user_id=current_user.id,
+            account_id=current_user.account_id,
+            db=db,
+        )
+    except ValueError as e:
+        logger.warning("session_to_flow validation error: %s", e)
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=str(e),
+        )
+    except Exception as e:
+        logger.exception("session_to_flow failed: %s", e)
+        await record_ai_usage(
+            user_id=current_user.id,
+            account_id=current_user.account_id,
+            conversation_id=None,
+            generation_type="session_to_flow",
+            tier=plan,
+            input_tokens=0,
+            output_tokens=0,
+            estimated_cost=0,
+            succeeded=False,
+            counts_toward_quota=False,
+            error_code=type(e).__name__,
+            extra_data={"session_id": data.session_id, "error": str(e)},
+            db=db,
+        )
+        await db.commit()
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Failed to generate flow: {type(e).__name__}. Please try again.",
+        )
+
+    # Record successful quota-consuming usage
+    await record_ai_usage(
+        user_id=current_user.id,
+        account_id=current_user.account_id,
+        conversation_id=None,
+        generation_type="session_to_flow",
+        tier=plan,
+        input_tokens=0,
+        output_tokens=0,
+        estimated_cost=0,
+        succeeded=True,
+        counts_toward_quota=True,
+        error_code=None,
+        extra_data={"session_id": data.session_id},
+        db=db,
+    )
+    await db.commit()
+
+    return SessionToFlowResponse(**result)

backend/app/api/endpoints/integrations.py

@@ -319,6 +319,61 @@ async def search_tickets(
         raise HTTPException(status_code=502, detail=str(e))
 
 
+@router.get("/tickets/{ticket_id}/context")
+async def get_ticket_context(
+    ticket_id: int,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+):
+    """Get rich ticket context (company, contact, configs, notes, related tickets) for AI prompt injection."""
+    from app.services.psa.registry import get_provider_for_account
+    from app.services.psa.exceptions import (
+        PSAError,
+        PSAAuthError,
+        PSAPermissionError,
+        PSANotFoundError,
+        PSAConnectionError,
+    )
+    from app.schemas.psa_context import TicketContext
+
+    if not current_user.account_id:
+        raise HTTPException(status_code=400, detail="User has no account")
+
+    # Look up the active connection for connection_id
+    conn_result = await db.execute(
+        select(PsaConnection).where(
+            PsaConnection.account_id == current_user.account_id,
+            PsaConnection.is_active.is_(True),
+        )
+    )
+    connection = conn_result.scalar_one_or_none()
+    if not connection:
+        raise HTTPException(status_code=404, detail="No active PSA connection configured")
+
+    try:
+        provider = await get_provider_for_account(current_user.account_id, db)
+    except PSAConnectionError:
+        raise HTTPException(status_code=404, detail="No active PSA connection configured")
+    except PSAError as e:
+        raise HTTPException(status_code=502, detail=str(e))
+
+    try:
+        ctx: TicketContext = await provider.get_ticket_context(
+            ticket_id=ticket_id,
+            connection_id=str(connection.id),
+        )
+        return ctx
+    except (PSAAuthError, PSAPermissionError):
+        raise HTTPException(
+            status_code=502,
+            detail={"error": "psa_auth_failed", "message": "PSA credentials may have expired."},
+        )
+    except PSANotFoundError:
+        raise HTTPException(status_code=404, detail="Ticket not found")
+    except PSAError as e:
+        raise HTTPException(status_code=502, detail=str(e))
+
+
 @router.get("/tickets/{ticket_id}")
 async def get_ticket(
     ticket_id: str,

backend/app/core/ai_tree_validator.py

@@ -301,6 +301,31 @@ def validate_generated_procedural_steps(tree: dict[str, Any]) -> list[str]:
                 f"Must be one of: {', '.join(sorted(VALID_CONTENT_TYPES))}"
             )
 
+        # Validate fallback_steps if present (one level deep only)
+        fallback_steps = step.get("fallback_steps")
+        if fallback_steps is not None:
+            if not isinstance(fallback_steps, list):
+                errors.append(f"Step '{step_id or f'index {i}'}' fallback_steps must be an array")
+            else:
+                fallback_ids: set[str] = set()
+                for j, fb_step in enumerate(fallback_steps):
+                    if not isinstance(fb_step, dict):
+                        errors.append(f"Fallback step at {step_id}[{j}] is not an object")
+                        continue
+                    fb_id = fb_step.get("id")
+                    if not fb_id or not isinstance(fb_id, str):
+                        errors.append(f"Fallback step at {step_id}[{j}] missing or invalid 'id'")
+                    elif fb_id in all_ids or fb_id in fallback_ids:
+                        errors.append(f"Duplicate fallback step ID: '{fb_id}' (collides with primary or other fallback steps)")
+                    else:
+                        fallback_ids.add(fb_id)
+                        all_ids.add(fb_id)
+                    fb_title = fb_step.get("title")
+                    if not fb_title or not isinstance(fb_title, str):
+                        errors.append(f"Fallback step '{fb_id or f'{step_id}[{j}]'}' missing or invalid 'title'")
+                    if fb_step.get("fallback_steps"):
+                        errors.append(f"Fallback step '{fb_id or f'{step_id}[{j}]'}' cannot have its own fallback_steps (one level deep only)")
+
     # Must have exactly one procedure_end as the last step
     if procedure_end_count == 0:
         errors.append("Procedural flow must have exactly one 'procedure_end' step")

backend/app/core/tree_validation.py

@@ -208,6 +208,34 @@ def validate_procedural_structure(tree_structure: dict[str, Any]) -> tuple[bool,
         if content_type and content_type not in VALID_CONTENT_TYPES:
             errors.append({"field": f"{path}.content_type", "message": f"Invalid content_type: {content_type}. Must be one of: {', '.join(VALID_CONTENT_TYPES)}"})
 
+        # Validate fallback_steps if present (one level deep only)
+        fallback_steps = step.get("fallback_steps")
+        if fallback_steps is not None:
+            if not isinstance(fallback_steps, list):
+                errors.append({"field": f"{path}.fallback_steps", "message": "fallback_steps must be an array"})
+            else:
+                fallback_ids: set[str] = set()
+                for j, fb_step in enumerate(fallback_steps):
+                    fb_path = f"{path}.fallback_steps[{j}]"
+                    if not isinstance(fb_step, dict):
+                        errors.append({"field": fb_path, "message": "Fallback step must be an object"})
+                        continue
+                    fb_id = fb_step.get("id")
+                    if not fb_id:
+                        errors.append({"field": f"{fb_path}.id", "message": "Fallback step must have an id"})
+                    elif fb_id in seen_ids or fb_id in fallback_ids:
+                        errors.append({"field": f"{fb_path}.id", "message": f"Duplicate fallback step id: {fb_id}"})
+                    else:
+                        fallback_ids.add(fb_id)
+                        seen_ids.add(fb_id)
+                    if not fb_step.get("title"):
+                        errors.append({"field": f"{fb_path}.title", "message": "Fallback step must have a non-empty title"})
+                    fb_type = fb_step.get("type")
+                    if fb_type and fb_type not in VALID_STEP_TYPES:
+                        errors.append({"field": f"{fb_path}.type", "message": f"Invalid fallback step type: {fb_type}"})
+                    if fb_step.get("fallback_steps"):
+                        errors.append({"field": f"{fb_path}.fallback_steps", "message": "Fallback steps cannot have their own fallback_steps (one level deep only)"})
+
     # Must have exactly one end step
     if end_count == 0:
         errors.append({"field": "tree_structure.steps", "message": "Procedural tree must have a procedure_end step as the last step"})

backend/app/schemas/psa_context.py

@@ -0,0 +1,68 @@
+"""Pydantic schemas for PSA ticket context enrichment."""
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel
+
+
+class TicketDetails(BaseModel):
+    id: int
+    summary: str
+    status: str
+    priority: str
+    board: str
+    sla: str | None = None
+    date_entered: datetime
+    resources: str | None = None
+
+
+class CompanyInfo(BaseModel):
+    id: int
+    name: str
+    site: str | None = None
+    address: str | None = None
+    phone: str | None = None
+    type: str | None = None
+    territory: str | None = None
+
+
+class ContactInfo(BaseModel):
+    name: str
+    email: str | None = None
+    phone: str | None = None
+    title: str | None = None
+
+
+class ConfigItem(BaseModel):
+    device_identifier: str
+    type: str | None = None
+    os_type: str | None = None
+    serial_number: str | None = None
+    ip_address: str | None = None
+    model_number: str | None = None
+
+
+class TicketNote(BaseModel):
+    text: str
+    member: str | None = None
+    date_created: datetime
+    internal_analysis_flag: bool = False
+
+
+class RelatedTicket(BaseModel):
+    id: int
+    summary: str
+    status: str
+    priority: str
+    board: str
+
+
+class TicketContext(BaseModel):
+    ticket: TicketDetails
+    company: CompanyInfo
+    contact: ContactInfo | None = None
+    configurations: list[ConfigItem] = []
+    notes: list[TicketNote] = []
+    related_tickets: list[RelatedTicket] = []
+    fetched_at: datetime

backend/app/schemas/session.py

@@ -98,6 +98,9 @@ class SessionResponse(BaseModel):
     psa_ticket_id: Optional[str] = None
     psa_connection_id: Optional[UUID] = None
 
+    # Fallback step decisions
+    fallback_decisions: list[dict[str, Any]] = Field(default_factory=list)
+
     class Config:
         from_attributes = True
 
@@ -123,6 +126,14 @@ class SessionComplete(BaseModel):
     next_steps: Optional[str] = None
 
 
+class FallbackStepRecord(BaseModel):
+    parent_step_id: str
+    fallback_step_id: str
+    completed_at: str | None = None
+    notes: str | None = None
+    outcome: Literal['resolved', 'not_resolved', 'skipped']
+
+
 class SessionVariablesUpdate(BaseModel):
     """Partial update to session variables (dict merge)."""
     variables: dict[str, str] = Field(..., description="Key-value pairs to merge into session_variables")

backend/app/schemas/session_to_flow.py

@@ -0,0 +1,13 @@
+from pydantic import BaseModel
+
+
+class SessionToFlowRequest(BaseModel):
+    session_id: str
+
+
+class SessionToFlowResponse(BaseModel):
+    name: str
+    description: str
+    tree_type: str  # Always "procedural"
+    tags: list[str]
+    tree_structure: dict  # Procedural steps with optional fallback_steps

backend/app/services/copilot_service.py

@@ -180,6 +180,40 @@ async def send_message(
     system_prompt += _build_flow_context(tree, conversation.current_node_id)
     system_prompt += build_rag_context(rag_results)
 
+    # Inject PSA ticket context if session has a linked ticket
+    if conversation.session_id:
+        try:
+            from app.models.session import Session as SessionModel
+            session_result = await db.execute(
+                select(SessionModel).where(SessionModel.id == conversation.session_id)
+            )
+            session = session_result.scalar_one_or_none()
+            if session and session.psa_ticket_id:
+                try:
+                    from app.services.psa.registry import get_provider_for_account
+                    from app.services.psa.ticket_context import format_ticket_context_for_prompt
+
+                    provider = await get_provider_for_account(conversation.account_id, db)
+                    connection_id = str(session.psa_connection_id) if session.psa_connection_id else None
+                    ticket_ctx = await provider.get_ticket_context(
+                        ticket_id=int(session.psa_ticket_id),
+                        connection_id=connection_id,
+                    )
+                    system_prompt += "\n\n" + format_ticket_context_for_prompt(ticket_ctx)
+                except Exception as psa_err:
+                    logger.warning(
+                        "Failed to fetch PSA ticket context for copilot (session=%s, ticket=%s): %s",
+                        conversation.session_id,
+                        session.psa_ticket_id,
+                        psa_err,
+                    )
+        except Exception as session_err:
+            logger.warning(
+                "Failed to look up session for copilot PSA context (session_id=%s): %s",
+                conversation.session_id,
+                session_err,
+            )
+
     # Build messages for AI
     ai_messages = []
     for msg in conversation.messages:

backend/app/services/psa/connectwise/provider.py

@@ -1,6 +1,10 @@
 """ConnectWise implementation of PSAProvider."""
 from __future__ import annotations
 
+import asyncio
+import logging
+from datetime import datetime, timezone
+
 from app.services.psa.base import PSAProvider
 from app.services.psa.cache import psa_cache
 from app.services.psa.types import (
@@ -14,6 +18,8 @@ from app.services.psa.types import (
 )
 from .client import ConnectWiseClient
 
+logger = logging.getLogger(__name__)
+
 
 class ConnectWiseProvider(PSAProvider):
     """ConnectWise PSA provider implementation."""
@@ -263,6 +269,251 @@ class ConnectWiseProvider(PSAProvider):
         psa_cache.set(cache_key, result, ttl_seconds=900)
         return result
 
+    # ── Ticket Context ────────────────────────────────────────────────
+
+    async def get_ticket_context(
+        self, ticket_id: int, connection_id: str | None = None
+    ):
+        """Fetch rich ticket context for AI prompt injection.
+
+        Returns a TicketContext with ticket details, company, contact,
+        configurations, recent notes, and related open tickets.
+        Results are cached for 5 minutes per ticket.
+        """
+        from app.schemas.psa_context import (
+            TicketContext,
+            TicketDetails,
+            CompanyInfo,
+            ContactInfo,
+            ConfigItem,
+            TicketNote,
+            RelatedTicket,
+        )
+
+        cache_key = f"{connection_id or 'default'}:ticket_context:{ticket_id}"
+        cached = psa_cache.get(cache_key)
+        if cached is not None:
+            return cached
+
+        # Fetch ticket first to get company_id and contact_id
+        ticket_data = await self.client.get(
+            f"/service/tickets/{ticket_id}",
+            params={
+                "fields": "id,summary,status,priority,board,sla,dateEntered,resources,company,contact"
+            },
+        )
+
+        company_id = ticket_data.get("company", {}).get("id") if ticket_data.get("company") else None
+        contact_id = ticket_data.get("contact", {}).get("id") if ticket_data.get("contact") else None
+
+        # Build parallel fetch tasks
+        configs_task = asyncio.create_task(
+            self.client.get(
+                f"/service/tickets/{ticket_id}/configurations",
+                params={
+                    "fields": "id,deviceIdentifier,type,osType,serialNumber,ipAddress,modelNumber"
+                },
+            )
+        )
+        notes_task = asyncio.create_task(
+            self.client.get(
+                f"/service/tickets/{ticket_id}/notes",
+                params={
+                    "pageSize": "20",
+                    "orderBy": "dateCreated desc",
+                    "fields": "id,text,member,dateCreated,internalAnalysisFlag",
+                },
+            )
+        )
+        company_task = asyncio.create_task(
+            self.client.get(
+                f"/company/companies/{company_id}",
+                params={
+                    "fields": "id,name,site,addressLine1,city,state,zip,phoneNumber,type,territory"
+                },
+            )
+        ) if company_id else None
+
+        related_task = asyncio.create_task(
+            self.client.get(
+                "/service/tickets",
+                params={
+                    "conditions": f"company/id={company_id} AND closedFlag=false AND id != {ticket_id}",
+                    "pageSize": "5",
+                    "orderBy": "id desc",
+                    "fields": "id,summary,status,priority,board",
+                },
+            )
+        ) if company_id else None
+
+        contact_task = asyncio.create_task(
+            self.client.get(
+                f"/company/contacts/{contact_id}",
+                params={
+                    "fields": "id,firstName,lastName,title,defaultPhoneNbr,communicationItems"
+                },
+            )
+        ) if contact_id else None
+
+        # Gather all tasks with partial failure tolerance
+        tasks_to_await = [t for t in [configs_task, notes_task, company_task, related_task, contact_task] if t is not None]
+        task_results = await asyncio.gather(*tasks_to_await, return_exceptions=True)
+
+        # Unpack results in order (skipping None tasks)
+        result_iter = iter(task_results)
+        configs_raw = next(result_iter)
+        notes_raw = next(result_iter)
+        company_raw = next(result_iter) if company_task else None
+        related_raw = next(result_iter) if related_task else None
+        contact_raw = next(result_iter) if contact_task else None
+
+        # Map ticket details
+        def _parse_dt(val: str | None) -> datetime:
+            if not val:
+                return datetime.now(timezone.utc)
+            try:
+                # CW returns ISO 8601 strings — ensure timezone aware
+                dt = datetime.fromisoformat(val.replace("Z", "+00:00"))
+                if dt.tzinfo is None:
+                    dt = dt.replace(tzinfo=timezone.utc)
+                return dt
+            except (ValueError, AttributeError):
+                return datetime.now(timezone.utc)
+
+        ticket_details = TicketDetails(
+            id=ticket_data["id"],
+            summary=ticket_data.get("summary", ""),
+            status=ticket_data.get("status", {}).get("name", "") if isinstance(ticket_data.get("status"), dict) else str(ticket_data.get("status", "")),
+            priority=ticket_data.get("priority", {}).get("name", "") if isinstance(ticket_data.get("priority"), dict) else str(ticket_data.get("priority", "")),
+            board=ticket_data.get("board", {}).get("name", "") if isinstance(ticket_data.get("board"), dict) else str(ticket_data.get("board", "")),
+            sla=ticket_data.get("sla", {}).get("name") if isinstance(ticket_data.get("sla"), dict) else ticket_data.get("sla"),
+            date_entered=_parse_dt(ticket_data.get("dateEntered")),
+            resources=ticket_data.get("resources"),
+        )
+
+        # Map company
+        company_info: CompanyInfo
+        if isinstance(company_raw, dict):
+            addr_parts = [
+                company_raw.get("addressLine1"),
+                company_raw.get("city"),
+                company_raw.get("state"),
+                company_raw.get("zip"),
+            ]
+            address = ", ".join(p for p in addr_parts if p) or None
+            company_info = CompanyInfo(
+                id=company_raw["id"],
+                name=company_raw.get("name", ""),
+                site=company_raw.get("site", {}).get("name") if isinstance(company_raw.get("site"), dict) else company_raw.get("site"),
+                address=address,
+                phone=company_raw.get("phoneNumber"),
+                type=company_raw.get("type", {}).get("name") if isinstance(company_raw.get("type"), dict) else company_raw.get("type"),
+                territory=company_raw.get("territory", {}).get("name") if isinstance(company_raw.get("territory"), dict) else company_raw.get("territory"),
+            )
+        else:
+            if isinstance(company_raw, Exception):
+                logger.warning("Failed to fetch company for ticket %s: %s", ticket_id, company_raw)
+            # Fallback: use data from ticket itself
+            company_info = CompanyInfo(
+                id=company_id or 0,
+                name=ticket_data.get("company", {}).get("name", "") if isinstance(ticket_data.get("company"), dict) else "",
+            )
+
+        # Map contact
+        contact_info: ContactInfo | None = None
+        if isinstance(contact_raw, dict):
+            first = contact_raw.get("firstName", "")
+            last = contact_raw.get("lastName", "")
+            full_name = f"{first} {last}".strip() or "Unknown"
+
+            # Extract email from communicationItems
+            email: str | None = None
+            comm_items = contact_raw.get("communicationItems", [])
+            if isinstance(comm_items, list):
+                for item in comm_items:
+                    if isinstance(item, dict) and item.get("communicationType") == "Email":
+                        email = item.get("value")
+                        break
+
+            contact_info = ContactInfo(
+                name=full_name,
+                email=email,
+                phone=contact_raw.get("defaultPhoneNbr"),
+                title=contact_raw.get("title"),
+            )
+        elif isinstance(contact_raw, Exception):
+            logger.warning("Failed to fetch contact for ticket %s: %s", ticket_id, contact_raw)
+
+        # Map configurations
+        configurations: list[ConfigItem] = []
+        if isinstance(configs_raw, list):
+            for cfg in configs_raw:
+                if not isinstance(cfg, dict):
+                    continue
+                configurations.append(ConfigItem(
+                    device_identifier=cfg.get("deviceIdentifier", ""),
+                    type=cfg.get("type", {}).get("name") if isinstance(cfg.get("type"), dict) else cfg.get("type"),
+                    os_type=cfg.get("osType", {}).get("name") if isinstance(cfg.get("osType"), dict) else cfg.get("osType"),
+                    serial_number=cfg.get("serialNumber"),
+                    ip_address=cfg.get("ipAddress"),
+                    model_number=cfg.get("modelNumber"),
+                ))
+        elif isinstance(configs_raw, Exception):
+            logger.warning("Failed to fetch configs for ticket %s: %s", ticket_id, configs_raw)
+
+        # Map notes
+        notes: list[TicketNote] = []
+        if isinstance(notes_raw, list):
+            for note in notes_raw:
+                if not isinstance(note, dict):
+                    continue
+                member_name: str | None = None
+                member_obj = note.get("member")
+                if isinstance(member_obj, dict):
+                    first = member_obj.get("firstName", "")
+                    last = member_obj.get("lastName", "")
+                    member_name = f"{first} {last}".strip() or member_obj.get("identifier")
+                elif isinstance(member_obj, str):
+                    member_name = member_obj
+
+                notes.append(TicketNote(
+                    text=note.get("text", ""),
+                    member=member_name,
+                    date_created=_parse_dt(note.get("dateCreated")),
+                    internal_analysis_flag=note.get("internalAnalysisFlag", False),
+                ))
+        elif isinstance(notes_raw, Exception):
+            logger.warning("Failed to fetch notes for ticket %s: %s", ticket_id, notes_raw)
+
+        # Map related tickets
+        related_tickets: list[RelatedTicket] = []
+        if isinstance(related_raw, list):
+            for rt in related_raw:
+                if not isinstance(rt, dict):
+                    continue
+                related_tickets.append(RelatedTicket(
+                    id=rt["id"],
+                    summary=rt.get("summary", ""),
+                    status=rt.get("status", {}).get("name", "") if isinstance(rt.get("status"), dict) else str(rt.get("status", "")),
+                    priority=rt.get("priority", {}).get("name", "") if isinstance(rt.get("priority"), dict) else str(rt.get("priority", "")),
+                    board=rt.get("board", {}).get("name", "") if isinstance(rt.get("board"), dict) else str(rt.get("board", "")),
+                ))
+        elif isinstance(related_raw, Exception):
+            logger.warning("Failed to fetch related tickets for ticket %s: %s", ticket_id, related_raw)
+
+        ctx = TicketContext(
+            ticket=ticket_details,
+            company=company_info,
+            contact=contact_info,
+            configurations=configurations,
+            notes=notes,
+            related_tickets=related_tickets,
+            fetched_at=datetime.now(timezone.utc),
+        )
+
+        psa_cache.set(cache_key, ctx, ttl_seconds=300)
+        return ctx
+
     # ── Private helpers ───────────────────────────────────────────────
 
     @staticmethod

backend/app/services/psa/ticket_context.py

@@ -0,0 +1,84 @@
+"""Format PSA ticket context as structured text for AI system prompts."""
+from __future__ import annotations
+
+from app.schemas.psa_context import TicketContext
+
+
+def format_ticket_context_for_prompt(ctx: TicketContext) -> str:
+    """Serialize a TicketContext into a structured text block for AI prompts."""
+    lines: list[str] = ["=== TICKET CONTEXT ==="]
+
+    # Ticket summary line
+    t = ctx.ticket
+    lines.append(f'Ticket: #{t.id} — "{t.summary}"')
+    lines.append(f"Status: {t.status} | Priority: {t.priority}")
+    lines.append(f"Board: {t.board}")
+    if t.sla:
+        lines.append(f"SLA Deadline: {t.sla}")
+    if t.resources:
+        lines.append(f"Assigned To: {t.resources}")
+
+    # Company block
+    lines.append("")
+    c = ctx.company
+    lines.append(f"Client: {c.name}")
+    if c.site:
+        lines.append(f"Site: {c.site}")
+    if c.address:
+        lines.append(f"Address: {c.address}")
+    if c.phone:
+        lines.append(f"Phone: {c.phone}")
+    if c.type:
+        lines.append(f"Type: {c.type}")
+    if c.territory:
+        lines.append(f"Territory: {c.territory}")
+
+    # Contact block
+    if ctx.contact:
+        contact = ctx.contact
+        contact_parts = [contact.name]
+        if contact.email:
+            contact_parts.append(f"({contact.email})")
+        if contact.title:
+            contact_parts.append(f"— {contact.title}")
+        contact_line = " ".join(contact_parts)
+        if contact.phone:
+            contact_line += f" — {contact.phone}"
+        lines.append("")
+        lines.append(f"Contact: {contact_line}")
+
+    # Devices
+    if ctx.configurations:
+        lines.append("")
+        lines.append("Devices:")
+        for cfg in ctx.configurations:
+            parts = [cfg.device_identifier]
+            if cfg.type:
+                parts.append(cfg.type)
+            if cfg.os_type:
+                parts.append(cfg.os_type)
+            if cfg.ip_address:
+                parts.append(cfg.ip_address)
+            lines.append("- " + " | ".join(parts))
+
+    # Recent Notes (limit 10, text preview 200 chars)
+    if ctx.notes:
+        lines.append("")
+        lines.append("Recent Notes:")
+        for note in ctx.notes[:10]:
+            date_str = note.date_created.strftime("%b %d, %I:%M %p")
+            member_str = f"{note.member}: " if note.member else ""
+            text_preview = note.text[:200]
+            if len(note.text) > 200:
+                text_preview += "..."
+            lines.append(f"- [{date_str}] {member_str}{text_preview}")
+
+    # Related open tickets
+    if ctx.related_tickets:
+        lines.append("")
+        lines.append("Related Open Tickets:")
+        for rt in ctx.related_tickets:
+            lines.append(f'- #{rt.id}: "{rt.summary}" ({rt.status}, {rt.priority})')
+
+    lines.append("=== END CONTEXT ===")
+    return "\n".join(lines)

backend/app/services/session_to_flow_service.py

@@ -0,0 +1,254 @@
+"""Session-to-Flow AI generation service.
+
+Converts a completed troubleshooting session into a reusable procedural
+flow with fallback branches, powered by AI.
+"""
+import json
+import logging
+import re
+import uuid
+from typing import Any, Optional
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.ai_provider import get_ai_provider
+from app.core.config import settings
+from app.core.ai_tree_validator import validate_generated_procedural_steps
+from app.models.session import Session
+from app.models.tree import Tree
+
+logger = logging.getLogger(__name__)
+
+# AI system prompt for session-to-flow conversion
+SESSION_TO_FLOW_SYSTEM_PROMPT = """You are an expert MSP engineer and IT process documentation specialist.
+
+Your task is to convert a completed IT troubleshooting session into a reusable procedural flow with optional fallback branches.
+
+You will receive:
+- The session outcome and engineer notes
+- An ordered list of decisions the engineer made (questions/answers, actions, notes, command output)
+- The original troubleshooting tree structure (for context on alternative paths)
+- Optional PSA ticket context
+
+Generate a procedural flow that can be replicated for similar issues in the future. Each step should:
+1. Be concrete and actionable — include exact commands, paths, or config values
+2. Have a clear verification criterion
+3. Include 1-3 fallback_steps per step (alternatives to try if the primary action fails)
+4. End with a procedure_end step summarizing the resolution
+
+Return ONLY a valid JSON object with this exact structure:
+{
+  "name": "Short descriptive title (5-10 words)",
+  "description": "2-3 sentence description of what this flow resolves",
+  "tags": ["tag1", "tag2"],
+  "steps": [
+    {
+      "id": "step-1",
+      "type": "procedure_step",
+      "title": "Step title",
+      "description": "Detailed instructions with exact commands/paths",
+      "content_type": "text",
+      "fallback_steps": [
+        {
+          "id": "step-1-fb-1",
+          "type": "procedure_step",
+          "title": "Alternative: ...",
+          "description": "Alternative approach if primary step fails",
+          "content_type": "text"
+        }
+      ]
+    },
+    {
+      "id": "step-end",
+      "type": "procedure_end",
+      "title": "Resolution Complete",
+      "description": "Summary of what was resolved and any follow-up actions"
+    }
+  ]
+}
+
+Rules:
+- Use unique string IDs for all steps (e.g. "step-1", "step-2", "step-1-fb-1")
+- Include 3-10 procedure_step entries before the procedure_end
+- Each step should be 1 concrete action, not a vague suggestion
+- Fallback steps use the same schema as procedure_steps but represent alternative approaches
+- Tags should be 2-5 relevant keywords (technology, vendor, symptom)
+- Do NOT wrap JSON in markdown code fences
+- Return only valid JSON, nothing else
+"""
+
+
+def _strip_markdown_fences(text: str) -> str:
+    """Strip markdown code fences if the model wrapped its JSON response."""
+    text = text.strip()
+    match = re.match(r"^```(?:json)?\s*([\s\S]*?)```$", text)
+    if match:
+        return match.group(1).strip()
+    return text
+
+
+def _build_session_context(session: Session, tree: Optional[Tree]) -> str:
+    """Build a context string from session data for the AI prompt."""
+    parts: list[str] = []
+
+    # Flow info
+    tree_name = session.tree_snapshot.get("name", "Unknown Flow") if session.tree_snapshot else "Unknown Flow"
+    parts.append(f"Flow: {tree_name}")
+    parts.append(f"Outcome: {session.outcome or 'Unknown'}")
+
+    if session.outcome_notes:
+        parts.append(f"Outcome Notes: {session.outcome_notes}")
+
+    # Session decisions (the troubleshooting path taken)
+    if session.decisions:
+        parts.append("\n--- TROUBLESHOOTING PATH ---")
+        for i, decision in enumerate(session.decisions):
+            step_parts: list[str] = [f"\nStep {i + 1}:"]
+            if decision.get("question"):
+                step_parts.append(f"  Question: {decision['question']}")
+            if decision.get("answer"):
+                step_parts.append(f"  Answer: {decision['answer']}")
+            if decision.get("action_performed"):
+                step_parts.append(f"  Action: {decision['action_performed']}")
+            if decision.get("notes"):
+                step_parts.append(f"  Notes: {decision['notes']}")
+            if decision.get("command_output"):
+                # Truncate long command output
+                output = decision["command_output"]
+                if len(output) > 500:
+                    output = output[:500] + "... [truncated]"
+                step_parts.append(f"  Command Output: {output}")
+            parts.append("\n".join(step_parts))
+
+    # Scratchpad
+    if session.scratchpad and session.scratchpad.strip():
+        parts.append(f"\n--- ENGINEER SCRATCHPAD ---\n{session.scratchpad[:1000]}")
+
+    # Original tree structure (for branch context, truncated)
+    if tree and tree.tree_structure:
+        tree_json = json.dumps(tree.tree_structure, indent=None)
+        if len(tree_json) > 3000:
+            tree_json = tree_json[:3000] + "... [truncated]"
+        parts.append(f"\n--- ORIGINAL TREE STRUCTURE (for alternative paths) ---\n{tree_json}")
+    elif session.tree_snapshot:
+        snapshot_json = json.dumps(session.tree_snapshot, indent=None)
+        if len(snapshot_json) > 3000:
+            snapshot_json = snapshot_json[:3000] + "... [truncated]"
+        parts.append(f"\n--- TREE SNAPSHOT (for alternative paths) ---\n{snapshot_json}")
+
+    return "\n".join(parts)
+
+
+async def generate_flow_from_session(
+    session_id: str,
+    user_id: UUID,
+    account_id: UUID,
+    db: AsyncSession,
+) -> dict[str, Any]:
+    """Generate a procedural flow from a completed session.
+
+    Returns a dict with keys: name, description, tree_type, tags, tree_structure.
+    Raises ValueError on validation failures, Exception on AI/DB errors.
+    """
+    # Load the session
+    session_uuid = UUID(session_id) if isinstance(session_id, str) else session_id
+    result = await db.execute(
+        select(Session).where(
+            Session.id == session_uuid,
+            Session.user_id == user_id,
+        )
+    )
+    session = result.scalar_one_or_none()
+    if not session:
+        raise ValueError(f"Session '{session_id}' not found or access denied")
+
+    # Load the original tree for branch context
+    tree: Optional[Tree] = None
+    if session.tree_id:
+        tree_result = await db.execute(
+            select(Tree).where(Tree.id == session.tree_id)
+        )
+        tree = tree_result.scalar_one_or_none()
+
+    # Build session context
+    session_context = _build_session_context(session, tree)
+
+    # Optionally fetch PSA ticket context
+    psa_context = ""
+    if session.psa_ticket_id and session.psa_connection_id:
+        try:
+            from app.services.psa.registry import get_provider_for_account
+            from app.services.psa.ticket_context import format_ticket_context_for_prompt
+
+            psa_provider = await get_provider_for_account(account_id, db)
+            connection_id = str(session.psa_connection_id)
+            ticket_ctx = await psa_provider.get_ticket_context(
+                ticket_id=int(session.psa_ticket_id),
+                connection_id=connection_id,
+            )
+            psa_context = "\n\n--- PSA TICKET CONTEXT ---\n" + format_ticket_context_for_prompt(ticket_ctx)
+        except Exception as psa_err:
+            logger.warning(
+                "Failed to fetch PSA ticket context for session-to-flow (session=%s, ticket=%s): %s",
+                session_id,
+                session.psa_ticket_id,
+                psa_err,
+            )
+
+    # Build user message
+    user_message = (
+        "Please convert the following completed troubleshooting session into a reusable procedural flow:\n\n"
+        f"{session_context}"
+        f"{psa_context}"
+    )
+
+    # Call AI
+    model = settings.get_model_for_action("generate_steps")
+    provider = get_ai_provider(model=model)
+
+    raw_text, input_tokens, output_tokens = await provider.generate_json(
+        system_prompt=SESSION_TO_FLOW_SYSTEM_PROMPT,
+        messages=[{"role": "user", "content": user_message}],
+        max_tokens=4096,
+    )
+
+    logger.info(
+        "session_to_flow AI response (tokens in=%d out=%d, session=%s)",
+        input_tokens,
+        output_tokens,
+        session_id,
+    )
+
+    # Strip markdown fences and parse JSON
+    raw_text = _strip_markdown_fences(raw_text)
+    try:
+        generated = json.loads(raw_text)
+    except json.JSONDecodeError as e:
+        raise ValueError(f"AI returned invalid JSON: {e}") from e
+
+    # Validate the generated steps
+    val_errors = validate_generated_procedural_steps(generated)
+    if val_errors:
+        raise ValueError(f"Generated flow failed validation: {'; '.join(val_errors)}")
+
+    # Ensure procedure_end exists; add if missing
+    steps = generated.get("steps", [])
+    has_end = any(s.get("type") == "procedure_end" for s in steps)
+    if not has_end:
+        steps.append({
+            "id": f"step-end-{uuid.uuid4().hex[:8]}",
+            "type": "procedure_end",
+            "title": "Procedure Complete",
+            "description": "All steps completed successfully.",
+        })
+        generated["steps"] = steps
+
+    return {
+        "name": generated.get("name", "AI-Generated Flow"),
+        "description": generated.get("description", ""),
+        "tree_type": "procedural",
+        "tags": generated.get("tags", []),
+        "tree_structure": {"steps": steps},
+    }