feat(models): Phase 1 SQLAlchemy models — SessionFact, SessionSuggestedFix, DraftTemplate, AccountSettings

Backs the schema added in 210d310 with SQLAlchemy 2.0 models.

- SessionFact: "What we know" facts with polymorphic source_ref pointing
  at task-lane item UUIDs inside ai_sessions.pending_task_lane (not a FK
  per Section 4.2).
- SessionSuggestedFix: AI-proposed resolutions with supersession tracking
  and the full user_decision state machine.
- DraftTemplate: post-resolve templatization queue with promotion to
  script_templates.
- AccountSettings: per-account JSONB preferences grab-bag with async
  classmethod helpers — get_setting(db, account_id, key, default) reads
  without creating, set_setting(db, account_id, key, value) upserts via
  Postgres ON CONFLICT + jsonb `||` merge so existing keys are preserved.
  Lazy row creation matches the Phase 1 design.

Column additions on existing models to mirror the migration:
- AISession: resolution_note_* / escalation_package_* / state_version
  (the preview-cache-invalidation counter consumed by Phase 3).
- ScriptTemplate: source_session_id / source_user_id / source_ticket_ref
  (provenance for templates promoted from DraftTemplate).

All four new models registered in app.models.__init__ and __all__.
TYPE_CHECKING-guarded relationship imports throughout, matching the
repo's existing model style.

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

backend/app/models/__init__.py

@@ -58,6 +58,10 @@ from .template_tree import TemplateTree
 from .platform_step import PlatformStep
 from .device_type import DeviceType
 from .network_diagram import NetworkDiagram
+from .session_fact import SessionFact
+from .session_suggested_fix import SessionSuggestedFix
+from .draft_template import DraftTemplate
+from .account_settings import AccountSettings
 
 __all__ = [
     "User",
@@ -130,4 +134,8 @@ __all__ = [
     "PlatformStep",
     "DeviceType",
     "NetworkDiagram",
+    "SessionFact",
+    "SessionSuggestedFix",
+    "DraftTemplate",
+    "AccountSettings",
 ]

backend/app/models/account_settings.py

@@ -0,0 +1,99 @@
+"""Per-account settings with a JSONB preferences grab-bag.
+
+Rows are created lazily on first write. Reads of a missing row return the
+caller-supplied default — no upfront row creation per account.
+
+Settings live in `preferences` until they meet the promotion criteria in
+Section 4.6 of FLOWPILOT-MIGRATION.md (hot path / validation / joins), at
+which point a future migration adds a typed column and the helpers prefer it.
+"""
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timezone
+from typing import Any, TYPE_CHECKING
+
+from sqlalchemy import DateTime, ForeignKey, text
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy.dialects.postgresql import UUID, JSONB, insert as pg_insert
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.sql import select
+
+from app.core.database import Base
+
+if TYPE_CHECKING:
+    from app.models.account import Account
+
+
+class AccountSettings(Base):
+    """One row per account. Created lazily on first `set_setting` call."""
+    __tablename__ = "account_settings"
+
+    account_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("accounts.id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    preferences: Mapped[dict[str, Any]] = mapped_column(
+        JSONB, nullable=False, default=dict, server_default=text("'{}'::jsonb")
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.now(timezone.utc),
+        onupdate=lambda: datetime.now(timezone.utc),
+    )
+
+    account: Mapped["Account"] = relationship("Account", foreign_keys=[account_id])
+
+    @classmethod
+    async def get_setting(
+        cls,
+        db: AsyncSession,
+        account_id: uuid.UUID,
+        key: str,
+        default: Any = None,
+    ) -> Any:
+        """Return preferences[key] for the account, or `default` if no row/no key.
+
+        Never creates a row — this is the pure-read path.
+        """
+        result = await db.execute(
+            select(cls.preferences).where(cls.account_id == account_id)
+        )
+        prefs = result.scalar_one_or_none()
+        if prefs is None:
+            return default
+        return prefs.get(key, default)
+
+    @classmethod
+    async def set_setting(
+        cls,
+        db: AsyncSession,
+        account_id: uuid.UUID,
+        key: str,
+        value: Any,
+    ) -> None:
+        """Upsert preferences[key] = value for the account.
+
+        Creates the row on first write; on subsequent writes, merges the key
+        into the existing preferences JSON without clobbering other keys.
+        Uses PostgreSQL's `||` jsonb merge operator via ON CONFLICT DO UPDATE.
+        """
+        stmt = pg_insert(cls).values(
+            account_id=account_id,
+            preferences={key: value},
+        )
+        stmt = stmt.on_conflict_do_update(
+            index_elements=[cls.account_id],
+            set_={
+                # Merge the new {key: value} into the existing preferences.
+                # The `||` operator on jsonb overwrites matching keys and keeps
+                # all other keys intact.
+                "preferences": cls.preferences.op("||")(stmt.excluded.preferences),
+                "updated_at": text("now()"),
+            },
+        )
+        await db.execute(stmt)

backend/app/models/ai_session.py

@@ -214,6 +214,38 @@ class AISession(Base):
         comment="Current task lane state: {questions: [...], actions: [...]}",
     )
 
+    # ── Resolution / Escalation artifacts (Phase 1 — FlowPilot migration) ──
+    # Markdown of the posted note + PSA external ID for round-trip traceability.
+    resolution_note_markdown: Mapped[Optional[str]] = mapped_column(
+        Text, nullable=True,
+        comment="Final Resolve note markdown, as posted to the PSA",
+    )
+    resolution_note_posted_at: Mapped[Optional[datetime]] = mapped_column(
+        DateTime(timezone=True), nullable=True,
+    )
+    resolution_note_external_id: Mapped[Optional[str]] = mapped_column(
+        String(128), nullable=True,
+        comment="PSA (e.g. CW) ticket-note ID returned at post time",
+    )
+    escalation_package_markdown: Mapped[Optional[str]] = mapped_column(
+        Text, nullable=True,
+        comment="Final Escalate handoff package markdown, as posted to the PSA",
+    )
+    escalation_package_posted_at: Mapped[Optional[datetime]] = mapped_column(
+        DateTime(timezone=True), nullable=True,
+    )
+    escalation_package_external_id: Mapped[Optional[str]] = mapped_column(
+        String(128), nullable=True,
+        comment="PSA ticket-note ID for the escalation package",
+    )
+    # Incremented atomically by any write that invalidates the resolution
+    # note preview cache (facts, suggested fixes, script generations).
+    # See FLOWPILOT-MIGRATION.md Section 5.5.
+    state_version: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0"),
+        comment="Monotonic preview-cache version; bumped on state-changing writes",
+    )
+
     # ── Branching ──
     is_branching: Mapped[bool] = mapped_column(
         default=False,

backend/app/models/draft_template.py

@@ -0,0 +1,91 @@
+"""Draft template model — scripts generated during a session, pending templatization.
+
+Created when an engineer picks "Run now, templatize after resolve" in the
+three-option dialog. Post-resolve, the TemplatizePrompt component reads pending
+drafts and lets the engineer accept (promotes to `script_templates`) or reject.
+"""
+import uuid
+from datetime import datetime, timezone
+from typing import Any, TYPE_CHECKING
+
+from sqlalchemy import (
+    Text, DateTime, ForeignKey, String, CheckConstraint,
+)
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy.dialects.postgresql import UUID, JSONB
+
+from app.core.database import Base
+
+if TYPE_CHECKING:
+    from app.models.account import Account
+    from app.models.ai_session import AISession
+    from app.models.user import User
+    from app.models.script_template import ScriptCategory, ScriptTemplate
+
+
+class DraftTemplate(Base):
+    """A session-generated script pending conversion to a reusable template."""
+    __tablename__ = "draft_templates"
+    __table_args__ = (
+        CheckConstraint(
+            "status IN ('pending', 'accepted', 'rejected')",
+            name="ck_draft_templates_status",
+        ),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    account_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("accounts.id"),
+        nullable=False,
+    )
+    source_session_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("ai_sessions.id"),
+        nullable=False,
+    )
+    source_user_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("users.id"),
+        nullable=False,
+    )
+    script_body: Mapped[str] = mapped_column(Text, nullable=False)
+    proposed_parameters: Mapped[dict[str, Any]] = mapped_column(
+        JSONB, nullable=False
+    )
+    proposed_name: Mapped[str | None] = mapped_column(String(200), nullable=True)
+    proposed_category_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("script_categories.id"),
+        nullable=True,
+    )
+    status: Mapped[str] = mapped_column(
+        String(32), nullable=False, default="pending"
+    )
+    resolved_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    # Set when status transitions to 'accepted' and the draft is promoted
+    # to a real script_templates row.
+    promoted_template_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("script_templates.id"),
+        nullable=True,
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
+    )
+
+    account: Mapped["Account"] = relationship("Account", foreign_keys=[account_id])
+    source_session: Mapped["AISession"] = relationship(
+        "AISession", foreign_keys=[source_session_id]
+    )
+    source_user: Mapped["User"] = relationship("User", foreign_keys=[source_user_id])
+    proposed_category: Mapped["ScriptCategory | None"] = relationship(
+        "ScriptCategory", foreign_keys=[proposed_category_id]
+    )
+    promoted_template: Mapped["ScriptTemplate | None"] = relationship(
+        "ScriptTemplate", foreign_keys=[promoted_template_id]
+    )

backend/app/models/script_template.py

@@ -78,6 +78,20 @@ class ScriptTemplate(Base):
     is_gallery_featured: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default=text("false"), index=True)
     gallery_sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default=text("0"))
     usage_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default=text("0"))
+    # ── Provenance (Phase 1 — FlowPilot migration) ──
+    # Populated when a template is promoted from a post-resolve draft_templates row.
+    # Powers the Script Library provenance chip:
+    # "generated from CW #X · resolved by Y · used N times"
+    source_session_id: Mapped[Optional[uuid.UUID]] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("ai_sessions.id"), nullable=True,
+    )
+    source_user_id: Mapped[Optional[uuid.UUID]] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("users.id"), nullable=True,
+    )
+    source_ticket_ref: Mapped[Optional[str]] = mapped_column(
+        String(64), nullable=True,
+        comment="Human-readable PSA ticket ref for display, e.g. 'CW #48307'",
+    )
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
     )

backend/app/models/session_fact.py

@@ -0,0 +1,79 @@
+"""Session fact model — the "What we know" backing store for a FlowPilot session.
+
+A fact is an atomic, engineer-readable statement of what has been confirmed
+during troubleshooting. Facts accumulate across the session and drive the
+resolution note preview.
+
+`source_ref` is a polymorphic pointer to a task-lane item inside
+`ai_sessions.pending_task_lane` JSON — it is NOT a FK. Integrity is enforced
+at the service layer per the FLOWPILOT-MIGRATION design doc Section 4.2.
+Phase 2 assigns stable UUIDs to those task-lane items so `source_ref` has
+something reliable to point to.
+"""
+import uuid
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING
+
+from sqlalchemy import Text, DateTime, ForeignKey, String, CheckConstraint
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy.dialects.postgresql import UUID
+
+from app.core.database import Base
+
+if TYPE_CHECKING:
+    from app.models.ai_session import AISession
+    from app.models.account import Account
+    from app.models.user import User
+
+
+class SessionFact(Base):
+    """A single fact in the What-we-know section of a session's task lane."""
+    __tablename__ = "session_facts"
+    __table_args__ = (
+        CheckConstraint(
+            "source_type IN ('question', 'diagnostic_check', 'user_note', 'ai_synthesis')",
+            name="ck_session_facts_source_type",
+        ),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    session_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("ai_sessions.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    account_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("accounts.id"),
+        nullable=False,
+    )
+    text: Mapped[str] = mapped_column(Text, nullable=False)
+    source_type: Mapped[str] = mapped_column(String(32), nullable=False)
+    # Pointer to a task-lane item UUID inside ai_sessions.pending_task_lane.
+    # NOT a FK. Null for `user_note` and `ai_synthesis` sources.
+    source_ref: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), nullable=True
+    )
+    source_summary: Mapped[str | None] = mapped_column(Text, nullable=True)
+    created_by: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("users.id"),
+        nullable=False,
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.now(timezone.utc),
+        onupdate=lambda: datetime.now(timezone.utc),
+    )
+    deleted_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+
+    session: Mapped["AISession"] = relationship("AISession", foreign_keys=[session_id])
+    account: Mapped["Account"] = relationship("Account", foreign_keys=[account_id])
+    creator: Mapped["User"] = relationship("User", foreign_keys=[created_by])

backend/app/models/session_suggested_fix.py

@@ -0,0 +1,80 @@
+"""Session suggested-fix model — AI-proposed resolution path for a session.
+
+A session can have multiple suggested fixes over its lifetime as the AI's
+understanding evolves. Only one is active at a time (superseded_at IS NULL);
+emitting a new [SUGGEST_FIX] marker supersedes the prior active one.
+"""
+import uuid
+from datetime import datetime, timezone
+from typing import Any, TYPE_CHECKING
+
+from sqlalchemy import (
+    Text, DateTime, ForeignKey, String, Integer, CheckConstraint,
+)
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy.dialects.postgresql import UUID, JSONB
+
+from app.core.database import Base
+
+if TYPE_CHECKING:
+    from app.models.ai_session import AISession
+    from app.models.account import Account
+    from app.models.script_template import ScriptTemplate
+
+
+class SessionSuggestedFix(Base):
+    """One AI-proposed fix for a FlowPilot session."""
+    __tablename__ = "session_suggested_fixes"
+    __table_args__ = (
+        CheckConstraint(
+            "confidence_pct BETWEEN 0 AND 100",
+            name="ck_session_suggested_fixes_confidence_pct",
+        ),
+        CheckConstraint(
+            "user_decision IS NULL OR user_decision IN ("
+            "'one_off', 'draft_template', 'build_template', 'dismissed')",
+            name="ck_session_suggested_fixes_user_decision",
+        ),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    session_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("ai_sessions.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    account_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("accounts.id"),
+        nullable=False,
+    )
+    title: Mapped[str] = mapped_column(String(200), nullable=False)
+    description: Mapped[str] = mapped_column(Text, nullable=False)
+    confidence_pct: Mapped[int] = mapped_column(Integer, nullable=False)
+    script_template_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("script_templates.id"),
+        nullable=True,
+    )
+    # Populated only when there's no matching template and the AI has
+    # drafted a session-specific script.
+    ai_drafted_script: Mapped[str | None] = mapped_column(Text, nullable=True)
+    ai_drafted_parameters: Mapped[dict[str, Any] | None] = mapped_column(
+        JSONB, nullable=True
+    )
+    user_decision: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    # Set when a newer suggested fix supersedes this one.
+    superseded_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
+    )
+
+    session: Mapped["AISession"] = relationship("AISession", foreign_keys=[session_id])
+    account: Mapped["Account"] = relationship("Account", foreign_keys=[account_id])
+    script_template: Mapped["ScriptTemplate | None"] = relationship(
+        "ScriptTemplate", foreign_keys=[script_template_id]
+    )