feat(ai-session): add FlowPilot AI-powered troubleshooting sessions

Implements Phase 1 of the FlowPilot-First pivot — the core AI session
experience where engineers describe a problem and FlowPilot guides them
through structured diagnosis with selectable options, free-text escape
hatches, and auto-generated documentation on resolution.

Backend: AISession + AISessionStep models, FlowPilot Engine (LLM
orchestration with structured JSON output), Flow Matching Engine v1
(semantic + keyword + recency scoring), 8 API endpoints with auth,
rate limiting, and AI quota enforcement.

Frontend: Intake screen, conversational session view with sidebar,
step cards with options/actions/resolution suggestions, resolve/escalate
modals, documentation view with rating, session history integration,
and /pilot route with sidebar navigation.

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

backend/alembic/env.py

@@ -20,6 +20,8 @@ from app.models.ai_suggestion import AISuggestion  # noqa: F401
 from app.models.kb_import import KBImport, KBImportNode  # noqa: F401
 from app.models.script_template import ScriptCategory, ScriptTemplate, ScriptGeneration  # noqa: F401
 from app.models.psa_connection import PsaConnection  # noqa: F401
+from app.models.ai_session import AISession  # noqa: F401
+from app.models.ai_session_step import AISessionStep  # noqa: F401
 from app.models.psa_post_log import PsaPostLog  # noqa: F401
 from app.models.psa_member_mapping import PsaMemberMapping  # noqa: F401
 from app.core.config import settings

backend/alembic/versions/f1a2b3c4d5e6_add_ai_sessions_and_ai_session_steps.py

@@ -0,0 +1,129 @@
+"""add ai_sessions and ai_session_steps tables
+
+Revision ID: f1a2b3c4d5e6
+Revises: ee98013dd18c
+Create Date: 2026-03-18
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import UUID, JSONB
+
+# revision identifiers, used by Alembic.
+revision = "f1a2b3c4d5e6"
+down_revision = "ee98013dd18c"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ── ai_sessions table ──
+    op.create_table(
+        "ai_sessions",
+        sa.Column("id", UUID(as_uuid=True), primary_key=True),
+        sa.Column("user_id", UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True),
+        sa.Column("account_id", UUID(as_uuid=True), sa.ForeignKey("accounts.id", ondelete="CASCADE"), nullable=False, index=True),
+        sa.Column("team_id", UUID(as_uuid=True), sa.ForeignKey("teams.id", ondelete="SET NULL"), nullable=True, index=True),
+        # Intake
+        sa.Column("intake_type", sa.String(20), nullable=False, server_default="free_text"),
+        sa.Column("intake_content", JSONB, nullable=False, server_default="{}"),
+        sa.Column("problem_summary", sa.Text, nullable=True),
+        sa.Column("problem_domain", sa.String(100), nullable=True),
+        # Session state
+        sa.Column("status", sa.String(20), nullable=False, server_default="active", index=True),
+        sa.Column("confidence_tier", sa.String(20), nullable=False, server_default="discovery"),
+        sa.Column("confidence_score", sa.Float, nullable=False, server_default="0.0"),
+        # Flow matching
+        sa.Column("matched_flow_id", UUID(as_uuid=True), sa.ForeignKey("trees.id", ondelete="SET NULL"), nullable=True),
+        sa.Column("match_score", sa.Float, nullable=True),
+        # PSA link
+        sa.Column("psa_ticket_id", sa.String(100), nullable=True),
+        sa.Column("psa_connection_id", UUID(as_uuid=True), sa.ForeignKey("psa_connections.id", ondelete="SET NULL"), nullable=True),
+        sa.Column("ticket_data", JSONB, nullable=True),
+        # Resolution / Escalation
+        sa.Column("resolution_summary", sa.Text, nullable=True),
+        sa.Column("resolution_action", sa.Text, nullable=True),
+        sa.Column("escalation_reason", sa.Text, nullable=True),
+        sa.Column("escalation_package", JSONB, nullable=True),
+        sa.Column("escalated_to_id", UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True),
+        # Feedback
+        sa.Column("session_rating", sa.Integer, nullable=True),
+        sa.Column("session_feedback", sa.Text, nullable=True),
+        # AI tracking
+        sa.Column("total_input_tokens", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("total_output_tokens", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("step_count", sa.Integer, nullable=False, server_default="0"),
+        # Timestamps
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
+        sa.Column("resolved_at", sa.DateTime(timezone=True), nullable=True),
+        # LLM context
+        sa.Column("system_prompt_snapshot", sa.Text, nullable=True),
+        sa.Column("conversation_messages", JSONB, nullable=False, server_default="[]"),
+        # Check constraints
+        sa.CheckConstraint(
+            "intake_type IN ('free_text', 'psa_ticket', 'screenshot', 'log_paste', 'combined')",
+            name="ck_ai_sessions_intake_type",
+        ),
+        sa.CheckConstraint(
+            "status IN ('active', 'paused', 'resolved', 'escalated', 'abandoned')",
+            name="ck_ai_sessions_status",
+        ),
+        sa.CheckConstraint(
+            "confidence_tier IN ('guided', 'exploring', 'discovery')",
+            name="ck_ai_sessions_confidence_tier",
+        ),
+    )
+
+    # ── ai_session_steps table ──
+    op.create_table(
+        "ai_session_steps",
+        sa.Column("id", UUID(as_uuid=True), primary_key=True),
+        sa.Column("session_id", UUID(as_uuid=True), sa.ForeignKey("ai_sessions.id", ondelete="CASCADE"), nullable=False, index=True),
+        sa.Column("step_order", sa.Integer, nullable=False),
+        sa.Column("step_type", sa.String(30), nullable=False),
+        # Content
+        sa.Column("content", JSONB, nullable=False, server_default="{}"),
+        sa.Column("context_message", sa.Text, nullable=True),
+        # Options
+        sa.Column("options_presented", JSONB, nullable=True),
+        # Engineer response
+        sa.Column("selected_option", sa.String(500), nullable=True),
+        sa.Column("free_text_input", sa.Text, nullable=True),
+        sa.Column("was_free_text", sa.Boolean, nullable=False, server_default="false"),
+        sa.Column("was_skipped", sa.Boolean, nullable=False, server_default="false"),
+        # Action results
+        sa.Column("action_result", JSONB, nullable=True),
+        # Script generation link
+        sa.Column("script_generation_id", UUID(as_uuid=True), sa.ForeignKey("script_generations.id", ondelete="SET NULL"), nullable=True),
+        # AI internals
+        sa.Column("confidence_at_step", sa.Float, nullable=False, server_default="0.0"),
+        sa.Column("ai_reasoning", sa.Text, nullable=True),
+        sa.Column("input_tokens", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("output_tokens", sa.Integer, nullable=False, server_default="0"),
+        # Timestamps
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
+        sa.Column("responded_at", sa.DateTime(timezone=True), nullable=True),
+        # Check constraint
+        sa.CheckConstraint(
+            "step_type IN ('question', 'action', 'script_generation', 'verification', "
+            "'info_request', 'note', 'intake_analysis')",
+            name="ck_ai_session_steps_step_type",
+        ),
+    )
+
+    # ── Add flow matching columns to trees table ──
+    op.add_column("trees", sa.Column("origin", sa.String(20), nullable=True, comment="manual | ai_generated | ai_enhanced"))
+    op.add_column("trees", sa.Column("source_session_id", UUID(as_uuid=True), nullable=True))
+    op.add_column("trees", sa.Column("match_keywords", JSONB, nullable=True, server_default="[]"))
+    op.add_column("trees", sa.Column("success_rate", sa.Float, nullable=True))
+    op.add_column("trees", sa.Column("last_matched_at", sa.DateTime(timezone=True), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("trees", "last_matched_at")
+    op.drop_column("trees", "success_rate")
+    op.drop_column("trees", "match_keywords")
+    op.drop_column("trees", "source_session_id")
+    op.drop_column("trees", "origin")
+    op.drop_table("ai_session_steps")
+    op.drop_table("ai_sessions")

backend/app/api/endpoints/ai_sessions.py

@@ -0,0 +1,411 @@
+"""FlowPilot AI session endpoints.
+
+CRUD and interaction endpoints for AI-powered troubleshooting sessions:
+  POST   /ai-sessions                    — Start a new session
+  POST   /ai-sessions/{id}/respond       — Submit step response, get next step
+  POST   /ai-sessions/{id}/resolve       — Resolve the session
+  POST   /ai-sessions/{id}/escalate      — Escalate the session
+  GET    /ai-sessions                    — List user's sessions (paginated)
+  GET    /ai-sessions/{id}              — Get session detail with all steps
+  GET    /ai-sessions/{id}/documentation — Get auto-generated documentation
+  POST   /ai-sessions/{id}/rate         — Submit post-session rating
+"""
+import logging
+from typing import Annotated, Optional
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
+from sqlalchemy import select, func
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from app.core.rate_limit import limiter
+from app.api.deps import get_current_active_user, get_db, require_engineer_or_admin
+from app.core.config import settings
+from app.core.ai_quota_service import check_ai_quota, record_ai_usage, get_user_plan
+from app.models.user import User
+from app.models.ai_session import AISession
+from app.schemas.ai_session import (
+    AISessionCreateRequest,
+    AISessionCreateResponse,
+    StepResponseRequest,
+    StepResponseResponse,
+    ResolveSessionRequest,
+    EscalateSessionRequest,
+    SessionCloseResponse,
+    SessionDocumentation,
+    RateSessionRequest,
+    AISessionSummary,
+    AISessionDetail,
+    AISessionStepResponse,
+    StepOptionSchema,
+)
+from app.services import flowpilot_engine
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/ai-sessions", tags=["ai-sessions"])
+
+
+def _require_ai_enabled() -> None:
+    if not settings.ai_enabled:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="AI is not configured. Set GOOGLE_AI_API_KEY or ANTHROPIC_API_KEY.",
+        )
+
+
+async def _check_quota(user: User, db: AsyncSession) -> None:
+    """Check AI quota and raise 429 if exceeded."""
+    allowed, quota_status = await check_ai_quota(
+        user_id=user.id,
+        account_id=user.account_id,
+        db=db,
+        billing_anchor=user.ai_billing_cycle_anchor_at,
+        is_super_admin=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 limit exceeded ({quota_status['deny_reason']})",
+                "reset_at": quota_status.get(reset_key),
+                "quota": quota_status,
+            },
+        )
+
+
+async def _record_usage(
+    user: User,
+    db: AsyncSession,
+    generation_type: str,
+    input_tokens: int,
+    output_tokens: int,
+    succeeded: bool,
+    session_id: Optional[UUID] = None,
+    error_code: Optional[str] = None,
+) -> None:
+    """Record AI usage after an LLM call."""
+    plan = await get_user_plan(user.account_id, db)
+    estimated_cost = (
+        input_tokens * 3.0 / 1_000_000
+        + output_tokens * 15.0 / 1_000_000
+    )
+    await record_ai_usage(
+        user_id=user.id,
+        account_id=user.account_id,
+        conversation_id=None,
+        generation_type=generation_type,
+        tier=plan,
+        input_tokens=input_tokens,
+        output_tokens=output_tokens,
+        estimated_cost=estimated_cost,
+        succeeded=succeeded,
+        counts_toward_quota=True,
+        error_code=error_code,
+        extra_data={"ai_session_id": str(session_id)} if session_id else None,
+        db=db,
+    )
+
+
+# ── Create session ──
+
+@router.post("", response_model=AISessionCreateResponse, status_code=201)
+@limiter.limit("5/minute")
+async def create_session(
+    request: Request,
+    data: AISessionCreateRequest,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+):
+    """Start a new FlowPilot troubleshooting session."""
+    _require_ai_enabled()
+    await _check_quota(current_user, db)
+
+    try:
+        result = await flowpilot_engine.start_session(
+            request=data,
+            user_id=current_user.id,
+            account_id=current_user.account_id,
+            team_id=current_user.team_id,
+            db=db,
+        )
+    except Exception as e:
+        logger.exception("FlowPilot session start failed: %s", e)
+        await _record_usage(
+            current_user, db,
+            generation_type="flowpilot_start",
+            input_tokens=0, output_tokens=0,
+            succeeded=False, error_code=type(e).__name__,
+        )
+        await db.commit()
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=f"AI provider error ({type(e).__name__}). Please try again.",
+        )
+
+    await _record_usage(
+        current_user, db,
+        generation_type="flowpilot_start",
+        input_tokens=result.first_step.confidence_score and 0,  # Tracked on session
+        output_tokens=0,
+        succeeded=True,
+        session_id=result.session_id,
+    )
+    await db.commit()
+
+    return result
+
+
+# ── Respond to step ──
+
+@router.post("/{session_id}/respond", response_model=StepResponseResponse)
+@limiter.limit("15/minute")
+async def respond_to_step(
+    request: Request,
+    session_id: UUID,
+    data: StepResponseRequest,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+):
+    """Submit an engineer's response to a FlowPilot step and get the next step."""
+    _require_ai_enabled()
+    await _check_quota(current_user, db)
+
+    try:
+        result = await flowpilot_engine.process_response(
+            session_id=session_id,
+            request=data,
+            user_id=current_user.id,
+            db=db,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
+    except PermissionError as e:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(e))
+    except Exception as e:
+        logger.exception("FlowPilot response failed: %s", e)
+        await _record_usage(
+            current_user, db,
+            generation_type="flowpilot_respond",
+            input_tokens=0, output_tokens=0,
+            succeeded=False,
+            session_id=session_id,
+            error_code=type(e).__name__,
+        )
+        await db.commit()
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=f"AI provider error ({type(e).__name__}). Please try again.",
+        )
+
+    await _record_usage(
+        current_user, db,
+        generation_type="flowpilot_respond",
+        input_tokens=0, output_tokens=0,
+        succeeded=True,
+        session_id=session_id,
+    )
+    await db.commit()
+
+    return result
+
+
+# ── Resolve ──
+
+@router.post("/{session_id}/resolve", response_model=SessionCloseResponse)
+@limiter.limit("15/minute")
+async def resolve_session(
+    request: Request,
+    session_id: UUID,
+    data: ResolveSessionRequest,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+):
+    """Resolve a FlowPilot session and generate documentation."""
+    try:
+        result = await flowpilot_engine.resolve_session(
+            session_id=session_id,
+            request=data,
+            user_id=current_user.id,
+            db=db,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
+    except PermissionError as e:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(e))
+
+    await db.commit()
+    return result
+
+
+# ── Escalate ──
+
+@router.post("/{session_id}/escalate", response_model=SessionCloseResponse)
+@limiter.limit("15/minute")
+async def escalate_session(
+    request: Request,
+    session_id: UUID,
+    data: EscalateSessionRequest,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+):
+    """Escalate a FlowPilot session to another engineer."""
+    try:
+        result = await flowpilot_engine.escalate_session(
+            session_id=session_id,
+            request=data,
+            user_id=current_user.id,
+            db=db,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
+    except PermissionError as e:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(e))
+
+    await db.commit()
+    return result
+
+
+# ── List sessions ──
+
+@router.get("", response_model=list[AISessionSummary])
+@limiter.limit("30/minute")
+async def list_sessions(
+    request: Request,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    session_status: Optional[str] = Query(None, alias="status"),
+    skip: int = Query(0, ge=0),
+    limit: int = Query(20, ge=1, le=100),
+):
+    """List the current user's AI sessions."""
+    query = (
+        select(AISession)
+        .where(AISession.user_id == current_user.id)
+        .order_by(AISession.created_at.desc())
+        .offset(skip)
+        .limit(limit)
+    )
+
+    if session_status:
+        query = query.where(AISession.status == session_status)
+
+    result = await db.execute(query)
+    sessions = result.scalars().all()
+
+    return [AISessionSummary.model_validate(s) for s in sessions]
+
+
+# ── Get session detail ──
+
+@router.get("/{session_id}", response_model=AISessionDetail)
+@limiter.limit("30/minute")
+async def get_session(
+    request: Request,
+    session_id: UUID,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+):
+    """Get full session detail with all steps."""
+    result = await db.execute(
+        select(AISession)
+        .options(selectinload(AISession.steps))
+        .where(AISession.id == session_id)
+    )
+    session = result.scalar_one_or_none()
+
+    if not session:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Session not found")
+
+    # Allow access if user is owner or escalation target
+    if session.user_id != current_user.id and session.escalated_to_id != current_user.id:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized")
+
+    # Build step responses
+    step_responses = []
+    for step in session.steps:
+        options = []
+        if step.options_presented:
+            options = [
+                StepOptionSchema(
+                    label=opt.get("label", ""),
+                    value=opt.get("value", ""),
+                    followup_hint=opt.get("followup_hint"),
+                )
+                for opt in step.options_presented
+            ]
+        content = step.content or {}
+        step_responses.append(AISessionStepResponse(
+            step_id=step.id,
+            step_order=step.step_order,
+            step_type=step.step_type,
+            content=content,
+            context_message=step.context_message,
+            options=options,
+            allow_free_text=content.get("allow_free_text", True),
+            allow_skip=content.get("allow_skip", True),
+            confidence_tier=session.confidence_tier,
+            confidence_score=step.confidence_at_step,
+        ))
+
+    detail = AISessionDetail.model_validate(session)
+    detail.steps = step_responses
+    return detail
+
+
+# ── Documentation ──
+
+@router.get("/{session_id}/documentation", response_model=SessionDocumentation)
+@limiter.limit("30/minute")
+async def get_documentation(
+    request: Request,
+    session_id: UUID,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+):
+    """Get auto-generated documentation for a session."""
+    try:
+        return await flowpilot_engine.get_session_documentation(
+            session_id=session_id,
+            user_id=current_user.id,
+            db=db,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
+    except PermissionError as e:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(e))
+
+
+# ── Rate ──
+
+@router.post("/{session_id}/rate", status_code=204)
+@limiter.limit("15/minute")
+async def rate_session(
+    request: Request,
+    session_id: UUID,
+    data: RateSessionRequest,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+):
+    """Submit a post-session rating."""
+    try:
+        await flowpilot_engine.rate_session(
+            session_id=session_id,
+            rating=data.rating,
+            feedback=data.feedback,
+            user_id=current_user.id,
+            db=db,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
+    except PermissionError as e:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(e))
+
+    await db.commit()

backend/app/api/router.py

@@ -21,6 +21,7 @@ from app.api.endpoints import integrations
 from app.api.endpoints import onboarding
 from app.api.endpoints import branding
 from app.api.endpoints import supporting_data
+from app.api.endpoints import ai_sessions
 
 api_router = APIRouter()
 
@@ -67,3 +68,4 @@ api_router.include_router(integrations.router)
 api_router.include_router(onboarding.router)
 api_router.include_router(branding.router)
 api_router.include_router(supporting_data.router)
+api_router.include_router(ai_sessions.router)

backend/app/models/__init__.py

@@ -36,6 +36,8 @@ from .survey_response import SurveyResponse
 from .survey_invite import SurveyInvite
 from .kb_import import KBImport, KBImportNode
 from .script_template import ScriptCategory, ScriptTemplate, ScriptGeneration
+from .ai_session import AISession
+from .ai_session_step import AISessionStep
 from .psa_connection import PsaConnection
 from .psa_post_log import PsaPostLog
 from .psa_member_mapping import PsaMemberMapping
@@ -90,6 +92,8 @@ __all__ = [
     "ScriptCategory",
     "ScriptTemplate",
     "ScriptGeneration",
+    "AISession",
+    "AISessionStep",
     "PsaConnection",
     "PsaPostLog",
     "PsaMemberMapping",

backend/app/models/ai_session.py

@@ -0,0 +1,204 @@
+"""AI-powered troubleshooting session model.
+
+Represents a complete FlowPilot interaction from intake to resolution/escalation.
+This is the central entity of the FlowPilot-First pivot.
+"""
+import uuid
+from datetime import datetime, timezone
+from typing import Optional, Any, TYPE_CHECKING
+
+from sqlalchemy import String, Text, DateTime, ForeignKey, Boolean, Integer, Float, CheckConstraint
+import sqlalchemy as sa
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy.dialects.postgresql import UUID, JSONB
+
+from app.core.database import Base
+
+if TYPE_CHECKING:
+    from app.models.user import User
+    from app.models.team import Team
+    from app.models.account import Account
+    from app.models.tree import Tree
+    from app.models.psa_connection import PsaConnection
+
+
+class AISession(Base):
+    """A FlowPilot-guided troubleshooting session.
+
+    Lifecycle: active → resolved | escalated | abandoned
+    Sessions may be paused and resumed (e.g., escalation handoff).
+    """
+    __tablename__ = "ai_sessions"
+    __table_args__ = (
+        CheckConstraint(
+            "intake_type IN ('free_text', 'psa_ticket', 'screenshot', 'log_paste', 'combined')",
+            name="ck_ai_sessions_intake_type",
+        ),
+        CheckConstraint(
+            "status IN ('active', 'paused', 'resolved', 'escalated', 'abandoned')",
+            name="ck_ai_sessions_status",
+        ),
+        CheckConstraint(
+            "confidence_tier IN ('guided', 'exploring', 'discovery')",
+            name="ck_ai_sessions_confidence_tier",
+        ),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    user_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("users.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    account_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("accounts.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    team_id: Mapped[Optional[uuid.UUID]] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("teams.id", ondelete="SET NULL"),
+        nullable=True,
+        index=True,
+    )
+
+    # ── Intake ──
+    intake_type: Mapped[str] = mapped_column(
+        String(20), nullable=False, default="free_text"
+    )
+    intake_content: Mapped[dict[str, Any]] = mapped_column(
+        JSONB, nullable=False, default=dict,
+        comment="Original intake data: {text, image_urls, log_content, ticket_data}",
+    )
+    problem_summary: Mapped[Optional[str]] = mapped_column(
+        Text, nullable=True,
+        comment="AI-generated one-line problem summary from intake",
+    )
+    problem_domain: Mapped[Optional[str]] = mapped_column(
+        String(100), nullable=True,
+        comment="Classified domain: active_directory, networking, m365, hardware, etc.",
+    )
+
+    # ── Session state ──
+    status: Mapped[str] = mapped_column(
+        String(20), nullable=False, default="active", index=True,
+    )
+    confidence_tier: Mapped[str] = mapped_column(
+        String(20), nullable=False, default="discovery",
+        comment="Current AI confidence: guided (>80%), exploring (40-80%), discovery (<40%)",
+    )
+    confidence_score: Mapped[float] = mapped_column(
+        Float, nullable=False, default=0.0,
+        comment="Numeric confidence 0.0-1.0 for internal tracking",
+    )
+
+    # ── Flow matching ──
+    matched_flow_id: Mapped[Optional[uuid.UUID]] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("trees.id", ondelete="SET NULL"),
+        nullable=True,
+        comment="If following an existing flow, which one",
+    )
+    match_score: Mapped[Optional[float]] = mapped_column(
+        Float, nullable=True,
+        comment="Similarity score of the matched flow (0.0-1.0)",
+    )
+
+    # ── PSA link ──
+    psa_ticket_id: Mapped[Optional[str]] = mapped_column(
+        String(100), nullable=True,
+        comment="External PSA ticket ID if session was started from a ticket",
+    )
+    psa_connection_id: Mapped[Optional[uuid.UUID]] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("psa_connections.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+    ticket_data: Mapped[Optional[dict[str, Any]]] = mapped_column(
+        JSONB, nullable=True,
+        comment="Snapshot of PSA ticket data at session start",
+    )
+
+    # ── Resolution / Escalation ──
+    resolution_summary: Mapped[Optional[str]] = mapped_column(
+        Text, nullable=True,
+        comment="What fixed the issue (set on resolution)",
+    )
+    resolution_action: Mapped[Optional[str]] = mapped_column(
+        Text, nullable=True,
+        comment="The specific action/step that resolved the issue",
+    )
+    escalation_reason: Mapped[Optional[str]] = mapped_column(
+        Text, nullable=True,
+        comment="Why escalated (set on escalation)",
+    )
+    escalation_package: Mapped[Optional[dict[str, Any]]] = mapped_column(
+        JSONB, nullable=True,
+        comment="Context package for receiving engineer: steps_tried, hypotheses, suggestions",
+    )
+    escalated_to_id: Mapped[Optional[uuid.UUID]] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("users.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+
+    # ── Feedback ──
+    session_rating: Mapped[Optional[int]] = mapped_column(
+        Integer, nullable=True,
+        comment="1-5 engineer feedback rating",
+    )
+    session_feedback: Mapped[Optional[str]] = mapped_column(
+        Text, nullable=True,
+        comment="Optional feedback text from engineer",
+    )
+
+    # ── AI tracking ──
+    total_input_tokens: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0,
+    )
+    total_output_tokens: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0,
+    )
+    step_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0,
+    )
+
+    # ── Timestamps ──
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.now(timezone.utc),
+        onupdate=lambda: datetime.now(timezone.utc),
+    )
+    resolved_at: Mapped[Optional[datetime]] = mapped_column(
+        DateTime(timezone=True), nullable=True,
+    )
+
+    # ── LLM conversation context ──
+    system_prompt_snapshot: Mapped[Optional[str]] = mapped_column(
+        Text, nullable=True,
+        comment="Snapshot of the system prompt used (for debugging/training)",
+    )
+    conversation_messages: Mapped[list[dict[str, Any]]] = mapped_column(
+        JSONB, nullable=False, default=list,
+        comment="Full LLM message history for context continuity",
+    )
+
+    # ── Relationships ──
+    user: Mapped["User"] = relationship("User", foreign_keys=[user_id])
+    account: Mapped["Account"] = relationship("Account")
+    team: Mapped[Optional["Team"]] = relationship("Team")
+    matched_flow: Mapped[Optional["Tree"]] = relationship("Tree", foreign_keys=[matched_flow_id])
+    escalated_to: Mapped[Optional["User"]] = relationship("User", foreign_keys=[escalated_to_id])
+    psa_connection: Mapped[Optional["PsaConnection"]] = relationship("PsaConnection")
+    steps: Mapped[list["AISessionStep"]] = relationship(
+        "AISessionStep", back_populates="session",
+        cascade="all, delete-orphan",
+        order_by="AISessionStep.step_order",
+    )

backend/app/models/ai_session_step.py

@@ -0,0 +1,133 @@
+"""AI session step model.
+
+Every interaction within an AI session is captured as a step.
+Steps are the raw material that becomes flow nodes in the Knowledge Flywheel.
+"""
+import uuid
+from datetime import datetime, timezone
+from typing import Optional, Any, TYPE_CHECKING
+
+from sqlalchemy import String, Text, DateTime, ForeignKey, Integer, Float, CheckConstraint
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy.dialects.postgresql import UUID, JSONB
+
+from app.core.database import Base
+
+if TYPE_CHECKING:
+    from app.models.ai_session import AISession
+    from app.models.script_template import ScriptGeneration
+
+
+class AISessionStep(Base):
+    """A single interaction step within a FlowPilot session.
+
+    Step types:
+    - question: FlowPilot asks a diagnostic question with options
+    - action: FlowPilot suggests an action for the engineer to perform
+    - script_generation: FlowPilot invokes the Script Generator
+    - verification: FlowPilot asks engineer to verify a condition
+    - info_request: FlowPilot asks engineer to gather specific data
+    - note: Engineer or FlowPilot adds a contextual note
+    - intake_analysis: Initial analysis of the intake content
+    """
+    __tablename__ = "ai_session_steps"
+    __table_args__ = (
+        CheckConstraint(
+            "step_type IN ('question', 'action', 'script_generation', 'verification', "
+            "'info_request', 'note', 'intake_analysis')",
+            name="ck_ai_session_steps_step_type",
+        ),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    session_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("ai_sessions.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    step_order: Mapped[int] = mapped_column(
+        Integer, nullable=False,
+        comment="Sequential position in the session (0-indexed)",
+    )
+    step_type: Mapped[str] = mapped_column(
+        String(30), nullable=False,
+    )
+
+    # ── Content presented to engineer ──
+    content: Mapped[dict[str, Any]] = mapped_column(
+        JSONB, nullable=False, default=dict,
+        comment="The question/action content rendered in the session UI",
+    )
+    context_message: Mapped[Optional[str]] = mapped_column(
+        Text, nullable=True,
+        comment="Why FlowPilot is asking this (shown above the question)",
+    )
+
+    # ── Options (for question steps) ──
+    options_presented: Mapped[Optional[list[dict[str, Any]]]] = mapped_column(
+        JSONB, nullable=True,
+        comment="Array of {label, value, followup_hint} options shown to engineer",
+    )
+
+    # ── Engineer response ──
+    selected_option: Mapped[Optional[str]] = mapped_column(
+        String(500), nullable=True,
+        comment="Which option the engineer selected (value field)",
+    )
+    free_text_input: Mapped[Optional[str]] = mapped_column(
+        Text, nullable=True,
+        comment="If engineer typed a custom response instead of selecting an option",
+    )
+    was_free_text: Mapped[bool] = mapped_column(
+        default=False,
+        comment="True if the engineer used the free-text escape hatch",
+    )
+    was_skipped: Mapped[bool] = mapped_column(
+        default=False,
+        comment="True if engineer selected 'I don't know / Can't check'",
+    )
+
+    # ── Action results ──
+    action_result: Mapped[Optional[dict[str, Any]]] = mapped_column(
+        JSONB, nullable=True,
+        comment="Outcome of action step: {success: bool, details: str, next_hint: str}",
+    )
+
+    # ── Script generation link ──
+    script_generation_id: Mapped[Optional[uuid.UUID]] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("script_generations.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+
+    # ── AI internals ──
+    confidence_at_step: Mapped[float] = mapped_column(
+        Float, nullable=False, default=0.0,
+        comment="FlowPilot confidence level at this point (0.0-1.0)",
+    )
+    ai_reasoning: Mapped[Optional[str]] = mapped_column(
+        Text, nullable=True,
+        comment="Why FlowPilot chose this step (internal, for debugging/training)",
+    )
+    input_tokens: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0,
+    )
+    output_tokens: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0,
+    )
+
+    # ── Timestamps ──
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
+    )
+    responded_at: Mapped[Optional[datetime]] = mapped_column(
+        DateTime(timezone=True), nullable=True,
+        comment="When the engineer responded to this step",
+    )
+
+    # ── Relationships ──
+    session: Mapped["AISession"] = relationship("AISession", back_populates="steps")
+    script_generation: Mapped[Optional["ScriptGeneration"]] = relationship("ScriptGeneration")

backend/app/models/tree.py

@@ -1,7 +1,7 @@
 import uuid
 from datetime import datetime, timezone
 from typing import Optional, Any, TYPE_CHECKING
-from sqlalchemy import String, Text, DateTime, ForeignKey, Boolean, Integer, Index, CheckConstraint
+from sqlalchemy import String, Text, DateTime, ForeignKey, Boolean, Integer, Float, Index, CheckConstraint
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 from sqlalchemy.dialects.postgresql import UUID, JSONB
 from app.core.database import Base
@@ -161,6 +161,25 @@ class Tree(Base):
         comment="Provenance metadata from .rfflow file import"
     )
 
+    # Flow matching (FlowPilot AI sessions)
+    origin: Mapped[Optional[str]] = mapped_column(
+        String(20), nullable=True,
+        comment="manual | ai_generated | ai_enhanced"
+    )
+    source_session_id: Mapped[Optional[uuid.UUID]] = mapped_column(
+        UUID(as_uuid=True), nullable=True,
+    )
+    match_keywords: Mapped[Optional[list[Any]]] = mapped_column(
+        JSONB, nullable=True,
+        comment="Keywords for FlowPilot flow matching"
+    )
+    success_rate: Mapped[Optional[float]] = mapped_column(
+        Float, nullable=True,
+    )
+    last_matched_at: Mapped[Optional[datetime]] = mapped_column(
+        DateTime(timezone=True), nullable=True,
+    )
+
     # Relationships
     author: Mapped[Optional["User"]] = relationship("User", foreign_keys=[author_id], back_populates="trees")
     team: Mapped[Optional["Team"]] = relationship("Team", back_populates="trees")

backend/app/schemas/ai_session.py

@@ -0,0 +1,171 @@
+"""Pydantic schemas for FlowPilot AI sessions."""
+from __future__ import annotations
+
+from typing import Optional, Any
+from uuid import UUID
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+
+# ── Intake ──
+
+class AISessionCreateRequest(BaseModel):
+    """Start a new FlowPilot session."""
+    intake_type: str = Field(
+        "free_text",
+        pattern="^(free_text|psa_ticket|screenshot|log_paste|combined)$",
+    )
+    intake_content: dict[str, Any] = Field(
+        ...,
+        description=(
+            "Intake payload. Shape depends on intake_type: "
+            "{text: str} for free_text, "
+            "{text?: str, image_urls?: list[str]} for screenshot, "
+            "{text?: str, log_content?: str} for log_paste, "
+            "{ticket_id: str, psa_connection_id: str} for psa_ticket, "
+            "any combination for combined."
+        ),
+    )
+    psa_ticket_id: Optional[str] = None
+    psa_connection_id: Optional[UUID] = None
+
+
+class AISessionCreateResponse(BaseModel):
+    """Response after starting a session — includes the first FlowPilot step."""
+    session_id: UUID
+    status: str
+    confidence_tier: str
+    problem_summary: str | None = None
+    problem_domain: str | None = None
+    matched_flow_id: UUID | None = None
+    matched_flow_name: str | None = None
+    match_score: float | None = None
+    first_step: AISessionStepResponse
+
+
+# ── Step interaction ──
+
+class StepOptionSchema(BaseModel):
+    """A selectable option presented to the engineer."""
+    label: str
+    value: str
+    followup_hint: str | None = None
+
+
+class AISessionStepResponse(BaseModel):
+    """A FlowPilot step rendered in the session UI."""
+    step_id: UUID
+    step_order: int
+    step_type: str
+    content: dict[str, Any]
+    context_message: str | None = None
+    options: list[StepOptionSchema] = []
+    allow_free_text: bool = True
+    allow_skip: bool = True
+    confidence_tier: str
+    confidence_score: float
+
+    model_config = {"from_attributes": True}
+
+
+class StepResponseRequest(BaseModel):
+    """Engineer's response to a FlowPilot step."""
+    selected_option: str | None = None
+    free_text_input: str | None = None
+    was_skipped: bool = False
+    action_result: dict[str, Any] | None = None
+
+
+class StepResponseResponse(BaseModel):
+    """FlowPilot's next step after processing the engineer's response."""
+    session_id: UUID
+    status: str
+    confidence_tier: str
+    confidence_score: float
+    next_step: AISessionStepResponse | None = None
+    resolution_suggested: bool = False
+    resolution_summary: str | None = None
+
+
+# ── Resolution / Escalation ──
+
+class ResolveSessionRequest(BaseModel):
+    """Close a session as resolved."""
+    resolution_summary: str = Field(..., min_length=5, max_length=2000)
+    resolution_action: str | None = None
+    session_rating: int | None = Field(None, ge=1, le=5)
+    session_feedback: str | None = None
+
+
+class EscalateSessionRequest(BaseModel):
+    """Escalate a session to another engineer."""
+    escalation_reason: str = Field(..., min_length=5, max_length=2000)
+    escalated_to_id: UUID | None = None
+
+
+class DocumentationStep(BaseModel):
+    """A step in the documentation trail."""
+    step_number: int
+    step_type: str
+    description: str
+    engineer_response: str | None = None
+    outcome: str | None = None
+
+
+class SessionDocumentation(BaseModel):
+    """Auto-generated session documentation."""
+    problem_summary: str
+    problem_domain: str | None = None
+    intake_summary: str
+    diagnostic_steps: list[DocumentationStep]
+    resolution_summary: str | None = None
+    escalation_reason: str | None = None
+    total_steps: int
+    duration_display: str | None = None
+    generated_at: datetime
+
+
+class SessionCloseResponse(BaseModel):
+    """Response after resolving or escalating."""
+    session_id: UUID
+    status: str
+    documentation: SessionDocumentation
+
+
+class RateSessionRequest(BaseModel):
+    """Submit post-session rating."""
+    rating: int = Field(..., ge=1, le=5)
+    feedback: str | None = None
+
+
+# ── List / Detail ──
+
+class AISessionSummary(BaseModel):
+    """Compact session for list views."""
+    id: UUID
+    status: str
+    intake_type: str
+    problem_summary: str | None = None
+    problem_domain: str | None = None
+    confidence_tier: str
+    step_count: int
+    session_rating: int | None = None
+    created_at: datetime
+    resolved_at: datetime | None = None
+
+    model_config = {"from_attributes": True}
+
+
+class AISessionDetail(AISessionSummary):
+    """Full session detail with steps."""
+    intake_content: dict[str, Any]
+    matched_flow_id: UUID | None = None
+    match_score: float | None = None
+    resolution_summary: str | None = None
+    resolution_action: str | None = None
+    escalation_reason: str | None = None
+    session_feedback: str | None = None
+    steps: list[AISessionStepResponse] = []
+
+    model_config = {"from_attributes": True}

backend/app/services/flow_matching_engine.py

@@ -0,0 +1,278 @@
+"""Flow Matching Engine v1 — find existing flows relevant to an AI session's intake.
+
+Combines keyword matching, semantic search (via RAG embeddings), and recency
+scoring to rank flows. Deliberately simple for v1; v2 (Phase 3) adds deeper
+semantic matching.
+
+Scoring weights: semantic 0.5, keyword 0.3, recency 0.2.
+Threshold: only return matches with composite score > 0.5.
+"""
+import logging
+from datetime import datetime, timezone, timedelta
+from typing import Any, Optional
+from uuid import UUID
+
+from sqlalchemy import select, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.tree import Tree
+from app.services.rag_service import search as rag_search
+
+logger = logging.getLogger(__name__)
+
+# Scoring weights
+SEMANTIC_WEIGHT = 0.5
+KEYWORD_WEIGHT = 0.3
+RECENCY_WEIGHT = 0.2
+
+# Only return matches above this composite score
+SCORE_THRESHOLD = 0.5
+
+
+async def find_matches(
+    intake_text: str,
+    problem_domain: Optional[str],
+    account_id: UUID,
+    db: AsyncSession,
+    limit: int = 5,
+) -> list[dict[str, Any]]:
+    """Find existing flows that match the intake description.
+
+    Returns list of dicts sorted by composite score:
+    {tree_id, tree_name, score, match_reason}
+    """
+    candidates: dict[str, dict[str, Any]] = {}
+
+    # 1. Semantic search via existing RAG embeddings
+    try:
+        rag_results = await rag_search(
+            query=intake_text,
+            account_id=account_id,
+            db=db,
+            limit=10,
+        )
+        for r in rag_results:
+            tree_id = str(r["tree_id"])
+            similarity = r.get("similarity", 0.0)
+            if tree_id not in candidates:
+                candidates[tree_id] = {
+                    "tree_id": tree_id,
+                    "tree_name": r["tree_name"],
+                    "semantic_score": similarity,
+                    "keyword_score": 0.0,
+                    "recency_score": 0.0,
+                    "match_reasons": [],
+                }
+            else:
+                # Take the best semantic score across chunks
+                candidates[tree_id]["semantic_score"] = max(
+                    candidates[tree_id]["semantic_score"], similarity
+                )
+            if similarity > 0.5:
+                candidates[tree_id]["match_reasons"].append(
+                    f"semantic match ({similarity:.0%})"
+                )
+    except Exception as e:
+        logger.warning("Semantic search failed during flow matching: %s", e)
+
+    # 2. Keyword matching against trees.match_keywords
+    try:
+        keyword_matches = await _keyword_match(intake_text, account_id, db)
+        for km in keyword_matches:
+            tree_id = str(km["tree_id"])
+            if tree_id not in candidates:
+                candidates[tree_id] = {
+                    "tree_id": tree_id,
+                    "tree_name": km["tree_name"],
+                    "semantic_score": 0.0,
+                    "keyword_score": km["score"],
+                    "recency_score": 0.0,
+                    "match_reasons": [],
+                }
+            else:
+                candidates[tree_id]["keyword_score"] = km["score"]
+            if km["score"] > 0.3:
+                candidates[tree_id]["match_reasons"].append(
+                    f"keyword match: {', '.join(km.get('matched_keywords', []))}"
+                )
+    except Exception as e:
+        logger.warning("Keyword matching failed: %s", e)
+
+    # 3. Category/domain match
+    if problem_domain:
+        try:
+            domain_matches = await _domain_match(problem_domain, account_id, db)
+            for dm in domain_matches:
+                tree_id = str(dm["tree_id"])
+                if tree_id not in candidates:
+                    candidates[tree_id] = {
+                        "tree_id": tree_id,
+                        "tree_name": dm["tree_name"],
+                        "semantic_score": 0.0,
+                        "keyword_score": 0.2,  # Small boost for domain match
+                        "recency_score": 0.0,
+                        "match_reasons": [],
+                    }
+                else:
+                    candidates[tree_id]["keyword_score"] = max(
+                        candidates[tree_id]["keyword_score"], 0.2
+                    )
+                candidates[tree_id]["match_reasons"].append(f"domain match: {problem_domain}")
+        except Exception as e:
+            logger.warning("Domain matching failed: %s", e)
+
+    # 4. Apply recency boost
+    now = datetime.now(timezone.utc)
+    for tree_id, candidate in candidates.items():
+        # We'll compute recency from the tree data if available
+        candidate["recency_score"] = 0.0  # Default, updated below
+
+    # Fetch recency data for all candidates
+    if candidates:
+        try:
+            recency_data = await _get_recency_scores(
+                list(candidates.keys()), db
+            )
+            for tree_id, recency_score in recency_data.items():
+                if tree_id in candidates:
+                    candidates[tree_id]["recency_score"] = recency_score
+        except Exception as e:
+            logger.warning("Recency scoring failed: %s", e)
+
+    # 5. Compute composite scores and filter
+    results = []
+    for tree_id, c in candidates.items():
+        composite = (
+            c["semantic_score"] * SEMANTIC_WEIGHT
+            + c["keyword_score"] * KEYWORD_WEIGHT
+            + c["recency_score"] * RECENCY_WEIGHT
+        )
+        if composite > SCORE_THRESHOLD:
+            results.append({
+                "tree_id": UUID(tree_id),
+                "tree_name": c["tree_name"],
+                "score": round(composite, 3),
+                "match_reason": "; ".join(c["match_reasons"][:3]) if c["match_reasons"] else "composite match",
+            })
+
+    # Sort by score descending, take top N
+    results.sort(key=lambda x: x["score"], reverse=True)
+    return results[:limit]
+
+
+async def _keyword_match(
+    intake_text: str,
+    account_id: UUID,
+    db: AsyncSession,
+) -> list[dict[str, Any]]:
+    """Match intake text against trees.match_keywords JSONB arrays.
+
+    Simple approach: tokenize intake text, check overlap with each tree's keywords.
+    """
+    # Extract meaningful tokens from intake (lowercase, 3+ chars)
+    tokens = set()
+    for word in intake_text.lower().split():
+        cleaned = "".join(c for c in word if c.isalnum())
+        if len(cleaned) >= 3:
+            tokens.add(cleaned)
+
+    if not tokens:
+        return []
+
+    # Find trees with match_keywords set
+    result = await db.execute(
+        select(Tree.id, Tree.name, Tree.match_keywords)
+        .where(
+            Tree.account_id == account_id,
+            Tree.deleted_at.is_(None),
+            Tree.status == "published",
+            Tree.match_keywords.isnot(None),
+        )
+    )
+    rows = result.all()
+
+    matches = []
+    for row in rows:
+        tree_keywords = row.match_keywords or []
+        if not isinstance(tree_keywords, list):
+            continue
+
+        # Lowercase keywords for comparison
+        kw_lower = {str(kw).lower() for kw in tree_keywords}
+
+        # Calculate overlap
+        matched = tokens & kw_lower
+        if matched:
+            score = len(matched) / max(len(kw_lower), 1)
+            matches.append({
+                "tree_id": row.id,
+                "tree_name": row.name,
+                "score": min(score, 1.0),
+                "matched_keywords": list(matched)[:5],
+            })
+
+    return matches
+
+
+async def _domain_match(
+    problem_domain: str,
+    account_id: UUID,
+    db: AsyncSession,
+) -> list[dict[str, Any]]:
+    """Find trees whose category matches the classified problem domain."""
+    result = await db.execute(
+        select(Tree.id, Tree.name)
+        .where(
+            Tree.account_id == account_id,
+            Tree.deleted_at.is_(None),
+            Tree.status == "published",
+            Tree.category.ilike(f"%{problem_domain}%"),
+        )
+        .limit(10)
+    )
+    rows = result.all()
+    return [{"tree_id": row.id, "tree_name": row.name} for row in rows]
+
+
+async def _get_recency_scores(
+    tree_ids: list[str],
+    db: AsyncSession,
+) -> dict[str, float]:
+    """Calculate recency scores based on last_matched_at.
+
+    Trees matched within the last 7 days get full recency boost (0.2 → 1.0).
+    Trees matched within 30 days get partial boost.
+    Older or never-matched trees get 0.
+    """
+    if not tree_ids:
+        return {}
+
+    result = await db.execute(
+        select(Tree.id, Tree.last_matched_at, Tree.success_rate)
+        .where(Tree.id.in_([UUID(tid) for tid in tree_ids]))
+    )
+    rows = result.all()
+
+    now = datetime.now(timezone.utc)
+    scores = {}
+    for row in rows:
+        tree_id = str(row.id)
+        if row.last_matched_at is None:
+            scores[tree_id] = 0.0
+            continue
+
+        days_since = (now - row.last_matched_at).days
+        if days_since <= 7:
+            recency = 1.0
+        elif days_since <= 30:
+            recency = 1.0 - ((days_since - 7) / 23)  # Linear decay 7-30 days
+        else:
+            recency = 0.0
+
+        # Factor in success rate if available
+        if row.success_rate is not None:
+            recency *= row.success_rate
+
+        scores[tree_id] = max(0.0, min(1.0, recency))
+
+    return scores

backend/app/services/flowpilot_engine.py

@@ -0,0 +1,737 @@
+"""FlowPilot Engine — core LLM orchestration for AI troubleshooting sessions.
+
+Manages structured diagnostic conversations: intake analysis, step generation,
+confidence tracking, and auto-documentation. All LLM responses are structured
+JSON validated against known output shapes.
+"""
+import json
+import logging
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Optional
+from uuid import UUID
+
+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.models.ai_session import AISession
+from app.models.ai_session_step import AISessionStep
+from app.schemas.ai_session import (
+    AISessionCreateRequest,
+    AISessionCreateResponse,
+    AISessionStepResponse,
+    StepOptionSchema,
+    StepResponseRequest,
+    StepResponseResponse,
+    ResolveSessionRequest,
+    EscalateSessionRequest,
+    SessionCloseResponse,
+    SessionDocumentation,
+    DocumentationStep,
+)
+
+logger = logging.getLogger(__name__)
+
+# Maximum steps per session as a safety limit
+MAX_STEPS_PER_SESSION = 30
+
+STRUCTURED_OUTPUT_SCHEMA = """\
+Your response MUST be a valid JSON object with one of these shapes:
+
+1. Diagnostic question:
+{"type": "question", "content": "Brief description", "reasoning": "Internal why", "context_message": "Shown to engineer", "options": [{"label": "Human text", "value": "machine_value", "followup_hint": "or null"}], "allow_free_text": true, "allow_skip": true, "confidence": 0.65}
+
+2. Suggested action:
+{"type": "action", "content": "What to do", "reasoning": "Internal why", "context_message": "Here's what to try", "action_type": "instruction | script_generation | verification | info_request", "expected_outcome": "What success looks like", "confidence": 0.78}
+
+3. Resolution suggestion:
+{"type": "resolution_suggestion", "content": "Summary of what we did", "reasoning": "Internal why", "resolution_summary": "Issue was caused by X, fixed by Y", "confidence": 0.92, "follow_up_recommendations": ["Monitor for 24 hours"]}\
+"""
+
+FLOWPILOT_SYSTEM_PROMPT = """\
+You are FlowPilot, an expert MSP troubleshooting assistant embedded in ResolutionFlow. You guide engineers through structured diagnosis of IT issues.
+
+## YOUR ROLE
+- Conduct systematic troubleshooting through targeted questions and actions
+- Start broad, narrow down based on responses
+- Never guess — ask clarifying questions when uncertain
+- Suggest specific, actionable steps the engineer can verify
+- When confidence is high, suggest resolution; when low, keep investigating
+
+## RESPONSE FORMAT
+You MUST respond with ONLY a valid JSON object. No markdown, no prose, no code fences.
+Every response must have a "type" field: "question", "action", or "resolution_suggestion".
+
+{structured_output_schema}
+
+## RULES
+- Maximum 5 options per question. Options should be the most likely scenarios.
+- Always include relevant context in context_message — explain WHY you're asking
+- confidence is a float 0.0-1.0 reflecting how certain you are about the diagnosis path
+- When multiple symptoms point to one root cause with >90% confidence, suggest resolution
+- If you detect the engineer needs a PowerShell script, suggest a script_generation action
+- Never suggest restarting or rebooting as a first step — diagnose first
+- Be specific: "Check Event Viewer > System > source NTFS" not "check the logs"
+
+{team_context}
+
+{matched_flow_context}\
+"""
+
+INTAKE_CLASSIFICATION_PROMPT = """\
+You are a triage classifier for IT support issues in an MSP environment.
+
+Analyze the following intake and respond with ONLY a JSON object:
+{
+  "problem_summary": "One-line summary of the issue (max 120 chars)",
+  "problem_domain": "One of: active_directory, networking, m365, hardware, endpoint, virtualization, security, backup, email, printing, cloud, other",
+  "key_symptoms": ["symptom1", "symptom2"],
+  "urgency": "low | medium | high | critical"
+}\
+"""
+
+
+def _confidence_to_tier(confidence: float) -> str:
+    """Map numeric confidence to tier label."""
+    if confidence >= 0.8:
+        return "guided"
+    elif confidence >= 0.4:
+        return "exploring"
+    return "discovery"
+
+
+def _parse_structured_output(raw_text: str) -> dict[str, Any]:
+    """Parse and validate structured JSON from LLM response.
+
+    Handles common LLM quirks: markdown fences, trailing commas, etc.
+    """
+    text = raw_text.strip()
+
+    # Strip markdown code fences if present
+    if text.startswith("```"):
+        lines = text.split("\n")
+        # Remove first line (```json or ```) and last line (```)
+        lines = [l for l in lines if not l.strip().startswith("```")]
+        text = "\n".join(lines).strip()
+
+    try:
+        data = json.loads(text)
+    except json.JSONDecodeError as e:
+        logger.warning("Failed to parse LLM JSON output: %s — raw: %.200s", e, text)
+        raise ValueError(f"Invalid JSON from LLM: {e}") from e
+
+    if not isinstance(data, dict) or "type" not in data:
+        raise ValueError("LLM response missing required 'type' field")
+
+    valid_types = {"question", "action", "resolution_suggestion"}
+    if data["type"] not in valid_types:
+        raise ValueError(f"Unknown response type: {data['type']}")
+
+    return data
+
+
+def _build_step_response(step: AISessionStep, session: AISession) -> AISessionStepResponse:
+    """Convert a model step + session state into an API response."""
+    options = []
+    if step.options_presented:
+        options = [
+            StepOptionSchema(
+                label=opt.get("label", ""),
+                value=opt.get("value", ""),
+                followup_hint=opt.get("followup_hint"),
+            )
+            for opt in step.options_presented
+        ]
+
+    content = step.content or {}
+    return AISessionStepResponse(
+        step_id=step.id,
+        step_order=step.step_order,
+        step_type=step.step_type,
+        content=content,
+        context_message=step.context_message,
+        options=options,
+        allow_free_text=content.get("allow_free_text", True),
+        allow_skip=content.get("allow_skip", True),
+        confidence_tier=session.confidence_tier,
+        confidence_score=session.confidence_score,
+    )
+
+
+async def start_session(
+    request: AISessionCreateRequest,
+    user_id: UUID,
+    account_id: UUID,
+    team_id: Optional[UUID],
+    db: AsyncSession,
+) -> AISessionCreateResponse:
+    """Start a new FlowPilot session: classify intake, match flows, get first step."""
+
+    # 1. Classify intake via fast LLM call
+    intake_text = _extract_intake_text(request.intake_content)
+    classification = await _classify_intake(intake_text)
+
+    # 2. Try to match existing flows
+    from app.services.flow_matching_engine import find_matches
+
+    matches = await find_matches(
+        intake_text=intake_text,
+        problem_domain=classification.get("problem_domain"),
+        account_id=account_id,
+        db=db,
+    )
+
+    top_match = matches[0] if matches else None
+    matched_flow_id = top_match["tree_id"] if top_match else None
+    match_score = top_match["score"] if top_match else None
+    matched_flow_name = top_match["tree_name"] if top_match else None
+
+    # 3. Build system prompt
+    matched_flow_context = ""
+    if top_match and top_match.get("score", 0) > 0.5:
+        matched_flow_context = (
+            f"## MATCHED FLOW\n"
+            f"A similar flow exists: \"{top_match['tree_name']}\" "
+            f"(match score: {top_match['score']:.0%}). "
+            f"Use it as a guide but adapt to the specific situation."
+        )
+
+    system_prompt = FLOWPILOT_SYSTEM_PROMPT.format(
+        structured_output_schema=STRUCTURED_OUTPUT_SCHEMA,
+        team_context="",  # Phase 2: team-specific context
+        matched_flow_context=matched_flow_context,
+    )
+
+    # 4. Build first user message from intake
+    user_message = _format_intake_message(request.intake_content, classification)
+
+    messages = [{"role": "user", "content": user_message}]
+
+    # 5. Call LLM for first diagnostic step
+    provider = get_ai_provider(settings.get_model_for_action("open_chat"))
+    raw_response, input_tokens, output_tokens = await provider.generate_json(
+        system_prompt=system_prompt,
+        messages=messages,
+        max_tokens=2048,
+    )
+
+    # Parse with retry on failure
+    try:
+        parsed = _parse_structured_output(raw_response)
+    except ValueError:
+        # Retry once with nudge
+        retry_messages = messages + [
+            {"role": "assistant", "content": raw_response},
+            {"role": "user", "content": "Please respond with ONLY valid JSON matching the required schema. No markdown or prose."},
+        ]
+        raw_response, retry_in, retry_out = await provider.generate_json(
+            system_prompt=system_prompt,
+            messages=retry_messages,
+            max_tokens=2048,
+        )
+        input_tokens += retry_in
+        output_tokens += retry_out
+        parsed = _parse_structured_output(raw_response)
+
+    confidence = parsed.get("confidence", 0.0)
+    confidence_tier = _confidence_to_tier(confidence)
+
+    # Initial confidence from match + classification
+    if top_match and top_match.get("score", 0) > 0.8:
+        confidence_tier = "guided"
+        confidence = max(confidence, 0.8)
+
+    # 6. Create session
+    session = AISession(
+        id=uuid.uuid4(),
+        user_id=user_id,
+        account_id=account_id,
+        team_id=team_id,
+        intake_type=request.intake_type,
+        intake_content=request.intake_content,
+        problem_summary=classification.get("problem_summary"),
+        problem_domain=classification.get("problem_domain"),
+        status="active",
+        confidence_tier=confidence_tier,
+        confidence_score=confidence,
+        matched_flow_id=matched_flow_id,
+        match_score=match_score,
+        psa_ticket_id=request.psa_ticket_id,
+        psa_connection_id=request.psa_connection_id,
+        total_input_tokens=input_tokens,
+        total_output_tokens=output_tokens,
+        step_count=1,
+        system_prompt_snapshot=system_prompt,
+        conversation_messages=[
+            {"role": "user", "content": user_message},
+            {"role": "assistant", "content": raw_response},
+        ],
+    )
+    db.add(session)
+
+    # 7. Create first step
+    step = _create_step_from_parsed(
+        session_id=session.id,
+        step_order=0,
+        parsed=parsed,
+        input_tokens=input_tokens,
+        output_tokens=output_tokens,
+    )
+    db.add(step)
+
+    await db.flush()
+
+    return AISessionCreateResponse(
+        session_id=session.id,
+        status=session.status,
+        confidence_tier=session.confidence_tier,
+        problem_summary=session.problem_summary,
+        problem_domain=session.problem_domain,
+        matched_flow_id=matched_flow_id,
+        matched_flow_name=matched_flow_name,
+        match_score=match_score,
+        first_step=_build_step_response(step, session),
+    )
+
+
+async def process_response(
+    session_id: UUID,
+    request: StepResponseRequest,
+    user_id: UUID,
+    db: AsyncSession,
+) -> StepResponseResponse:
+    """Process an engineer's response and generate the next FlowPilot step."""
+
+    session = await _load_session(session_id, user_id, db)
+
+    if session.status != "active":
+        raise ValueError(f"Session is {session.status}, not active")
+
+    if session.step_count >= MAX_STEPS_PER_SESSION:
+        raise ValueError("Maximum steps reached for this session")
+
+    # Update the current (latest) step with engineer's response
+    latest_step = session.steps[-1] if session.steps else None
+    if latest_step and latest_step.responded_at is None:
+        latest_step.selected_option = request.selected_option
+        latest_step.free_text_input = request.free_text_input
+        latest_step.was_free_text = bool(request.free_text_input and not request.selected_option)
+        latest_step.was_skipped = request.was_skipped
+        latest_step.action_result = request.action_result
+        latest_step.responded_at = datetime.now(timezone.utc)
+
+    # Build the conversation message for the engineer's response
+    response_text = _format_engineer_response(request)
+    session.conversation_messages = session.conversation_messages + [
+        {"role": "user", "content": response_text}
+    ]
+
+    # Call LLM with full conversation
+    provider = get_ai_provider(settings.get_model_for_action("open_chat"))
+    raw_response, input_tokens, output_tokens = await provider.generate_json(
+        system_prompt=session.system_prompt_snapshot or "",
+        messages=session.conversation_messages,
+        max_tokens=2048,
+    )
+
+    try:
+        parsed = _parse_structured_output(raw_response)
+    except ValueError:
+        retry_messages = session.conversation_messages + [
+            {"role": "assistant", "content": raw_response},
+            {"role": "user", "content": "Please respond with ONLY valid JSON matching the required schema."},
+        ]
+        raw_response, retry_in, retry_out = await provider.generate_json(
+            system_prompt=session.system_prompt_snapshot or "",
+            messages=retry_messages,
+            max_tokens=2048,
+        )
+        input_tokens += retry_in
+        output_tokens += retry_out
+        parsed = _parse_structured_output(raw_response)
+
+    # Append assistant response to conversation
+    session.conversation_messages = session.conversation_messages + [
+        {"role": "assistant", "content": raw_response}
+    ]
+
+    # Update session confidence
+    confidence = parsed.get("confidence", session.confidence_score)
+    session.confidence_score = confidence
+    session.confidence_tier = _confidence_to_tier(confidence)
+    session.total_input_tokens += input_tokens
+    session.total_output_tokens += output_tokens
+    session.step_count += 1
+
+    # Create new step
+    step = _create_step_from_parsed(
+        session_id=session.id,
+        step_order=session.step_count - 1,
+        parsed=parsed,
+        input_tokens=input_tokens,
+        output_tokens=output_tokens,
+    )
+    db.add(step)
+
+    await db.flush()
+
+    # Check if resolution was suggested
+    resolution_suggested = parsed["type"] == "resolution_suggestion"
+    resolution_summary = parsed.get("resolution_summary") if resolution_suggested else None
+
+    return StepResponseResponse(
+        session_id=session.id,
+        status=session.status,
+        confidence_tier=session.confidence_tier,
+        confidence_score=session.confidence_score,
+        next_step=_build_step_response(step, session),
+        resolution_suggested=resolution_suggested,
+        resolution_summary=resolution_summary,
+    )
+
+
+async def resolve_session(
+    session_id: UUID,
+    request: ResolveSessionRequest,
+    user_id: UUID,
+    db: AsyncSession,
+) -> SessionCloseResponse:
+    """Close a session as resolved and generate documentation."""
+    session = await _load_session(session_id, user_id, db)
+
+    if session.status not in ("active", "paused"):
+        raise ValueError(f"Cannot resolve session in status: {session.status}")
+
+    session.status = "resolved"
+    session.resolved_at = datetime.now(timezone.utc)
+    session.resolution_summary = request.resolution_summary
+    session.resolution_action = request.resolution_action
+
+    if request.session_rating is not None:
+        session.session_rating = request.session_rating
+    if request.session_feedback is not None:
+        session.session_feedback = request.session_feedback
+
+    documentation = _generate_documentation(session)
+
+    await db.flush()
+
+    return SessionCloseResponse(
+        session_id=session.id,
+        status=session.status,
+        documentation=documentation,
+    )
+
+
+async def escalate_session(
+    session_id: UUID,
+    request: EscalateSessionRequest,
+    user_id: UUID,
+    db: AsyncSession,
+) -> SessionCloseResponse:
+    """Escalate a session to another engineer."""
+    session = await _load_session(session_id, user_id, db)
+
+    if session.status not in ("active", "paused"):
+        raise ValueError(f"Cannot escalate session in status: {session.status}")
+
+    session.status = "escalated"
+    session.resolved_at = datetime.now(timezone.utc)
+    session.escalation_reason = request.escalation_reason
+    session.escalated_to_id = request.escalated_to_id
+
+    # Build escalation package
+    session.escalation_package = _build_escalation_package(session)
+
+    documentation = _generate_documentation(session)
+
+    await db.flush()
+
+    return SessionCloseResponse(
+        session_id=session.id,
+        status=session.status,
+        documentation=documentation,
+    )
+
+
+async def rate_session(
+    session_id: UUID,
+    rating: int,
+    feedback: Optional[str],
+    user_id: UUID,
+    db: AsyncSession,
+) -> None:
+    """Submit post-session rating."""
+    session = await _load_session(session_id, user_id, db)
+    session.session_rating = rating
+    session.session_feedback = feedback
+    await db.flush()
+
+
+async def get_session_documentation(
+    session_id: UUID,
+    user_id: UUID,
+    db: AsyncSession,
+) -> SessionDocumentation:
+    """Get auto-generated documentation for a session."""
+    session = await _load_session(session_id, user_id, db)
+    return _generate_documentation(session)
+
+
+# ── Internal helpers ──
+
+async def _load_session(
+    session_id: UUID,
+    user_id: UUID,
+    db: AsyncSession,
+) -> AISession:
+    """Load session with steps, verifying ownership."""
+    result = await db.execute(
+        select(AISession)
+        .options(selectinload(AISession.steps))
+        .where(AISession.id == session_id)
+    )
+    session = result.scalar_one_or_none()
+
+    if not session:
+        raise ValueError("Session not found")
+
+    # Allow access if user is the session owner or the escalation target
+    if session.user_id != user_id and session.escalated_to_id != user_id:
+        raise PermissionError("Not authorized to access this session")
+
+    return session
+
+
+async def _classify_intake(intake_text: str) -> dict[str, Any]:
+    """Quick LLM call to classify intake content."""
+    try:
+        provider = get_ai_provider(settings.get_model_for_action("quick_action"))
+        raw, _, _ = await provider.generate_json(
+            system_prompt=INTAKE_CLASSIFICATION_PROMPT,
+            messages=[{"role": "user", "content": intake_text}],
+            max_tokens=512,
+        )
+        return json.loads(raw.strip())
+    except Exception as e:
+        logger.warning("Intake classification failed: %s", e)
+        return {
+            "problem_summary": intake_text[:120],
+            "problem_domain": "other",
+            "key_symptoms": [],
+            "urgency": "medium",
+        }
+
+
+def _extract_intake_text(intake_content: dict[str, Any]) -> str:
+    """Extract searchable text from intake content."""
+    parts = []
+    if text := intake_content.get("text"):
+        parts.append(text)
+    if log := intake_content.get("log_content"):
+        parts.append(f"Log output:\n{log}")
+    if ticket := intake_content.get("ticket_data"):
+        if isinstance(ticket, dict):
+            parts.append(f"Ticket: {ticket.get('summary', '')}")
+    return "\n\n".join(parts) if parts else str(intake_content)
+
+
+def _format_intake_message(
+    intake_content: dict[str, Any],
+    classification: dict[str, Any],
+) -> str:
+    """Format intake + classification into the first user message."""
+    parts = ["I need help troubleshooting an issue."]
+
+    if text := intake_content.get("text"):
+        parts.append(f"\n**Problem description:**\n{text}")
+
+    if log := intake_content.get("log_content"):
+        parts.append(f"\n**Log output:**\n```\n{log}\n```")
+
+    if summary := classification.get("problem_summary"):
+        parts.append(f"\n**Classified as:** {summary}")
+
+    if domain := classification.get("problem_domain"):
+        parts.append(f"**Domain:** {domain}")
+
+    symptoms = classification.get("key_symptoms", [])
+    if symptoms:
+        parts.append(f"**Key symptoms:** {', '.join(symptoms)}")
+
+    return "\n".join(parts)
+
+
+def _format_engineer_response(request: StepResponseRequest) -> str:
+    """Format engineer's step response into a conversation message."""
+    if request.was_skipped:
+        return "I can't check this right now / I don't know."
+
+    parts = []
+    if request.selected_option:
+        parts.append(f"Selected: {request.selected_option}")
+
+    if request.free_text_input:
+        parts.append(request.free_text_input)
+
+    if request.action_result:
+        result = request.action_result
+        success = "succeeded" if result.get("success") else "did not work"
+        parts.append(f"Action {success}.")
+        if details := result.get("details"):
+            parts.append(f"Details: {details}")
+
+    return "\n".join(parts) if parts else "No response provided."
+
+
+def _create_step_from_parsed(
+    session_id: UUID,
+    step_order: int,
+    parsed: dict[str, Any],
+    input_tokens: int,
+    output_tokens: int,
+) -> AISessionStep:
+    """Create an AISessionStep from parsed LLM output."""
+    step_type = parsed["type"]
+    if step_type == "resolution_suggestion":
+        step_type = "action"  # Store as action in DB, UI distinguishes via content
+
+    # Build content dict (everything the UI needs to render)
+    content = {
+        "text": parsed.get("content", ""),
+        "type": parsed["type"],
+    }
+    if parsed["type"] == "action":
+        content["action_type"] = parsed.get("action_type", "instruction")
+        content["expected_outcome"] = parsed.get("expected_outcome")
+    elif parsed["type"] == "resolution_suggestion":
+        content["resolution_summary"] = parsed.get("resolution_summary")
+        content["follow_up_recommendations"] = parsed.get("follow_up_recommendations", [])
+        content["allow_free_text"] = False
+        content["allow_skip"] = False
+
+    # Extract options for question type
+    options = None
+    if parsed["type"] == "question" and "options" in parsed:
+        options = parsed["options"]
+        content["allow_free_text"] = parsed.get("allow_free_text", True)
+        content["allow_skip"] = parsed.get("allow_skip", True)
+
+    return AISessionStep(
+        id=uuid.uuid4(),
+        session_id=session_id,
+        step_order=step_order,
+        step_type=step_type if parsed["type"] != "resolution_suggestion" else "action",
+        content=content,
+        context_message=parsed.get("context_message"),
+        options_presented=options,
+        confidence_at_step=parsed.get("confidence", 0.0),
+        ai_reasoning=parsed.get("reasoning"),
+        input_tokens=input_tokens,
+        output_tokens=output_tokens,
+    )
+
+
+def _generate_documentation(session: AISession) -> SessionDocumentation:
+    """Generate structured documentation from a session's steps."""
+    diagnostic_steps = []
+
+    for step in session.steps:
+        content = step.content or {}
+        description = content.get("text", "")
+
+        # Determine engineer response
+        engineer_response = None
+        if step.was_skipped:
+            engineer_response = "Skipped"
+        elif step.selected_option:
+            # Find the label for the selected option
+            if step.options_presented:
+                for opt in step.options_presented:
+                    if opt.get("value") == step.selected_option:
+                        engineer_response = opt.get("label", step.selected_option)
+                        break
+                else:
+                    engineer_response = step.selected_option
+            else:
+                engineer_response = step.selected_option
+        elif step.free_text_input:
+            engineer_response = step.free_text_input
+
+        # Determine outcome
+        outcome = None
+        if step.action_result:
+            result = step.action_result
+            outcome = "Succeeded" if result.get("success") else "Did not resolve"
+            if details := result.get("details"):
+                outcome += f" — {details}"
+
+        diagnostic_steps.append(DocumentationStep(
+            step_number=step.step_order + 1,
+            step_type=step.step_type,
+            description=description,
+            engineer_response=engineer_response,
+            outcome=outcome,
+        ))
+
+    # Calculate duration
+    duration_display = None
+    if session.resolved_at and session.created_at:
+        delta = session.resolved_at - session.created_at
+        minutes = int(delta.total_seconds() / 60)
+        if minutes < 60:
+            duration_display = f"{minutes}m"
+        else:
+            hours = minutes // 60
+            remaining = minutes % 60
+            duration_display = f"{hours}h {remaining}m"
+
+    # Build intake summary
+    intake = session.intake_content or {}
+    intake_summary = intake.get("text", "")[:500]
+    if not intake_summary:
+        intake_summary = str(intake)[:500]
+
+    return SessionDocumentation(
+        problem_summary=session.problem_summary or "No summary available",
+        problem_domain=session.problem_domain,
+        intake_summary=intake_summary,
+        diagnostic_steps=diagnostic_steps,
+        resolution_summary=session.resolution_summary,
+        escalation_reason=session.escalation_reason,
+        total_steps=session.step_count,
+        duration_display=duration_display,
+        generated_at=datetime.now(timezone.utc),
+    )
+
+
+def _build_escalation_package(session: AISession) -> dict[str, Any]:
+    """Build context package for the receiving engineer."""
+    steps_tried = []
+    for step in session.steps:
+        content = step.content or {}
+        entry = {
+            "step_type": step.step_type,
+            "description": content.get("text", ""),
+        }
+        if step.selected_option:
+            entry["response"] = step.selected_option
+        elif step.free_text_input:
+            entry["response"] = step.free_text_input
+        elif step.was_skipped:
+            entry["response"] = "Skipped"
+        if step.action_result:
+            entry["action_result"] = step.action_result
+        steps_tried.append(entry)
+
+    return {
+        "problem_summary": session.problem_summary,
+        "problem_domain": session.problem_domain,
+        "intake_content": session.intake_content,
+        "confidence_at_escalation": session.confidence_score,
+        "steps_tried": steps_tried,
+        "escalation_reason": session.escalation_reason,
+    }