fix(pilot): reorder Phase 3 useCallbacks to avoid TDZ on render

refreshSessionDerived's dep array referenced refreshActiveFix and
schedulePreviewRefresh before they were declared. React evaluates
useCallback deps synchronously during render, so the page blew up with
"Cannot access 'refreshActiveFix' before initialization" before a single
render completed. Moved the three leaf helpers above the aggregator.

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

backend/app/api/endpoints/scripts.py

@@ -5,7 +5,7 @@ import re
 
 from fastapi import APIRouter, Depends, HTTPException, Query, status
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy import select, func, or_, literal
+from sqlalchemy import select, func, or_, literal, update as sa_update
 
 from app.core.database import get_db
 from app.api.deps import get_current_active_user
@@ -374,6 +374,20 @@ async def generate_script(
     )
     db.add(generation)
     template.usage_count += 1
+
+    # FlowPilot Phase 3: bump the linked AI session's state_version so the
+    # resolution-note preview cache invalidates. One-off scripts run outside
+    # any FlowPilot session — in that case the UPDATE matches zero rows.
+    if data.ai_session_id is not None:
+        # Local import: scripts endpoint stays independent of AI-session
+        # imports for non-AI generation paths.
+        from app.models.ai_session import AISession
+        await db.execute(
+            sa_update(AISession)
+            .where(AISession.id == data.ai_session_id)
+            .values(state_version=AISession.state_version + 1)
+        )
+
     await db.commit()
     await db.refresh(generation)
 

backend/app/api/endpoints/session_facts.py

@@ -0,0 +1,315 @@
+"""Session fact endpoints — the "What we know" CRUD surface for a FlowPilot session.
+
+All routes are sub-resources of `/ai-sessions/{session_id}`. Tenant isolation is
+enforced by RLS on `session_facts.account_id`; a user from another account
+literally cannot see or write facts for this session.
+
+Editability rule (per FLOWPILOT-MIGRATION.md Section 7.3):
+- `user_note` and `ai_synthesis` facts are editable at the card level.
+- `question` and `diagnostic_check` facts are read-only at the card level —
+  edit the source question/check instead. PATCH returns 403 for those.
+
+Fact promotion writes always bump `ai_sessions.state_version` so the
+resolution-note preview cache invalidates (Section 5.5).
+"""
+import logging
+from datetime import datetime, timezone
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.deps import get_current_active_user, get_db, require_engineer_or_admin
+from app.models.ai_session import AISession
+from app.models.session_fact import SessionFact
+from app.models.user import User
+from app.schemas.session_fact import (
+    SessionFactCreateRequest,
+    SessionFactListResponse,
+    SessionFactPromoteRequest,
+    SessionFactResponse,
+    SessionFactUpdateRequest,
+)
+from app.services.fact_synthesis_service import (
+    FactSynthesisService,
+    list_facts_for_session,
+)
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/ai-sessions/{session_id}", tags=["session-facts"])
+
+# Source types whose facts can be edited at the card level (Section 7.3).
+_EDITABLE_SOURCE_TYPES = frozenset({"user_note", "ai_synthesis"})
+
+
+def _to_response(fact: SessionFact) -> SessionFactResponse:
+    """Wrap an ORM SessionFact in the response model with the editable flag."""
+    return SessionFactResponse(
+        id=fact.id,
+        session_id=fact.session_id,
+        text=fact.text,
+        source_type=fact.source_type,  # type: ignore[arg-type]
+        source_ref=fact.source_ref,
+        source_summary=fact.source_summary,
+        created_by=fact.created_by,
+        created_at=fact.created_at,
+        updated_at=fact.updated_at,
+        editable=fact.source_type in _EDITABLE_SOURCE_TYPES,
+    )
+
+
+async def _load_session_or_404(db: AsyncSession, session_id: UUID) -> AISession:
+    """Load the session via RLS-scoped SELECT. Returns 404 if missing/cross-tenant.
+
+    Tenant isolation: RLS on `ai_sessions` filters by current account, so a
+    cross-tenant access returns no rows and we 404 (rather than 403, which
+    would leak the row's existence).
+    """
+    result = await db.execute(select(AISession).where(AISession.id == session_id))
+    session = result.scalar_one_or_none()
+    if session is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Session not found")
+    return session
+
+
+async def _load_fact_or_404(
+    db: AsyncSession, session_id: UUID, fact_id: UUID
+) -> SessionFact:
+    """Load a non-deleted fact for the session. 404 if missing or already deleted."""
+    result = await db.execute(
+        select(SessionFact).where(
+            SessionFact.id == fact_id,
+            SessionFact.session_id == session_id,
+            SessionFact.deleted_at.is_(None),
+        )
+    )
+    fact = result.scalar_one_or_none()
+    if fact is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Fact not found")
+    return fact
+
+
+# ── List ──
+
+@router.get("/facts", response_model=SessionFactListResponse)
+async def list_facts(
+    session_id: UUID,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+) -> SessionFactListResponse:
+    """List facts for a session, oldest first."""
+    await _load_session_or_404(db, session_id)
+    facts = await list_facts_for_session(db, session_id)
+    return SessionFactListResponse(facts=[_to_response(f) for f in facts])
+
+
+# ── Create (manual user note) ──
+
+@router.post("/facts", response_model=SessionFactResponse, status_code=201)
+async def create_fact(
+    session_id: UUID,
+    body: SessionFactCreateRequest,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+) -> SessionFactResponse:
+    """Create a manual fact (the "+ Add a note" UI affordance).
+
+    Always recorded as `source_type=user_note`. Source-typed creation goes
+    through `/facts/promote` so the originating item ID is captured.
+    """
+    session = await _load_session_or_404(db, session_id)
+    service = FactSynthesisService(db)
+    try:
+        fact = await service.create_fact(
+            session_id=session.id,
+            account_id=session.account_id,
+            user_id=current_user.id,
+            source_type="user_note",
+            text=body.text,
+            summary=body.summary,
+            source_ref=None,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
+    await db.commit()
+    await db.refresh(fact)
+    return _to_response(fact)
+
+
+# ── Update ──
+
+@router.patch("/facts/{fact_id}", response_model=SessionFactResponse)
+async def update_fact(
+    session_id: UUID,
+    fact_id: UUID,
+    body: SessionFactUpdateRequest,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+) -> SessionFactResponse:
+    """Edit fact text or summary.
+
+    Returns 403 for `question` and `diagnostic_check`-sourced facts: the
+    source item is the canonical input, so editing the fact card would
+    desync the two. Engineers edit the source instead.
+    """
+    fact = await _load_fact_or_404(db, session_id, fact_id)
+    if fact.source_type not in _EDITABLE_SOURCE_TYPES:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=(
+                f"Facts sourced from {fact.source_type!r} are read-only at the "
+                "card level. Edit the originating question or diagnostic check instead."
+            ),
+        )
+
+    if body.text is None and body.summary is None:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="At least one of `text` or `summary` must be provided",
+        )
+
+    service = FactSynthesisService(db)
+    try:
+        fact = await service.update_fact(fact, text=body.text, summary=body.summary)
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
+    await db.commit()
+    await db.refresh(fact)
+    return _to_response(fact)
+
+
+# ── Soft delete ──
+
+@router.delete("/facts/{fact_id}", status_code=204)
+async def delete_fact(
+    session_id: UUID,
+    fact_id: UUID,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+) -> None:
+    """Soft-delete a fact. All source types are deletable.
+
+    Soft delete (rather than hard) preserves provenance for audit and lets
+    accidental deletes be recovered if needed. The `editable` flag does NOT
+    control deletion — even read-only facts can be removed when the
+    underlying question/check turned out to be wrong.
+    """
+    fact = await _load_fact_or_404(db, session_id, fact_id)
+    service = FactSynthesisService(db)
+    await service.soft_delete_fact(fact)
+    await db.commit()
+
+
+# ── Promote (AI marker + engineer-driven) ──
+
+@router.post("/facts/promote", response_model=SessionFactResponse, status_code=201)
+async def promote_fact(
+    session_id: UUID,
+    body: SessionFactPromoteRequest,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+) -> SessionFactResponse:
+    """Convert a question answer / check result into a fact.
+
+    Two modes:
+
+    - `proposed_text` provided → persisted as-is.
+    - `raw_input` provided → server drafts text/summary via FactSynthesisService.
+
+    Exactly one of the two must be set. The engineer-facing UI typically uses
+    `proposed_text` after letting the engineer review/edit a draft.
+    """
+    if (body.proposed_text is None) == (body.raw_input is None):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Exactly one of `proposed_text` or `raw_input` must be provided",
+        )
+    if body.source_type == "ai_synthesis" and body.source_ref is not None:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="`source_ref` must be null for source_type=ai_synthesis",
+        )
+
+    session = await _load_session_or_404(db, session_id)
+    service = FactSynthesisService(db)
+
+    text = body.proposed_text
+    summary = body.proposed_summary
+    if text is None:
+        # Synthesize via LLM. Caller must hint which task-lane item the input
+        # came from so we can shape the prompt appropriately.
+        raw = body.raw_input or ""
+        if body.source_type == "question":
+            draft = await service.synthesize_from_question(
+                question_text=_lookup_task_lane_text(session, body.source_ref, "questions"),
+                raw_answer=raw,
+            )
+        elif body.source_type == "diagnostic_check":
+            draft = await service.synthesize_from_check(
+                check_label=_lookup_task_lane_text(session, body.source_ref, "actions"),
+                check_output=raw,
+            )
+        else:
+            # ai_synthesis with raw_input: the raw input IS the synthesis.
+            # Re-run through the question synthesizer with an empty question
+            # so the conservative prompt still applies.
+            draft = await service.synthesize_from_question(
+                question_text="(none — synthesizing from engineer summary)",
+                raw_answer=raw,
+            )
+        text = draft["text"]
+        summary = summary or draft["summary"]
+        if not text:
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+                detail=(
+                    "Synthesizer found no substantive fact in the input. "
+                    "Edit the input or supply `proposed_text` directly."
+                ),
+            )
+
+    try:
+        fact = await service.create_fact(
+            session_id=session.id,
+            account_id=session.account_id,
+            user_id=current_user.id,
+            source_type=body.source_type,
+            text=text,
+            summary=summary,
+            source_ref=body.source_ref,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
+
+    await db.commit()
+    await db.refresh(fact)
+    return _to_response(fact)
+
+
+def _lookup_task_lane_text(
+    session: AISession, source_ref: UUID | None, list_key: str
+) -> str:
+    """Find the originating question text / action label from pending_task_lane.
+
+    Falls back to a generic placeholder if the source item is no longer in
+    the lane (e.g., the AI dropped it from a later turn). The synthesizer is
+    forgiving — an empty/generic question still produces a useful fact when
+    the engineer's answer is substantive on its own.
+    """
+    if source_ref is None:
+        return ""
+    lane = session.pending_task_lane or {}
+    items = lane.get(list_key) or []
+    sref = str(source_ref)
+    for item in items:
+        if isinstance(item, dict) and str(item.get("id")) == sref:
+            return str(item.get("text") or item.get("label") or "")
+    return ""

backend/app/api/endpoints/session_suggested_fixes.py

@@ -0,0 +1,183 @@
+"""Suggested-fix and resolution-note preview endpoints (Phase 3).
+
+Per FLOWPILOT-MIGRATION.md Sections 5.2 + 5.4. The preview is keyed on
+`(session_id, ai_sessions.state_version)` so repeat fetches against the same
+state hit the in-process cache instead of paying for a Sonnet call.
+"""
+import logging
+from datetime import datetime, timezone
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.deps import get_current_active_user, get_db, require_engineer_or_admin
+from app.models.ai_session import AISession
+from app.models.session_suggested_fix import SessionSuggestedFix
+from app.models.user import User
+from app.schemas.session_suggested_fix import (
+    ResolutionNotePreviewResponse,
+    SessionSuggestedFixDecisionRequest,
+    SessionSuggestedFixDecisionResponse,
+    SessionSuggestedFixResponse,
+)
+from app.services.preview_cache import preview_cache
+from app.services.resolution_note_generator import ResolutionNoteGeneratorService
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/ai-sessions/{session_id}", tags=["session-suggested-fixes"])
+
+
+async def _load_session_or_404(db: AsyncSession, session_id: UUID) -> AISession:
+    """RLS-scoped session load. 404 covers both missing and cross-tenant."""
+    result = await db.execute(select(AISession).where(AISession.id == session_id))
+    session = result.scalar_one_or_none()
+    if session is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Session not found")
+    return session
+
+
+# ── Suggested fix: active ──────────────────────────────────────────────────
+
+@router.get(
+    "/suggested-fixes/active",
+    response_model=SessionSuggestedFixResponse,
+)
+async def get_active_suggested_fix(
+    session_id: UUID,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+) -> SessionSuggestedFixResponse:
+    """Return the current active suggested fix (`superseded_at IS NULL`) or 404.
+
+    A session has at most one active fix. Multiple historical rows persist
+    for audit, but only the most-recent un-superseded one is returned here.
+    """
+    await _load_session_or_404(db, session_id)
+    result = await db.execute(
+        select(SessionSuggestedFix)
+        .where(
+            SessionSuggestedFix.session_id == session_id,
+            SessionSuggestedFix.superseded_at.is_(None),
+        )
+        .order_by(SessionSuggestedFix.created_at.desc())
+    )
+    fix = result.scalars().first()
+    if fix is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="No active suggested fix for this session",
+        )
+    return SessionSuggestedFixResponse.model_validate(fix)
+
+
+# ── Suggested fix: decision ────────────────────────────────────────────────
+
+@router.post(
+    "/suggested-fixes/{fix_id}/decision",
+    response_model=SessionSuggestedFixDecisionResponse,
+)
+async def record_decision(
+    session_id: UUID,
+    fix_id: UUID,
+    body: SessionSuggestedFixDecisionRequest,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+) -> SessionSuggestedFixDecisionResponse:
+    """Record the engineer's path choice on a suggested fix.
+
+    Phase 3 only persists the decision and (for `dismissed`) supersedes the
+    row. Side effects — script generation for `one_off` / `draft_template`,
+    redirect for `build_template` — land in Phase 5 alongside the inline
+    Script Generator integration. The response shape is forward-compatible.
+    """
+    await _load_session_or_404(db, session_id)
+
+    result = await db.execute(
+        select(SessionSuggestedFix).where(
+            SessionSuggestedFix.id == fix_id,
+            SessionSuggestedFix.session_id == session_id,
+        )
+    )
+    fix = result.scalar_one_or_none()
+    if fix is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Suggested fix not found"
+        )
+
+    # Once a fix has been superseded we still record the engineer's
+    # decision (it's a historical signal — "engineer dismissed the
+    # interim hypothesis"), but `dismissed` on a superseded row would
+    # be redundant noise.
+    if fix.superseded_at is not None and body.decision == "dismissed":
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="This fix is already superseded by a newer suggestion",
+        )
+
+    fix.user_decision = body.decision
+    if body.decision == "dismissed" and fix.superseded_at is None:
+        fix.superseded_at = datetime.now(timezone.utc)
+
+    # Engineer's choice changes the bundle the resolution-note preview sees,
+    # so bump state_version too.
+    await db.execute(
+        update(AISession)
+        .where(AISession.id == session_id)
+        .values(state_version=AISession.state_version + 1)
+    )
+    await db.commit()
+    await db.refresh(fix)
+
+    return SessionSuggestedFixDecisionResponse(
+        id=fix.id,
+        user_decision=fix.user_decision,  # type: ignore[arg-type]
+    )
+
+
+# ── Resolution note preview ────────────────────────────────────────────────
+
+@router.post(
+    "/resolution-note/preview",
+    response_model=ResolutionNotePreviewResponse,
+)
+async def resolution_note_preview(
+    session_id: UUID,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+) -> ResolutionNotePreviewResponse:
+    """Generate (or return cached) draft markdown for the Resolve note.
+
+    Cache key: `(resolution_note, session_id, state_version)`. State_version is
+    bumped by every fact / suggested-fix / script-generation write, so two
+    consecutive calls with no intervening writes return the same cached
+    payload (and won't pay for a Sonnet call).
+
+    Posted to PSA in Phase 4. Until then, this endpoint is read-only.
+    """
+    await _load_session_or_404(db, session_id)
+    gen = ResolutionNoteGeneratorService(db)
+    try:
+        payload = await gen.generate_or_get_cached(session_id)
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
+    except Exception as e:
+        logger.exception("Resolution note preview failed for session %s", session_id)
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=f"Resolution-note generator error ({type(e).__name__})",
+        )
+    return ResolutionNotePreviewResponse(**payload)
+
+
+# ── Helper used by tests ───────────────────────────────────────────────────
+
+def _clear_preview_cache_for_tests() -> None:
+    """Reset the singleton cache between tests."""
+    preview_cache._store.clear()  # noqa: SLF001 — test-only access

backend/app/api/router.py

@@ -41,8 +41,10 @@ from app.api.endpoints import (
     scripts,
     script_builder,
     session_branches,
+    session_facts,
     session_handoffs,
     session_resolutions,
+    session_suggested_fixes,
     sessions,
     shared,
     shares,
@@ -135,6 +137,10 @@ api_router.include_router(network_diagrams.router, dependencies=_tenant_deps)
 # session_handoffs queue router must come before ai_sessions to avoid conflict
 api_router.include_router(session_handoffs.queue_router, dependencies=_tenant_deps)
 api_router.include_router(session_resolutions.router, dependencies=_tenant_deps)
+# session_facts mounts under /ai-sessions/{id}/facts — register before ai_sessions
+# so the {session_id}/facts subpaths take precedence over any future generic catchalls.
+api_router.include_router(session_facts.router, dependencies=_tenant_deps)
+api_router.include_router(session_suggested_fixes.router, dependencies=_tenant_deps)
 api_router.include_router(ai_sessions.router, dependencies=_tenant_deps)
 api_router.include_router(flow_proposals.router, dependencies=_tenant_deps)
 api_router.include_router(flowpilot_analytics.router, dependencies=_tenant_deps)

backend/app/core/config.py

@@ -129,6 +129,16 @@ class Settings(BaseSettings):
         "kb_convert": "standard",
         "script_build": "standard",
         "network_diagram_generate": "standard",
+        # FlowPilot migration Phase 2 — short, latency-sensitive transformation
+        # of an engineer's answer/check output into a candidate fact.
+        # Doc Section 6.6 sets Haiku as the default; instrumentation tracks
+        # disputed_fact_rate so we can escalate to Sonnet if quality drops.
+        "fact_synthesis": "fast",
+        # FlowPilot migration Phase 3 — resolution-note preview that ships to
+        # the customer ticket. Sonnet because customer-facing artifact quality
+        # matters more than latency; the in-process state_version cache keeps
+        # cost manageable.
+        "resolution_note": "standard",
     }
 
     def get_model_for_action(self, action_type: str) -> str:

backend/app/schemas/session_fact.py

@@ -0,0 +1,81 @@
+"""Pydantic schemas for the FlowPilot "What we know" session facts.
+
+See FLOWPILOT-MIGRATION.md Section 4.2 for the data model rationale.
+"""
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Literal
+from uuid import UUID
+
+from pydantic import BaseModel, Field
+
+# AI-emittable source types are a subset (`user_note` is engineer-only).
+AIEmittableSourceType = Literal["question", "diagnostic_check", "ai_synthesis"]
+SourceType = Literal["question", "diagnostic_check", "user_note", "ai_synthesis"]
+
+
+class SessionFactResponse(BaseModel):
+    """A single fact card in the What-we-know panel."""
+    id: UUID
+    session_id: UUID
+    text: str
+    source_type: SourceType
+    source_ref: UUID | None
+    source_summary: str | None
+    created_by: UUID
+    created_at: datetime
+    updated_at: datetime
+    # `editable` is computed server-side so the client doesn't have to
+    # re-encode the editability rule. It mirrors the PATCH endpoint's
+    # 403 policy: only user_note and ai_synthesis facts are editable.
+    editable: bool
+
+    model_config = {"from_attributes": False}
+
+
+class SessionFactListResponse(BaseModel):
+    facts: list[SessionFactResponse]
+
+
+class SessionFactCreateRequest(BaseModel):
+    """Engineer-created manual fact (the "+ Add a note" affordance).
+
+    The endpoint hard-codes source_type="user_note" — manual creation cannot
+    spoof a question/check origin. Source-type-bound creation goes through
+    `/promote` instead.
+    """
+    text: str = Field(..., min_length=1, max_length=2000)
+    summary: str | None = Field(None, max_length=200)
+
+
+class SessionFactUpdateRequest(BaseModel):
+    """Edit an existing fact's text or summary.
+
+    The endpoint returns 403 when the fact's source_type is `question` or
+    `diagnostic_check` — those facts must be edited at the source item.
+    """
+    text: str | None = Field(None, min_length=1, max_length=2000)
+    summary: str | None = Field(None, max_length=200)
+
+
+class SessionFactPromoteRequest(BaseModel):
+    """Promote a question answer / check result into a fact.
+
+    Two modes:
+    - **Direct**: caller provides `proposed_text` (and optionally `proposed_summary`).
+      The fact is persisted as-is. Used by the AI [PROMOTE] marker path and by the
+      engineer's "edit then save" affordance.
+    - **Synthesize**: caller provides `raw_input` (the engineer's typed answer or
+      the check output) and the server drafts `text`/`summary` via the
+      FactSynthesisService. The draft is persisted immediately for now —
+      the supervisor-staging review is a future enhancement (out of scope per
+      Section 12).
+
+    Exactly one of `proposed_text` or `raw_input` must be set.
+    """
+    source_type: AIEmittableSourceType
+    source_ref: UUID | None = None
+    proposed_text: str | None = Field(None, min_length=1, max_length=2000)
+    proposed_summary: str | None = Field(None, max_length=200)
+    raw_input: str | None = Field(None, min_length=1, max_length=10_000)

backend/app/schemas/session_suggested_fix.py

@@ -0,0 +1,63 @@
+"""Pydantic schemas for session suggested fixes (Phase 3).
+
+See FLOWPILOT-MIGRATION.md Section 5.2.
+"""
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Literal
+from uuid import UUID
+
+from pydantic import BaseModel, Field
+
+UserDecision = Literal["one_off", "draft_template", "build_template", "dismissed"]
+
+
+class SessionSuggestedFixResponse(BaseModel):
+    id: UUID
+    session_id: UUID
+    title: str
+    description: str
+    confidence_pct: int
+    script_template_id: UUID | None
+    ai_drafted_script: str | None
+    ai_drafted_parameters: dict[str, Any] | None
+    user_decision: UserDecision | None
+    superseded_at: datetime | None
+    created_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class SessionSuggestedFixDecisionRequest(BaseModel):
+    """Engineer's path choice on a suggested fix.
+
+    Server-side side effects per Section 5.2:
+    - one_off: render the script (Phase 5), no template created.
+    - draft_template: render + queue a draft_templates row (Phase 5/6).
+    - build_template: redirect to full template creation (Phase 5).
+    - dismissed: mark the fix superseded so a fresh suggestion can take over.
+    """
+    decision: UserDecision
+
+
+class SessionSuggestedFixDecisionResponse(BaseModel):
+    """Returned after recording a decision; richer payloads land in Phase 5."""
+    id: UUID
+    user_decision: UserDecision
+    # Set when the decision triggered side effects (e.g. a script generation).
+    # Phase 3 only records the choice; this stays None until Phase 5 wires it.
+    rendered_script: str | None = None
+    redirect_path: str | None = Field(
+        None,
+        description="Where to send the engineer next (e.g. /scripts/builder?... for build_template)",
+    )
+
+
+# ── Resolution note preview ────────────────────────────────────────────────
+
+class ResolutionNotePreviewResponse(BaseModel):
+    markdown: str
+    target_ticket_ref: str | None
+    state_version: int
+    from_cache: bool

backend/app/services/assistant_chat_service.py

@@ -62,9 +62,12 @@ Every response you write MUST follow this exact structure:
 1. **1-3 sentences of analysis** (what the symptoms tell you)
 2. **[QUESTIONS] marker** with 1-3 questions for the engineer (if you need info)
 3. **[ACTIONS] marker** with 1-4 diagnostic commands to run (if applicable)
+4. **[PROMOTE] marker(s)** when the engineer's most recent message confirmed a fact \
+worth recording (optional; see "Promoting facts" below)
 
 You MUST include at least one marker ([QUESTIONS] or [ACTIONS]) in every response. \
-A response with only prose and no markers is INVALID and will break the UI.
+A response with only prose and no markers is INVALID and will break the UI. \
+[PROMOTE] is optional and IN ADDITION to the required markers, never a replacement.
 
 ### Complete example of a correct first response:
 
@@ -112,6 +115,88 @@ information is no longer needed to resolve the issue. Default to keeping them.
 **Both markers are stripped from display** — the engineer sees them as interactive UI cards, \
 not raw JSON. Put analysis BEFORE markers. Markers go at the END of your response.
 
+## Promoting facts to "What we know"
+
+The engineer has a "What we know" panel that holds confirmed facts about this \
+session. Each confirmed fact stays visible to the engineer for the rest of the \
+session and feeds the resolution note posted to the customer ticket. Surface \
+facts there using a `[PROMOTE]` marker.
+
+**When to emit [PROMOTE]:**
+- The engineer just answered a [QUESTIONS] item with a substantive answer that \
+rules something in or out
+- The engineer just shared diagnostic-check output that confirmed a finding
+- You synthesized a new conclusion from two or more prior facts
+
+**When NOT to emit [PROMOTE]:**
+- The engineer's answer was "unknown", "I don't know", or a clarifying question \
+back to you
+- The diagnostic output was empty, errored, or inconclusive
+- You're re-stating something already in What we know
+- The "fact" is your own hypothesis, not something the engineer confirmed
+
+**[PROMOTE] marker format:**
+Each fact is its own block. You may emit multiple blocks per response.
+
+[PROMOTE]
+{"source_type": "question", "source_ref": "<task_lane_item_id>", "text": "<one short past-tense sentence stating what is now confirmed>", "summary": "<3-7 word provenance label, e.g. 'rules out tenant/license'>"}
+[/PROMOTE]
+
+- `source_type` is one of: `"question"` (fact derived from a question's answer), \
+`"diagnostic_check"` (fact derived from a check's output), or `"ai_synthesis"` \
+(you combined prior facts).
+- `source_ref` is the `id` field of the originating task-lane item — the \
+[QUESTIONS] and [ACTIONS] payloads you receive in conversation context include \
+an `id` for each item. Copy that UUID verbatim. For `ai_synthesis`, OMIT \
+`source_ref` (or set it to null).
+- `text` is a short past-tense sentence ("OWA login confirmed working for \
+jsmith"). Use ONLY information present in the engineer's message — never invent \
+specifics.
+- `summary` names the diagnostic value (what the fact rules in or out), 3-7 \
+words, no period.
+
+**Strict rule:** [PROMOTE] is for confirmed facts only. If you're not certain \
+the engineer's message confirms the fact, do not emit a [PROMOTE]. Hallucinated \
+facts get posted to customer tickets and will erode trust in the system.
+
+## Proposing a fix with [SUGGEST_FIX]
+
+When you have a concrete proposed resolution path with reasonable confidence, \
+emit a `[SUGGEST_FIX]` marker. This populates the "Suggested fix" card the \
+engineer can act on (run a script, build a template, etc.). A new \
+[SUGGEST_FIX] supersedes any prior suggested fix on the session — emit a fresh \
+one whenever your top hypothesis changes meaningfully.
+
+**When to emit [SUGGEST_FIX]:**
+- You have a concrete resolution path (not just "investigate further")
+- Confidence is at least ~50% — below that, keep diagnosing
+- Either a known Script Library template applies, OR you can draft a script \
+that resolves the issue end-to-end
+
+**When NOT to emit [SUGGEST_FIX]:**
+- You're still narrowing causes and the fix depends on the next answer
+- The "fix" is just running another diagnostic — that goes in [ACTIONS]
+- Two paths are equally likely — fork or ask first, suggest later
+
+**[SUGGEST_FIX] marker format (one block per response, last one wins):**
+
+[SUGGEST_FIX]
+{"title": "Clear cached credentials + rebuild Outlook profile", "description": "Stale cached credential in Credential Manager is holding the pre-reset token. Clearing it and recreating the profile completes the password change.", "confidence": 94, "script_template_slug": "clear-outlook-credentials"}
+[/SUGGEST_FIX]
+
+- `title`: short imperative summary, ≤ 200 chars
+- `description`: one short paragraph explaining the root cause and the fix
+- `confidence`: integer 0-100 (what you'd bet this resolves the ticket)
+- `script_template_slug`: slug of an existing Script Library template if one \
+applies; OMIT or set null otherwise
+- `ai_drafted_script`: full script body if no template matches (only when \
+`script_template_slug` is null/omitted)
+- `ai_drafted_parameters`: optional JSON object of suggested parameter values \
+for the drafted script
+
+The marker is stripped from display — the engineer sees the suggested fix as \
+an interactive card with confidence badge, not raw JSON.
+
 ## Using the Team's Flow Library
 Your team has built troubleshooting flows in ResolutionFlow. When relevant flows \
 appear in the context below, reference them by name so the engineer can launch them \
@@ -182,6 +267,12 @@ No exceptions. Not even when forking. A response without at least one of these m
 will crash the UI. If you are unsure, include both. The markers are REQUIRED output, not optional.
 If any tasks in the engineer's message are marked `_(not yet completed)_`, re-include them \
 in your markers unless you are ≥75% confident that information is no longer relevant.
+[PROMOTE] markers are OPTIONAL and IN ADDITION to the required ones — emit them only \
+when the engineer's most recent message confirmed something worth recording, and copy \
+the originating item's `id` into `source_ref` verbatim.
+[SUGGEST_FIX] is OPTIONAL — emit one at most per response, only when you have a \
+concrete proposed resolution at ~50%+ confidence. A new [SUGGEST_FIX] supersedes \
+any prior suggested fix.
 """
 
 

backend/app/services/fact_synthesis_service.py

@@ -0,0 +1,285 @@
+"""FactSynthesisService — converts engineer answers and check output into facts.
+
+Two paths feed this service:
+
+1. **AI marker path (the common case).** When the model emits a `[PROMOTE]`
+   marker in the chat stream, `unified_chat_service` parses the marker (which
+   already contains the engineer-readable `text` and short provenance `summary`)
+   and calls `create_fact` directly. No LLM call is needed — the model already
+   wrote the fact.
+
+2. **Engineer-driven synthesize path.** The "+ Promote to What we know" affordance
+   in the UI sends a raw answer or check output and asks the server to draft
+   `text` + `summary` for review. `synthesize_from_question` /
+   `synthesize_from_check` make a small Haiku call for that draft. The engineer
+   confirms (or edits) before persistence, so the LLM output is never
+   silently posted to a customer ticket.
+
+Either way, persistence funnels through `create_fact`, which ALSO bumps
+`ai_sessions.state_version` so the resolution-note preview cache invalidates
+(see FLOWPILOT-MIGRATION.md Section 5.5).
+
+Model tier is `fact_synthesis` in `settings.ACTION_MODEL_MAP` (Haiku per
+Section 6.6). MCP is intentionally disabled for synthesis — these are
+pure transformations of input, not research calls.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import re
+from typing import Any
+from uuid import UUID
+
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+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.session_fact import SessionFact
+
+logger = logging.getLogger(__name__)
+
+
+# Conservative synthesis prompt. Hallucinated specifics are a trust-killer
+# because facts feed the resolution note posted to customer tickets — the
+# prompt makes "no fact" an explicit, valid output.
+_SYNTHESIS_SYSTEM_PROMPT = """\
+You convert one engineer answer or one diagnostic-check output into a single \
+candidate fact for a troubleshooting session's "What we know" log.
+
+Return strict JSON with this shape:
+{
+  "text": "<one short sentence stating what is now known, in past tense>",
+  "summary": "<3-7 word provenance label, e.g. 'rules out tenant/license'>"
+}
+
+If the answer/output does NOT contain a substantive fact (e.g. the engineer \
+typed 'unknown', the command failed, the output is empty), return:
+{
+  "text": null,
+  "summary": null
+}
+
+Strict rules:
+- Use ONLY information present in the input. Never add details that were not stated.
+- Do not speculate, infer causes, or extrapolate. State only what the input proves.
+- The text is a fact a colleague could verify by looking at the original answer/output.
+- The summary names the diagnostic value (what this fact rules in or out), not the topic.
+- Output ONLY the JSON object, no prose, no markdown fences.
+"""
+
+
+class FactSynthesisService:
+    """Persists session facts and (optionally) drafts them via an LLM call.
+
+    Methods that touch the database take an `AsyncSession` and assume the
+    caller commits. `create_fact` flushes so the returned row has a primary key.
+    """
+
+    def __init__(self, db: AsyncSession) -> None:
+        self.db = db
+
+    # ── Persistence ────────────────────────────────────────────────────────
+
+    async def create_fact(
+        self,
+        *,
+        session_id: UUID,
+        account_id: UUID,
+        user_id: UUID,
+        source_type: str,
+        text: str,
+        summary: str | None = None,
+        source_ref: UUID | None = None,
+    ) -> SessionFact:
+        """Persist a fact and bump the session's preview-cache version.
+
+        `source_ref` MUST be None for `user_note` and `ai_synthesis` sources;
+        for `question` and `diagnostic_check` it should point at the stable
+        UUID of the originating task-lane item. The DB has no FK constraint
+        on `source_ref` (the target lives inside JSONB) — integrity is enforced
+        here.
+        """
+        if source_type not in ("question", "diagnostic_check", "user_note", "ai_synthesis"):
+            raise ValueError(f"Invalid source_type: {source_type}")
+
+        if source_type in ("user_note", "ai_synthesis") and source_ref is not None:
+            # `source_ref` is a back-pointer to a question/check; user notes
+            # and AI-synthesized facts have no source item to point at.
+            raise ValueError(
+                f"source_ref must be None for source_type={source_type}"
+            )
+
+        text = (text or "").strip()
+        if not text:
+            raise ValueError("Fact text cannot be empty")
+
+        fact = SessionFact(
+            session_id=session_id,
+            account_id=account_id,
+            text=text,
+            source_type=source_type,
+            source_ref=source_ref,
+            source_summary=(summary or "").strip() or None,
+            created_by=user_id,
+        )
+        self.db.add(fact)
+
+        # Invalidate any preview cached against the prior state_version.
+        # Single UPDATE so the bump is atomic relative to the fact insert
+        # within this transaction; concurrent writers serialize on the row.
+        await self.db.execute(
+            update(AISession)
+            .where(AISession.id == session_id)
+            .values(state_version=AISession.state_version + 1)
+        )
+        await self.db.flush()
+        return fact
+
+    async def soft_delete_fact(self, fact: SessionFact) -> None:
+        """Mark a fact deleted and bump state_version."""
+        from datetime import datetime, timezone
+
+        fact.deleted_at = datetime.now(timezone.utc)
+        await self.db.execute(
+            update(AISession)
+            .where(AISession.id == fact.session_id)
+            .values(state_version=AISession.state_version + 1)
+        )
+        await self.db.flush()
+
+    async def update_fact(
+        self,
+        fact: SessionFact,
+        *,
+        text: str | None = None,
+        summary: str | None = None,
+    ) -> SessionFact:
+        """Update an editable fact and bump state_version.
+
+        Caller is responsible for the editability check — only `user_note`
+        and `ai_synthesis` facts may be edited at the card level. The
+        endpoint enforces this and returns 403 for the read-only types.
+        """
+        if text is not None:
+            stripped = text.strip()
+            if not stripped:
+                raise ValueError("Fact text cannot be empty")
+            fact.text = stripped
+        if summary is not None:
+            fact.source_summary = summary.strip() or None
+
+        await self.db.execute(
+            update(AISession)
+            .where(AISession.id == fact.session_id)
+            .values(state_version=AISession.state_version + 1)
+        )
+        await self.db.flush()
+        return fact
+
+    # ── LLM-backed drafting ────────────────────────────────────────────────
+
+    async def synthesize_from_question(
+        self, *, question_text: str, raw_answer: str
+    ) -> dict[str, str | None]:
+        """Draft `{text, summary}` from a question + engineer's free-text answer.
+
+        Returns `{"text": None, "summary": None}` when the answer doesn't
+        contain a substantive fact — caller should not persist in that case.
+        """
+        return await self._synthesize(
+            user_input=(
+                f"Question asked: {question_text.strip()}\n"
+                f"Engineer's answer: {raw_answer.strip()}"
+            ),
+        )
+
+    async def synthesize_from_check(
+        self, *, check_label: str, check_output: str
+    ) -> dict[str, str | None]:
+        """Draft `{text, summary}` from a diagnostic check label + its output."""
+        return await self._synthesize(
+            user_input=(
+                f"Diagnostic check: {check_label.strip()}\n"
+                f"Output:\n{check_output.strip()}"
+            ),
+        )
+
+    async def _synthesize(self, *, user_input: str) -> dict[str, str | None]:
+        """Single Haiku call with the conservative synthesis prompt."""
+        model = settings.get_model_for_action("fact_synthesis")
+        provider = get_ai_provider(model=model)
+
+        # Cache the system prompt — it's identical across every synthesis call.
+        system_blocks: list[dict[str, Any]] = [
+            {
+                "type": "text",
+                "text": _SYNTHESIS_SYSTEM_PROMPT,
+                "cache_control": {"type": "ephemeral"},
+                # cacheable: identical across all fact-synthesis calls
+            },
+        ]
+
+        try:
+            text, _in, _out = await provider.generate_json(
+                system_prompt=system_blocks,
+                messages=[{"role": "user", "content": user_input}],
+                max_tokens=200,
+            )
+        except Exception:
+            logger.exception("Fact synthesis LLM call failed")
+            return {"text": None, "summary": None}
+
+        return self._parse_synthesis_response(text)
+
+    @staticmethod
+    def _parse_synthesis_response(raw: str) -> dict[str, str | None]:
+        """Tolerant parse: strip fences, accept null fields, ignore extras."""
+        cleaned = raw.strip()
+        if cleaned.startswith("```"):
+            cleaned = re.sub(r"^```(?:json)?\s*", "", cleaned)
+            cleaned = re.sub(r"\s*```$", "", cleaned)
+
+        try:
+            data = json.loads(cleaned)
+        except (json.JSONDecodeError, ValueError):
+            logger.warning("Fact synthesis returned non-JSON: %r", raw[:200])
+            return {"text": None, "summary": None}
+
+        if not isinstance(data, dict):
+            return {"text": None, "summary": None}
+
+        text = data.get("text")
+        summary = data.get("summary")
+        if text is not None and not isinstance(text, str):
+            text = None
+        if summary is not None and not isinstance(summary, str):
+            summary = None
+
+        # Treat empty strings the same as null — "no substantive fact".
+        if isinstance(text, str) and not text.strip():
+            text = None
+        if isinstance(summary, str) and not summary.strip():
+            summary = None
+
+        return {"text": text, "summary": summary}
+
+
+async def list_facts_for_session(
+    db: AsyncSession, session_id: UUID
+) -> list[SessionFact]:
+    """List non-deleted facts for a session, oldest first.
+
+    RLS filters by tenant; the explicit account_id check is unnecessary here.
+    """
+    result = await db.execute(
+        select(SessionFact)
+        .where(
+            SessionFact.session_id == session_id,
+            SessionFact.deleted_at.is_(None),
+        )
+        .order_by(SessionFact.created_at.asc())
+    )
+    return list(result.scalars().all())

backend/app/services/preview_cache.py

@@ -0,0 +1,52 @@
+"""In-process preview cache for FlowPilot resolution-note / escalation-package previews.
+
+Phase 3 implementation per FLOWPILOT-MIGRATION.md Section 5.5:
+- Cache key: `(kind, session_id, state_version)` — no TTL needed, state_version
+  is the source of truth.
+- Invalidation: any write to session_facts, session_suggested_fixes, or
+  script_generations bumps `ai_sessions.state_version`. Old entries simply
+  stop being looked up and leak harmlessly until process restart.
+- Storage: plain dict, single-process. When Session Sharing brings Redis,
+  swap the storage without changing the call sites.
+
+Bound: best-effort soft cap of 5000 entries. When exceeded we drop the
+oldest insertion. Not a TTL — at current scale, the cap is more about
+resident-memory hygiene than correctness.
+"""
+from __future__ import annotations
+
+from collections import OrderedDict
+from typing import Any
+from uuid import UUID
+
+_MAX_ENTRIES = 5000
+
+
+class _PreviewCache:
+    def __init__(self) -> None:
+        self._store: OrderedDict[tuple[str, UUID, int], Any] = OrderedDict()
+
+    def get(self, kind: str, session_id: UUID, state_version: int) -> Any | None:
+        key = (kind, session_id, state_version)
+        if key not in self._store:
+            return None
+        # Touch on access so LRU eviction is meaningful.
+        self._store.move_to_end(key)
+        return self._store[key]
+
+    def set(self, kind: str, session_id: UUID, state_version: int, value: Any) -> None:
+        key = (kind, session_id, state_version)
+        self._store[key] = value
+        self._store.move_to_end(key)
+        # Evict oldest if over cap. OrderedDict.popitem(last=False) is O(1).
+        while len(self._store) > _MAX_ENTRIES:
+            self._store.popitem(last=False)
+
+    def invalidate_session(self, session_id: UUID) -> None:
+        """Drop all entries for a session — used when the session is deleted."""
+        keys = [k for k in self._store if k[1] == session_id]
+        for k in keys:
+            del self._store[k]
+
+
+preview_cache = _PreviewCache()

backend/app/services/resolution_note_generator.py

@@ -0,0 +1,320 @@
+"""ResolutionNoteGeneratorService — drafts the structured Resolve note for a session.
+
+Produces the four-section markdown that ships to the customer ticket (per
+FLOWPILOT-MIGRATION.md Section 6.2):
+
+    ## Problem
+    ## What we confirmed
+    ## Root cause
+    ## Resolution
+
+The output is the *draft* — engineers review and edit in the preview popover
+before clicking Confirm & post (Phase 4). Caching is keyed on
+`(session_id, ai_sessions.state_version)` per Section 5.5; the cache lives in
+`preview_cache` and invalidates automatically when any fact / suggested fix /
+script generation bumps the session's state_version.
+
+Model: Sonnet (`resolution_note` action tier — quality matters because the
+output is customer-facing). MCP intentionally disabled — this is a summary
+of existing state, not a research task.
+
+Sensitive parameter values in script_generations are redacted using the
+script template's `parameters_schema` (`field_type: "password"`). Existing
+ScriptTemplateEngine.redact_sensitive handles the substitution.
+"""
+from __future__ import annotations
+
+import logging
+from typing import Any
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.ai_provider import get_ai_provider
+from app.core.config import settings
+from app.models.ai_session import AISession
+from app.models.script_template import ScriptGeneration, ScriptTemplate
+from app.models.session_fact import SessionFact
+from app.models.session_suggested_fix import SessionSuggestedFix
+from app.services.preview_cache import preview_cache
+from app.services.script_template_engine import ScriptTemplateEngine
+
+logger = logging.getLogger(__name__)
+
+
+_RESOLUTION_NOTE_SYSTEM_PROMPT = """\
+You produce structured resolution notes for an MSP troubleshooting platform. \
+The notes are posted as ticket notes in the customer's PSA, so they must read \
+like a competent senior engineer summarized the work — not like an AI \
+narration. Your output goes in front of paying customers.
+
+Output exactly this markdown structure, no preamble, no closing remarks, no \
+extra headings:
+
+## Problem
+<one short paragraph stating the issue the engineer worked on, derived from the \
+session's intake/title and the incident header. Past tense. No "user reported" \
+hedging — state the problem directly.>
+
+## What we confirmed
+<bulleted list of facts from the "What we know" section, each one a short line. \
+Group similar facts together; do not invent connecting prose. If there are no \
+facts, write "Nothing was confirmed." and skip to Root cause.>
+
+## Root cause
+<one short paragraph naming the root cause based on the active suggested fix \
+and confirmed facts. If the suggested fix is low-confidence (<60%) or absent, \
+say "Root cause not definitively isolated." and explain what is suspected based \
+on facts.>
+
+## Resolution
+<one short paragraph describing the resolution applied. If a script ran during \
+the session, mention it (e.g. "Cleared cached credentials via the \
+clear-outlook-credentials script."). If no resolution has been performed yet, \
+write "Resolution not yet applied — fix proposed: <fix title>." Pull verbatim \
+script names and template references when available.>
+
+Strict rules:
+- Use ONLY the facts and state I provide. Never invent specifics that are not \
+in the input.
+- Do not include placeholder text like "TBD", "TODO", or empty bullets.
+- Do not include the engineer's name, the AI's name, internal session IDs, or \
+the session's chat transcript.
+- Markdown headings exactly as shown (## level), no bolding the headings.
+- No trailing whitespace, no double-blank lines, no horizontal rules.
+"""
+
+
+class ResolutionNoteGeneratorService:
+    """Generates and caches the four-section Resolve note markdown."""
+
+    KIND = "resolution_note"
+
+    def __init__(self, db: AsyncSession) -> None:
+        self.db = db
+
+    async def generate_or_get_cached(
+        self, session_id: UUID, *, force: bool = False,
+    ) -> dict[str, Any]:
+        """Return the preview for the session.
+
+        Reads `(KIND, session_id, state_version)` from the in-process cache;
+        on miss, generates fresh markdown and stores under the same key.
+        `force=True` bypasses the cache and refreshes the cached entry.
+
+        Returns `{"markdown": str, "target_ticket_ref": str | None,
+                   "state_version": int, "from_cache": bool}`.
+        """
+        session = await self._load_session(session_id)
+        cached = preview_cache.get(self.KIND, session.id, session.state_version) if not force else None
+        if cached is not None:
+            return {**cached, "from_cache": True}
+
+        markdown = await self._render(session)
+        target = self._target_ticket_ref(session)
+        payload = {
+            "markdown": markdown,
+            "target_ticket_ref": target,
+            "state_version": session.state_version,
+        }
+        preview_cache.set(self.KIND, session.id, session.state_version, payload)
+        return {**payload, "from_cache": False}
+
+    # ── Internals ─────────────────────────────────────────────────────────
+
+    async def _load_session(self, session_id: UUID) -> AISession:
+        result = await self.db.execute(
+            select(AISession).where(AISession.id == session_id)
+        )
+        session = result.scalar_one_or_none()
+        if session is None:
+            raise ValueError(f"Session {session_id} not found")
+        return session
+
+    async def _render(self, session: AISession) -> str:
+        """Build the prompt input bundle, call the model, return markdown."""
+        facts = await self._load_facts(session.id)
+        active_fix = await self._load_active_fix(session.id)
+        gens = await self._load_redacted_generations(session.id)
+
+        bundle = self._build_input_bundle(session, facts, active_fix, gens)
+
+        model = settings.get_model_for_action("resolution_note")
+        provider = get_ai_provider(model=model)
+
+        # Cache the system prompt — identical across every preview call for
+        # every session. Per-session bundle is in the user message, uncached.
+        system_blocks: list[dict[str, Any]] = [
+            {
+                "type": "text",
+                "text": _RESOLUTION_NOTE_SYSTEM_PROMPT,
+                "cache_control": {"type": "ephemeral"},
+                # cacheable: identical across every resolution-note preview call
+            },
+        ]
+
+        try:
+            text, _in, _out = await provider.generate_text(
+                system_prompt=system_blocks,
+                messages=[{"role": "user", "content": bundle}],
+                max_tokens=1200,
+            )
+        except Exception:
+            logger.exception("Resolution note generation failed for session %s", session.id)
+            raise
+        return text.strip()
+
+    async def _load_facts(self, session_id: UUID) -> list[SessionFact]:
+        result = await self.db.execute(
+            select(SessionFact)
+            .where(
+                SessionFact.session_id == session_id,
+                SessionFact.deleted_at.is_(None),
+            )
+            .order_by(SessionFact.created_at.asc())
+        )
+        return list(result.scalars().all())
+
+    async def _load_active_fix(self, session_id: UUID) -> SessionSuggestedFix | None:
+        result = await self.db.execute(
+            select(SessionSuggestedFix)
+            .where(
+                SessionSuggestedFix.session_id == session_id,
+                SessionSuggestedFix.superseded_at.is_(None),
+            )
+            .order_by(SessionSuggestedFix.created_at.desc())
+        )
+        return result.scalars().first()
+
+    async def _load_redacted_generations(
+        self, session_id: UUID
+    ) -> list[dict[str, Any]]:
+        """Pull script_generations for the session, redacting password params.
+
+        Password fields are inferred from the linked template's
+        `parameters_schema` (`field_type: "password"`). The existing
+        ScriptTemplateEngine.redact_sensitive handles the substitution.
+        """
+        result = await self.db.execute(
+            select(ScriptGeneration)
+            .where(ScriptGeneration.ai_session_id == session_id)
+            .order_by(ScriptGeneration.created_at.asc())
+        )
+        gens = list(result.scalars().all())
+        if not gens:
+            return []
+
+        template_ids = {g.template_id for g in gens}
+        tpl_result = await self.db.execute(
+            select(ScriptTemplate).where(ScriptTemplate.id.in_(template_ids))
+        )
+        templates_by_id = {t.id: t for t in tpl_result.scalars().all()}
+
+        engine = ScriptTemplateEngine()
+        out: list[dict[str, Any]] = []
+        for g in gens:
+            tpl = templates_by_id.get(g.template_id)
+            sensitive_keys = self._sensitive_keys_from_schema(
+                (tpl.parameters_schema if tpl else {}) or {}
+            )
+            redacted_params = engine.redact_sensitive(g.parameters_used or {}, sensitive_keys)
+            out.append({
+                "template_name": tpl.name if tpl else "(unknown template)",
+                "template_slug": tpl.slug if tpl else None,
+                "parameters_used": redacted_params,
+                "created_at": g.created_at.isoformat(),
+            })
+        return out
+
+    @staticmethod
+    def _sensitive_keys_from_schema(schema: dict[str, Any]) -> set[str]:
+        """Extract password-typed parameter keys from a template's schema.
+
+        The schema shape is `{"parameters": [{"key": "...", "field_type": "password", ...}]}`
+        per the existing Script Generator convention. Tolerate both that shape
+        and the simpler `{"key": {"field_type": "password"}}` form.
+        """
+        keys: set[str] = set()
+        params = schema.get("parameters") if isinstance(schema, dict) else None
+        if isinstance(params, list):
+            for p in params:
+                if isinstance(p, dict) and p.get("field_type") == "password":
+                    k = p.get("key") or p.get("variable_name")
+                    if isinstance(k, str):
+                        keys.add(k)
+        elif isinstance(schema, dict):
+            for k, v in schema.items():
+                if isinstance(v, dict) and v.get("field_type") == "password":
+                    keys.add(k)
+        return keys
+
+    @staticmethod
+    def _target_ticket_ref(session: AISession) -> str | None:
+        """Display ref for the linked PSA ticket, e.g. 'CW #48291'.
+
+        ConnectWise is the only PSA wired today (per the Phase 1 constraint),
+        so a CW prefix is reasonable. Other PSAs will need provider-aware
+        formatting in Phase 4.
+        """
+        if not session.psa_ticket_id:
+            return None
+        return f"CW #{session.psa_ticket_id}"
+
+    @staticmethod
+    def _build_input_bundle(
+        session: AISession,
+        facts: list[SessionFact],
+        active_fix: SessionSuggestedFix | None,
+        generations: list[dict[str, Any]],
+    ) -> str:
+        """Compose the structured input the LLM sees for one preview call."""
+        lines: list[str] = []
+        lines.append("# Session context")
+        lines.append(f"Title: {session.title or '(untitled)'}")
+        if session.problem_summary:
+            lines.append(f"Problem summary: {session.problem_summary}")
+        if session.problem_domain:
+            lines.append(f"Domain: {session.problem_domain}")
+        intake_text = (session.intake_content or {}).get("text") if isinstance(session.intake_content, dict) else None
+        if intake_text:
+            lines.append(f"Intake message: {intake_text}")
+        if session.psa_ticket_id:
+            lines.append(f"Linked PSA ticket: CW #{session.psa_ticket_id}")
+
+        lines.append("")
+        lines.append("# Confirmed facts (What we know)")
+        if not facts:
+            lines.append("(none)")
+        else:
+            for f in facts:
+                tag = f.source_type
+                summary = f" — {f.source_summary}" if f.source_summary else ""
+                lines.append(f"- [{tag}] {f.text}{summary}")
+
+        lines.append("")
+        lines.append("# Active suggested fix")
+        if active_fix is None:
+            lines.append("(no active suggested fix)")
+        else:
+            lines.append(f"Title: {active_fix.title}")
+            lines.append(f"Confidence: {active_fix.confidence_pct}%")
+            lines.append(f"Description: {active_fix.description}")
+            if active_fix.user_decision:
+                lines.append(f"Engineer decision: {active_fix.user_decision}")
+
+        lines.append("")
+        lines.append("# Scripts run during the session (passwords redacted)")
+        if not generations:
+            lines.append("(none)")
+        else:
+            for g in generations:
+                lines.append(f"- {g['template_name']} (slug={g['template_slug']})")
+                if g["parameters_used"]:
+                    lines.append(f"  parameters: {g['parameters_used']}")
+
+        lines.append("")
+        lines.append(
+            "Produce the four-section resolution note now. Use only the input above."
+        )
+        return "\n".join(lines)

backend/app/services/unified_chat_service.py

@@ -3,22 +3,41 @@
 Replaces assistant_chat_service for new chat sessions. Messages are stored
 in ai_sessions.conversation_messages JSONB. Reuses the same AI calling
 infrastructure and system prompt from assistant_chat_service.
+
+## Markers parsed here
+- `[QUESTIONS]` / `[ACTIONS]` — task-lane items shown to the engineer
+- `[FORK]` — diagnostic forking, creates SessionBranch rows
+- `[PROMOTE]` (Phase 2) — surfaces a fact to the What-we-know section.
+  Items in pending_task_lane carry stable UUIDs (assigned here) so PROMOTE
+  source_refs survive across turns even when the model re-emits the same
+  question/action.
+- `[SUGGEST_FIX]` (Phase 3) — proposes a resolution path for the session.
+  Each new emission supersedes the previous active row (sets superseded_at)
+  so there's exactly one active fix at a time.
 """
 import json
 import logging
 import re
+import uuid as _uuid
 from typing import Any
 from uuid import UUID
 
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from datetime import datetime, timezone
+
+from sqlalchemy import update
+
 from app.models.ai_session import AISession
+from app.models.script_template import ScriptTemplate
+from app.models.session_suggested_fix import SessionSuggestedFix
 from app.services.assistant_chat_service import (
     ASSISTANT_SYSTEM_PROMPT,
     _call_ai,
     _auto_title,
 )
+from app.services.fact_synthesis_service import FactSynthesisService
 from app.services.rag_service import search as rag_search, build_rag_context, extract_suggested_flows
 
 logger = logging.getLogger(__name__)
@@ -147,6 +166,295 @@ def _parse_questions_marker(ai_content: str) -> tuple[str, list[dict[str, Any]]
     return cleaned, valid_questions
 
 
+def _parse_promote_marker(ai_content: str) -> tuple[str, list[dict[str, Any]] | None]:
+    """Extract one or more [PROMOTE]...[/PROMOTE] JSON blocks from AI response.
+
+    Each block contains a JSON object describing a candidate fact:
+      {"source_type": "question"|"diagnostic_check"|"ai_synthesis",
+       "source_ref": "<task_lane_item_uuid>" | null,
+       "text": "<fact text>",
+       "summary": "<short provenance, optional>"}
+
+    Returns (cleaned_content, list_of_items_or_None). All matched blocks are
+    stripped from display text. Invalid items are dropped silently with a
+    warning — a malformed PROMOTE should never break the chat response.
+
+    Per FLOWPILOT-MIGRATION.md Section 8.1, the model emits text + summary
+    inline so no LLM round-trip is needed to persist the fact.
+    """
+    blocks = list(re.finditer(r"\[PROMOTE\]\s*([\s\S]*?)\s*\[/PROMOTE\]", ai_content))
+    if not blocks:
+        return ai_content, None
+
+    items: list[dict[str, Any]] = []
+    for m in blocks:
+        raw = m.group(1).strip()
+        if raw.startswith("```"):
+            raw = re.sub(r"^```(?:json)?\s*", "", raw)
+            raw = re.sub(r"\s*```$", "", raw)
+        try:
+            data = json.loads(raw)
+        except (json.JSONDecodeError, ValueError) as e:
+            logger.warning("Failed to parse [PROMOTE] block: %s", e)
+            continue
+
+        if not isinstance(data, dict):
+            logger.warning("[PROMOTE] block must be a JSON object, got %s", type(data).__name__)
+            continue
+
+        source_type = data.get("source_type")
+        text = (data.get("text") or "").strip()
+        summary = (data.get("summary") or "").strip() or None
+        source_ref_raw = data.get("source_ref")
+
+        if source_type not in ("question", "diagnostic_check", "ai_synthesis"):
+            # `user_note` is engineer-only, not an AI-emittable type.
+            logger.warning("Invalid [PROMOTE] source_type=%r, skipping", source_type)
+            continue
+        if not text:
+            logger.warning("[PROMOTE] block missing text, skipping")
+            continue
+
+        source_ref: UUID | None = None
+        if source_ref_raw:
+            try:
+                source_ref = UUID(str(source_ref_raw))
+            except (ValueError, AttributeError):
+                logger.warning("[PROMOTE] source_ref %r is not a valid UUID, dropping ref", source_ref_raw)
+                source_ref = None
+
+        # `ai_synthesis` must NEVER carry a source_ref (no question/check item
+        # to point at) — surface mistakes from the model rather than tripping
+        # the FactSynthesisService validation later.
+        if source_type == "ai_synthesis":
+            source_ref = None
+
+        items.append({
+            "source_type": source_type,
+            "source_ref": source_ref,
+            "text": text,
+            "summary": summary,
+        })
+
+    # Strip all PROMOTE blocks from display content — engineers see facts in
+    # the What-we-know panel, not as raw markers in the chat.
+    cleaned = re.sub(r"\[PROMOTE\]\s*[\s\S]*?\s*\[/PROMOTE\]", "", ai_content).strip()
+
+    return cleaned, items or None
+
+
+def _assign_stable_task_lane_ids(
+    prev_lane: dict[str, Any] | None,
+    questions: list[dict[str, Any]] | None,
+    actions: list[dict[str, Any]] | None,
+) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    """Assign stable UUIDs to task-lane items, preserving them across turns.
+
+    The model often re-emits the same question/action across multiple turns
+    (it is told to keep `_(not yet completed)_` items alive). When the
+    question text matches a prior turn's, we keep the prior UUID so any
+    `session_facts.source_ref` pointing at it stays valid.
+
+    Match key:
+    - Questions: exact `text`
+    - Actions: exact `label`
+
+    Returns the questions/actions lists augmented with an `id` field.
+    """
+    prev_questions = (prev_lane or {}).get("questions") or []
+    prev_actions = (prev_lane or {}).get("actions") or []
+
+    prev_q_ids: dict[str, str] = {
+        str(q.get("text") or "").strip(): str(q["id"])
+        for q in prev_questions
+        if isinstance(q, dict) and q.get("id") and q.get("text")
+    }
+    prev_a_ids: dict[str, str] = {
+        str(a.get("label") or "").strip(): str(a["id"])
+        for a in prev_actions
+        if isinstance(a, dict) and a.get("id") and a.get("label")
+    }
+
+    out_questions: list[dict[str, Any]] = []
+    for q in questions or []:
+        text = str(q.get("text") or "").strip()
+        existing = prev_q_ids.get(text) if text else None
+        out_questions.append({
+            **q,
+            "id": existing or str(_uuid.uuid4()),
+        })
+
+    out_actions: list[dict[str, Any]] = []
+    for a in actions or []:
+        label = str(a.get("label") or "").strip()
+        existing = prev_a_ids.get(label) if label else None
+        out_actions.append({
+            **a,
+            "id": existing or str(_uuid.uuid4()),
+        })
+
+    return out_questions, out_actions
+
+
+def _parse_suggest_fix_marker(
+    ai_content: str,
+) -> tuple[str, dict[str, Any] | None]:
+    """Extract a single [SUGGEST_FIX]...[/SUGGEST_FIX] JSON block from AI response.
+
+    The block contains:
+      {"title": "...", "description": "...", "confidence": 0..100,
+       "script_template_slug": "..." | null,
+       "ai_drafted_script": "..." | null,
+       "ai_drafted_parameters": {...} | null}
+
+    Per FLOWPILOT-MIGRATION.md Section 8.2. Only the LAST block in the response
+    is honored — if the model emits multiple, only its final view of the fix
+    matters; earlier ones in the same turn are stale even before persistence.
+
+    Returns (cleaned_content, fix_dict_or_None). Marker stripped from display.
+    """
+    blocks = list(re.finditer(r"\[SUGGEST_FIX\]\s*([\s\S]*?)\s*\[/SUGGEST_FIX\]", ai_content))
+    if not blocks:
+        return ai_content, None
+
+    # Take the last block — most-recent intent wins within a single turn.
+    last = blocks[-1]
+    raw = last.group(1).strip()
+    if raw.startswith("```"):
+        raw = re.sub(r"^```(?:json)?\s*", "", raw)
+        raw = re.sub(r"\s*```$", "", raw)
+    try:
+        data = json.loads(raw)
+    except (json.JSONDecodeError, ValueError) as e:
+        logger.warning("Failed to parse [SUGGEST_FIX] block: %s", e)
+        return re.sub(r"\[SUGGEST_FIX\]\s*[\s\S]*?\s*\[/SUGGEST_FIX\]", "", ai_content).strip(), None
+
+    if not isinstance(data, dict):
+        return re.sub(r"\[SUGGEST_FIX\]\s*[\s\S]*?\s*\[/SUGGEST_FIX\]", "", ai_content).strip(), None
+
+    title = (data.get("title") or "").strip()
+    description = (data.get("description") or "").strip()
+    confidence = data.get("confidence")
+    if not title or not description or not isinstance(confidence, (int, float)):
+        logger.warning("[SUGGEST_FIX] missing required fields, dropping")
+        return re.sub(r"\[SUGGEST_FIX\]\s*[\s\S]*?\s*\[/SUGGEST_FIX\]", "", ai_content).strip(), None
+
+    confidence_int = max(0, min(100, int(round(float(confidence)))))
+
+    parsed = {
+        "title": title[:200],
+        "description": description,
+        "confidence_pct": confidence_int,
+        "script_template_slug": (data.get("script_template_slug") or None),
+        "ai_drafted_script": (data.get("ai_drafted_script") or None),
+        "ai_drafted_parameters": data.get("ai_drafted_parameters") if isinstance(data.get("ai_drafted_parameters"), dict) else None,
+    }
+
+    cleaned = re.sub(r"\[SUGGEST_FIX\]\s*[\s\S]*?\s*\[/SUGGEST_FIX\]", "", ai_content).strip()
+    return cleaned, parsed
+
+
+async def _persist_suggested_fix(
+    *,
+    db: AsyncSession,
+    session: AISession,
+    fix: dict[str, Any],
+) -> None:
+    """Supersede the prior active fix and insert the new one. Bumps state_version.
+
+    A session has at most one active suggested fix (`superseded_at IS NULL`).
+    Emitting [SUGGEST_FIX] is the only way to introduce a new one; the
+    engineer's user_decision is recorded via the decision endpoint.
+    """
+    now = datetime.now(timezone.utc)
+
+    # Mark any prior active rows for this session as superseded.
+    await db.execute(
+        update(SessionSuggestedFix)
+        .where(
+            SessionSuggestedFix.session_id == session.id,
+            SessionSuggestedFix.superseded_at.is_(None),
+        )
+        .values(superseded_at=now)
+    )
+
+    # Resolve script_template_slug → script_template_id if provided.
+    script_template_id = None
+    slug = fix.get("script_template_slug")
+    if slug:
+        result = await db.execute(
+            select(ScriptTemplate).where(ScriptTemplate.slug == slug)
+        )
+        tpl = result.scalar_one_or_none()
+        if tpl is not None:
+            script_template_id = tpl.id
+        else:
+            logger.warning(
+                "SUGGEST_FIX referenced unknown script_template_slug=%r — "
+                "treating as no template match", slug,
+            )
+
+    new_fix = SessionSuggestedFix(
+        session_id=session.id,
+        account_id=session.account_id,
+        title=fix["title"],
+        description=fix["description"],
+        confidence_pct=fix["confidence_pct"],
+        script_template_id=script_template_id,
+        ai_drafted_script=fix.get("ai_drafted_script"),
+        ai_drafted_parameters=fix.get("ai_drafted_parameters"),
+    )
+    db.add(new_fix)
+
+    # Bump preview-cache version atomically with the supersession+insert.
+    await db.execute(
+        update(AISession)
+        .where(AISession.id == session.id)
+        .values(state_version=AISession.state_version + 1)
+    )
+    await db.flush()
+
+
+async def _persist_promote_items(
+    *,
+    db: AsyncSession,
+    session: AISession,
+    user_id: UUID,
+    items: list[dict[str, Any]],
+) -> None:
+    """Persist parsed [PROMOTE] items as session_facts. Failures are logged.
+
+    A malformed PROMOTE must never break the chat response — the engineer
+    still gets the AI's analysis; the missing fact can be added manually.
+    """
+    if not items:
+        return
+    service = FactSynthesisService(db)
+    for item in items:
+        try:
+            await service.create_fact(
+                session_id=session.id,
+                account_id=session.account_id,
+                user_id=user_id,
+                source_type=item["source_type"],
+                text=item["text"],
+                summary=item["summary"],
+                source_ref=item["source_ref"],
+            )
+        except ValueError:
+            # Validation failure (e.g. empty text after strip, or
+            # source_ref-on-ai_synthesis race). Log and continue — losing
+            # one fact is better than aborting the whole chat turn.
+            logger.warning(
+                "Skipping invalid PROMOTE item for session %s: %r",
+                session.id, item, exc_info=True,
+            )
+        except Exception:
+            logger.exception(
+                "Failed to persist PROMOTE item for session %s", session.id
+            )
+
+
 async def create_chat_session(
     user_id: UUID,
     account_id: UUID,
@@ -251,10 +559,13 @@ async def send_chat_message(
             if session.status == "paused":
                 session.status = "active"
 
-            # Check for fork, actions, and questions markers in branch response too
+            # Check for fork, actions, questions, promote, and suggest_fix markers
+            # in branch response too
             branch_display, branch_fork_data = _parse_fork_marker(ai_content)
             branch_display, branch_actions_data = _parse_actions_marker(branch_display)
             branch_display, branch_questions_data = _parse_questions_marker(branch_display)
+            branch_display, branch_promote_items = _parse_promote_marker(branch_display)
+            branch_display, branch_suggest_fix = _parse_suggest_fix_marker(branch_display)
             if branch_display != ai_content:
                 # Store stripped content in branch history
                 msgs[-1] = {"role": "assistant", "content": branch_display}
@@ -288,15 +599,36 @@ async def send_chat_message(
                 except Exception:
                     logger.exception("Failed to create fork within branch for session %s", session.id)
 
-            # Persist task lane state on session
+            # Persist task lane state on session — assign stable UUIDs so any
+            # PROMOTE marker emitted later can reference the same items.
             if branch_questions_data or branch_actions_data:
+                stable_qs, stable_as = _assign_stable_task_lane_ids(
+                    session.pending_task_lane,
+                    branch_questions_data,
+                    branch_actions_data,
+                )
                 session.pending_task_lane = {
-                    "questions": branch_questions_data or [],
-                    "actions": branch_actions_data or [],
+                    "questions": stable_qs,
+                    "actions": stable_as,
                 }
             else:
                 session.pending_task_lane = None
 
+            # Persist any PROMOTE items emitted in this turn. Done AFTER the
+            # task-lane write so source_refs to brand-new items would still
+            # land on persisted UUIDs (the model can also reference IDs from
+            # the previous turn, which were already persisted).
+            if branch_promote_items:
+                await _persist_promote_items(
+                    db=db, session=session, user_id=user_id, items=branch_promote_items,
+                )
+
+            # Persist a [SUGGEST_FIX] if the branch turn included one.
+            if branch_suggest_fix:
+                await _persist_suggested_fix(
+                    db=db, session=session, fix=branch_suggest_fix,
+                )
+
             suggested_flows = extract_suggested_flows(
                 await rag_search(query=message, account_id=account_id, db=db, limit=8)
             )
@@ -343,9 +675,17 @@ async def send_chat_message(
     # Check for questions marker in AI response
     display_content, questions_data = _parse_questions_marker(display_content)
 
+    # Check for promote markers — facts the AI is surfacing to What we know.
+    display_content, promote_items = _parse_promote_marker(display_content)
+
+    # Check for a [SUGGEST_FIX] marker — supersedes the prior active fix.
+    display_content, suggest_fix_data = _parse_suggest_fix_marker(display_content)
+
     logger.info(
-        "Marker parsing results — actions: %s, questions: %s, fork: %s, raw_length: %d, display_length: %d",
+        "Marker parsing results — actions: %s, questions: %s, fork: %s, "
+        "promote: %d, suggest_fix: %s, raw_length: %d, display_length: %d",
         bool(actions_data), bool(questions_data), bool(fork_data),
+        len(promote_items or []), bool(suggest_fix_data),
         len(ai_content), len(display_content),
     )
 
@@ -410,15 +750,30 @@ async def send_chat_message(
             logger.exception("Failed to create fork for session %s", session_id)
             # Fork failed but chat message still sent — don't break the response
 
-    # Persist task lane state on session
+    # Persist task lane state on session — assign stable UUIDs so any PROMOTE
+    # marker (this turn or a later one) can reference the same items.
     if questions_data or actions_data:
+        stable_qs, stable_as = _assign_stable_task_lane_ids(
+            session.pending_task_lane, questions_data, actions_data,
+        )
         session.pending_task_lane = {
-            "questions": questions_data or [],
-            "actions": actions_data or [],
+            "questions": stable_qs,
+            "actions": stable_as,
         }
     else:
         session.pending_task_lane = None
 
+    # Persist any PROMOTE items emitted in this turn. Done after task-lane
+    # assignment so source_refs the model invented this turn already exist.
+    if promote_items:
+        await _persist_promote_items(
+            db=db, session=session, user_id=user_id, items=promote_items,
+        )
+
+    # Persist a [SUGGEST_FIX] if this turn included one — supersedes prior fix.
+    if suggest_fix_data:
+        await _persist_suggested_fix(db=db, session=session, fix=suggest_fix_data)
+
     suggested_flows = extract_suggested_flows(rag_results)
 
     return display_content, suggested_flows, session, fork_metadata, actions_data, questions_data

backend/tests/test_session_facts_api.py

@@ -0,0 +1,455 @@
+"""API + service tests for the FlowPilot Phase 2 "What we know" facts surface.
+
+Covers:
+- /api/v1/ai-sessions/{id}/facts CRUD
+- Editability rule (403 on PATCH for question/diagnostic_check facts)
+- /facts/promote with `proposed_text` (no LLM call) and via synthesis (mocked)
+- state_version increments on every fact write
+- Stable-UUID assignment for pending_task_lane items
+- [PROMOTE] marker parser shape
+"""
+from __future__ import annotations
+
+import uuid
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+
+from app.models.ai_session import AISession
+from app.models.session_fact import SessionFact
+from app.services.fact_synthesis_service import FactSynthesisService
+from app.services.unified_chat_service import (
+    _assign_stable_task_lane_ids,
+    _parse_promote_marker,
+)
+
+
+# ── Fixtures ────────────────────────────────────────────────────────────────
+
+async def _make_session(test_db, user, *, pending_task_lane=None) -> AISession:
+    session = AISession(
+        user_id=user["user_data"]["id"],
+        account_id=user["user_data"]["account_id"],
+        session_type="chat",
+        intake_type="free_text",
+        intake_content={"text": "test"},
+        status="active",
+        confidence_tier="discovery",
+        conversation_messages=[],
+        pending_task_lane=pending_task_lane,
+    )
+    test_db.add(session)
+    await test_db.commit()
+    await test_db.refresh(session)
+    return session
+
+
+# ── [PROMOTE] marker parser ─────────────────────────────────────────────────
+
+class TestPromoteMarkerParser:
+    def test_no_marker_returns_unchanged(self):
+        text = "Just an analysis sentence."
+        cleaned, items = _parse_promote_marker(text)
+        assert cleaned == text
+        assert items is None
+
+    def test_single_block(self):
+        ref = uuid.uuid4()
+        text = (
+            "Some analysis.\n\n"
+            f'[PROMOTE]\n{{"source_type":"question","source_ref":"{ref}",'
+            '"text":"OWA login confirmed working","summary":"rules out tenant"}\n'
+            "[/PROMOTE]"
+        )
+        cleaned, items = _parse_promote_marker(text)
+        assert cleaned == "Some analysis."
+        assert items is not None and len(items) == 1
+        assert items[0]["source_type"] == "question"
+        assert items[0]["source_ref"] == ref
+        assert items[0]["text"] == "OWA login confirmed working"
+        assert items[0]["summary"] == "rules out tenant"
+
+    def test_multiple_blocks(self):
+        text = (
+            '[PROMOTE]\n{"source_type":"question","source_ref":null,'
+            '"text":"a","summary":"x"}\n[/PROMOTE]\n'
+            '[PROMOTE]\n{"source_type":"diagnostic_check","source_ref":null,'
+            '"text":"b","summary":"y"}\n[/PROMOTE]'
+        )
+        cleaned, items = _parse_promote_marker(text)
+        assert items is not None and len(items) == 2
+        assert items[0]["text"] == "a"
+        assert items[1]["text"] == "b"
+        assert "[PROMOTE]" not in cleaned
+
+    def test_ai_synthesis_strips_source_ref(self):
+        # The model should not provide source_ref for synthesis facts —
+        # the parser drops it defensively even if the model does.
+        ref = uuid.uuid4()
+        text = (
+            f'[PROMOTE]\n{{"source_type":"ai_synthesis","source_ref":"{ref}",'
+            '"text":"Combined finding","summary":"synth"}\n[/PROMOTE]'
+        )
+        _, items = _parse_promote_marker(text)
+        assert items is not None and items[0]["source_ref"] is None
+
+    def test_invalid_source_type_dropped(self):
+        text = (
+            '[PROMOTE]\n{"source_type":"bogus","text":"x"}\n[/PROMOTE]\n'
+            '[PROMOTE]\n{"source_type":"question","source_ref":null,"text":"good"}\n[/PROMOTE]'
+        )
+        _, items = _parse_promote_marker(text)
+        assert items is not None and len(items) == 1
+        assert items[0]["text"] == "good"
+
+    def test_missing_text_dropped(self):
+        text = '[PROMOTE]\n{"source_type":"question","source_ref":null,"text":""}\n[/PROMOTE]'
+        _, items = _parse_promote_marker(text)
+        assert items is None  # empty list collapses to None
+
+    def test_invalid_uuid_drops_ref_keeps_item(self):
+        text = '[PROMOTE]\n{"source_type":"question","source_ref":"not-a-uuid","text":"keep"}\n[/PROMOTE]'
+        _, items = _parse_promote_marker(text)
+        assert items is not None and items[0]["source_ref"] is None
+        assert items[0]["text"] == "keep"
+
+    def test_malformed_json_dropped(self):
+        text = "[PROMOTE]\nnot json at all\n[/PROMOTE]"
+        cleaned, items = _parse_promote_marker(text)
+        assert items is None
+        # Block is still stripped from display so the engineer doesn't see it.
+        assert "[PROMOTE]" not in cleaned
+
+
+# ── Stable-UUID assignment ──────────────────────────────────────────────────
+
+class TestAssignStableTaskLaneIds:
+    def test_empty_prev_assigns_fresh_uuids(self):
+        qs, acts = _assign_stable_task_lane_ids(
+            None,
+            [{"text": "Q1", "context": "c1"}],
+            [{"label": "A1", "command": "cmd"}],
+        )
+        assert len(qs) == 1 and uuid.UUID(qs[0]["id"])
+        assert len(acts) == 1 and uuid.UUID(acts[0]["id"])
+
+    def test_prev_uuid_preserved_on_text_match(self):
+        qid = str(uuid.uuid4())
+        prev = {
+            "questions": [{"id": qid, "text": "Same text"}],
+            "actions": [],
+        }
+        qs, _ = _assign_stable_task_lane_ids(prev, [{"text": "Same text"}], [])
+        assert qs[0]["id"] == qid
+
+    def test_prev_uuid_replaced_when_text_changes(self):
+        qid = str(uuid.uuid4())
+        prev = {"questions": [{"id": qid, "text": "Original"}], "actions": []}
+        qs, _ = _assign_stable_task_lane_ids(prev, [{"text": "Different"}], [])
+        assert qs[0]["id"] != qid
+
+    def test_action_label_match_preserves_uuid(self):
+        aid = str(uuid.uuid4())
+        prev = {"questions": [], "actions": [{"id": aid, "label": "Run X"}]}
+        _, acts = _assign_stable_task_lane_ids(prev, [], [{"label": "Run X"}])
+        assert acts[0]["id"] == aid
+
+
+# ── FactSynthesisService.create_fact validation ─────────────────────────────
+
+@pytest.mark.asyncio
+async def test_create_fact_rejects_source_ref_for_user_note(test_db, test_user):
+    session = await _make_session(test_db, test_user)
+    svc = FactSynthesisService(test_db)
+    with pytest.raises(ValueError, match="source_ref must be None"):
+        await svc.create_fact(
+            session_id=session.id,
+            account_id=session.account_id,
+            user_id=session.user_id,
+            source_type="user_note",
+            text="x",
+            source_ref=uuid.uuid4(),
+        )
+
+
+@pytest.mark.asyncio
+async def test_create_fact_rejects_invalid_source_type(test_db, test_user):
+    session = await _make_session(test_db, test_user)
+    svc = FactSynthesisService(test_db)
+    with pytest.raises(ValueError, match="Invalid source_type"):
+        await svc.create_fact(
+            session_id=session.id,
+            account_id=session.account_id,
+            user_id=session.user_id,
+            source_type="not_a_type",
+            text="x",
+        )
+
+
+@pytest.mark.asyncio
+async def test_create_fact_bumps_state_version(test_db, test_user):
+    session = await _make_session(test_db, test_user)
+    initial_version = session.state_version
+    svc = FactSynthesisService(test_db)
+    await svc.create_fact(
+        session_id=session.id,
+        account_id=session.account_id,
+        user_id=session.user_id,
+        source_type="user_note",
+        text="A confirmed observation",
+    )
+    await test_db.commit()
+    await test_db.refresh(session)
+    assert session.state_version == initial_version + 1
+
+
+# ── Endpoint tests ──────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_list_facts_empty(client: AsyncClient, test_user, auth_headers, test_db):
+    session = await _make_session(test_db, test_user)
+    resp = await client.get(
+        f"/api/v1/ai-sessions/{session.id}/facts",
+        headers=auth_headers,
+    )
+    assert resp.status_code == 200
+    assert resp.json()["facts"] == []
+
+
+@pytest.mark.asyncio
+async def test_create_user_note_fact(client: AsyncClient, test_user, auth_headers, test_db):
+    session = await _make_session(test_db, test_user)
+    resp = await client.post(
+        f"/api/v1/ai-sessions/{session.id}/facts",
+        headers=auth_headers,
+        json={"text": "Customer is on a laptop", "summary": "endpoint type"},
+    )
+    assert resp.status_code == 201
+    body = resp.json()
+    assert body["source_type"] == "user_note"
+    assert body["editable"] is True
+    assert body["source_ref"] is None
+    assert body["text"] == "Customer is on a laptop"
+
+
+@pytest.mark.asyncio
+async def test_patch_user_note_succeeds(client: AsyncClient, test_user, auth_headers, test_db):
+    session = await _make_session(test_db, test_user)
+    create = await client.post(
+        f"/api/v1/ai-sessions/{session.id}/facts",
+        headers=auth_headers,
+        json={"text": "original"},
+    )
+    fact_id = create.json()["id"]
+
+    patch_resp = await client.patch(
+        f"/api/v1/ai-sessions/{session.id}/facts/{fact_id}",
+        headers=auth_headers,
+        json={"text": "edited", "summary": "new label"},
+    )
+    assert patch_resp.status_code == 200
+    assert patch_resp.json()["text"] == "edited"
+    assert patch_resp.json()["source_summary"] == "new label"
+
+
+@pytest.mark.asyncio
+async def test_patch_question_fact_returns_403(client: AsyncClient, test_user, auth_headers, test_db):
+    """Question/check-sourced facts must be edited at the source item, not the card."""
+    session = await _make_session(test_db, test_user)
+    # Insert a question-sourced fact directly so the editability rule applies.
+    fact = SessionFact(
+        session_id=session.id,
+        account_id=session.account_id,
+        text="Pre-existing question fact",
+        source_type="question",
+        source_ref=uuid.uuid4(),
+        created_by=session.user_id,
+    )
+    test_db.add(fact)
+    await test_db.commit()
+    await test_db.refresh(fact)
+
+    resp = await client.patch(
+        f"/api/v1/ai-sessions/{session.id}/facts/{fact.id}",
+        headers=auth_headers,
+        json={"text": "trying to edit"},
+    )
+    assert resp.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_delete_fact_soft_deletes(client: AsyncClient, test_user, auth_headers, test_db):
+    session = await _make_session(test_db, test_user)
+    create = await client.post(
+        f"/api/v1/ai-sessions/{session.id}/facts",
+        headers=auth_headers,
+        json={"text": "to be removed"},
+    )
+    fact_id = create.json()["id"]
+
+    del_resp = await client.delete(
+        f"/api/v1/ai-sessions/{session.id}/facts/{fact_id}",
+        headers=auth_headers,
+    )
+    assert del_resp.status_code == 204
+
+    # Listed facts should not include the soft-deleted one.
+    list_resp = await client.get(
+        f"/api/v1/ai-sessions/{session.id}/facts",
+        headers=auth_headers,
+    )
+    assert list_resp.status_code == 200
+    assert all(f["id"] != fact_id for f in list_resp.json()["facts"])
+
+    # Row still exists in DB (deleted_at set), proving it was soft-deleted.
+    row = (
+        await test_db.execute(
+            select(SessionFact).where(SessionFact.id == uuid.UUID(fact_id))
+        )
+    ).scalar_one()
+    assert row.deleted_at is not None
+
+
+@pytest.mark.asyncio
+async def test_promote_with_proposed_text(client: AsyncClient, test_user, auth_headers, test_db):
+    qid = uuid.uuid4()
+    session = await _make_session(
+        test_db, test_user,
+        pending_task_lane={
+            "questions": [{"id": str(qid), "text": "Is OWA working?"}],
+            "actions": [],
+        },
+    )
+    resp = await client.post(
+        f"/api/v1/ai-sessions/{session.id}/facts/promote",
+        headers=auth_headers,
+        json={
+            "source_type": "question",
+            "source_ref": str(qid),
+            "proposed_text": "OWA confirmed working for jsmith",
+            "proposed_summary": "rules out tenant/license",
+        },
+    )
+    assert resp.status_code == 201
+    body = resp.json()
+    assert body["source_type"] == "question"
+    assert body["source_ref"] == str(qid)
+    assert body["editable"] is False  # question-sourced facts are read-only at the card
+
+
+@pytest.mark.asyncio
+async def test_promote_via_synthesis(client: AsyncClient, test_user, auth_headers, test_db):
+    qid = uuid.uuid4()
+    session = await _make_session(
+        test_db, test_user,
+        pending_task_lane={
+            "questions": [{"id": str(qid), "text": "Is the user on a laptop?"}],
+            "actions": [],
+        },
+    )
+
+    # Mock the LLM call to avoid hitting the network in tests.
+    fake_provider = AsyncMock()
+    fake_provider.generate_json = AsyncMock(return_value=(
+        '{"text": "User confirmed on a laptop", "summary": "endpoint type"}',
+        50, 20,
+    ))
+
+    with patch(
+        "app.services.fact_synthesis_service.get_ai_provider",
+        return_value=fake_provider,
+    ):
+        resp = await client.post(
+            f"/api/v1/ai-sessions/{session.id}/facts/promote",
+            headers=auth_headers,
+            json={
+                "source_type": "question",
+                "source_ref": str(qid),
+                "raw_input": "Yes, it's a Lenovo X1 Carbon",
+            },
+        )
+
+    assert resp.status_code == 201
+    assert resp.json()["text"] == "User confirmed on a laptop"
+    assert resp.json()["source_summary"] == "endpoint type"
+
+
+@pytest.mark.asyncio
+async def test_promote_synthesis_returning_null_returns_422(
+    client: AsyncClient, test_user, auth_headers, test_db
+):
+    """When the synthesizer judges the input has no fact, the endpoint surfaces 422."""
+    qid = uuid.uuid4()
+    session = await _make_session(
+        test_db, test_user,
+        pending_task_lane={
+            "questions": [{"id": str(qid), "text": "Is OWA working?"}],
+            "actions": [],
+        },
+    )
+
+    fake_provider = AsyncMock()
+    fake_provider.generate_json = AsyncMock(return_value=(
+        '{"text": null, "summary": null}', 30, 10,
+    ))
+
+    with patch(
+        "app.services.fact_synthesis_service.get_ai_provider",
+        return_value=fake_provider,
+    ):
+        resp = await client.post(
+            f"/api/v1/ai-sessions/{session.id}/facts/promote",
+            headers=auth_headers,
+            json={
+                "source_type": "question",
+                "source_ref": str(qid),
+                "raw_input": "unknown",
+            },
+        )
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_promote_rejects_both_or_neither_inputs(
+    client: AsyncClient, test_user, auth_headers, test_db
+):
+    session = await _make_session(test_db, test_user)
+    # Neither
+    resp = await client.post(
+        f"/api/v1/ai-sessions/{session.id}/facts/promote",
+        headers=auth_headers,
+        json={"source_type": "question"},
+    )
+    assert resp.status_code == 400
+
+    # Both
+    resp2 = await client.post(
+        f"/api/v1/ai-sessions/{session.id}/facts/promote",
+        headers=auth_headers,
+        json={
+            "source_type": "question",
+            "proposed_text": "x",
+            "raw_input": "y",
+        },
+    )
+    assert resp2.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_state_version_bumps_on_create_via_endpoint(
+    client: AsyncClient, test_user, auth_headers, test_db
+):
+    session = await _make_session(test_db, test_user)
+    initial = session.state_version
+
+    await client.post(
+        f"/api/v1/ai-sessions/{session.id}/facts",
+        headers=auth_headers,
+        json={"text": "a"},
+    )
+
+    # Reload — refresh fetches the latest persisted row.
+    await test_db.refresh(session)
+    assert session.state_version == initial + 1

backend/tests/test_session_suggested_fixes_api.py

@@ -0,0 +1,356 @@
+"""API + service tests for the FlowPilot Phase 3 suggested-fix + preview surface.
+
+Covers:
+- /api/v1/ai-sessions/{id}/suggested-fixes/active (200 + 404)
+- /api/v1/ai-sessions/{id}/suggested-fixes/{fix_id}/decision (one_off,
+  draft_template, build_template, dismissed; 409 on dismissing a superseded
+  fix; state_version bump)
+- /api/v1/ai-sessions/{id}/resolution-note/preview (LLM mocked; cache hit on
+  same state_version, miss after a fact write)
+- [SUGGEST_FIX] marker parser shape
+- _persist_suggested_fix supersession + state_version bump
+"""
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+
+from app.api.endpoints.session_suggested_fixes import _clear_preview_cache_for_tests
+from app.models.ai_session import AISession
+from app.models.session_suggested_fix import SessionSuggestedFix
+from app.services.unified_chat_service import (
+    _parse_suggest_fix_marker,
+    _persist_suggested_fix,
+)
+
+
+@pytest.fixture(autouse=True)
+def _isolate_preview_cache():
+    _clear_preview_cache_for_tests()
+    yield
+    _clear_preview_cache_for_tests()
+
+
+async def _make_session(test_db, user) -> AISession:
+    session = AISession(
+        user_id=user["user_data"]["id"],
+        account_id=user["user_data"]["account_id"],
+        session_type="chat",
+        intake_type="free_text",
+        intake_content={"text": "phase 3 test"},
+        status="active",
+        confidence_tier="discovery",
+        conversation_messages=[],
+    )
+    test_db.add(session)
+    await test_db.commit()
+    await test_db.refresh(session)
+    return session
+
+
+# ── [SUGGEST_FIX] parser ────────────────────────────────────────────────────
+
+class TestSuggestFixParser:
+    def test_no_marker(self):
+        cleaned, fix = _parse_suggest_fix_marker("just analysis")
+        assert cleaned == "just analysis"
+        assert fix is None
+
+    def test_well_formed_block(self):
+        text = (
+            "Analysis sentence.\n\n"
+            '[SUGGEST_FIX]\n'
+            '{"title": "Reset password", "description": "Stale credential.", '
+            '"confidence": 87, "script_template_slug": "reset-cw"}\n'
+            '[/SUGGEST_FIX]'
+        )
+        cleaned, fix = _parse_suggest_fix_marker(text)
+        assert cleaned == "Analysis sentence."
+        assert fix is not None
+        assert fix["title"] == "Reset password"
+        assert fix["confidence_pct"] == 87
+        assert fix["script_template_slug"] == "reset-cw"
+        assert fix["ai_drafted_script"] is None
+
+    def test_confidence_clamped_and_rounded(self):
+        text = (
+            '[SUGGEST_FIX]\n{"title":"x","description":"y","confidence":120.7}\n[/SUGGEST_FIX]'
+        )
+        _, fix = _parse_suggest_fix_marker(text)
+        assert fix is not None and fix["confidence_pct"] == 100
+
+        text2 = (
+            '[SUGGEST_FIX]\n{"title":"x","description":"y","confidence":-3}\n[/SUGGEST_FIX]'
+        )
+        _, fix2 = _parse_suggest_fix_marker(text2)
+        assert fix2 is not None and fix2["confidence_pct"] == 0
+
+    def test_only_last_block_wins(self):
+        # Stale early block plus a final intent — the parser keeps the LAST one.
+        text = (
+            '[SUGGEST_FIX]\n{"title":"old","description":"o","confidence":50}\n[/SUGGEST_FIX]\n'
+            '[SUGGEST_FIX]\n{"title":"new","description":"n","confidence":80}\n[/SUGGEST_FIX]'
+        )
+        cleaned, fix = _parse_suggest_fix_marker(text)
+        assert fix is not None and fix["title"] == "new"
+        assert "[SUGGEST_FIX]" not in cleaned
+
+    def test_missing_required_field_dropped(self):
+        text = '[SUGGEST_FIX]\n{"title":"only title"}\n[/SUGGEST_FIX]'
+        cleaned, fix = _parse_suggest_fix_marker(text)
+        assert fix is None
+        # Marker still stripped from display.
+        assert "[SUGGEST_FIX]" not in cleaned
+
+    def test_malformed_json_dropped(self):
+        text = "[SUGGEST_FIX]\nnot json\n[/SUGGEST_FIX]"
+        cleaned, fix = _parse_suggest_fix_marker(text)
+        assert fix is None
+        assert "[SUGGEST_FIX]" not in cleaned
+
+
+# ── _persist_suggested_fix ──────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_persist_supersedes_prior_active_and_bumps_state_version(test_db, test_user):
+    session = await _make_session(test_db, test_user)
+    initial_version = session.state_version
+
+    # Insert an existing active fix so we can verify supersession.
+    existing = SessionSuggestedFix(
+        session_id=session.id,
+        account_id=session.account_id,
+        title="Old fix",
+        description="prior",
+        confidence_pct=60,
+    )
+    test_db.add(existing)
+    await test_db.commit()
+
+    await _persist_suggested_fix(
+        db=test_db,
+        session=session,
+        fix={
+            "title": "New fix",
+            "description": "current best",
+            "confidence_pct": 88,
+            "script_template_slug": None,
+            "ai_drafted_script": None,
+            "ai_drafted_parameters": None,
+        },
+    )
+    await test_db.commit()
+    await test_db.refresh(existing)
+    await test_db.refresh(session)
+
+    assert existing.superseded_at is not None
+    assert session.state_version == initial_version + 1
+
+    # Exactly one active row remains — and it's the new one.
+    result = await test_db.execute(
+        select(SessionSuggestedFix).where(
+            SessionSuggestedFix.session_id == session.id,
+            SessionSuggestedFix.superseded_at.is_(None),
+        )
+    )
+    actives = list(result.scalars().all())
+    assert len(actives) == 1
+    assert actives[0].title == "New fix"
+
+
+# ── /suggested-fixes/active endpoint ────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_get_active_returns_404_when_none(client: AsyncClient, test_user, auth_headers, test_db):
+    session = await _make_session(test_db, test_user)
+    r = await client.get(
+        f"/api/v1/ai-sessions/{session.id}/suggested-fixes/active",
+        headers=auth_headers,
+    )
+    assert r.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_get_active_returns_active_fix(client: AsyncClient, test_user, auth_headers, test_db):
+    session = await _make_session(test_db, test_user)
+    fix = SessionSuggestedFix(
+        session_id=session.id,
+        account_id=session.account_id,
+        title="Active fix",
+        description="d",
+        confidence_pct=72,
+    )
+    test_db.add(fix)
+    await test_db.commit()
+
+    r = await client.get(
+        f"/api/v1/ai-sessions/{session.id}/suggested-fixes/active",
+        headers=auth_headers,
+    )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["title"] == "Active fix"
+    assert body["confidence_pct"] == 72
+    assert body["superseded_at"] is None
+
+
+# ── /decision endpoint ─────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_record_decision_persists_and_bumps_state_version(
+    client: AsyncClient, test_user, auth_headers, test_db
+):
+    session = await _make_session(test_db, test_user)
+    initial_version = session.state_version
+    fix = SessionSuggestedFix(
+        session_id=session.id,
+        account_id=session.account_id,
+        title="x",
+        description="y",
+        confidence_pct=50,
+    )
+    test_db.add(fix)
+    await test_db.commit()
+
+    r = await client.post(
+        f"/api/v1/ai-sessions/{session.id}/suggested-fixes/{fix.id}/decision",
+        headers=auth_headers,
+        json={"decision": "draft_template"},
+    )
+    assert r.status_code == 200
+    assert r.json()["user_decision"] == "draft_template"
+
+    await test_db.refresh(session)
+    assert session.state_version == initial_version + 1
+
+
+@pytest.mark.asyncio
+async def test_dismissed_supersedes_the_fix(
+    client: AsyncClient, test_user, auth_headers, test_db
+):
+    session = await _make_session(test_db, test_user)
+    fix = SessionSuggestedFix(
+        session_id=session.id,
+        account_id=session.account_id,
+        title="x",
+        description="y",
+        confidence_pct=50,
+    )
+    test_db.add(fix)
+    await test_db.commit()
+
+    r = await client.post(
+        f"/api/v1/ai-sessions/{session.id}/suggested-fixes/{fix.id}/decision",
+        headers=auth_headers,
+        json={"decision": "dismissed"},
+    )
+    assert r.status_code == 200
+    await test_db.refresh(fix)
+    assert fix.superseded_at is not None
+
+
+@pytest.mark.asyncio
+async def test_dismiss_already_superseded_returns_409(
+    client: AsyncClient, test_user, auth_headers, test_db
+):
+    session = await _make_session(test_db, test_user)
+    fix = SessionSuggestedFix(
+        session_id=session.id,
+        account_id=session.account_id,
+        title="x",
+        description="y",
+        confidence_pct=50,
+        superseded_at=datetime.now(timezone.utc),
+    )
+    test_db.add(fix)
+    await test_db.commit()
+
+    r = await client.post(
+        f"/api/v1/ai-sessions/{session.id}/suggested-fixes/{fix.id}/decision",
+        headers=auth_headers,
+        json={"decision": "dismissed"},
+    )
+    assert r.status_code == 409
+
+
+# ── /resolution-note/preview endpoint ──────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_preview_uses_state_version_cache(
+    client: AsyncClient, test_user, auth_headers, test_db
+):
+    session = await _make_session(test_db, test_user)
+
+    fake_provider = AsyncMock()
+    fake_provider.generate_text = AsyncMock(return_value=(
+        "## Problem\nx\n\n## What we confirmed\n(none)\n\n## Root cause\ny\n\n## Resolution\nz",
+        100, 50,
+    ))
+
+    with patch(
+        "app.services.resolution_note_generator.get_ai_provider",
+        return_value=fake_provider,
+    ):
+        # First call — cache miss, generates fresh.
+        r1 = await client.post(
+            f"/api/v1/ai-sessions/{session.id}/resolution-note/preview",
+            headers=auth_headers,
+        )
+        assert r1.status_code == 200
+        assert r1.json()["from_cache"] is False
+        assert fake_provider.generate_text.await_count == 1
+
+        # Second call, no state change — must hit the cache (no extra LLM call).
+        r2 = await client.post(
+            f"/api/v1/ai-sessions/{session.id}/resolution-note/preview",
+            headers=auth_headers,
+        )
+        assert r2.status_code == 200
+        assert r2.json()["from_cache"] is True
+        assert r2.json()["markdown"] == r1.json()["markdown"]
+        assert fake_provider.generate_text.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_preview_invalidates_after_fact_write(
+    client: AsyncClient, test_user, auth_headers, test_db
+):
+    """A new fact bumps state_version → next preview is a fresh generation, not cached."""
+    session = await _make_session(test_db, test_user)
+
+    fake_provider = AsyncMock()
+    fake_provider.generate_text = AsyncMock(return_value=(
+        "## Problem\nx\n\n## What we confirmed\n(none)\n\n## Root cause\ny\n\n## Resolution\nz",
+        100, 50,
+    ))
+
+    with patch(
+        "app.services.resolution_note_generator.get_ai_provider",
+        return_value=fake_provider,
+    ):
+        await client.post(
+            f"/api/v1/ai-sessions/{session.id}/resolution-note/preview",
+            headers=auth_headers,
+        )
+        assert fake_provider.generate_text.await_count == 1
+
+        # Add a fact — bumps state_version on the session.
+        await client.post(
+            f"/api/v1/ai-sessions/{session.id}/facts",
+            headers=auth_headers,
+            json={"text": "a confirmed observation"},
+        )
+
+        # Next preview must regenerate (cache key includes state_version).
+        r = await client.post(
+            f"/api/v1/ai-sessions/{session.id}/resolution-note/preview",
+            headers=auth_headers,
+        )
+        assert r.status_code == 200
+        assert r.json()["from_cache"] is False
+        assert fake_provider.generate_text.await_count == 2

docs/FlowAssist_Migration/FLOWPILOT-MIGRATION.md

@@ -2,8 +2,8 @@
 
 > **Target:** Transform `/assistant` (ResolutionAssist) into the new unified `/pilot` (FlowPilot) surface.
 > **Audience:** Claude Code (implementation) and Codex (review) reviewed by Michael (owner).
-> **Status:** Phase 0 in progress. Phases 1–7 awaiting Phase 0 completion for the AI-dependent work; Phase 1 can run in parallel.
-> **Last updated:** April 17, 2026 (post-Codex plan review, reflects Phase 0 audit findings and in-flight implementation decisions)
+> **Status:** Phases 0–3 implemented and verified end-to-end against the dev stack. Phase 4 next.
+> **Last updated:** April 22, 2026 (Phase 3 — Suggested fix + Resolve preview — committed; live Sonnet preview + state_version cache verified)
 
 ---
 
@@ -30,6 +30,10 @@ This document was originally written against a set of assumptions about the code
 - **`pending_task_lane` items do not have stable IDs today.** Phase 2 must assign stable UUIDs when questions/checks are first persisted. `session_facts.source_ref` points to those JSON item IDs.
 - **`account_settings` table did not exist.** Phase 1 creates it with a JSONB `preferences` column; settings live in `preferences` until they need their own column.
 - **`/tickets/ai-parse` endpoint does not exist.** Phase 0.2 became a doc-only note; no code change.
+- **`[PROMOTE]` marker uses JSON, not key:value.** The doc's original example showed
+  `key: value` lines; implementation uses a JSON object inside each `[PROMOTE]...[/PROMOTE]`
+  block (matching the `[QUESTIONS]` / `[ACTIONS]` parser pattern). Multi-line text values would
+  have been ambiguous in the key:value form. Section 8.1 below has been updated.
 
 Any further drift found during implementation should be flagged by the implementer and reconciled in this doc before writing code that assumes the drifted spec.
 
@@ -572,18 +576,15 @@ The existing FlowPilot / ResolutionAssist system prompt needs updates to emit th
 
 ### 8.1 New marker: `[PROMOTE]`
 
-Used to surface facts to What we know. Syntax:
+Used to surface facts to What we know. Syntax — each block contains a single JSON object; multiple blocks may appear in one response:
 
 ```
 [PROMOTE]
-source_type: question
-source_ref: {task_lane_item_uuid}
-text: OWA login and send/receive confirmed working for jsmith
-summary: rules out tenant/license
+{"source_type": "question", "source_ref": "{task_lane_item_uuid}", "text": "OWA login and send/receive confirmed working for jsmith", "summary": "rules out tenant/license"}
 [/PROMOTE]
 ```
 
-The AI should emit `[PROMOTE]` blocks in the same message that answers or processes a question/check, so the fact appears in What we know simultaneously with the chat acknowledgment. `source_ref` points to the stable UUID of the task lane item being promoted (assigned in Phase 2).
+The AI should emit `[PROMOTE]` blocks in the same message that answers or processes a question/check, so the fact appears in What we know simultaneously with the chat acknowledgment. `source_ref` points to the stable UUID of the task lane item being promoted (assigned in Phase 2). For `source_type: "ai_synthesis"`, omit `source_ref` (or set it to null) — the parser drops it defensively even if the model includes one.
 
 ### 8.2 New marker: `[SUGGEST_FIX]`
 
@@ -709,6 +710,11 @@ git commit -m "feat(pilot): rename /assistant to /pilot, add session_facts/sugge
 - Attempt to PATCH a question-sourced fact → 403.
 - PATCH a user_note fact → succeeds.
 
+**Verification deferred** — same constraint as Phase 0: no live dev environment was
+available at authoring time. Backend pytest suite (`tests/test_session_facts_api.py`)
+and the manual scenarios above must run when the dev env is up. Failures should be
+treated as normal bugs, not blockers for Phase 3.
+
 ```
 git commit -m "feat(pilot): add What we know section with fact synthesis and stable task-lane item IDs"
 ```
@@ -733,6 +739,15 @@ git commit -m "feat(pilot): add What we know section with fact synthesis and sta
 - Preview contains no hallucinated information not present in session state (human review of 5 real-ish sessions).
 - Incrementing `state_version` invalidates the preview cache; reading the same version returns the cached markdown.
 
+**Verified end-to-end** against the dev stack on 2026-04-22:
+- `/suggested-fixes/active` → 404 when no fix; 200 with payload when one exists.
+- Fact write bumps `state_version`; preview cache invalidates as expected.
+- Sonnet generates well-formed four-section markdown grounded only in
+  provided facts (single-fact session correctly says "Root cause not
+  definitively isolated").
+- Second consecutive preview call with no state change returns
+  `from_cache=true` and emits no LLM call.
+
 ```
 git commit -m "feat(pilot): add suggested fix tracking and Resolve note preview with state_version caching"
 ```

frontend/src/api/sessionFacts.ts

@@ -0,0 +1,89 @@
+/**
+ * Session facts API — the "What we know" CRUD surface for a FlowPilot session.
+ *
+ * Mirrors backend endpoints at `/api/v1/ai-sessions/{id}/facts`.
+ * See FLOWPILOT-MIGRATION.md Section 5.1.
+ */
+import apiClient from './client'
+
+export type SessionFactSourceType =
+  | 'question'
+  | 'diagnostic_check'
+  | 'user_note'
+  | 'ai_synthesis'
+
+export interface SessionFact {
+  id: string
+  session_id: string
+  text: string
+  source_type: SessionFactSourceType
+  source_ref: string | null
+  source_summary: string | null
+  created_by: string
+  created_at: string
+  updated_at: string
+  // Server-computed: false for question/diagnostic_check (PATCH returns 403),
+  // true for user_note/ai_synthesis. Drives the edit affordance in the UI.
+  editable: boolean
+}
+
+export interface SessionFactCreateRequest {
+  text: string
+  summary?: string | null
+}
+
+export interface SessionFactUpdateRequest {
+  text?: string | null
+  summary?: string | null
+}
+
+export interface SessionFactPromoteRequest {
+  source_type: 'question' | 'diagnostic_check' | 'ai_synthesis'
+  source_ref?: string | null
+  proposed_text?: string | null
+  proposed_summary?: string | null
+  raw_input?: string | null
+}
+
+export const sessionFactsApi = {
+  async list(sessionId: string): Promise<SessionFact[]> {
+    const r = await apiClient.get<{ facts: SessionFact[] }>(
+      `/ai-sessions/${sessionId}/facts`,
+    )
+    return r.data.facts
+  },
+
+  async create(sessionId: string, data: SessionFactCreateRequest): Promise<SessionFact> {
+    const r = await apiClient.post<SessionFact>(
+      `/ai-sessions/${sessionId}/facts`,
+      data,
+    )
+    return r.data
+  },
+
+  async update(
+    sessionId: string,
+    factId: string,
+    data: SessionFactUpdateRequest,
+  ): Promise<SessionFact> {
+    const r = await apiClient.patch<SessionFact>(
+      `/ai-sessions/${sessionId}/facts/${factId}`,
+      data,
+    )
+    return r.data
+  },
+
+  async remove(sessionId: string, factId: string): Promise<void> {
+    await apiClient.delete(`/ai-sessions/${sessionId}/facts/${factId}`)
+  },
+
+  async promote(sessionId: string, data: SessionFactPromoteRequest): Promise<SessionFact> {
+    const r = await apiClient.post<SessionFact>(
+      `/ai-sessions/${sessionId}/facts/promote`,
+      data,
+    )
+    return r.data
+  },
+}
+
+export default sessionFactsApi

frontend/src/api/sessionSuggestedFixes.ts

@@ -0,0 +1,84 @@
+/**
+ * Session suggested-fix + resolution-note preview API (Phase 3).
+ *
+ * Mirrors backend endpoints under /api/v1/ai-sessions/{id}/...
+ * See FLOWPILOT-MIGRATION.md Sections 5.2 + 5.4.
+ */
+import apiClient from './client'
+
+export type UserDecision = 'one_off' | 'draft_template' | 'build_template' | 'dismissed'
+
+export interface SessionSuggestedFix {
+  id: string
+  session_id: string
+  title: string
+  description: string
+  confidence_pct: number
+  script_template_id: string | null
+  ai_drafted_script: string | null
+  ai_drafted_parameters: Record<string, unknown> | null
+  user_decision: UserDecision | null
+  superseded_at: string | null
+  created_at: string
+}
+
+export interface DecisionResponse {
+  id: string
+  user_decision: UserDecision
+  rendered_script: string | null
+  redirect_path: string | null
+}
+
+export interface ResolutionNotePreview {
+  markdown: string
+  target_ticket_ref: string | null
+  state_version: number
+  from_cache: boolean
+}
+
+export const sessionSuggestedFixesApi = {
+  /**
+   * Returns the active suggested fix for a session, or `null` if there isn't one.
+   * The endpoint returns 404 in the no-fix case, which is normal — we coerce
+   * to null so callers don't have to distinguish "no fix" from "request failed".
+   */
+  async getActive(sessionId: string): Promise<SessionSuggestedFix | null> {
+    try {
+      const r = await apiClient.get<SessionSuggestedFix>(
+        `/ai-sessions/${sessionId}/suggested-fixes/active`,
+      )
+      return r.data
+    } catch (err) {
+      const status = (err as { response?: { status?: number } })?.response?.status
+      if (status === 404) return null
+      throw err
+    }
+  },
+
+  async recordDecision(
+    sessionId: string,
+    fixId: string,
+    decision: UserDecision,
+  ): Promise<DecisionResponse> {
+    const r = await apiClient.post<DecisionResponse>(
+      `/ai-sessions/${sessionId}/suggested-fixes/${fixId}/decision`,
+      { decision },
+    )
+    return r.data
+  },
+
+  /**
+   * Fetch (or get cached) draft markdown for the Resolve note. Backend cache
+   * is keyed on state_version, so calling this back-to-back without intervening
+   * fact / suggested-fix / script-generation writes returns the same payload
+   * cheaply (no Sonnet call).
+   */
+  async getResolutionNotePreview(sessionId: string): Promise<ResolutionNotePreview> {
+    const r = await apiClient.post<ResolutionNotePreview>(
+      `/ai-sessions/${sessionId}/resolution-note/preview`,
+    )
+    return r.data
+  },
+}
+
+export default sessionSuggestedFixesApi

frontend/src/components/assistant/TaskLane.tsx

@@ -38,6 +38,17 @@ interface TaskLaneProps {
   onSubmit: (responses: TaskResponse[]) => void
   onClose: () => void
   loading?: boolean
+  // Slot for the FlowPilot Phase 2 "What we know" section. Rendered above
+  // Questions in the body (per FLOWPILOT-MIGRATION.md Section 3.1). The slot
+  // shape lets the parent own fact-fetching and state-version polling without
+  // pulling that concern into TaskLane.
+  whatWeKnowSlot?: React.ReactNode
+  // Phase 3: Suggested fix card, rendered below Diagnostic Checks.
+  suggestedFixSlot?: React.ReactNode
+  // Phase 3: bottom-of-lane slot for the Resolve action bar + preview popover
+  // (parent owns state). Renders inside the scrollable body so the popover
+  // stays anchored as the lane scrolls.
+  bottomSlot?: React.ReactNode
 }
 
 // ── Storage helpers ──
@@ -64,7 +75,7 @@ export function clearTaskState(sessionId: string) {
 
 // ── Component ──
 
-export function TaskLane({ questions, actions, sessionId, onSubmit, onClose, loading }: TaskLaneProps) {
+export function TaskLane({ questions, actions, sessionId, onSubmit, onClose, loading, whatWeKnowSlot, suggestedFixSlot, bottomSlot }: TaskLaneProps) {
   const [tasks, setTasks] = useState<TaskResponse[]>(() => {
     // Try to restore saved state for this session (preserves user's in-progress answers)
     if (sessionId) {
@@ -269,6 +280,9 @@ export function TaskLane({ questions, actions, sessionId, onSubmit, onClose, loa
       {/* Body */}
       <div className="flex-1 overflow-y-auto p-3 space-y-4">
 
+        {/* ── What we know (Phase 2) ── */}
+        {whatWeKnowSlot}
+
         {/* ── Questions Section ── */}
         {questionTasks.length > 0 && (
           <section>
@@ -495,6 +509,12 @@ export function TaskLane({ questions, actions, sessionId, onSubmit, onClose, loa
             })}
           </section>
         )}
+
+        {/* ── Suggested fix (Phase 3) ── */}
+        {suggestedFixSlot}
+
+        {/* ── Resolve action bar + preview popover (Phase 3) ── */}
+        {bottomSlot}
       </div>
 
       {/* Footer */}

frontend/src/components/pilot/ResolutionNotePreview.tsx

@@ -0,0 +1,107 @@
+/**
+ * ResolutionNotePreview — Phase 3 popover anchored to the Resolve action area.
+ *
+ * Persistent (not modal) popover showing the four-section draft markdown that
+ * would be posted to the customer ticket on Resolve. Per FLOWPILOT-MIGRATION.md
+ * Section 3.1, the engineer reviews/edits the draft inline and Confirm & post
+ * fires the PSA writeback (wired in Phase 4 — for now this is read-only).
+ *
+ * Refresh policy: parent triggers `onRefresh` when state_version changes.
+ * Backend caches by state_version, so repeat fetches are cheap (no Sonnet
+ * call) when no facts/fixes/scripts have changed.
+ */
+import { useState } from 'react'
+import { Loader2, RefreshCw, X, FileText } from 'lucide-react'
+import { MarkdownContent } from '@/components/ui/MarkdownContent'
+import type { ResolutionNotePreview as PreviewData } from '@/api/sessionSuggestedFixes'
+
+interface ResolutionNotePreviewProps {
+  open: boolean
+  loading: boolean
+  preview: PreviewData | null
+  error: string | null
+  onClose: () => void
+  onRefresh: () => Promise<void> | void
+}
+
+export function ResolutionNotePreview({
+  open,
+  loading,
+  preview,
+  error,
+  onClose,
+  onRefresh,
+}: ResolutionNotePreviewProps) {
+  const [refreshing, setRefreshing] = useState(false)
+
+  if (!open) return null
+
+  const handleRefresh = async () => {
+    setRefreshing(true)
+    try { await onRefresh() } finally { setRefreshing(false) }
+  }
+
+  return (
+    // The popover is positioned absolutely against its anchor by the parent.
+    // We render full-width inside the task lane below the Resolve action bar.
+    <div className="rounded-lg border border-default bg-elevated/30 mx-3 mb-3 overflow-hidden shadow-lg">
+      <div className="flex items-center justify-between px-3 py-2 border-b border-default bg-bg-page">
+        <div className="flex items-center gap-2">
+          <FileText size={13} className="text-accent" />
+          <span className="text-[0.75rem] font-semibold text-heading">
+            Resolution note preview
+          </span>
+          {preview?.target_ticket_ref && (
+            <span className="text-[0.6875rem] font-mono text-accent-text">
+              → {preview.target_ticket_ref}
+            </span>
+          )}
+          {preview?.from_cache && (
+            <span className="text-[0.6875rem] text-muted-foreground italic">cached</span>
+          )}
+        </div>
+        <div className="flex items-center gap-1">
+          <button
+            onClick={handleRefresh}
+            disabled={refreshing || loading}
+            className="p-1 rounded text-muted-foreground hover:text-heading hover:bg-elevated/40 transition-colors disabled:opacity-40"
+            title="Refresh preview"
+          >
+            {refreshing ? <Loader2 size={11} className="animate-spin" /> : <RefreshCw size={11} />}
+          </button>
+          <button
+            onClick={onClose}
+            className="p-1 rounded text-muted-foreground hover:text-heading hover:bg-elevated/40 transition-colors"
+            title="Close preview"
+          >
+            <X size={11} />
+          </button>
+        </div>
+      </div>
+
+      <div className="px-3 py-3 max-h-[40vh] overflow-y-auto">
+        {loading && !preview && (
+          <div className="flex items-center gap-2 text-[0.75rem] text-muted-foreground">
+            <Loader2 size={12} className="animate-spin" />
+            Drafting note from session state...
+          </div>
+        )}
+        {error && (
+          <div className="text-[0.75rem] text-danger">{error}</div>
+        )}
+        {preview && (
+          <div className="prose prose-invert prose-sm max-w-none text-[0.8125rem] leading-relaxed">
+            <MarkdownContent content={preview.markdown} />
+          </div>
+        )}
+        {!loading && !error && !preview && (
+          <div className="text-[0.75rem] text-muted-foreground italic">
+            No preview yet — add a fact or accept a suggested fix to populate.
+          </div>
+        )}
+      </div>
+    </div>
+  )
+}
+
+export default ResolutionNotePreview

frontend/src/components/pilot/sections/AddNoteButton.tsx

@@ -0,0 +1,87 @@
+/**
+ * "+ Add a note" affordance for the What-we-know section.
+ *
+ * Inline composer that posts a `user_note` fact when the engineer wants to
+ * record something the AI didn't surface (a hunch, an observation, a piece
+ * of customer context). Per FLOWPILOT-MIGRATION.md Section 3.1.
+ */
+import { useState } from 'react'
+import { Plus, Check } from 'lucide-react'
+
+interface AddNoteButtonProps {
+  onAdd: (text: string, summary: string | null) => Promise<void> | void
+}
+
+export function AddNoteButton({ onAdd }: AddNoteButtonProps) {
+  const [open, setOpen] = useState(false)
+  const [text, setText] = useState('')
+  const [summary, setSummary] = useState('')
+  const [busy, setBusy] = useState(false)
+
+  const reset = () => {
+    setText('')
+    setSummary('')
+    setOpen(false)
+  }
+
+  const handleSubmit = async () => {
+    if (!text.trim()) return
+    setBusy(true)
+    try {
+      await onAdd(text.trim(), summary.trim() || null)
+      reset()
+    } finally {
+      setBusy(false)
+    }
+  }
+
+  if (!open) {
+    return (
+      <button
+        onClick={() => setOpen(true)}
+        className="flex items-center gap-1.5 text-[0.75rem] text-muted-foreground hover:text-accent-text transition-colors pl-1 mt-1"
+      >
+        <Plus size={12} />
+        Add a note
+      </button>
+    )
+  }
+
+  return (
+    <div className="rounded-lg border border-default bg-card p-3 mb-2">
+      <textarea
+        autoFocus
+        value={text}
+        onChange={(e) => setText(e.target.value)}
+        placeholder="What did you observe or confirm?"
+        className="w-full rounded-md border border-default bg-input px-2.5 py-1.5 text-[0.8125rem] text-heading placeholder:text-muted-foreground resize-y min-h-[44px] max-h-[140px] focus:outline-none focus:border-accent focus:ring-1 focus:ring-accent/30"
+        rows={2}
+      />
+      <input
+        type="text"
+        value={summary}
+        onChange={(e) => setSummary(e.target.value)}
+        placeholder="Short label (optional)"
+        className="mt-1.5 w-full rounded-md border border-default bg-input px-2.5 py-1 text-[0.75rem] text-heading placeholder:text-muted-foreground focus:outline-none focus:border-accent focus:ring-1 focus:ring-accent/30"
+      />
+      <div className="mt-1.5 flex items-center gap-2">
+        <button
+          onClick={handleSubmit}
+          disabled={busy || !text.trim()}
+          className="flex items-center gap-1 rounded-md bg-accent px-2.5 py-1 text-[0.75rem] font-medium text-white disabled:opacity-40 hover:bg-accent-hover transition-colors"
+        >
+          <Check size={11} /> Add
+        </button>
+        <button
+          onClick={reset}
+          disabled={busy}
+          className="text-[0.75rem] text-muted-foreground hover:text-heading"
+        >
+          Cancel
+        </button>
+      </div>
+    </div>
+  )
+}
+
+export default AddNoteButton

frontend/src/components/pilot/sections/SuggestedFix.tsx

@@ -0,0 +1,82 @@
+/**
+ * SuggestedFix card — Phase 3 task-lane section.
+ *
+ * Renders the active AI-proposed resolution path for the session
+ * (per FLOWPILOT-MIGRATION.md Section 3.1, "Suggested fix"). Amber-accented
+ * to match the mockup; clicking opens the Script Generator flow in Phase 5.
+ *
+ * For Phase 3, the card is informational + a Dismiss action. The three-option
+ * dialog (one_off / draft_template / build_template) is wired in Phase 5
+ * via a separate component.
+ */
+import { useState } from 'react'
+import { Sparkles, X } from 'lucide-react'
+import type { SessionSuggestedFix } from '@/api/sessionSuggestedFixes'
+
+interface SuggestedFixProps {
+  fix: SessionSuggestedFix
+  onDismiss: () => Promise<void> | void
+}
+
+function confidenceBucket(pct: number): { label: string; tone: string } {
+  if (pct >= 80) return { label: 'high', tone: 'text-success' }
+  if (pct >= 50) return { label: 'medium', tone: 'text-warning' }
+  return { label: 'low', tone: 'text-muted-foreground' }
+}
+
+export function SuggestedFix({ fix, onDismiss }: SuggestedFixProps) {
+  const [busy, setBusy] = useState(false)
+  const conf = confidenceBucket(fix.confidence_pct)
+
+  const handleDismiss = async () => {
+    setBusy(true)
+    try { await onDismiss() } finally { setBusy(false) }
+  }
+
+  return (
+    <section>
+      <div className="sticky top-0 z-10 pb-2" style={{ background: 'var(--color-bg-page)' }}>
+        <div className="flex items-center gap-2 text-[10px] font-semibold uppercase tracking-[1.2px] text-muted-foreground pl-0.5">
+          <span className="w-1.5 h-1.5 rounded-full bg-warning" />
+          Suggested fix
+          <span className="text-muted-foreground">·</span>
+          <span className={`tabular-nums ${conf.tone}`}>{fix.confidence_pct}% confidence</span>
+        </div>
+      </div>
+
+      <div className="rounded-lg border-l-[3px] border-l-warning border border-warning/25 bg-warning-dim/15 p-3 mb-2">
+        <div className="flex items-start gap-2">
+          <Sparkles size={14} className="text-warning shrink-0 mt-0.5" />
+          <div className="min-w-0 flex-1">
+            <div className="text-[0.8125rem] font-medium text-heading leading-snug">
+              {fix.title}
+            </div>
+            <div className="mt-1 text-[0.75rem] text-muted-foreground leading-relaxed">
+              {fix.description}
+            </div>
+            {fix.script_template_id && (
+              <div className="mt-1.5 text-[0.6875rem] text-success">
+                ✓ Matches an existing Script Library template
+              </div>
+            )}
+            {!fix.script_template_id && fix.ai_drafted_script && (
+              <div className="mt-1.5 text-[0.6875rem] text-accent-text">
+                Custom script drafted (no template match)
+              </div>
+            )}
+          </div>
+          <button
+            onClick={handleDismiss}
+            disabled={busy}
+            className="p-1 rounded text-muted-foreground hover:text-danger hover:bg-elevated/40 transition-colors"
+            title="Dismiss this suggestion"
+          >
+            <X size={11} />
+          </button>
+        </div>
+      </div>
+    </section>
+  )
+}
+
+export default SuggestedFix

frontend/src/components/pilot/sections/WhatWeKnow.tsx

@@ -0,0 +1,74 @@
+/**
+ * What-we-know section — the load-bearing structural feature of the FlowPilot
+ * task lane (per FLOWPILOT-MIGRATION.md Section 0 architectural claim #2).
+ *
+ * Owns the list of confirmed facts for a session. Facts arrive via three paths:
+ *   1. AI [PROMOTE] markers parsed in unified_chat_service (most common)
+ *   2. Engineer "+ Add a note" (manual user_note)
+ *   3. Engineer-initiated promotion of a question/check (future affordance)
+ *
+ * This component is the parent's contract: it takes a fact list + handlers
+ * and renders the section. Loading/refresh logic lives in the parent
+ * (AssistantChatPage) so it can coordinate with the chat send cycle.
+ */
+import { cn } from '@/lib/utils'
+import type { SessionFact } from '@/api/sessionFacts'
+import { WhatWeKnowItem } from './WhatWeKnowItem'
+import { AddNoteButton } from './AddNoteButton'
+
+interface WhatWeKnowProps {
+  facts: SessionFact[]
+  onAddNote: (text: string, summary: string | null) => Promise<void> | void
+  onUpdateFact: (factId: string, text: string, summary: string | null) => Promise<void> | void
+  onDeleteFact: (factId: string) => Promise<void> | void
+  loading?: boolean
+}
+
+export function WhatWeKnow({
+  facts,
+  onAddNote,
+  onUpdateFact,
+  onDeleteFact,
+  loading,
+}: WhatWeKnowProps) {
+  const count = facts.length
+
+  return (
+    <section
+      className={cn(
+        'rounded-lg p-3 -mx-1 mb-1',
+        // Subtle green-to-transparent gradient distinguishes this section
+        // from the rest of the lane (mockup 01-session-primary.png).
+        'bg-gradient-to-b from-success/[0.05] to-transparent',
+      )}
+    >
+      <div className="sticky top-0 z-10 pb-2" style={{ background: 'var(--color-bg-page)' }}>
+        <div className="flex items-center gap-2 text-[10px] font-semibold uppercase tracking-[1.2px] text-muted-foreground pl-0.5">
+          <span className="w-1.5 h-1.5 rounded-full bg-success" />
+          What we know
+          <span className="text-muted-foreground">·</span>
+          <span className="tabular-nums">{count}</span>
+        </div>
+      </div>
+
+      {count === 0 && !loading && (
+        <div className="text-[0.75rem] text-muted-foreground italic px-1 py-2">
+          Nothing confirmed yet — facts appear here as the engineer answers questions and runs checks.
+        </div>
+      )}
+
+      {facts.map((fact) => (
+        <WhatWeKnowItem
+          key={fact.id}
+          fact={fact}
+          onSave={(text, summary) => onUpdateFact(fact.id, text, summary)}
+          onDelete={() => onDeleteFact(fact.id)}
+        />
+      ))}
+
+      <AddNoteButton onAdd={onAddNote} />
+    </section>
+  )
+}
+
+export default WhatWeKnow

frontend/src/components/pilot/sections/WhatWeKnowItem.tsx

@@ -0,0 +1,152 @@
+/**
+ * Single fact card in the What-we-know section.
+ *
+ * Layout per FLOWPILOT-MIGRATION.md Section 3.1:
+ *   [✓] <fact text>
+ *       <provenance line: 'from question · rules out tenant/license'>
+ *
+ * Editability: the server marks `editable=false` for `question` and
+ * `diagnostic_check` facts (those are edited at the source item, not here).
+ * Manual notes and AI-synthesized facts are editable inline.
+ *
+ * Delete is allowed for all source types — even read-only facts may be
+ * removed when the underlying question or check turned out to be wrong.
+ */
+import { useState } from 'react'
+import { Check, Pencil, Trash2, X } from 'lucide-react'
+import { cn } from '@/lib/utils'
+import type { SessionFact, SessionFactSourceType } from '@/api/sessionFacts'
+
+interface WhatWeKnowItemProps {
+  fact: SessionFact
+  onSave: (text: string, summary: string | null) => Promise<void> | void
+  onDelete: () => Promise<void> | void
+}
+
+const SOURCE_LABEL: Record<SessionFactSourceType, string> = {
+  question: 'from question',
+  diagnostic_check: 'from check',
+  user_note: 'manual note',
+  ai_synthesis: 'synthesized',
+}
+
+export function WhatWeKnowItem({ fact, onSave, onDelete }: WhatWeKnowItemProps) {
+  const [editing, setEditing] = useState(false)
+  const [draftText, setDraftText] = useState(fact.text)
+  const [draftSummary, setDraftSummary] = useState(fact.source_summary ?? '')
+  const [busy, setBusy] = useState(false)
+
+  const handleSave = async () => {
+    if (!draftText.trim()) return
+    setBusy(true)
+    try {
+      await onSave(draftText.trim(), draftSummary.trim() || null)
+      setEditing(false)
+    } finally {
+      setBusy(false)
+    }
+  }
+
+  const handleCancel = () => {
+    setDraftText(fact.text)
+    setDraftSummary(fact.source_summary ?? '')
+    setEditing(false)
+  }
+
+  const handleDelete = async () => {
+    setBusy(true)
+    try {
+      await onDelete()
+    } finally {
+      setBusy(false)
+    }
+  }
+
+  const provenanceLabel = SOURCE_LABEL[fact.source_type]
+
+  if (editing) {
+    return (
+      <div className="rounded-lg border border-default bg-card p-3 mb-2">
+        <textarea
+          autoFocus
+          value={draftText}
+          onChange={(e) => setDraftText(e.target.value)}
+          className="w-full rounded-md border border-default bg-input px-2.5 py-1.5 text-[0.8125rem] text-heading resize-y min-h-[44px] max-h-[140px] focus:outline-none focus:border-accent focus:ring-1 focus:ring-accent/30"
+          rows={2}
+        />
+        <input
+          type="text"
+          value={draftSummary}
+          onChange={(e) => setDraftSummary(e.target.value)}
+          placeholder="Short label (e.g. 'rules out tenant/license')"
+          className="mt-1.5 w-full rounded-md border border-default bg-input px-2.5 py-1 text-[0.75rem] text-heading placeholder:text-muted-foreground focus:outline-none focus:border-accent focus:ring-1 focus:ring-accent/30"
+        />
+        <div className="mt-1.5 flex items-center gap-2">
+          <button
+            onClick={handleSave}
+            disabled={busy || !draftText.trim()}
+            className="flex items-center gap-1 rounded-md bg-accent px-2.5 py-1 text-[0.75rem] font-medium text-white disabled:opacity-40 hover:bg-accent-hover transition-colors"
+          >
+            <Check size={11} /> Save
+          </button>
+          <button
+            onClick={handleCancel}
+            disabled={busy}
+            className="text-[0.75rem] text-muted-foreground hover:text-heading"
+          >
+            Cancel
+          </button>
+        </div>
+      </div>
+    )
+  }
+
+  return (
+    <div
+      className={cn(
+        'group rounded-lg border-l-[3px] border-l-success border border-success/25 bg-success-dim/20 p-3 mb-2',
+        busy && 'opacity-60',
+      )}
+    >
+      <div className="flex items-start gap-2">
+        <div className="mt-0.5 flex h-4 w-4 shrink-0 items-center justify-center rounded-full border border-dashed border-success">
+          <Check size={9} className="text-success" />
+        </div>
+        <div className="min-w-0 flex-1">
+          <div className="text-[0.8125rem] text-heading leading-relaxed">{fact.text}</div>
+          <div className="mt-1 text-[0.6875rem] text-muted-foreground">
+            <span>{provenanceLabel}</span>
+            {fact.source_summary && (
+              <>
+                <span className="mx-1">·</span>
+                <span className="italic">{fact.source_summary}</span>
+              </>
+            )}
+          </div>
+        </div>
+        <div className="flex shrink-0 items-center gap-1 opacity-0 group-hover:opacity-100 transition-opacity">
+          {fact.editable && (
+            <button
+              onClick={() => setEditing(true)}
+              disabled={busy}
+              className="p-1 rounded text-muted-foreground hover:text-heading hover:bg-elevated/40 transition-colors"
+              title="Edit fact"
+            >
+              <Pencil size={11} />
+            </button>
+          )}
+          <button
+            onClick={handleDelete}
+            disabled={busy}
+            className="p-1 rounded text-muted-foreground hover:text-danger hover:bg-elevated/40 transition-colors"
+            title="Remove fact"
+          >
+            {busy ? <X size={11} /> : <Trash2 size={11} />}
+          </button>
+        </div>
+      </div>
+    </div>
+  )
+}
+
+export default WhatWeKnowItem

frontend/src/pages/AssistantChatPage.tsx

@@ -13,6 +13,15 @@ import { toast } from '@/lib/toast'
 import { ChatSidebar, ChatSidebarCollapsedBar } from '@/components/assistant/ChatSidebar'
 import { ChatMessage } from '@/components/assistant/ChatMessage'
 import { TaskLane, clearTaskState } from '@/components/assistant/TaskLane'
+import { WhatWeKnow } from '@/components/pilot/sections/WhatWeKnow'
+import { SuggestedFix } from '@/components/pilot/sections/SuggestedFix'
+import { ResolutionNotePreview as ResolutionNotePreviewPopover } from '@/components/pilot/ResolutionNotePreview'
+import { sessionFactsApi, type SessionFact } from '@/api/sessionFacts'
+import {
+  sessionSuggestedFixesApi,
+  type SessionSuggestedFix,
+  type ResolutionNotePreview as ResolutionNotePreviewData,
+} from '@/api/sessionSuggestedFixes'
 import { ConcludeSessionModal } from '@/components/assistant/ConcludeSessionModal'
 import { StatusUpdateModal } from '@/components/flowpilot/StatusUpdateModal'
 import type { ChatListItem, ConclusionOutcome } from '@/types/assistant-chat'
@@ -74,6 +83,20 @@ export default function AssistantChatPage() {
   )
   const [activeSessionStatus, setActiveSessionStatus] = useState<string | null>(null)
   const [activePsaTicketId, setActivePsaTicketId] = useState<string | null>(null)
+  // Phase 2: "What we know" facts for the active session. Refreshed on
+  // selectChat and after each chat send (the AI may have emitted [PROMOTE]
+  // markers that synthesized new facts server-side).
+  const [facts, setFacts] = useState<SessionFact[]>([])
+  // Phase 3: active suggested fix + resolution-note preview state.
+  const [activeFix, setActiveFix] = useState<SessionSuggestedFix | null>(null)
+  const [previewOpen, setPreviewOpen] = useState(false)
+  const [previewData, setPreviewData] = useState<ResolutionNotePreviewData | null>(null)
+  const [previewLoading, setPreviewLoading] = useState(false)
+  const [previewError, setPreviewError] = useState<string | null>(null)
+  // Debounce timer for preview refresh — Phase 3 spec calls for 500ms client-
+  // side debounce so rapid edits don't fan out to the LLM (cache absorbs the
+  // dups, but the request itself still costs HTTP RTT).
+  const previewDebounceRef = useRef<ReturnType<typeof setTimeout> | null>(null)
   const [showOverflow, setShowOverflow] = useState(false)
   const toggleSidebarCollapse = () => {
     const next = !sidebarCollapsed
@@ -178,6 +201,8 @@ export default function AssistantChatPage() {
           setActiveActions(response.actions || [])
           setShowTaskLane(true)
         }
+        // Refetch facts + active fix — the AI may have emitted markers.
+        refreshSessionDerived(session.session_id)
       } catch {
         toast.error('Failed to start AI conversation')
       } finally {
@@ -222,6 +247,134 @@ export default function AssistantChatPage() {
     }
   }
 
+  // Phase 2 facts — fetch + handlers. `refreshFacts` is called from selectChat
+  // and after each chat send, because the AI may have emitted [PROMOTE] markers
+  // that synthesized new facts server-side (see unified_chat_service.
+  // _persist_promote_items).
+  const refreshFacts = useCallback(async (chatId: string) => {
+    try {
+      const list = await sessionFactsApi.list(chatId)
+      // Guard: discard stale fetch if the user switched chats mid-flight.
+      if (currentChatRef.current !== chatId) return
+      setFacts(list)
+      // Auto-open the task lane when the session has facts so the engineer
+      // can see them — without this, a session with only facts (no open
+      // questions) would hide the lane and the facts would be invisible.
+      if (list.length > 0) setShowTaskLane(true)
+    } catch {
+      // Best-effort — facts are accessory state. Surfacing a toast on every
+      // refetch failure would be noisy; the empty state explains the absence.
+    }
+  }, [])
+
+  // Phase 3 — active suggested fix + resolution-note preview.
+  // Declared BEFORE refreshSessionDerived / handleAddNote so the useCallback
+  // dep arrays don't hit a temporal dead zone on React's synchronous render.
+  const refreshActiveFix = useCallback(async (chatId: string) => {
+    try {
+      const fix = await sessionSuggestedFixesApi.getActive(chatId)
+      if (currentChatRef.current !== chatId) return
+      setActiveFix(fix)
+    } catch {
+      // No-fix-yet (404) is normalized to null inside the client. Genuine
+      // failures stay silent — accessory state, not load-bearing.
+    }
+  }, [])
+
+  const refreshPreview = useCallback(async (chatId: string) => {
+    setPreviewLoading(true)
+    setPreviewError(null)
+    try {
+      const p = await sessionSuggestedFixesApi.getResolutionNotePreview(chatId)
+      if (currentChatRef.current !== chatId) return
+      setPreviewData(p)
+    } catch (err: unknown) {
+      const status = (err as { response?: { status?: number } })?.response?.status
+      setPreviewError(
+        status === 502
+          ? 'AI provider error drafting the note. Try again in a few seconds.'
+          : 'Could not load preview.',
+      )
+    } finally {
+      setPreviewLoading(false)
+    }
+  }, [])
+
+  // Trigger preview refresh with a 500ms debounce. The backend cache short-
+  // circuits same-state calls, but the network round-trip is still avoidable
+  // when the user is typing quickly (e.g. editing a fact).
+  const schedulePreviewRefresh = useCallback((chatId: string) => {
+    if (previewDebounceRef.current) clearTimeout(previewDebounceRef.current)
+    previewDebounceRef.current = setTimeout(() => {
+      if (previewOpen && currentChatRef.current === chatId) {
+        refreshPreview(chatId)
+      }
+    }, 500)
+  }, [previewOpen, refreshPreview])
+
+  // Phase 3: convenience helper — refresh fact list, active fix, and (if open)
+  // schedule a preview refresh. Called after every chat send so the new state
+  // (PROMOTE-synthesized facts, new SUGGEST_FIX) appears in the lane.
+  const refreshSessionDerived = useCallback(async (chatId: string) => {
+    await Promise.all([refreshFacts(chatId), refreshActiveFix(chatId)])
+    if (previewOpen) schedulePreviewRefresh(chatId)
+  }, [refreshFacts, refreshActiveFix, previewOpen, schedulePreviewRefresh])
+
+  const handleAddNote = async (text: string, summary: string | null) => {
+    if (!activeChatId) return
+    try {
+      const fact = await sessionFactsApi.create(activeChatId, { text, summary })
+      setFacts(prev => [...prev, fact])
+      schedulePreviewRefresh(activeChatId)
+    } catch {
+      toast.error('Failed to add note')
+    }
+  }
+
+  const handleUpdateFact = async (factId: string, text: string, summary: string | null) => {
+    if (!activeChatId) return
+    try {
+      const updated = await sessionFactsApi.update(activeChatId, factId, { text, summary })
+      setFacts(prev => prev.map(f => f.id === factId ? updated : f))
+      schedulePreviewRefresh(activeChatId)
+    } catch {
+      toast.error('Failed to update fact')
+    }
+  }
+
+  const handleDeleteFact = async (factId: string) => {
+    if (!activeChatId) return
+    try {
+      await sessionFactsApi.remove(activeChatId, factId)
+      setFacts(prev => prev.filter(f => f.id !== factId))
+      schedulePreviewRefresh(activeChatId)
+    } catch {
+      toast.error('Failed to remove fact')
+    }
+  }
+
+  const handleDismissFix = async () => {
+    if (!activeChatId || !activeFix) return
+    try {
+      await sessionSuggestedFixesApi.recordDecision(activeChatId, activeFix.id, 'dismissed')
+      setActiveFix(null)
+      // Dismissal bumps state_version on the server; reflect in preview.
+      schedulePreviewRefresh(activeChatId)
+    } catch {
+      toast.error('Failed to dismiss suggestion')
+    }
+  }
+
+  const handleTogglePreview = () => {
+    if (!activeChatId) return
+    const next = !previewOpen
+    setPreviewOpen(next)
+    if (next && !previewData) {
+      // First open — fetch immediately, no debounce.
+      refreshPreview(activeChatId)
+    }
+  }
+
   const selectChat = useCallback(async (chatId: string) => {
     currentChatRef.current = chatId
     setActiveChatId(chatId)
@@ -231,6 +384,13 @@ export default function AssistantChatPage() {
     setActiveActions([])
     setActiveSessionStatus(null)
     setActivePsaTicketId(null)
+    setFacts([])
+    setActiveFix(null)
+    setPreviewData(null)
+    setPreviewError(null)
+    setPreviewOpen(false)
+    // Fire facts + active-fix fetches in parallel with session detail.
+    refreshSessionDerived(chatId)
     try {
       const detail = await aiSessionsApi.getSession(chatId)
       // Guard: if the user switched to a different chat while this API call was
@@ -266,7 +426,7 @@ export default function AssistantChatPage() {
     } catch {
       setMessages([])
     }
-  }, [])
+  }, [refreshSessionDerived])
 
   const handleNewChat = async () => {
     // Invalidate currentChatRef BEFORE the await so any in-flight handleSend/handleTaskSubmit
@@ -277,6 +437,7 @@ export default function AssistantChatPage() {
     setShowTaskLane(false)
     setActiveQuestions([])
     setActiveActions([])
+    setFacts([])
     setMessages([])
     setActiveSessionStatus('active')
     setActivePsaTicketId(null)
@@ -366,6 +527,8 @@ export default function AssistantChatPage() {
         setActiveActions(response.actions || [])
         setShowTaskLane(true)
       }
+      // Refetch facts + active fix; preview refreshes if open.
+      refreshSessionDerived(sentForChatId)
     } catch (err: unknown) {
       console.error('[AssistantChat] sendChatMessage failed:', err)
       const status = (err as { response?: { status?: number } })?.response?.status
@@ -434,6 +597,8 @@ export default function AssistantChatPage() {
         setActiveQuestions([])
         setActiveActions([])
       }
+      // Refetch facts + active fix; answering tasks is the primary trigger.
+      refreshSessionDerived(sentForChatId)
     } catch (err: unknown) {
       console.error('[AssistantChat] handleTaskSubmit failed:', err)
       const status = (err as { response?: { status?: number } })?.response?.status
@@ -523,6 +688,8 @@ export default function AssistantChatPage() {
         setActiveActions(response.actions || [])
         setShowTaskLane(true)
       }
+      // Refetch facts + active fix — resume turn may emit markers.
+      refreshSessionDerived(session.session_id)
     } catch {
       toast.error('Failed to create resume chat')
     } finally {
@@ -1011,8 +1178,11 @@ export default function AssistantChatPage() {
         )}
       </div>
 
-      {/* Task lane — slides in when AI sends questions or actions */}
-      {showTaskLane && (activeQuestions.length > 0 || activeActions.length > 0) && (
+      {/* Task lane — slides in when AI sends questions/actions OR when the
+          session has any "What we know" facts OR an active suggested fix.
+          Phase 2/3 make the lane the structural home of session diagnostic
+          state, not a transient questions panel. */}
+      {showTaskLane && (activeQuestions.length > 0 || activeActions.length > 0 || facts.length > 0 || activeFix !== null) && (
         <TaskLane
           questions={activeQuestions}
           actions={activeActions}
@@ -1022,6 +1192,38 @@ export default function AssistantChatPage() {
             setShowTaskLane(false)
           }}
           loading={loading}
+          whatWeKnowSlot={
+            <WhatWeKnow
+              facts={facts}
+              onAddNote={handleAddNote}
+              onUpdateFact={handleUpdateFact}
+              onDeleteFact={handleDeleteFact}
+            />
+          }
+          suggestedFixSlot={
+            activeFix && (
+              <SuggestedFix fix={activeFix} onDismiss={handleDismissFix} />
+            )
+          }
+          bottomSlot={
+            <>
+              <button
+                onClick={handleTogglePreview}
+                className="flex items-center gap-1.5 text-[0.75rem] font-medium text-accent-text hover:text-heading transition-colors px-3 mt-1"
+              >
+                <FileText size={12} />
+                {previewOpen ? 'Hide' : 'Preview'} Resolve note
+              </button>
+              <ResolutionNotePreviewPopover
+                open={previewOpen}
+                loading={previewLoading}
+                preview={previewData}
+                error={previewError}
+                onClose={() => setPreviewOpen(false)}
+                onRefresh={() => activeChatId && refreshPreview(activeChatId)}
+              />
+            </>
+          }
         />
       )}