feat(knowledge-flywheel): add Phase 3 Knowledge Flywheel — AI analysis, review queue, analytics

Phase 3 implementation:
- AI session analysis service that generates flow proposals from resolved sessions
- APScheduler job for batch processing pending analyses (max_instances=1)
- Knowledge gap detection (weak options, high escalation signals)
- Flow proposals CRUD with team admin review workflow (approve/edit/dismiss/reject)
- FlowPilot analytics dashboard with confidence tiers, PSA metrics, knowledge gaps
- In-session script generator component
- Review queue page with filtering and proposal detail panel

Bug fixes from review (12 total):
- Fix "Edit & Publish" navigating to non-existent /editor/new route
- Hide Approve button for enhancement proposals (require Edit & Publish)
- Add max_instances=1 to scheduler to prevent TOCTOU race
- Fix eventual_success case() double-counting failed retries
- Add tree_structure validation before creating tree from proposal
- Simplify script generator rendering condition
- Add severity style fallback, toFixed on rates, Link instead of <a href>
- Add toast.warning on dismiss failure, fix dedup for domain-less sessions
- Cast Decimal to int in knowledge gap evidence dicts

Also updates CLAUDE.md with lessons 67-71 and Phase 3 project structure.

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

backend/alembic/versions/3266dd9d8111_add_ai_session_id_to_script_generations.py

@@ -0,0 +1,40 @@
+"""add ai_session_id to script_generations
+
+Revision ID: 3266dd9d8111
+Revises: a0b871cb0c5e
+Create Date: 2026-03-19 03:52:09.457961
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = '3266dd9d8111'
+down_revision: Union[str, None] = 'a0b871cb0c5e'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column('script_generations', sa.Column(
+        'ai_session_id', sa.UUID(), nullable=True,
+        comment='FlowPilot AI session that triggered this generation',
+    ))
+    op.create_index(
+        op.f('ix_script_generations_ai_session_id'),
+        'script_generations', ['ai_session_id'], unique=False,
+    )
+    op.create_foreign_key(
+        'fk_script_generations_ai_session_id',
+        'script_generations', 'ai_sessions',
+        ['ai_session_id'], ['id'],
+        ondelete='SET NULL',
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint('fk_script_generations_ai_session_id', 'script_generations', type_='foreignkey')
+    op.drop_index(op.f('ix_script_generations_ai_session_id'), table_name='script_generations')
+    op.drop_column('script_generations', 'ai_session_id')

backend/alembic/versions/47c3b4f42e88_add_flow_proposals_table.py

@@ -0,0 +1,64 @@
+"""add flow_proposals table
+
+Revision ID: 47c3b4f42e88
+Revises: cc3201489b72
+Create Date: 2026-03-19 03:11:33.663729
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = '47c3b4f42e88'
+down_revision: Union[str, None] = 'cc3201489b72'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table('flow_proposals',
+        sa.Column('id', sa.UUID(), nullable=False),
+        sa.Column('account_id', sa.UUID(), nullable=False),
+        sa.Column('team_id', sa.UUID(), nullable=True),
+        sa.Column('source_session_id', sa.UUID(), nullable=False),
+        sa.Column('proposal_type', sa.String(length=30), nullable=False),
+        sa.Column('target_flow_id', sa.UUID(), nullable=True, comment='For enhancements: which existing flow to modify'),
+        sa.Column('title', sa.String(length=255), nullable=False, comment='Human-readable title for the proposed flow'),
+        sa.Column('description', sa.Text(), nullable=True, comment='AI-generated description of what this flow covers'),
+        sa.Column('proposed_flow_data', postgresql.JSONB(astext_type=sa.Text()), nullable=False, comment='Complete flow/tree_structure definition (nodes, edges, conditions)'),
+        sa.Column('proposed_diff', postgresql.JSONB(astext_type=sa.Text()), nullable=True, comment='For enhancements: what changed vs existing flow'),
+        sa.Column('confidence_score', sa.Float(), nullable=False, comment='How confident the system is in this proposal (0.0-1.0)'),
+        sa.Column('supporting_session_count', sa.Integer(), nullable=False, comment='Number of sessions with similar resolution paths'),
+        sa.Column('supporting_session_ids', postgresql.JSONB(astext_type=sa.Text()), nullable=False, comment='Array of session IDs that support this proposal'),
+        sa.Column('problem_domain', sa.String(length=100), nullable=True),
+        sa.Column('status', sa.String(length=30), nullable=False),
+        sa.Column('reviewed_by', sa.UUID(), nullable=True),
+        sa.Column('reviewer_notes', sa.Text(), nullable=True),
+        sa.Column('published_flow_id', sa.UUID(), nullable=True, comment='The flow that was created/updated when this proposal was approved'),
+        sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+        sa.Column('reviewed_at', sa.DateTime(timezone=True), nullable=True),
+        sa.CheckConstraint("proposal_type IN ('new_flow', 'enhancement', 'branch_addition', 'auto_reinforced')", name='ck_flow_proposals_type'),
+        sa.CheckConstraint("status IN ('pending', 'approved', 'modified', 'rejected', 'dismissed', 'auto_reinforced')", name='ck_flow_proposals_status'),
+        sa.ForeignKeyConstraint(['account_id'], ['accounts.id'], ondelete='CASCADE'),
+        sa.ForeignKeyConstraint(['published_flow_id'], ['trees.id'], ondelete='SET NULL'),
+        sa.ForeignKeyConstraint(['reviewed_by'], ['users.id'], ondelete='SET NULL'),
+        sa.ForeignKeyConstraint(['source_session_id'], ['ai_sessions.id'], ondelete='CASCADE'),
+        sa.ForeignKeyConstraint(['target_flow_id'], ['trees.id'], ondelete='SET NULL'),
+        sa.ForeignKeyConstraint(['team_id'], ['teams.id'], ondelete='SET NULL'),
+        sa.PrimaryKeyConstraint('id'),
+    )
+    op.create_index(op.f('ix_flow_proposals_account_id'), 'flow_proposals', ['account_id'], unique=False)
+    op.create_index(op.f('ix_flow_proposals_source_session_id'), 'flow_proposals', ['source_session_id'], unique=False)
+    op.create_index(op.f('ix_flow_proposals_status'), 'flow_proposals', ['status'], unique=False)
+    op.create_index(op.f('ix_flow_proposals_team_id'), 'flow_proposals', ['team_id'], unique=False)
+
+
+def downgrade() -> None:
+    op.drop_index(op.f('ix_flow_proposals_team_id'), table_name='flow_proposals')
+    op.drop_index(op.f('ix_flow_proposals_status'), table_name='flow_proposals')
+    op.drop_index(op.f('ix_flow_proposals_source_session_id'), table_name='flow_proposals')
+    op.drop_index(op.f('ix_flow_proposals_account_id'), table_name='flow_proposals')
+    op.drop_table('flow_proposals')

backend/alembic/versions/a0b871cb0c5e_add_analysis_status_to_ai_sessions.py

@@ -0,0 +1,28 @@
+"""add analysis_status to ai_sessions
+
+Revision ID: a0b871cb0c5e
+Revises: 47c3b4f42e88
+Create Date: 2026-03-19 03:26:26.965134
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = 'a0b871cb0c5e'
+down_revision: Union[str, None] = '47c3b4f42e88'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column('ai_sessions', sa.Column(
+        'analysis_status', sa.String(length=20), nullable=True,
+        comment='Knowledge Flywheel status: null (N/A), pending, completed, failed',
+    ))
+
+
+def downgrade() -> None:
+    op.drop_column('ai_sessions', 'analysis_status')

backend/app/api/deps.py

@@ -145,6 +145,22 @@ async def require_engineer_or_admin(
     )
 
 
+async def require_team_admin(
+    current_user: Annotated[User, Depends(get_current_active_user)]
+) -> User:
+    """Require team admin, account owner, or super admin role."""
+    if current_user.is_super_admin:
+        return current_user
+    if current_user.is_team_admin:
+        return current_user
+    if current_user.account_role == "owner":
+        return current_user
+    raise HTTPException(
+        status_code=status.HTTP_403_FORBIDDEN,
+        detail="Team admin access required"
+    )
+
+
 async def require_account_owner(
     current_user: Annotated[User, Depends(get_current_active_user)]
 ) -> User:

backend/app/api/endpoints/flow_proposals.py

@@ -0,0 +1,306 @@
+"""Review Queue API — CRUD for flow proposals.
+
+Endpoints for listing, reviewing, and managing Knowledge Flywheel proposals:
+  GET    /flow-proposals              — List proposals (filterable)
+  GET    /flow-proposals/stats        — Dashboard stats
+  GET    /flow-proposals/{id}         — Get proposal detail
+  POST   /flow-proposals/{id}/review  — Approve, reject, modify, or dismiss
+"""
+import logging
+import uuid
+from datetime import datetime, timezone, timedelta
+from typing import Annotated, Optional
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
+from sqlalchemy import select, func, case
+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, require_team_admin
+from app.models.user import User
+from app.models.tree import Tree
+from app.models.flow_proposal import FlowProposal
+from app.schemas.flow_proposal import (
+    FlowProposalSummary,
+    FlowProposalDetail,
+    FlowProposalStats,
+    ReviewProposalRequest,
+)
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/flow-proposals", tags=["flow-proposals"])
+
+
+# ── List proposals ──
+
+@router.get("", response_model=list[FlowProposalSummary])
+@limiter.limit("30/minute")
+async def list_proposals(
+    request: Request,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+    proposal_status: Optional[str] = Query(None, alias="status"),
+    proposal_type: Optional[str] = Query(None, alias="type"),
+    domain: Optional[str] = Query(None),
+    sort_by: str = Query("newest", pattern="^(newest|confidence|sessions)$"),
+    skip: int = Query(0, ge=0),
+    limit: int = Query(20, ge=1, le=100),
+):
+    """List flow proposals for the current user's account."""
+    if not current_user.account_id:
+        return []
+
+    query = (
+        select(FlowProposal)
+        .where(FlowProposal.account_id == current_user.account_id)
+    )
+
+    if proposal_status:
+        query = query.where(FlowProposal.status == proposal_status)
+    if proposal_type:
+        query = query.where(FlowProposal.proposal_type == proposal_type)
+    if domain:
+        query = query.where(FlowProposal.problem_domain == domain)
+
+    # Sorting
+    if sort_by == "confidence":
+        query = query.order_by(FlowProposal.confidence_score.desc())
+    elif sort_by == "sessions":
+        query = query.order_by(FlowProposal.supporting_session_count.desc())
+    else:  # newest
+        query = query.order_by(FlowProposal.created_at.desc())
+
+    query = query.offset(skip).limit(limit)
+
+    result = await db.execute(query)
+    proposals = result.scalars().all()
+
+    return [FlowProposalSummary.model_validate(p) for p in proposals]
+
+
+# ── Stats ──
+
+@router.get("/stats", response_model=FlowProposalStats)
+@limiter.limit("30/minute")
+async def get_proposal_stats(
+    request: Request,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+):
+    """Get review queue dashboard stats."""
+    if not current_user.account_id:
+        return FlowProposalStats(
+            pending_count=0, approved_this_week=0, rejected_this_week=0,
+            auto_reinforced_this_week=0, top_domains=[],
+        )
+
+    week_ago = datetime.now(timezone.utc) - timedelta(days=7)
+
+    # Count pending
+    pending_result = await db.execute(
+        select(func.count(FlowProposal.id))
+        .where(
+            FlowProposal.account_id == current_user.account_id,
+            FlowProposal.status == "pending",
+        )
+    )
+    pending_count = pending_result.scalar() or 0
+
+    # Reviewed this week (approved/rejected/modified use reviewed_at)
+    reviewed_result = await db.execute(
+        select(
+            FlowProposal.status,
+            func.count(FlowProposal.id),
+        )
+        .where(
+            FlowProposal.account_id == current_user.account_id,
+            FlowProposal.reviewed_at >= week_ago,
+            FlowProposal.status.in_(["approved", "modified", "rejected", "dismissed"]),
+        )
+        .group_by(FlowProposal.status)
+    )
+    reviewed_counts = {row[0]: row[1] for row in reviewed_result.all()}
+
+    # Auto-reinforced this week (use created_at since they have no review)
+    reinforced_result = await db.execute(
+        select(func.count(FlowProposal.id))
+        .where(
+            FlowProposal.account_id == current_user.account_id,
+            FlowProposal.created_at >= week_ago,
+            FlowProposal.status == "auto_reinforced",
+        )
+    )
+    auto_reinforced_count = reinforced_result.scalar() or 0
+
+    # Top domains
+    domain_result = await db.execute(
+        select(
+            FlowProposal.problem_domain,
+            func.count(FlowProposal.id).label("count"),
+        )
+        .where(
+            FlowProposal.account_id == current_user.account_id,
+            FlowProposal.status == "pending",
+            FlowProposal.problem_domain.isnot(None),
+        )
+        .group_by(FlowProposal.problem_domain)
+        .order_by(func.count(FlowProposal.id).desc())
+        .limit(5)
+    )
+    top_domains = [{"domain": row[0], "count": row[1]} for row in domain_result.all()]
+
+    return FlowProposalStats(
+        pending_count=pending_count,
+        approved_this_week=reviewed_counts.get("approved", 0) + reviewed_counts.get("modified", 0),
+        rejected_this_week=reviewed_counts.get("rejected", 0),
+        auto_reinforced_this_week=auto_reinforced_count,
+        top_domains=top_domains,
+    )
+
+
+# ── Detail ──
+
+@router.get("/{proposal_id}", response_model=FlowProposalDetail)
+@limiter.limit("30/minute")
+async def get_proposal(
+    request: Request,
+    proposal_id: UUID,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+):
+    """Get full proposal detail."""
+    result = await db.execute(
+        select(FlowProposal).where(
+            FlowProposal.id == proposal_id,
+            FlowProposal.account_id == current_user.account_id,
+        )
+    )
+    proposal = result.scalar_one_or_none()
+    if not proposal:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Proposal not found")
+
+    return FlowProposalDetail.model_validate(proposal)
+
+
+# ── Review ──
+
+@router.post("/{proposal_id}/review", response_model=FlowProposalDetail)
+@limiter.limit("10/minute")
+async def review_proposal(
+    request: Request,
+    proposal_id: UUID,
+    data: ReviewProposalRequest,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_team_admin),
+):
+    """Review a proposal: approve, reject, modify, or dismiss."""
+    result = await db.execute(
+        select(FlowProposal).where(
+            FlowProposal.id == proposal_id,
+            FlowProposal.account_id == current_user.account_id,
+        )
+    )
+    proposal = result.scalar_one_or_none()
+    if not proposal:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Proposal not found")
+
+    if proposal.status not in ("pending", "dismissed"):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Cannot review proposal in status: {proposal.status}",
+        )
+
+    proposal.reviewed_by = current_user.id
+    proposal.reviewed_at = datetime.now(timezone.utc)
+    proposal.reviewer_notes = data.reviewer_notes
+
+    if data.action == "approve":
+        if proposal.proposal_type == "new_flow":
+            flow_data = proposal.proposed_flow_data
+            new_tree = await _create_tree_from_proposal(proposal, flow_data, current_user, db)
+            proposal.status = "approved"
+            proposal.published_flow_id = new_tree.id
+        elif proposal.proposal_type in ("enhancement", "branch_addition"):
+            # Enhancement proposals contain diffs, not complete tree structures.
+            # Direct approval requires modified_flow_data with the complete merged structure.
+            # Redirect reviewers to use "Edit & Publish" for enhancements.
+            if not data.modified_flow_data:
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail="Enhancement proposals require 'Edit & Publish' to merge changes into the existing flow. Use the modify action with modified_flow_data.",
+                )
+            new_tree = await _create_tree_from_proposal(proposal, data.modified_flow_data, current_user, db)
+            proposal.status = "approved"
+            proposal.published_flow_id = new_tree.id
+        else:
+            # auto_reinforced shouldn't reach here, but handle gracefully
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Cannot approve proposal of type: {proposal.proposal_type}",
+            )
+
+    elif data.action == "modify":
+        if not data.modified_flow_data:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="modified_flow_data is required for modify action",
+            )
+        new_tree = await _create_tree_from_proposal(proposal, data.modified_flow_data, current_user, db)
+        proposal.status = "modified"
+        proposal.published_flow_id = new_tree.id
+
+    elif data.action == "reject":
+        proposal.status = "rejected"
+
+    elif data.action == "dismiss":
+        proposal.status = "dismissed"
+
+    await db.commit()
+
+    return FlowProposalDetail.model_validate(proposal)
+
+
+async def _create_tree_from_proposal(
+    proposal: FlowProposal,
+    flow_data: dict,
+    user: User,
+    db: AsyncSession,
+) -> Tree:
+    """Create a new Tree from proposal flow data."""
+    tree_structure = flow_data.get("tree_structure", flow_data)
+    match_keywords = flow_data.get("match_keywords", [])
+
+    if not tree_structure or not isinstance(tree_structure, dict) or not tree_structure.get("id"):
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="Proposal has no valid tree structure. Use 'Edit & Publish' to build the flow manually.",
+        )
+
+    new_tree = Tree(
+        id=uuid.uuid4(),
+        name=proposal.title,
+        description=proposal.description,
+        tree_type="troubleshooting",
+        tree_structure=tree_structure,
+        author_id=user.id,
+        account_id=proposal.account_id,
+        team_id=proposal.team_id,
+        origin="ai_generated" if proposal.proposal_type == "new_flow" else "ai_enhanced",
+        source_session_id=proposal.source_session_id,
+        match_keywords=match_keywords,
+    )
+    db.add(new_tree)
+    await db.flush()
+
+    logger.info(
+        "Created tree %s from proposal %s (%s)",
+        new_tree.id, proposal.id, proposal.proposal_type,
+    )
+    return new_tree

backend/app/api/endpoints/flowpilot_analytics.py

@@ -0,0 +1,358 @@
+"""FlowPilot Analytics API — MTTR, resolution rates, knowledge coverage.
+
+Endpoints:
+  GET /analytics/flowpilot?period=30d — Main dashboard data
+  GET /analytics/flowpilot/knowledge-gaps — Knowledge gap report
+"""
+import logging
+from datetime import datetime, timezone, timedelta
+from typing import Annotated, Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
+from sqlalchemy import select, func, case, cast, Date, extract
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.rate_limit import limiter
+from app.api.deps import get_current_active_user, get_db, require_team_admin
+from app.models.user import User
+from app.models.tree import Tree
+from app.models.ai_session import AISession
+from app.models.flow_proposal import FlowProposal
+from app.models.psa_post_log import PsaPostLog
+from app.schemas.flowpilot_analytics import (
+    FlowPilotDashboard,
+    MTTRDataPoint,
+    DomainBreakdown,
+    ConfidenceBreakdown,
+    KnowledgeCoverage,
+    DomainCoverage,
+    PsaMetrics,
+)
+from app.services.knowledge_gap_service import get_knowledge_gaps, KnowledgeGapReport
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/analytics/flowpilot", tags=["flowpilot-analytics"])
+
+
+def _get_period_start(period: str) -> datetime:
+    days = {"7d": 7, "30d": 30, "90d": 90}.get(period, 30)
+    return datetime.now(timezone.utc) - timedelta(days=days)
+
+
+@router.get("", response_model=FlowPilotDashboard)
+@limiter.limit("15/minute")
+async def get_dashboard(
+    request: Request,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_team_admin),
+    period: str = Query("30d", pattern="^(7d|30d|90d)$"),
+):
+    """Get FlowPilot analytics dashboard data."""
+    if not current_user.account_id:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="No account")
+
+    account_id = current_user.account_id
+    period_start = _get_period_start(period)
+
+    # ── Session counts ──
+    counts_result = await db.execute(
+        select(
+            func.count(AISession.id).label("total"),
+            func.sum(case((AISession.status == "resolved", 1), else_=0)).label("resolved"),
+            func.sum(case((AISession.status.in_(["escalated", "requesting_escalation"]), 1), else_=0)).label("escalated"),
+            func.sum(case((AISession.status == "abandoned", 1), else_=0)).label("abandoned"),
+            func.avg(case((AISession.status == "resolved", AISession.step_count), else_=None)).label("avg_steps"),
+            func.avg(AISession.session_rating).label("avg_rating"),
+        )
+        .where(
+            AISession.account_id == account_id,
+            AISession.created_at >= period_start,
+        )
+    )
+    row = counts_result.one()
+    total = row.total or 0
+    resolved = row.resolved or 0
+    escalated = row.escalated or 0
+    abandoned = row.abandoned or 0
+    avg_steps = float(row.avg_steps or 0)
+    avg_rating = float(row.avg_rating) if row.avg_rating else None
+    resolution_rate = (resolved / total * 100) if total > 0 else 0.0
+
+    # ── MTTR ──
+    mttr_result = await db.execute(
+        select(
+            func.avg(
+                extract("epoch", AISession.resolved_at - AISession.created_at) / 60
+            ).label("avg_mttr"),
+        )
+        .where(
+            AISession.account_id == account_id,
+            AISession.created_at >= period_start,
+            AISession.status == "resolved",
+            AISession.resolved_at.isnot(None),
+        )
+    )
+    mttr_row = mttr_result.one()
+    mttr_minutes = float(mttr_row.avg_mttr) if mttr_row.avg_mttr else None
+
+    # ── Average duration ──
+    duration_result = await db.execute(
+        select(
+            func.avg(
+                extract("epoch", AISession.resolved_at - AISession.created_at) / 60
+            ).label("avg_duration"),
+        )
+        .where(
+            AISession.account_id == account_id,
+            AISession.created_at >= period_start,
+            AISession.resolved_at.isnot(None),
+        )
+    )
+    dur_row = duration_result.one()
+    avg_duration = float(dur_row.avg_duration) if dur_row.avg_duration else 0.0
+
+    # ── MTTR trend ──
+    mttr_trend_result = await db.execute(
+        select(
+            cast(AISession.resolved_at, Date).label("day"),
+            func.avg(
+                extract("epoch", AISession.resolved_at - AISession.created_at) / 60
+            ).label("mttr"),
+            func.count(AISession.id).label("count"),
+        )
+        .where(
+            AISession.account_id == account_id,
+            AISession.created_at >= period_start,
+            AISession.status == "resolved",
+            AISession.resolved_at.isnot(None),
+        )
+        .group_by(cast(AISession.resolved_at, Date))
+        .order_by(cast(AISession.resolved_at, Date))
+    )
+    mttr_trend = [
+        MTTRDataPoint(
+            date=str(r.day),
+            mttr_minutes=round(float(r.mttr or 0), 1),
+            session_count=r.count,
+        )
+        for r in mttr_trend_result.all()
+    ]
+
+    # ── Domain breakdown ──
+    domain_result = await db.execute(
+        select(
+            AISession.problem_domain,
+            func.count(AISession.id).label("total"),
+            func.sum(case((AISession.status == "resolved", 1), else_=0)).label("resolved"),
+            func.sum(case((AISession.status.in_(["escalated", "requesting_escalation"]), 1), else_=0)).label("escalated"),
+        )
+        .where(
+            AISession.account_id == account_id,
+            AISession.created_at >= period_start,
+            AISession.problem_domain.isnot(None),
+        )
+        .group_by(AISession.problem_domain)
+        .order_by(func.count(AISession.id).desc())
+    )
+    sessions_by_domain = [
+        DomainBreakdown(
+            domain=r.problem_domain or "unknown",
+            total=r.total,
+            resolved=r.resolved or 0,
+            escalated=r.escalated or 0,
+            resolution_rate=round((r.resolved or 0) / r.total * 100, 1) if r.total > 0 else 0.0,
+        )
+        for r in domain_result.all()
+    ]
+
+    # ── Confidence breakdown ──
+    confidence_result = await db.execute(
+        select(
+            AISession.confidence_tier,
+            func.count(AISession.id).label("total"),
+            func.sum(case((AISession.status == "resolved", 1), else_=0)).label("resolved"),
+        )
+        .where(
+            AISession.account_id == account_id,
+            AISession.created_at >= period_start,
+            AISession.status.in_(["resolved", "escalated", "requesting_escalation"]),
+        )
+        .group_by(AISession.confidence_tier)
+    )
+    conf_data = {r.confidence_tier: (r.total or 0, r.resolved or 0) for r in confidence_result.all()}
+
+    guided_total, guided_resolved = conf_data.get("guided", (0, 0))
+    exploring_total, exploring_resolved = conf_data.get("exploring", (0, 0))
+    discovery_total, discovery_resolved = conf_data.get("discovery", (0, 0))
+
+    confidence_breakdown = ConfidenceBreakdown(
+        guided_sessions=guided_total,
+        guided_resolution_rate=round(guided_resolved / guided_total * 100, 1) if guided_total > 0 else 0.0,
+        exploring_sessions=exploring_total,
+        exploring_resolution_rate=round(exploring_resolved / exploring_total * 100, 1) if exploring_total > 0 else 0.0,
+        discovery_sessions=discovery_total,
+        discovery_resolution_rate=round(discovery_resolved / discovery_total * 100, 1) if discovery_total > 0 else 0.0,
+    )
+
+    # ── Knowledge coverage ──
+    total_flows_result = await db.execute(
+        select(func.count(Tree.id)).where(Tree.account_id == account_id)
+    )
+    total_flows = total_flows_result.scalar() or 0
+
+    ai_flows_result = await db.execute(
+        select(func.count(Tree.id)).where(
+            Tree.account_id == account_id,
+            Tree.origin.in_(["ai_generated", "ai_enhanced"]),
+        )
+    )
+    ai_generated_flows = ai_flows_result.scalar() or 0
+
+    pending_proposals_result = await db.execute(
+        select(func.count(FlowProposal.id)).where(
+            FlowProposal.account_id == account_id,
+            FlowProposal.status == "pending",
+        )
+    )
+    total_proposals_pending = pending_proposals_result.scalar() or 0
+
+    approved_result = await db.execute(
+        select(func.count(FlowProposal.id)).where(
+            FlowProposal.account_id == account_id,
+            FlowProposal.reviewed_at >= period_start,
+            FlowProposal.status.in_(["approved", "modified"]),
+        )
+    )
+    proposals_approved = approved_result.scalar() or 0
+
+    rejected_result = await db.execute(
+        select(func.count(FlowProposal.id)).where(
+            FlowProposal.account_id == account_id,
+            FlowProposal.reviewed_at >= period_start,
+            FlowProposal.status == "rejected",
+        )
+    )
+    proposals_rejected = rejected_result.scalar() or 0
+
+    # Domain coverage
+    domain_coverage_result = await db.execute(
+        select(
+            AISession.problem_domain,
+            func.count(AISession.id).label("session_count"),
+            func.sum(case((AISession.confidence_tier == "guided", 1), else_=0)).label("guided_count"),
+        )
+        .where(
+            AISession.account_id == account_id,
+            AISession.created_at >= period_start,
+            AISession.problem_domain.isnot(None),
+        )
+        .group_by(AISession.problem_domain)
+    )
+    domain_flow_counts_result = await db.execute(
+        select(
+            Tree.tree_type,  # Reuse as domain proxy — not ideal but workable
+            func.count(Tree.id),
+        )
+        .where(Tree.account_id == account_id)
+        .group_by(Tree.tree_type)
+    )
+    # For now, flow_count per domain isn't directly available since Tree doesn't have problem_domain.
+    # Use match_keywords or just report 0. We'll improve this in Phase 4 with better flow categorization.
+    domain_cov_data = {}
+    for r in domain_coverage_result.all():
+        domain = r.problem_domain or "unknown"
+        sc = r.session_count or 0
+        gc = r.guided_count or 0
+        domain_cov_data[domain] = DomainCoverage(
+            domain=domain,
+            flow_count=0,  # TODO: match via category/tags in Phase 4
+            session_count=sc,
+            guided_rate=round(gc / sc * 100, 1) if sc > 0 else 0.0,
+        )
+
+    knowledge_coverage = KnowledgeCoverage(
+        total_flows=total_flows,
+        ai_generated_flows=ai_generated_flows,
+        total_proposals_pending=total_proposals_pending,
+        proposals_approved_this_period=proposals_approved,
+        proposals_rejected_this_period=proposals_rejected,
+        coverage_by_domain=list(domain_cov_data.values()),
+    )
+
+    # ── PSA metrics ──
+    psa_metrics = None
+    psa_linked = await db.execute(
+        select(func.count(AISession.id)).where(
+            AISession.account_id == account_id,
+            AISession.created_at >= period_start,
+            AISession.psa_ticket_id.isnot(None),
+        )
+    )
+    psa_linked_count = psa_linked.scalar() or 0
+
+    if psa_linked_count > 0 and total > 0:
+        psa_push_result = await db.execute(
+            select(
+                func.count(PsaPostLog.id).label("total_pushes"),
+                func.sum(case((PsaPostLog.status == "success", 1), else_=0)).label("first_success"),
+                func.sum(case(
+                    ((PsaPostLog.status == "success") & (PsaPostLog.retry_count > 0), 1),
+                    else_=0
+                )).label("retry_success"),
+            )
+            .join(AISession, PsaPostLog.ai_session_id == AISession.id)
+            .where(
+                AISession.account_id == account_id,
+                PsaPostLog.ai_session_id.isnot(None),
+                PsaPostLog.posted_at >= period_start,
+            )
+        )
+        push_row = psa_push_result.one()
+        total_pushes = push_row.total_pushes or 0
+        first_success = push_row.first_success or 0
+        retry_success = push_row.retry_success or 0
+
+        psa_metrics = PsaMetrics(
+            ticket_link_rate=round(psa_linked_count / total * 100, 1),
+            auto_push_success_rate=round(first_success / total_pushes * 100, 1) if total_pushes > 0 else 0.0,
+            auto_push_retry_success_rate=round(retry_success / total_pushes * 100, 1) if total_pushes > 0 else 0.0,
+            total_time_entries_logged=0,  # TODO: track from CW time entries
+            total_hours_logged=0.0,
+        )
+
+    return FlowPilotDashboard(
+        period=period,
+        total_sessions=total,
+        resolved_sessions=resolved,
+        escalated_sessions=escalated,
+        abandoned_sessions=abandoned,
+        resolution_rate=round(resolution_rate, 1),
+        avg_steps_to_resolution=round(avg_steps, 1),
+        avg_session_duration_minutes=round(avg_duration, 1),
+        avg_rating=round(avg_rating, 2) if avg_rating else None,
+        mttr_minutes=round(mttr_minutes, 1) if mttr_minutes else None,
+        mttr_trend=mttr_trend,
+        sessions_by_domain=sessions_by_domain,
+        confidence_breakdown=confidence_breakdown,
+        knowledge_coverage=knowledge_coverage,
+        psa_metrics=psa_metrics,
+    )
+
+
+@router.get("/knowledge-gaps", response_model=KnowledgeGapReport)
+@limiter.limit("10/minute")
+async def get_knowledge_gaps_endpoint(
+    request: Request,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_team_admin),
+    period: str = Query("30d", pattern="^(7d|30d|90d)$"),
+):
+    """Get knowledge gap analysis report."""
+    if not current_user.account_id:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="No account")
+
+    days = {"7d": 7, "30d": 30, "90d": 90}.get(period, 30)
+    return await get_knowledge_gaps(current_user.account_id, db, period_days=days)

backend/app/api/endpoints/scripts.py

@@ -355,6 +355,7 @@ async def generate_script(
         user_id=current_user.id,
         team_id=current_user.team_id,
         session_id=data.session_id,
+        ai_session_id=data.ai_session_id,
         parameters_used=redacted_params,
         generated_script=rendered_script,
     )

backend/app/api/router.py

@@ -22,6 +22,8 @@ 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
+from app.api.endpoints import flow_proposals
+from app.api.endpoints import flowpilot_analytics
 
 api_router = APIRouter()
 
@@ -69,3 +71,5 @@ 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)
+api_router.include_router(flow_proposals.router)
+api_router.include_router(flowpilot_analytics.router)

backend/app/main.py

@@ -190,6 +190,17 @@ async def lifespan(app: FastAPI):
         replace_existing=True,
     )
 
+    # Knowledge Flywheel analysis (every 5 minutes)
+    from app.services.knowledge_flywheel_scheduler import process_pending_analyses
+    scheduler.add_job(
+        process_pending_analyses,
+        trigger="interval",
+        minutes=5,
+        id="knowledge_flywheel_analysis",
+        replace_existing=True,
+        max_instances=1,
+    )
+
     # Auto-seed trees in background on PR environments
     seed_task = None
     if settings.SEED_ON_DEPLOY:

backend/app/models/__init__.py

@@ -42,6 +42,7 @@ from .psa_connection import PsaConnection
 from .psa_post_log import PsaPostLog
 from .psa_member_mapping import PsaMemberMapping
 from .supporting_data import SessionSupportingData
+from .flow_proposal import FlowProposal
 
 __all__ = [
     "User",
@@ -98,4 +99,5 @@ __all__ = [
     "PsaPostLog",
     "PsaMemberMapping",
     "SessionSupportingData",
+    "FlowProposal",
 ]

backend/app/models/ai_session.py

@@ -156,6 +156,12 @@ class AISession(Base):
         comment="Optional feedback text from engineer",
     )
 
+    # ── Knowledge Flywheel ──
+    analysis_status: Mapped[Optional[str]] = mapped_column(
+        String(20), nullable=True,
+        comment="Knowledge Flywheel status: null (N/A), pending, completed, failed",
+    )
+
     # ── AI tracking ──
     total_input_tokens: Mapped[int] = mapped_column(
         Integer, nullable=False, default=0,

backend/app/models/flow_proposal.py

@@ -0,0 +1,152 @@
+"""Flow proposal model.
+
+Generated by the Knowledge Flywheel after AI sessions resolve.
+Represents a proposed new flow or enhancement awaiting human review.
+"""
+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.user import User
+    from app.models.team import Team
+    from app.models.account import Account
+    from app.models.tree import Tree
+    from app.models.ai_session import AISession
+
+
+class FlowProposal(Base):
+    """A proposed new flow or enhancement generated from an AI session.
+
+    proposal_type:
+    - new_flow: No similar flow exists. Full flow definition proposed.
+    - enhancement: Similar flow exists but session discovered new branch/edge case.
+    - branch_addition: A single new branch to add to an existing flow.
+    - auto_reinforced: Session matched existing flow exactly (tracking only).
+
+    status:
+    - pending: Awaiting review
+    - approved: Reviewed and published to knowledge base
+    - modified: Reviewer edited before publishing
+    - rejected: Reviewer decided not to publish (bad quality)
+    - dismissed: Parked for later — not wrong, just not actionable now.
+    - auto_reinforced: Session matched existing flow exactly (no review needed)
+    """
+    __tablename__ = "flow_proposals"
+    __table_args__ = (
+        CheckConstraint(
+            "proposal_type IN ('new_flow', 'enhancement', 'branch_addition', 'auto_reinforced')",
+            name="ck_flow_proposals_type",
+        ),
+        CheckConstraint(
+            "status IN ('pending', 'approved', 'modified', 'rejected', 'dismissed', 'auto_reinforced')",
+            name="ck_flow_proposals_status",
+        ),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    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,
+    )
+    source_session_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("ai_sessions.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+
+    # ── Proposal details ──
+    proposal_type: Mapped[str] = mapped_column(
+        String(30), nullable=False,
+    )
+    target_flow_id: Mapped[Optional[uuid.UUID]] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("trees.id", ondelete="SET NULL"),
+        nullable=True,
+        comment="For enhancements: which existing flow to modify",
+    )
+    title: Mapped[str] = mapped_column(
+        String(255), nullable=False,
+        comment="Human-readable title for the proposed flow",
+    )
+    description: Mapped[Optional[str]] = mapped_column(
+        Text, nullable=True,
+        comment="AI-generated description of what this flow covers",
+    )
+    proposed_flow_data: Mapped[dict[str, Any]] = mapped_column(
+        JSONB, nullable=False,
+        comment="Complete flow/tree_structure definition (nodes, edges, conditions)",
+    )
+    proposed_diff: Mapped[Optional[dict[str, Any]]] = mapped_column(
+        JSONB, nullable=True,
+        comment="For enhancements: what changed vs existing flow",
+    )
+
+    # ── Scoring ──
+    confidence_score: Mapped[float] = mapped_column(
+        Float, nullable=False, default=0.0,
+        comment="How confident the system is in this proposal (0.0-1.0)",
+    )
+    supporting_session_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=1,
+        comment="Number of sessions with similar resolution paths",
+    )
+    supporting_session_ids: Mapped[list] = mapped_column(
+        JSONB, nullable=False, default=list,
+        comment="Array of session IDs that support this proposal",
+    )
+    problem_domain: Mapped[Optional[str]] = mapped_column(
+        String(100), nullable=True,
+    )
+
+    # ── Review ──
+    status: Mapped[str] = mapped_column(
+        String(30), nullable=False, default="pending", index=True,
+    )
+    reviewed_by: Mapped[Optional[uuid.UUID]] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("users.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+    reviewer_notes: Mapped[Optional[str]] = mapped_column(
+        Text, nullable=True,
+    )
+    published_flow_id: Mapped[Optional[uuid.UUID]] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("trees.id", ondelete="SET NULL"),
+        nullable=True,
+        comment="The flow that was created/updated when this proposal was approved",
+    )
+
+    # ── Timestamps ──
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
+    )
+    reviewed_at: Mapped[Optional[datetime]] = mapped_column(
+        DateTime(timezone=True), nullable=True,
+    )
+
+    # ── Relationships ──
+    account: Mapped["Account"] = relationship("Account")
+    team: Mapped[Optional["Team"]] = relationship("Team")
+    source_session: Mapped["AISession"] = relationship("AISession")
+    target_flow: Mapped[Optional["Tree"]] = relationship("Tree", foreign_keys=[target_flow_id])
+    published_flow: Mapped[Optional["Tree"]] = relationship("Tree", foreign_keys=[published_flow_id])
+    reviewer: Mapped[Optional["User"]] = relationship("User")

backend/app/models/script_template.py

@@ -97,6 +97,10 @@ class ScriptGeneration(Base):
     session_id: Mapped[Optional[uuid.UUID]] = mapped_column(
         UUID(as_uuid=True), ForeignKey("sessions.id", ondelete="SET NULL"), nullable=True, index=True
     )
+    ai_session_id: Mapped[Optional[uuid.UUID]] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("ai_sessions.id", ondelete="SET NULL"), nullable=True, index=True,
+        comment="FlowPilot AI session that triggered this generation",
+    )
     parameters_used: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
     generated_script: Mapped[str] = mapped_column(Text, nullable=False)
     created_at: Mapped[datetime] = mapped_column(

backend/app/schemas/flow_proposal.py

@@ -0,0 +1,51 @@
+"""Pydantic schemas for flow proposals (Knowledge Flywheel / Review Queue)."""
+from __future__ import annotations
+
+from typing import Optional, Any
+from uuid import UUID
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+
+class FlowProposalSummary(BaseModel):
+    """Compact proposal for list views."""
+    id: UUID
+    proposal_type: str
+    title: str
+    description: str | None = None
+    problem_domain: str | None = None
+    confidence_score: float
+    supporting_session_count: int
+    status: str
+    target_flow_id: UUID | None = None
+    source_session_id: UUID
+    created_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class FlowProposalDetail(FlowProposalSummary):
+    """Full proposal detail with flow data."""
+    proposed_flow_data: dict[str, Any]
+    proposed_diff: dict[str, Any] | None = None
+    supporting_session_ids: list[str] = []
+    reviewer_notes: str | None = None
+    reviewed_by: UUID | None = None
+    reviewed_at: datetime | None = None
+
+
+class ReviewProposalRequest(BaseModel):
+    """Review action on a proposal."""
+    action: str = Field(..., pattern="^(approve|reject|modify|dismiss)$")
+    reviewer_notes: str | None = None
+    modified_flow_data: dict[str, Any] | None = None  # Only for "modify"
+
+
+class FlowProposalStats(BaseModel):
+    """Dashboard stats for the review queue."""
+    pending_count: int
+    approved_this_week: int
+    rejected_this_week: int
+    auto_reinforced_this_week: int
+    top_domains: list[dict[str, Any]] = []  # [{domain, count}]

backend/app/schemas/flowpilot_analytics.py

@@ -0,0 +1,72 @@
+"""Pydantic schemas for FlowPilot analytics dashboard."""
+from __future__ import annotations
+
+from typing import Optional, Any
+from datetime import datetime
+
+from pydantic import BaseModel
+
+
+class MTTRDataPoint(BaseModel):
+    date: str
+    mttr_minutes: float
+    session_count: int
+
+
+class DomainBreakdown(BaseModel):
+    domain: str
+    total: int
+    resolved: int
+    escalated: int
+    resolution_rate: float
+
+
+class ConfidenceBreakdown(BaseModel):
+    guided_sessions: int
+    guided_resolution_rate: float
+    exploring_sessions: int
+    exploring_resolution_rate: float
+    discovery_sessions: int
+    discovery_resolution_rate: float
+
+
+class DomainCoverage(BaseModel):
+    domain: str
+    flow_count: int
+    session_count: int
+    guided_rate: float
+
+
+class KnowledgeCoverage(BaseModel):
+    total_flows: int
+    ai_generated_flows: int
+    total_proposals_pending: int
+    proposals_approved_this_period: int
+    proposals_rejected_this_period: int
+    coverage_by_domain: list[DomainCoverage] = []
+
+
+class PsaMetrics(BaseModel):
+    ticket_link_rate: float
+    auto_push_success_rate: float
+    auto_push_retry_success_rate: float
+    total_time_entries_logged: int
+    total_hours_logged: float
+
+
+class FlowPilotDashboard(BaseModel):
+    period: str
+    total_sessions: int
+    resolved_sessions: int
+    escalated_sessions: int
+    abandoned_sessions: int
+    resolution_rate: float
+    avg_steps_to_resolution: float
+    avg_session_duration_minutes: float
+    avg_rating: float | None = None
+    mttr_minutes: float | None = None
+    mttr_trend: list[MTTRDataPoint] = []
+    sessions_by_domain: list[DomainBreakdown] = []
+    confidence_breakdown: ConfidenceBreakdown
+    knowledge_coverage: KnowledgeCoverage
+    psa_metrics: PsaMetrics | None = None

backend/app/schemas/script_template.py

@@ -116,7 +116,8 @@ class ScriptTemplateDetail(ScriptTemplateListItem):
 class ScriptGenerateRequest(BaseModel):
     template_id: UUID
     parameters: dict[str, Any]
-    session_id: Optional[UUID] = None
+    session_id: Optional[UUID] = None  # Legacy tree-based session
+    ai_session_id: Optional[UUID] = None  # FlowPilot AI session
 
 class ScriptGenerateResponse(BaseModel):
     id: UUID

backend/app/services/flowpilot_engine.py

@@ -11,7 +11,7 @@ from datetime import datetime, timezone
 from typing import Any, Optional
 from uuid import UUID
 
-from sqlalchemy import select
+from sqlalchemy import select, or_
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
@@ -228,6 +228,11 @@ async def start_session(
     if ticket_context_block:
         ticket_prompt_section = f"\n## PSA TICKET CONTEXT\n{ticket_context_block}\n"
 
+    # Include available script templates for in-session script generation
+    script_context = await _build_script_context(team_id, db)
+    if script_context:
+        ticket_prompt_section += f"\n{script_context}\n"
+
     system_prompt = FLOWPILOT_SYSTEM_PROMPT.format(
         structured_output_schema=STRUCTURED_OUTPUT_SCHEMA,
         team_context=ticket_prompt_section,
@@ -448,6 +453,9 @@ async def resolve_session(
 
     documentation = _generate_documentation(session)
 
+    # Queue for Knowledge Flywheel analysis
+    session.analysis_status = "pending"
+
     await db.flush()
 
     # Push documentation to PSA if ticket is linked
@@ -909,6 +917,13 @@ def _create_step_from_parsed(
     if parsed["type"] == "action":
         content["action_type"] = parsed.get("action_type", "instruction")
         content["expected_outcome"] = parsed.get("expected_outcome")
+        # Script generation fields (populated when FlowPilot suggests a script)
+        if parsed.get("template_id"):
+            content["template_id"] = parsed["template_id"]
+        if parsed.get("pre_filled_params"):
+            content["pre_filled_params"] = parsed["pre_filled_params"]
+        if parsed.get("instructions"):
+            content["instructions"] = parsed["instructions"]
     elif parsed["type"] == "resolution_suggestion":
         content["resolution_summary"] = parsed.get("resolution_summary")
         content["follow_up_recommendations"] = parsed.get("follow_up_recommendations", [])
@@ -1066,6 +1081,51 @@ async def _process_ticket_intake(
         return None, None, "unavailable"
 
 
+async def _build_script_context(
+    team_id: Optional[UUID],
+    db: AsyncSession,
+) -> Optional[str]:
+    """Build script template context for the system prompt.
+
+    Includes available script templates so FlowPilot can suggest
+    script_generation actions with pre-filled parameters.
+    """
+    try:
+        from app.models.script_template import ScriptTemplate
+
+        result = await db.execute(
+            select(ScriptTemplate)
+            .where(
+                ScriptTemplate.is_active.is_(True),
+                or_(
+                    ScriptTemplate.team_id.is_(None),
+                    ScriptTemplate.team_id == team_id,
+                ),
+            )
+            .order_by(ScriptTemplate.usage_count.desc())
+            .limit(20)
+        )
+        templates = result.scalars().all()
+
+        if not templates:
+            return None
+
+        lines = ["## AVAILABLE SCRIPTS"]
+        lines.append("When the engineer needs to run a script, suggest an action with action_type='script_generation'.")
+        lines.append("Include template_id and pre_filled_params based on the diagnostic context.\n")
+        for t in templates:
+            params = t.parameters_schema.get("parameters", [])
+            param_keys = ", ".join(p.get("key", "") for p in params if p.get("key"))
+            lines.append(f"- {t.name} (ID: {t.id}): {t.description or 'No description'}")
+            if param_keys:
+                lines.append(f"  Parameters: {param_keys}")
+
+        return "\n".join(lines)
+    except Exception as e:
+        logger.warning("Failed to build script context: %s", e)
+        return None
+
+
 async def _build_escalation_package_enhanced(
     session: AISession,
     user_id: UUID,

backend/app/services/knowledge_flywheel.py

@@ -0,0 +1,454 @@
+"""Knowledge Flywheel — post-session analysis engine.
+
+Analyzes resolved AI sessions and generates flow proposals:
+- new_flow: Novel resolution path → propose a new troubleshooting flow
+- enhancement: Diverged from a matched flow → propose additions
+- auto_reinforced: Followed a flow exactly → update flow stats
+
+Called by the knowledge_flywheel_scheduler (APScheduler) after sessions resolve.
+"""
+import json
+import logging
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Optional
+from uuid import UUID
+
+from sqlalchemy import select, func
+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.models.flow_proposal import FlowProposal
+from app.models.tree import Tree
+
+logger = logging.getLogger(__name__)
+
+# Daily budget cap for proposal generation LLM calls per account
+MAX_PROPOSALS_PER_DAY = 50
+
+FLOW_GENERATION_PROMPT = """\
+You are a knowledge engineer converting a troubleshooting session into a reusable flow definition.
+
+Given the session transcript below, generate a JSON flow definition that captures the diagnostic logic so other engineers can follow the same path.
+
+## OUTPUT FORMAT
+Respond with ONLY valid JSON:
+{
+  "title": "Short descriptive title (5-10 words)",
+  "description": "When to use this flow (1-2 sentences)",
+  "match_keywords": ["keyword1", "keyword2", ...],
+  "problem_domain": "active_directory | networking | m365 | hardware | endpoint | virtualization | security | backup | email | printing | cloud | other",
+  "tree_structure": {
+    "id": "root",
+    "type": "decision",
+    "question": "First diagnostic question",
+    "help_text": "Context for the engineer",
+    "options": [
+      {"id": "opt1", "label": "Option text", "next_node_id": "node_id"}
+    ],
+    "children": [
+      {
+        "id": "node_id",
+        "type": "decision | action | solution",
+        "title": "Node title",
+        "question": "For decision nodes",
+        "description": "For action/solution nodes",
+        "options": [],
+        "next_node_id": "next_id or null for terminal nodes"
+      }
+    ]
+  }
+}
+
+## RULES
+- tree_structure uses a flat children array with id-based references via next_node_id
+- The root node has type "decision" with a question and options
+- Decision nodes have options with next_node_id pointing to child nodes
+- Action nodes describe what the engineer should do with a description field
+- Solution nodes describe the resolution (terminal — no next_node_id)
+- Every decision node must have 2-5 options
+- Include the key diagnostic questions that narrowed down the problem
+- Skip redundant or dead-end paths from the session
+- match_keywords should be symptoms, error messages, and technology names (5-10 keywords)
+- Do NOT wrap JSON in markdown code fences\
+"""
+
+ENHANCEMENT_PROMPT = """\
+You are a knowledge engineer analyzing how a troubleshooting session diverged from an existing flow.
+
+Given the session transcript and the existing flow structure, identify what should be added or changed.
+
+## OUTPUT FORMAT
+Respond with ONLY valid JSON:
+{
+  "title": "Enhancement: <what changed>",
+  "description": "Why this enhancement is needed",
+  "diff_description": "Human-readable summary of changes",
+  "new_nodes": [
+    {
+      "id": "new_node_id",
+      "type": "decision | action | solution",
+      "title": "Node title",
+      "question": "For decision nodes",
+      "description": "For action/solution nodes",
+      "options": [],
+      "attach_after_node_id": "existing node ID where this branches off",
+      "new_option_label": "Label for the new option on the parent node"
+    }
+  ],
+  "modified_options": [
+    {
+      "node_id": "existing node ID",
+      "add_option": {"id": "new_opt", "label": "New option text", "next_node_id": "new_node_id"}
+    }
+  ]
+}
+
+## RULES
+- Only propose changes supported by the session evidence
+- Minimize changes — add branches, don't restructure
+- new_nodes should follow the same format as the existing flow
+- Do NOT wrap JSON in markdown code fences\
+"""
+
+
+def _build_session_context(session: AISession) -> str:
+    """Build a text summary of a session for the LLM prompt."""
+    parts = [
+        f"Problem: {session.problem_summary or 'Unknown'}",
+        f"Domain: {session.problem_domain or 'Unknown'}",
+        f"Confidence at resolution: {session.confidence_tier} ({session.confidence_score:.0%})",
+        f"Resolution: {session.resolution_summary or 'No summary'}",
+    ]
+
+    if session.escalation_reason:
+        parts.append(f"Escalation reason: {session.escalation_reason}")
+
+    # Build step-by-step diagnostic trail
+    steps = sorted(session.steps, key=lambda s: s.step_order)
+    if steps:
+        parts.append("\n--- DIAGNOSTIC TRAIL ---")
+        for step in steps:
+            content = step.content or {}
+            step_desc = content.get("text", "")
+            step_type = content.get("type", step.step_type)
+
+            line = f"Step {step.step_order + 1} [{step_type}]: {step_desc}"
+
+            # Engineer response
+            if step.was_skipped:
+                line += "\n  → Skipped"
+            elif step.selected_option:
+                # Find label from options
+                label = step.selected_option
+                if step.options_presented:
+                    for opt in step.options_presented:
+                        if opt.get("value") == step.selected_option:
+                            label = opt.get("label", step.selected_option)
+                            break
+                line += f"\n  → Selected: {label}"
+            elif step.free_text_input:
+                line += f"\n  → Free text: {step.free_text_input}"
+
+            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}"
+                line += f"\n  → Result: {outcome}"
+
+            parts.append(line)
+
+    return "\n".join(parts)
+
+
+def _has_free_text_escapes(session: AISession) -> bool:
+    """Check if the session used free-text escapes (diverged from options)."""
+    return any(step.was_free_text for step in session.steps)
+
+
+async def _check_daily_budget(account_id: UUID, db: AsyncSession) -> bool:
+    """Check if the account has exceeded the daily proposal generation budget."""
+    today_start = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
+    result = await db.execute(
+        select(func.count(FlowProposal.id))
+        .where(
+            FlowProposal.account_id == account_id,
+            FlowProposal.created_at >= today_start,
+            FlowProposal.status != "auto_reinforced",  # Don't count no-LLM proposals
+        )
+    )
+    count = result.scalar() or 0
+    return count < MAX_PROPOSALS_PER_DAY
+
+
+async def _find_similar_pending_proposal(
+    title: str,
+    problem_domain: Optional[str],
+    account_id: UUID,
+    db: AsyncSession,
+) -> Optional[FlowProposal]:
+    """Find an existing pending proposal with similar title and domain.
+
+    Uses simple keyword overlap for now. Phase 4 will add embedding similarity.
+    """
+    # Build domain filter — match NULL domain proposals if domain is NULL
+    domain_filter = (
+        FlowProposal.problem_domain == problem_domain
+        if problem_domain
+        else FlowProposal.problem_domain.is_(None)
+    )
+
+    result = await db.execute(
+        select(FlowProposal)
+        .where(
+            FlowProposal.account_id == account_id,
+            FlowProposal.status == "pending",
+            domain_filter,
+        )
+        .limit(20)
+    )
+    candidates = result.scalars().all()
+
+    if not candidates:
+        return None
+
+    # Simple keyword overlap check
+    title_words = set(title.lower().split())
+    for candidate in candidates:
+        candidate_words = set(candidate.title.lower().split())
+        if len(title_words) > 0 and len(candidate_words) > 0:
+            overlap = len(title_words & candidate_words) / max(len(title_words), len(candidate_words))
+            if overlap > 0.6:
+                return candidate
+
+    return None
+
+
+async def analyze_session(session: AISession, db: AsyncSession) -> None:
+    """Analyze a resolved session and create appropriate flow proposal.
+
+    Dispatches to one of three outcomes:
+    1. new_flow — novel resolution, no matching flow
+    2. enhancement — matched flow but diverged
+    3. auto_reinforced — followed existing flow closely
+    """
+    # Re-fetch with eager-loaded steps to avoid async lazy-load errors
+    result = await db.execute(
+        select(AISession)
+        .where(AISession.id == session.id)
+        .options(selectinload(AISession.steps))
+    )
+    session = result.scalar_one()
+
+    # Determine which analysis path to take
+    has_match = session.matched_flow_id is not None
+    match_score = session.match_score or 0.0
+    has_divergence = _has_free_text_escapes(session)
+
+    if has_match and match_score > 0.8 and not has_divergence:
+        # Path 3: Auto-reinforcement
+        await _auto_reinforce(session, db)
+    elif has_match and match_score > 0.5 and has_divergence:
+        # Path 2: Enhancement proposal
+        await _propose_enhancement(session, db)
+    elif not has_match or match_score < 0.5:
+        # Path 1: New flow proposal
+        await _propose_new_flow(session, db)
+    else:
+        # Edge case: matched but moderate score, no divergence — reinforce
+        await _auto_reinforce(session, db)
+
+
+async def _auto_reinforce(session: AISession, db: AsyncSession) -> None:
+    """Update the matched flow's stats and create a tracking record."""
+    if session.matched_flow_id:
+        result = await db.execute(
+            select(Tree).where(Tree.id == session.matched_flow_id)
+        )
+        flow = result.scalar_one_or_none()
+        if flow:
+            # Update flow stats
+            current_rate = flow.success_rate or 0.0
+            # Simple moving average
+            flow.success_rate = round(current_rate * 0.9 + 1.0 * 0.1, 4)
+            flow.last_matched_at = datetime.now(timezone.utc)
+
+    # Create tracking record (no review needed)
+    proposal = FlowProposal(
+        id=uuid.uuid4(),
+        account_id=session.account_id,
+        team_id=session.team_id,
+        source_session_id=session.id,
+        proposal_type="auto_reinforced",
+        title=f"Reinforcement: {session.problem_summary or 'Session'}",
+        description="Session followed existing flow closely. No changes needed.",
+        proposed_flow_data={},
+        confidence_score=session.confidence_score,
+        supporting_session_ids=[str(session.id)],
+        problem_domain=session.problem_domain,
+        status="auto_reinforced",
+        target_flow_id=session.matched_flow_id,
+    )
+    db.add(proposal)
+    logger.info("Auto-reinforced flow %s from session %s", session.matched_flow_id, session.id)
+
+
+async def _propose_new_flow(session: AISession, db: AsyncSession) -> None:
+    """Generate a new flow proposal from a novel session."""
+    if not await _check_daily_budget(session.account_id, db):
+        logger.warning("Daily proposal budget exceeded for account %s", session.account_id)
+        return
+
+    session_context = _build_session_context(session)
+
+    try:
+        provider = get_ai_provider(settings.get_model_for_action("open_chat"))
+        raw_response, _, _ = await provider.generate_json(
+            system_prompt=FLOW_GENERATION_PROMPT,
+            messages=[{"role": "user", "content": session_context}],
+            max_tokens=4096,
+        )
+
+        parsed = _parse_llm_json(raw_response)
+    except Exception as e:
+        logger.warning("Knowledge Flywheel LLM call failed for session %s: %s", session.id, e)
+        return
+
+    title = parsed.get("title", session.problem_summary or "Untitled Flow")
+    domain = parsed.get("problem_domain", session.problem_domain)
+
+    # Check for similar pending proposals
+    existing = await _find_similar_pending_proposal(title, domain, session.account_id, db)
+    if existing:
+        # Merge into existing proposal
+        existing.supporting_session_count += 1
+        sids = existing.supporting_session_ids or []
+        sids.append(str(session.id))
+        existing.supporting_session_ids = sids
+        existing.confidence_score = min(1.0, existing.confidence_score + 0.1)
+        logger.info(
+            "Merged session %s into existing proposal %s (now %d supporting)",
+            session.id, existing.id, existing.supporting_session_count,
+        )
+        return
+
+    proposal = FlowProposal(
+        id=uuid.uuid4(),
+        account_id=session.account_id,
+        team_id=session.team_id,
+        source_session_id=session.id,
+        proposal_type="new_flow",
+        title=title,
+        description=parsed.get("description"),
+        proposed_flow_data={
+            "tree_structure": parsed.get("tree_structure", {}),
+            "match_keywords": parsed.get("match_keywords", []),
+        },
+        confidence_score=session.confidence_score,
+        supporting_session_ids=[str(session.id)],
+        problem_domain=domain,
+        status="pending",
+    )
+    db.add(proposal)
+    logger.info("Created new_flow proposal for session %s: %s", session.id, title)
+
+
+async def _propose_enhancement(session: AISession, db: AsyncSession) -> None:
+    """Generate an enhancement proposal for an existing flow."""
+    if not session.matched_flow_id:
+        # Fallback to new flow if no match
+        await _propose_new_flow(session, db)
+        return
+
+    if not await _check_daily_budget(session.account_id, db):
+        logger.warning("Daily proposal budget exceeded for account %s", session.account_id)
+        return
+
+    # Load the matched flow
+    result = await db.execute(
+        select(Tree).where(Tree.id == session.matched_flow_id)
+    )
+    matched_flow = result.scalar_one_or_none()
+    if not matched_flow:
+        await _propose_new_flow(session, db)
+        return
+
+    session_context = _build_session_context(session)
+    flow_json = json.dumps(matched_flow.tree_structure, indent=None)
+    if len(flow_json) > 4000:
+        flow_json = flow_json[:4000] + "... [truncated]"
+
+    prompt_content = (
+        f"## EXISTING FLOW\n"
+        f"Name: {matched_flow.name}\n"
+        f"Structure:\n{flow_json}\n\n"
+        f"## SESSION THAT DIVERGED\n"
+        f"{session_context}"
+    )
+
+    try:
+        provider = get_ai_provider(settings.get_model_for_action("open_chat"))
+        raw_response, _, _ = await provider.generate_json(
+            system_prompt=ENHANCEMENT_PROMPT,
+            messages=[{"role": "user", "content": prompt_content}],
+            max_tokens=4096,
+        )
+
+        parsed = _parse_llm_json(raw_response)
+    except Exception as e:
+        logger.warning("Knowledge Flywheel enhancement LLM call failed for session %s: %s", session.id, e)
+        return
+
+    title = parsed.get("title", f"Enhancement: {session.problem_summary or 'Flow update'}")
+    diff_description = parsed.get("diff_description", "Session diverged from existing flow")
+
+    proposal = FlowProposal(
+        id=uuid.uuid4(),
+        account_id=session.account_id,
+        team_id=session.team_id,
+        source_session_id=session.id,
+        proposal_type="enhancement",
+        target_flow_id=session.matched_flow_id,
+        title=title,
+        description=diff_description,
+        proposed_flow_data={
+            "new_nodes": parsed.get("new_nodes", []),
+            "modified_options": parsed.get("modified_options", []),
+        },
+        proposed_diff={
+            "diff_description": diff_description,
+            "new_nodes": parsed.get("new_nodes", []),
+            "modified_options": parsed.get("modified_options", []),
+        },
+        confidence_score=session.confidence_score,
+        supporting_session_ids=[str(session.id)],
+        problem_domain=session.problem_domain,
+        status="pending",
+    )
+    db.add(proposal)
+    logger.info(
+        "Created enhancement proposal for flow %s from session %s: %s",
+        session.matched_flow_id, session.id, title,
+    )
+
+
+def _parse_llm_json(raw_text: str) -> dict[str, Any]:
+    """Parse JSON from LLM response, handling common quirks."""
+    text = raw_text.strip()
+
+    # Strip markdown code fences if present
+    if text.startswith("```"):
+        lines = text.split("\n")
+        lines = [line for line in lines if not line.strip().startswith("```")]
+        text = "\n".join(lines).strip()
+
+    try:
+        return json.loads(text)
+    except json.JSONDecodeError as e:
+        logger.warning("Knowledge Flywheel JSON parse failed: %s — raw: %.300s", e, text)
+        raise ValueError(f"Invalid JSON from LLM: {e}") from e

backend/app/services/knowledge_flywheel_scheduler.py

@@ -0,0 +1,72 @@
+"""Background scheduler for Knowledge Flywheel analysis.
+
+Runs every 5 minutes via APScheduler, picks up AISession entries
+with analysis_status='pending' and runs flow proposal analysis.
+
+Each session is committed individually to prevent a single failure
+from rolling back all progress or causing duplicate proposals.
+"""
+import logging
+
+from sqlalchemy import select
+
+from app.core.database import async_session_maker
+from app.models.ai_session import AISession
+from app.services.knowledge_flywheel import analyze_session
+
+logger = logging.getLogger(__name__)
+
+
+async def process_pending_analyses() -> None:
+    """Process resolved sessions awaiting Knowledge Flywheel analysis."""
+    async with async_session_maker() as db:
+        try:
+            result = await db.execute(
+                select(AISession.id)
+                .where(AISession.analysis_status == "pending")
+                .order_by(AISession.resolved_at.asc())
+                .limit(10)
+            )
+            session_ids = [row[0] for row in result.all()]
+        except Exception as e:
+            logger.error("Knowledge Flywheel scheduler query error: %s", e)
+            return
+
+        if not session_ids:
+            return
+
+        logger.info("Processing %d pending Knowledge Flywheel analyses", len(session_ids))
+
+    # Process each session in its own DB session to isolate failures
+    for session_id in session_ids:
+        async with async_session_maker() as db:
+            try:
+                result = await db.execute(
+                    select(AISession).where(AISession.id == session_id)
+                )
+                session = result.scalar_one_or_none()
+                if not session or session.analysis_status != "pending":
+                    continue
+
+                await analyze_session(session, db)
+                session.analysis_status = "completed"
+                await db.commit()
+                logger.info("Knowledge Flywheel completed for session %s", session_id)
+            except Exception as e:
+                await db.rollback()
+                logger.warning(
+                    "Knowledge Flywheel failed for session %s: %s",
+                    session_id, e,
+                )
+                # Mark as failed in a separate transaction
+                try:
+                    async with async_session_maker() as db2:
+                        result = await db2.execute(
+                            select(AISession).where(AISession.id == session_id)
+                        )
+                        s = result.scalar_one_or_none()
+                        if s:
+                            s.analysis_status = "failed"
+                            await db2.commit()
+                except Exception:
+                    logger.error("Failed to mark session %s as failed", session_id)

backend/app/services/knowledge_gap_service.py

@@ -0,0 +1,334 @@
+"""Knowledge Gap Detection Service.
+
+Aggregates signals from AI sessions to identify gaps in the knowledge base.
+Results are served by the analytics API and cached for 1 hour.
+
+Signals:
+1. Frequent free-text escapes — FlowPilot's options didn't cover a common scenario
+2. High escalation rate by domain — domains where engineers can't self-resolve
+3. Discovery-mode resolutions — novel problems solved without flow guidance
+4. Repeated unmatched patterns — keyword-frequency based (Phase 4: embedding clustering)
+"""
+import logging
+from collections import Counter
+from datetime import datetime, timezone, timedelta
+from typing import Any, Optional
+from uuid import UUID
+
+from pydantic import BaseModel
+from sqlalchemy import select, func, case, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.ai_session import AISession
+from app.models.ai_session_step import AISessionStep
+from app.models.tree import Tree
+
+logger = logging.getLogger(__name__)
+
+# Cache for expensive gap analysis
+_cache: dict[str, Any] = {}
+_cache_expiry: dict[str, datetime] = {}
+CACHE_TTL = timedelta(hours=1)
+
+
+class KnowledgeGap(BaseModel):
+    gap_type: str  # "weak_options" | "high_escalation" | "uncharted_territory" | "repeated_pattern"
+    domain: str | None = None
+    severity: str  # "high" | "medium" | "low"
+    title: str
+    description: str
+    evidence: dict[str, Any] = {}
+    suggested_action: str
+
+
+class KnowledgeGapReport(BaseModel):
+    generated_at: datetime
+    gaps: list[KnowledgeGap]
+
+
+async def get_knowledge_gaps(
+    account_id: UUID,
+    db: AsyncSession,
+    period_days: int = 30,
+) -> KnowledgeGapReport:
+    """Generate a knowledge gap report for the account.
+
+    Results are cached for 1 hour per account.
+    """
+    cache_key = f"gaps:{account_id}:{period_days}"
+    now = datetime.now(timezone.utc)
+
+    if cache_key in _cache and _cache_expiry.get(cache_key, now) > now:
+        return _cache[cache_key]
+
+    period_start = now - timedelta(days=period_days)
+
+    gaps: list[KnowledgeGap] = []
+
+    # Signal 1: Frequent free-text escapes
+    signal1 = await _detect_weak_options(account_id, period_start, db)
+    gaps.extend(signal1)
+
+    # Signal 2: High escalation rate by domain
+    signal2 = await _detect_high_escalation(account_id, period_start, db)
+    gaps.extend(signal2)
+
+    # Signal 3: Discovery-mode resolutions
+    signal3 = await _detect_uncharted_territory(account_id, period_start, db)
+    gaps.extend(signal3)
+
+    # Signal 4: Repeated unmatched patterns (keyword-based for Phase 3)
+    signal4 = await _detect_repeated_patterns(account_id, period_start, db)
+    gaps.extend(signal4)
+
+    # Sort by severity (high > medium > low)
+    severity_order = {"high": 0, "medium": 1, "low": 2}
+    gaps.sort(key=lambda g: severity_order.get(g.severity, 3))
+
+    report = KnowledgeGapReport(generated_at=now, gaps=gaps)
+
+    _cache[cache_key] = report
+    _cache_expiry[cache_key] = now + CACHE_TTL
+
+    return report
+
+
+async def _detect_weak_options(
+    account_id: UUID,
+    period_start: datetime,
+    db: AsyncSession,
+) -> list[KnowledgeGap]:
+    """Signal 1: Find questions where engineers frequently use free-text escapes."""
+    # Count free-text usage per step context_message (the question asked)
+    result = await db.execute(
+        select(
+            AISessionStep.context_message,
+            func.count(AISessionStep.id).label("total"),
+            func.sum(case((AISessionStep.was_free_text.is_(True), 1), else_=0)).label("free_text_count"),
+        )
+        .join(AISession, AISessionStep.session_id == AISession.id)
+        .where(
+            AISession.account_id == account_id,
+            AISession.created_at >= period_start,
+            AISessionStep.step_type == "question",
+            AISessionStep.context_message.isnot(None),
+            AISessionStep.responded_at.isnot(None),
+        )
+        .group_by(AISessionStep.context_message)
+        .having(func.count(AISessionStep.id) >= 3)  # Minimum sample size
+        .order_by(func.sum(case((AISessionStep.was_free_text.is_(True), 1), else_=0)).desc())
+        .limit(5)
+    )
+
+    gaps = []
+    for row in result.all():
+        context_msg, total_raw, free_text_raw = row
+        total = int(total_raw or 0)
+        free_text_count = int(free_text_raw or 0)
+        if total == 0 or not free_text_count:
+            continue
+        rate = free_text_count / total
+        if rate < 0.3:
+            continue
+
+        severity = "high" if rate > 0.6 else "medium"
+        gaps.append(KnowledgeGap(
+            gap_type="weak_options",
+            severity=severity,
+            title=f"Weak options: {(context_msg or '')[:80]}",
+            description=(
+                f"Engineers used free-text input {free_text_count}/{total} times "
+                f"({rate:.0%}) when asked this question. The predefined options "
+                f"may not cover common scenarios."
+            ),
+            evidence={
+                "context_message": context_msg,
+                "total_responses": total,
+                "free_text_count": free_text_count,
+                "free_text_rate": round(rate, 3),
+            },
+            suggested_action="Review the free-text responses and add common answers as options.",
+        ))
+
+    return gaps
+
+
+async def _detect_high_escalation(
+    account_id: UUID,
+    period_start: datetime,
+    db: AsyncSession,
+) -> list[KnowledgeGap]:
+    """Signal 2: Find domains with >40% escalation rate."""
+    result = await db.execute(
+        select(
+            AISession.problem_domain,
+            func.count(AISession.id).label("total"),
+            func.sum(case(
+                (AISession.status == "resolved", 1), else_=0
+            )).label("resolved"),
+            func.sum(case(
+                (AISession.status.in_(["escalated", "requesting_escalation"]), 1), else_=0
+            )).label("escalated"),
+        )
+        .where(
+            AISession.account_id == account_id,
+            AISession.created_at >= period_start,
+            AISession.problem_domain.isnot(None),
+            AISession.status.in_(["resolved", "escalated", "requesting_escalation"]),
+        )
+        .group_by(AISession.problem_domain)
+        .having(func.count(AISession.id) >= 3)  # Minimum sample
+    )
+
+    gaps = []
+    for row in result.all():
+        domain, total_raw, resolved_raw, escalated_raw = row
+        total = int(total_raw or 0)
+        resolved = int(resolved_raw or 0)
+        escalated = int(escalated_raw or 0)
+        if total == 0 or not escalated:
+            continue
+        escalation_rate = escalated / total
+        if escalation_rate < 0.4:
+            continue
+
+        severity = "high" if escalation_rate > 0.6 else "medium"
+        gaps.append(KnowledgeGap(
+            gap_type="high_escalation",
+            domain=domain,
+            severity=severity,
+            title=f"High escalation rate in {domain}",
+            description=(
+                f"{escalated}/{total} sessions ({escalation_rate:.0%}) in {domain} "
+                f"were escalated. Only {resolved} resolved independently."
+            ),
+            evidence={
+                "domain": domain,
+                "total": total,
+                "resolved": resolved,
+                "escalated": escalated,
+                "escalation_rate": round(escalation_rate, 3),
+            },
+            suggested_action=f"Create or improve troubleshooting flows for {domain} issues.",
+        ))
+
+    return gaps
+
+
+async def _detect_uncharted_territory(
+    account_id: UUID,
+    period_start: datetime,
+    db: AsyncSession,
+) -> list[KnowledgeGap]:
+    """Signal 3: Find discovery-mode resolutions (novel problems solved without flows)."""
+    result = await db.execute(
+        select(
+            AISession.problem_domain,
+            func.count(AISession.id).label("count"),
+        )
+        .where(
+            AISession.account_id == account_id,
+            AISession.created_at >= period_start,
+            AISession.status == "resolved",
+            AISession.confidence_tier == "discovery",
+        )
+        .group_by(AISession.problem_domain)
+        .having(func.count(AISession.id) >= 2)
+        .order_by(func.count(AISession.id).desc())
+        .limit(5)
+    )
+
+    gaps = []
+    for row in result.all():
+        domain, count = row
+        severity = "high" if count >= 5 else "medium" if count >= 3 else "low"
+        domain_label = domain or "unknown domain"
+        gaps.append(KnowledgeGap(
+            gap_type="uncharted_territory",
+            domain=domain,
+            severity=severity,
+            title=f"Novel resolutions in {domain_label}",
+            description=(
+                f"{count} sessions in {domain_label} were resolved in discovery mode "
+                f"(no matching flow, low confidence). These represent knowledge capture "
+                f"opportunities — check the Review Queue for auto-generated proposals."
+            ),
+            evidence={
+                "domain": domain,
+                "discovery_resolution_count": count,
+            },
+            suggested_action="Review pending flow proposals or create flows from these session patterns.",
+        ))
+
+    return gaps
+
+
+async def _detect_repeated_patterns(
+    account_id: UUID,
+    period_start: datetime,
+    db: AsyncSession,
+) -> list[KnowledgeGap]:
+    """Signal 4: Find repeated unmatched intake patterns (keyword-frequency based).
+
+    Phase 3 uses keyword frequency on problem_summary. Phase 4 will use
+    embedding clustering for deeper semantic analysis.
+    """
+    # Get problem summaries from unmatched sessions
+    result = await db.execute(
+        select(AISession.problem_summary, AISession.problem_domain)
+        .where(
+            AISession.account_id == account_id,
+            AISession.created_at >= period_start,
+            AISession.problem_summary.isnot(None),
+            AISession.matched_flow_id.is_(None),
+        )
+        .limit(200)
+    )
+    rows = result.all()
+
+    if len(rows) < 3:
+        return []
+
+    # Extract keywords from summaries and count frequency
+    word_counts: Counter[str] = Counter()
+    domain_for_word: dict[str, str | None] = {}
+    for summary, domain in rows:
+        if not summary:
+            continue
+        words = set(summary.lower().split())
+        # Filter out common stop words and short words
+        stop_words = {"the", "a", "an", "is", "are", "was", "were", "in", "on", "at",
+                       "to", "for", "of", "and", "or", "not", "can", "can't", "with",
+                       "from", "by", "this", "that", "it", "its", "has", "have", "had",
+                       "user", "users", "issue", "error", "problem"}
+        keywords = {w for w in words if len(w) > 3 and w not in stop_words}
+        for kw in keywords:
+            word_counts[kw] += 1
+            if kw not in domain_for_word:
+                domain_for_word[kw] = domain
+
+    gaps = []
+    # Find keywords that appear in many unmatched sessions
+    for keyword, count in word_counts.most_common(3):
+        if count < 3:
+            continue
+        severity = "medium" if count >= 5 else "low"
+        domain = domain_for_word.get(keyword)
+        gaps.append(KnowledgeGap(
+            gap_type="repeated_pattern",
+            domain=domain,
+            severity=severity,
+            title=f"Recurring unmatched pattern: '{keyword}'",
+            description=(
+                f"The keyword '{keyword}' appeared in {count} sessions that had no "
+                f"matching flow. This may indicate a systematic knowledge gap."
+            ),
+            evidence={
+                "keyword": keyword,
+                "unmatched_session_count": count,
+                "domain": domain,
+            },
+            suggested_action=f"Search for '{keyword}' in recent sessions and consider creating a flow.",
+        ))
+
+    return gaps