feat: Step Library sync + service account for default tree ownership

* feat: maintenance flow UX redesign — batch status hub, context strip, detail page upgrades (#85) - Add BatchStatusPage (/flows/:id/batches/:batchId): per-target Start/Resume/View cards, progress bar, 5s polling while in-progress, completion outcome summary - Add BatchStatusCard: handles not-started/in-progress/complete states with step progress for in-progress targets - Add ActiveBatchBanner: amber banner on detail page when a batch is running, links to BatchStatusPage - Add MaintenanceContextStrip: amber strip in ProceduralNavigationPage for maintenance flows showing target name, batch progress (X/Y complete), and Back to Batch nav - Update MaintenanceFlowDetailPage: active batch banner, clickable run history rows with mini progress dots and outcome summaries, Run button loading state, post-launch navigates to BatchStatusPage - Update ProceduralNavigationPage: renders MaintenanceContextStrip between top bar and content when tree_type === 'maintenance'; fetches batch progress once on mount - Add batch_id filter to GET /sessions backend endpoint and SessionListParams frontend type - Add /flows/:id/batches/:batchId route to router Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: session detail page — completion action + outcome summary card - In-progress sessions: amber banner with "Complete Session" button opens SessionOutcomeModal to set outcome/notes/next-steps and finalize - Completed sessions: colored outcome summary card (icon + outcome label + duration + notes + next steps) replaces dense header metadata; "Copy for Ticket" promoted to primary action inside the card - Export toolbar de-emphasized to secondary row of smaller controls below the summary card Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: add library-page action props to StepCard (edit/delete/save) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: pass library-page action props through StepLibraryBrowser + refreshKey Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: StepFormModal wrapper + submitLabel/isSubmitting props on StepForm Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: Step Library page — create, edit, delete, save-to-library Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: add RuntimeStep union type for procedural custom steps Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: StepChecklist accepts RuntimeStep[], renders amber Custom badge Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: StepDetail accepts RuntimeStep, renders Custom Step badge for custom steps Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: custom step insertion in procedural flow sessions Engineers can add custom steps inline during execution. Steps are persisted to session.custom_steps and restored on resume. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix: suppress StepFeedback on custom steps, fix resume stepState seeding, functional updater for step index Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * docs: add tree forking UI design doc Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * docs: add tree fork UI implementation plan Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: add ForkInfo type and fork fields to Tree/TreeListItem Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix: align ForkInfo type with backend schema, remove redundant fork fields Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix: ForkInfo placement, required fork_info field, add JSDoc Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: add ForkModal component with name and reason fields Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix: ForkModal accessibility and UX (escape, click-outside, labels, maxLength) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: open ForkModal on fork action in TreeLibraryPage Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: add ForkModal to MyTreesPage Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: show Fork chip badge on forked tree cards Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * docs: add flow-to-library step sync design doc Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * docs: add flow-to-library sync implementation plan Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: add sync tracking columns to step_library Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: add sync columns and source_tree relationship to StepLibrary model Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: add group_label to StepContent, is_flow_synced/source_tree_name to StepLibraryResponse Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: include is_flow_synced and source_tree_name in step list/detail responses Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix: add is_flow_synced and source_tree_name to step list response Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix: add selectinload and sync fields to search and get_step endpoints Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: add step_sync module with extraction and upsert logic Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix: safe NOT IN placeholders for asyncpg, add deactivate docstring Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: trigger step library sync on tree publish and deactivate on delete - Call sync_steps_from_tree in update_tree whenever the tree is published (status transitions to 'published' or is already published and structure changes) - Call deactivate_synced_steps_for_tree in delete_tree before db.commit() so the FK SET NULL does not nullify source_tree_id before the WHERE clause runs - Fix ::jsonb cast syntax in step_sync.py (asyncpg rejects :: operator in text() queries; replaced with CAST(:content AS jsonb)) - Add UniqueConstraint('source_tree_id','source_node_id') to StepLibrary model so Base.metadata.create_all (used by tests) creates the constraint that the ON CONFLICT clause in sync_steps_from_tree depends on Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: add is_flow_synced and source_tree_name to Step types Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: show From Flow badge and lock icon on flow-synced StepCard Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: show source flow name in StepDetailModal for synced steps Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * feat: add Library Visibility select to procedural StepEditor Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix: address code review issues in flow-to-library sync - Fix sync trigger: only fire on publish transition, not every PUT - Add TestSyncOnPublish integration tests (2 tests, 16 total passing) - Add group_label to frontend StepContent interface - Guard Library Visibility select to procedure_step nodes only - Block API edits to flow-synced steps (400 read-only guard) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix: handle None author_id in step sync to avoid invalid UUID error When a system/default tree has no author (author_id is None), str(None) produces the literal string 'None' which asyncpg rejects as an invalid UUID for the created_by column. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix: add ResolutionFlow service account to own default tree steps in library Default/system trees had no author_id (NULL), causing a NOT NULL violation when syncing steps to step_library.created_by on publish. - Add is_service_account flag to users table (migration 4f4137ce) - Add service_account.py: idempotent ensure_service_account() creates noreply@resolutionflow.com with unusable password on startup - Cache service account ID on app.state at lifespan startup - Add get_service_account_id() FastAPI dep (returns None in tests) - sync_steps_from_tree: resolve author_id or service_account_id as created_by - create_tree: set author_id=service_account_id for is_default trees - Migration 1490781700bc: backfill author_id on 31 existing default trees Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> --------- Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-02-25 23:17:29 -05:00
parent a6abd23727
commit e6a0c0549b
45 changed files with 4261 additions and 270 deletions
--- a/backend/alembic/versions/1490781700bc_backfill_default_tree_author_id_to_.py
+++ b/backend/alembic/versions/1490781700bc_backfill_default_tree_author_id_to_.py
@@ -0,0 +1,94 @@
+"""backfill_default_tree_author_id_to_service_account
+
+Revision ID: 1490781700bc
+Revises: 4f4137ce79e5
+Create Date: 2026-02-25 21:26:00.000000
+
+Backfill author_id on is_default trees to the ResolutionFlow service account
+(noreply@resolutionflow.com). The service account is created here if it does
+not yet exist (idempotent), so this migration is safe to run before or after
+the app starts.
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import UUID as PG_UUID
+import uuid
+
+
+# revision identifiers, used by Alembic.
+revision: str = '1490781700bc'
+down_revision: Union[str, None] = '4f4137ce79e5'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+SERVICE_ACCOUNT_EMAIL = "noreply@resolutionflow.com"
+SERVICE_ACCOUNT_NAME = "ResolutionFlow"
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+
+    # Ensure service account exists
+    row = conn.execute(
+        sa.text("SELECT id FROM users WHERE email = :email"),
+        {"email": SERVICE_ACCOUNT_EMAIL},
+    ).fetchone()
+
+    if row is None:
+        service_id = str(uuid.uuid4())
+        conn.execute(
+            sa.text("""
+                INSERT INTO users (
+                    id, email, name, password_hash, role,
+                    is_super_admin, is_team_admin, is_active,
+                    is_service_account, must_change_password,
+                    account_role, created_at
+                ) VALUES (
+                    :id, :email, :name, :password_hash, 'engineer',
+                    false, false, true,
+                    true, false,
+                    'engineer', NOW()
+                )
+            """),
+            {
+                "id": service_id,
+                "email": SERVICE_ACCOUNT_EMAIL,
+                "name": SERVICE_ACCOUNT_NAME,
+                "password_hash": "!service-account-no-login",
+            },
+        )
+    else:
+        service_id = str(row[0])
+
+    # Backfill is_default trees that have no author
+    result = conn.execute(
+        sa.text("""
+            UPDATE trees
+            SET author_id = :service_id
+            WHERE author_id IS NULL AND is_default = true
+        """),
+        {"service_id": service_id},
+    )
+    print(f"[backfill] Set author_id to service account on {result.rowcount} default trees")
+
+
+def downgrade() -> None:
+    # Restore NULL on trees that were authored by the service account and are default
+    conn = op.get_bind()
+    row = conn.execute(
+        sa.text("SELECT id FROM users WHERE email = :email"),
+        {"email": SERVICE_ACCOUNT_EMAIL},
+    ).fetchone()
+    if row is None:
+        return
+    service_id = str(row[0])
+    conn.execute(
+        sa.text("""
+            UPDATE trees
+            SET author_id = NULL
+            WHERE author_id = :service_id AND is_default = true
+        """),
+        {"service_id": service_id},
+    )
--- a/backend/alembic/versions/4f4137ce79e5_add_is_service_account_to_users.py
+++ b/backend/alembic/versions/4f4137ce79e5_add_is_service_account_to_users.py
@@ -0,0 +1,34 @@
+"""add_is_service_account_to_users
+
+Revision ID: 4f4137ce79e5
+Revises: fb1481317ff6
+Create Date: 2026-02-25 20:28:46.075639
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '4f4137ce79e5'
+down_revision: Union[str, None] = 'fb1481317ff6'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        'users',
+        sa.Column(
+            'is_service_account',
+            sa.Boolean(),
+            nullable=False,
+            server_default='false',
+        )
+    )
+
+
+def downgrade() -> None:
+    op.drop_column('users', 'is_service_account')
--- a/backend/alembic/versions/fb1481317ff6_add_step_library_sync_fields.py
+++ b/backend/alembic/versions/fb1481317ff6_add_step_library_sync_fields.py
@@ -0,0 +1,47 @@
+"""add_step_library_sync_fields
+
+Revision ID: fb1481317ff6
+Revises: a1b2c3d4e5f6
+Create Date: 2026-02-25 03:19:52.600292
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'fb1481317ff6'
+down_revision: Union[str, None] = 'a1b2c3d4e5f6'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column('step_library', sa.Column('source_tree_id', sa.UUID(), nullable=True))
+    op.add_column('step_library', sa.Column('source_node_id', sa.String(255), nullable=True))
+    op.add_column('step_library', sa.Column('is_flow_synced', sa.Boolean(), nullable=False, server_default='false'))
+    op.add_column('step_library', sa.Column('last_synced_at', sa.DateTime(timezone=True), nullable=True))
+    op.create_foreign_key(
+        'fk_step_library_source_tree',
+        'step_library', 'trees',
+        ['source_tree_id'], ['id'],
+        ondelete='SET NULL'
+    )
+    op.create_unique_constraint(
+        'uq_step_library_source_node',
+        'step_library',
+        ['source_tree_id', 'source_node_id']
+    )
+    op.create_index('ix_step_library_source_tree_id', 'step_library', ['source_tree_id'])
+
+
+def downgrade() -> None:
+    op.drop_index('ix_step_library_source_tree_id', 'step_library')
+    op.drop_constraint('uq_step_library_source_node', 'step_library', type_='unique')
+    op.drop_constraint('fk_step_library_source_tree', 'step_library', type_='foreignkey')
+    op.drop_column('step_library', 'last_synced_at')
+    op.drop_column('step_library', 'is_flow_synced')
+    op.drop_column('step_library', 'source_node_id')
+    op.drop_column('step_library', 'source_tree_id')
--- a/backend/app/api/deps.py
+++ b/backend/app/api/deps.py
@@ -155,6 +155,14 @@ async def require_account_owner(
    )


+def get_service_account_id(request: Request) -> Optional[UUID]:
+    """Return the cached ResolutionFlow service account UUID from app.state.
+
+    Returns None in test environments where lifespan startup did not run.
+    """
+    return getattr(request.app.state, "service_account_id", None)
+
+
 async def get_plan_limits_for_user(
    current_user: Annotated[User, Depends(get_current_active_user)],
    db: Annotated[AsyncSession, Depends(get_db)],
--- a/backend/app/api/endpoints/sessions.py
+++ b/backend/app/api/endpoints/sessions.py
@@ -38,6 +38,7 @@ async def list_sessions(
    client_name: Optional[str] = Query(None, description="Search by client name (partial match)"),
    tree_name: Optional[str] = Query(None, description="Filter by tree name from snapshot"),
    tree_id: Optional[UUID] = Query(None, description="Filter by tree ID"),
+    batch_id: Optional[UUID] = Query(None, description="Filter by batch ID (maintenance batch runs)"),
    started_after: Optional[datetime] = Query(None, description="Filter sessions started after this datetime"),
    started_before: Optional[datetime] = Query(None, description="Filter sessions started before this datetime"),
    completed_after: Optional[datetime] = Query(None, description="Filter sessions completed after this datetime"),
@@ -73,6 +74,10 @@ async def list_sessions(
    if tree_id:
        query = query.where(Session.tree_id == tree_id)

+    # Batch ID filter
+    if batch_id:
+        query = query.where(Session.batch_id == batch_id)
+
    # Date range filters
    if started_after:
        query = query.where(Session.started_at >= started_after)
--- a/backend/app/api/endpoints/steps.py
+++ b/backend/app/api/endpoints/steps.py
@@ -5,6 +5,7 @@ from decimal import Decimal
 from fastapi import APIRouter, Depends, HTTPException, Query
 from sqlalchemy import select, func, desc, Integer, case
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload

 from app.core.database import get_db
 from app.api.deps import get_current_active_user, require_engineer_or_admin
@@ -39,7 +40,7 @@ async def get_step_or_404(
        select(StepLibrary).where(
            StepLibrary.id == step_id,
            StepLibrary.is_active == True
-        )
+        ).options(selectinload(StepLibrary.source_tree))
    )
    step = result.scalar_one_or_none()
    if not step:
@@ -72,7 +73,7 @@ async def list_steps(
    query = select(StepLibrary).where(
        StepLibrary.is_active == True,
        build_step_visibility_filter(current_user)
-    )
+    ).options(selectinload(StepLibrary.source_tree))

    # Apply filters
    if visibility:
@@ -117,6 +118,8 @@ async def list_steps(
            "is_featured": step.is_featured,
            "created_by": step.created_by,
            "created_at": step.created_at,
+            "is_flow_synced": step.is_flow_synced,
+            "source_tree_name": step.source_tree.name if step.source_tree else None,
        }

        # Get category name if exists
@@ -154,7 +157,7 @@ async def search_steps(
        StepLibrary.is_active == True,
        build_step_visibility_filter(current_user),
        func.to_tsvector('english', StepLibrary.title).match(search_query)
-    ).order_by(desc(StepLibrary.rating_average)).limit(limit)
+    ).options(selectinload(StepLibrary.source_tree)).order_by(desc(StepLibrary.rating_average)).limit(limit)

    result = await db.execute(query)
    steps = result.scalars().all()
@@ -174,6 +177,8 @@ async def search_steps(
            "is_featured": step.is_featured,
            "created_by": step.created_by,
            "created_at": step.created_at,
+            "is_flow_synced": step.is_flow_synced,
+            "source_tree_name": step.source_tree.name if step.source_tree else None,
        }

        if step.category_id:
@@ -247,6 +252,8 @@ async def get_step(
        "is_active": step.is_active,
        "created_at": step.created_at,
        "updated_at": step.updated_at,
+        "is_flow_synced": step.is_flow_synced,
+        "source_tree_name": step.source_tree.name if step.source_tree else None,
    }

    # Get category name if exists
@@ -346,6 +353,12 @@ async def update_step(
    """Update a step (owner or admin only)."""
    step = await get_step_or_404(step_id, db, current_user, check_edit=True)

+    if step.is_flow_synced:
+        raise HTTPException(
+            status_code=400,
+            detail="Flow-synced steps are read-only. Fork to customize."
+        )
+
    # Validate category if being updated
    if step_data.category_id:
        cat_result = await db.execute(
--- a/backend/app/api/endpoints/trees.py
+++ b/backend/app/api/endpoints/trees.py
@@ -21,13 +21,14 @@ from app.schemas.tree import (
    PinnedFlowResponse, PinnedFlowsListResponse, PinnedFlowReorderRequest
 )
 from app.models.user_pinned_tree import UserPinnedTree
-from app.api.deps import get_current_active_user, require_engineer_or_admin, require_admin
+from app.api.deps import get_current_active_user, require_engineer_or_admin, require_admin, get_service_account_id
 from app.core.permissions import can_edit_tree, can_access_tree
 from app.core.filters import build_tree_access_filter
 from app.core.subscriptions import check_tree_limit
 from app.core.audit import log_audit
 from app.core.config import settings
 from app.core.tree_validation import can_publish_tree
+from app.core.step_sync import sync_steps_from_tree, deactivate_synced_steps_for_tree

 router = APIRouter(prefix="/trees", tags=["trees"])

@@ -399,7 +400,8 @@ async def get_tree(
 async def create_tree(
    tree_data: TreeCreate,
    db: Annotated[AsyncSession, Depends(get_db)],
-    current_user: Annotated[User, Depends(require_engineer_or_admin)]
+    current_user: Annotated[User, Depends(require_engineer_or_admin)],
+    service_account_id: Annotated[Optional[UUID], Depends(get_service_account_id)],
 ):
    """Create a new tree (engineers and admins only).

@@ -464,7 +466,7 @@ async def create_tree(
        tree_type=tree_data.tree_type,
        tree_structure=tree_data.tree_structure,
        intake_form=intake_form_data,
-        author_id=None if is_default else current_user.id,  # Default trees have no author
+        author_id=service_account_id if is_default else current_user.id,
        account_id=None if is_default else current_user.account_id,
        is_public=True if is_default else tree_data.is_public,  # Default trees are always public
        is_default=is_default,
@@ -548,7 +550,8 @@ async def update_tree(
    tree_id: UUID,
    tree_data: TreeUpdate,
    db: Annotated[AsyncSession, Depends(get_db)],
-    current_user: Annotated[User, Depends(require_engineer_or_admin)]
+    current_user: Annotated[User, Depends(require_engineer_or_admin)],
+    service_account_id: Annotated[Optional[UUID], Depends(get_service_account_id)],
 ):
    """Update an existing tree (engineers and admins only).

@@ -640,6 +643,22 @@ async def update_tree(
    if "tree_structure" in update_data:
        tree.version += 1

+    # Sync steps to step library on publish transition only
+    if update_data.get("status") == 'published':
+        _structure = update_data.get("tree_structure", tree.tree_structure)
+        _type = update_data.get("tree_type", tree.tree_type)
+        _is_public = update_data.get("is_public", tree.is_public)
+        await sync_steps_from_tree(
+            db=db,
+            tree_id=tree.id,
+            tree_type=_type,
+            tree_structure=_structure,
+            author_id=tree.author_id,
+            account_id=tree.account_id,
+            is_public=_is_public,
+            service_account_id=service_account_id,
+        )
+
    # Handle tags replacement
    if tags_data is not None:
        from app.models.tag import tree_tag_assignments
@@ -753,6 +772,10 @@ async def delete_tree(
        tree_tag_assignments.delete().where(tree_tag_assignments.c.tree_id == tree.id)
    )

+    # Deactivate any synced step library entries before deletion
+    # (must happen before db.delete/commit — FK SET NULL would lose the reference)
+    await deactivate_synced_steps_for_tree(db, tree.id)
+
    await log_audit(db, current_user.id, "tree.delete", "tree", tree.id,
                    {"tree_name": tree.name})
    await db.commit()
--- a/backend/app/core/service_account.py
+++ b/backend/app/core/service_account.py
@@ -0,0 +1,60 @@
+"""ResolutionFlow system service account.
+
+This module manages the platform-level service account used as the author
+for system/default content (seeded trees, synced step library entries, etc.).
+
+The service account ID is resolved once at startup and cached on app.state
+so that sync operations can use it without a DB query per request.
+"""
+from __future__ import annotations
+
+import uuid
+import logging
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+logger = logging.getLogger(__name__)
+
+SERVICE_ACCOUNT_EMAIL = "noreply@resolutionflow.com"
+SERVICE_ACCOUNT_NAME = "ResolutionFlow"
+
+
+async def ensure_service_account(db: AsyncSession) -> uuid.UUID:
+    """Ensure the ResolutionFlow service account exists and return its ID.
+
+    Idempotent — safe to call on every startup. Creates the account if it
+    does not exist. The account has no usable password and is_service_account=True
+    so it can never log in via normal auth flows.
+    """
+    from app.models.user import User
+
+    result = await db.execute(
+        select(User).where(User.email == SERVICE_ACCOUNT_EMAIL)
+    )
+    user = result.scalar_one_or_none()
+
+    if user is not None:
+        if not user.is_service_account:
+            user.is_service_account = True
+            await db.commit()
+        return user.id
+
+    # Create the service account with a random, unusable password hash
+    new_user = User(
+        id=uuid.uuid4(),
+        email=SERVICE_ACCOUNT_EMAIL,
+        name=SERVICE_ACCOUNT_NAME,
+        password_hash="!service-account-no-login",  # bcrypt can't produce this prefix
+        role="engineer",
+        is_super_admin=False,
+        is_team_admin=False,
+        is_active=True,
+        is_service_account=True,
+        must_change_password=False,
+        account_role="engineer",
+    )
+    db.add(new_user)
+    await db.commit()
+    logger.info(f"[service_account] Created service account (id={new_user.id})")
+    return new_user.id
--- a/backend/app/core/step_sync.py
+++ b/backend/app/core/step_sync.py
@@ -0,0 +1,222 @@
+"""Sync steps from published flows into the step library."""
+from __future__ import annotations
+import json
+from typing import Any, Generator, Literal, Optional
+from uuid import UUID
+from datetime import datetime, timezone
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+
+StepVisibility = Literal['private', 'team', 'public']
+
+
+def resolve_step_visibility(
+    is_public: bool,
+    account_id: Optional[UUID],
+    node_override: Optional[str],
+) -> StepVisibility:
+    """Resolve the visibility for a synced step.
+
+    Priority: node-level library_visibility overrides flow visibility.
+    Flow visibility: 'public' if is_public, otherwise 'team'.
+    """
+    if node_override in ('team', 'public'):
+        return node_override  # type: ignore[return-value]
+    return 'public' if is_public else 'team'
+
+
+def _normalize_commands(raw: Any) -> list[dict]:
+    """Normalize the commands field to a list of StepCommand dicts."""
+    if not raw:
+        return []
+    if isinstance(raw, str):
+        return [{"label": "", "command": raw, "command_type": None}]
+    if isinstance(raw, list):
+        result = []
+        for item in raw:
+            if isinstance(item, str):
+                result.append({"label": "", "command": item, "command_type": None})
+            elif isinstance(item, dict):
+                result.append({
+                    "label": item.get("label", ""),
+                    "command": item.get("code", item.get("command", "")),
+                    "command_type": item.get("language", item.get("command_type")),
+                })
+        return result
+    return []
+
+
+def _walk_troubleshooting(node: dict) -> Generator[dict, None, None]:
+    """Recursively yield action and solution nodes from a troubleshooting tree."""
+    if node.get("type") in ("action", "solution"):
+        yield node
+    for child in node.get("children", []):
+        yield from _walk_troubleshooting(child)
+
+
+def extract_steps_for_sync(
+    tree_structure: dict,
+    tree_type: str,
+) -> Generator[dict, None, None]:
+    """Extract step dicts ready for upsert from a tree structure.
+
+    Yields dicts with keys:
+        source_node_id, title, step_type, content (dict), node_visibility_override
+    """
+    if tree_type in ("procedural", "maintenance"):
+        steps = tree_structure.get("steps", [])
+        current_section: Optional[str] = None
+        for node in steps:
+            node_type = node.get("type")
+            if node_type == "section_header":
+                current_section = node.get("title") or node.get("section_header")
+                continue
+            if node_type != "procedure_step":
+                continue
+            instructions = node.get("description") or node.get("title", "")
+            commands = _normalize_commands(node.get("commands")) or None
+            content: dict = {"instructions": instructions}
+            if node.get("expected_outcome"):
+                content["help_text"] = node["expected_outcome"]
+            if commands:
+                content["commands"] = commands
+            if current_section:
+                content["group_label"] = current_section
+            yield {
+                "source_node_id": node["id"],
+                "title": node.get("title", "Untitled step"),
+                "step_type": "action",
+                "content": content,
+                "node_visibility_override": node.get("library_visibility"),
+            }
+
+    elif tree_type == "troubleshooting":
+        for node in _walk_troubleshooting(tree_structure):
+            instructions = node.get("description") or node.get("title", "")
+            yield {
+                "source_node_id": node["id"],
+                "title": node.get("title", "Untitled step"),
+                "step_type": "action" if node["type"] == "action" else "solution",
+                "content": {"instructions": instructions},
+                "node_visibility_override": None,
+            }
+
+
+async def sync_steps_from_tree(
+    db: AsyncSession,
+    tree_id: UUID,
+    tree_type: str,
+    tree_structure: dict,
+    author_id: Optional[UUID],
+    account_id: Optional[UUID],
+    is_public: bool,
+    service_account_id: Optional[UUID] = None,
+) -> int:
+    """Upsert step library entries from a published tree.
+
+    Returns the number of steps synced.
+
+    For default/system trees that have no author_id, pass service_account_id
+    so that created_by is set to the ResolutionFlow service account.
+    """
+    resolved_author_id = author_id or service_account_id
+    if not resolved_author_id:
+        return 0
+
+    now = datetime.now(timezone.utc)
+    extracted = list(extract_steps_for_sync(tree_structure, tree_type))
+
+    for step_data in extracted:
+        visibility = resolve_step_visibility(
+            is_public=is_public,
+            account_id=account_id,
+            node_override=step_data["node_visibility_override"],
+        )
+        await db.execute(
+            text("""
+                INSERT INTO step_library (
+                    id, title, step_type, content, created_by, account_id,
+                    visibility, is_flow_synced, source_tree_id, source_node_id,
+                    last_synced_at, tags, is_active,
+                    usage_count, rating_average, rating_count,
+                    helpful_yes, helpful_no, is_featured, is_verified,
+                    created_at, updated_at
+                ) VALUES (
+                    gen_random_uuid(), :title, :step_type, CAST(:content AS jsonb),
+                    :created_by, :account_id, :visibility, true,
+                    :source_tree_id, :source_node_id, :last_synced_at,
+                    '{}', true,
+                    0, 0, 0, 0, 0, false, false,
+                    :now, :now
+                )
+                ON CONFLICT (source_tree_id, source_node_id)
+                DO UPDATE SET
+                    title = EXCLUDED.title,
+                    step_type = EXCLUDED.step_type,
+                    content = EXCLUDED.content,
+                    visibility = EXCLUDED.visibility,
+                    last_synced_at = EXCLUDED.last_synced_at,
+                    updated_at = EXCLUDED.updated_at,
+                    is_active = true
+            """),
+            {
+                "title": step_data["title"],
+                "step_type": step_data["step_type"],
+                "content": json.dumps(step_data["content"]),
+                "created_by": str(resolved_author_id),
+                "account_id": str(account_id) if account_id else None,
+                "visibility": visibility,
+                "source_tree_id": str(tree_id),
+                "source_node_id": step_data["source_node_id"],
+                "last_synced_at": now,
+                "now": now,
+            }
+        )
+
+    # Soft-delete previously synced steps that no longer exist in the tree
+    current_node_ids = [s["source_node_id"] for s in extracted]
+    if current_node_ids:
+        # Build NOT IN using explicit named placeholders — asyncpg does not
+        # auto-cast a Python list to a PostgreSQL array in text() queries.
+        placeholders = ", ".join(f":id_{i}" for i in range(len(current_node_ids)))
+        params = {f"id_{i}": nid for i, nid in enumerate(current_node_ids)}
+        params.update({"tree_id": str(tree_id), "now": now})
+        await db.execute(
+            text(f"""
+                UPDATE step_library
+                SET is_active = false, updated_at = :now
+                WHERE source_tree_id = :tree_id
+                  AND is_flow_synced = true
+                  AND source_node_id NOT IN ({placeholders})
+            """),
+            params
+        )
+    else:
+        await db.execute(
+            text("""
+                UPDATE step_library
+                SET is_active = false, updated_at = :now
+                WHERE source_tree_id = :tree_id AND is_flow_synced = true
+            """),
+            {"tree_id": str(tree_id), "now": now}
+        )
+
+    return len(extracted)
+
+
+async def deactivate_synced_steps_for_tree(db: AsyncSession, tree_id: UUID) -> None:
+    """Soft-delete all synced library entries for a tree (on tree delete/deactivate).
+
+    Must be called BEFORE deleting the tree row — after deletion the FK ondelete='SET NULL'
+    will null source_tree_id, making the WHERE clause match nothing.
+    """
+    await db.execute(
+        text("""
+            UPDATE step_library
+            SET is_active = false, updated_at = :now
+            WHERE source_tree_id = :tree_id AND is_flow_synced = true
+        """),
+        {"tree_id": str(tree_id), "now": datetime.now(timezone.utc)}
+    )
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -14,6 +14,7 @@ from app.core.middleware import RequestLoggingMiddleware, ErrorLoggingMiddleware
 from app.core.rate_limit import limiter
 from app.api.router import api_router
 from app.core.scheduler import scheduler, load_all_schedules, _cleanup_expired_ai_conversations
+from app.core.service_account import ensure_service_account

 # Initialize logging configuration
 setup_logging()
@@ -103,6 +104,12 @@ async def lifespan(app: FastAPI):
    # Note: In production, use Alembic migrations instead of init_db
    # await init_db()

+    # Ensure service account exists and cache its ID for sync operations
+    async with async_session_maker() as db:
+        service_account_id = await ensure_service_account(db)
+    app.state.service_account_id = service_account_id
+    logger.info(f"[service_account] Service account ready (id={service_account_id})")
+
    # Start maintenance schedule runner + AI conversation cleanup
    scheduler.start()
    async with async_session_maker() as db:
--- a/backend/app/models/step_library.py
+++ b/backend/app/models/step_library.py
@@ -2,7 +2,7 @@ import uuid
 from datetime import datetime, timezone
 from decimal import Decimal
 from typing import TYPE_CHECKING, Optional
-from sqlalchemy import String, DateTime, Integer, Boolean, Text, Numeric, ForeignKey, CheckConstraint
+from sqlalchemy import String, DateTime, Integer, Boolean, Text, Numeric, ForeignKey, CheckConstraint, UniqueConstraint
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 from sqlalchemy.dialects.postgresql import UUID, JSONB, ARRAY
 from app.core.database import Base
@@ -13,6 +13,7 @@ if TYPE_CHECKING:
    from app.models.account import Account
    from app.models.step_category import StepCategory
    from app.models.session import Session
+    from app.models.tree import Tree


 class StepLibrary(Base):
@@ -22,6 +23,7 @@ class StepLibrary(Base):
            "step_type IN ('decision', 'action', 'solution')",
            name='ck_step_library_step_type'
        ),
+        UniqueConstraint('source_tree_id', 'source_node_id', name='uq_step_library_source_node'),
    )

    id: Mapped[uuid.UUID] = mapped_column(
@@ -95,10 +97,26 @@ class StepLibrary(Base):
    # Soft delete
    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)

+    # Sync tracking (flow-sourced steps)
+    source_tree_id: Mapped[Optional[uuid.UUID]] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey('trees.id', ondelete='SET NULL'),
+        nullable=True,
+        index=True
+    )
+    source_node_id: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
+    is_flow_synced: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    last_synced_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+
    # Relationships
    creator: Mapped["User"] = relationship("User", foreign_keys=[created_by])
    team: Mapped[Optional["Team"]] = relationship("Team")
    account: Mapped[Optional["Account"]] = relationship("Account", foreign_keys=[account_id], back_populates="step_library")
+    source_tree: Mapped[Optional["Tree"]] = relationship(
+        "Tree",
+        foreign_keys=[source_tree_id],
+        lazy="select"
+    )
    category: Mapped[Optional["StepCategory"]] = relationship("StepCategory")
    ratings: Mapped[list["StepRating"]] = relationship("StepRating", back_populates="step", cascade="all, delete-orphan")
    usage_logs: Mapped[list["StepUsageLog"]] = relationship("StepUsageLog", back_populates="step", cascade="all, delete-orphan")
--- a/backend/app/models/user.py
+++ b/backend/app/models/user.py
@@ -39,6 +39,7 @@ class User(Base):
    is_super_admin: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    is_team_admin: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default="true")
+    is_service_account: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default="false")
    must_change_password: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default="false")

    # Account-based multi-tenancy (new)
--- a/backend/app/schemas/step_library.py
+++ b/backend/app/schemas/step_library.py
@@ -17,6 +17,7 @@ class StepContent(BaseModel):
    instructions: str = Field(..., min_length=1)
    help_text: Optional[str] = None
    commands: Optional[list[StepCommand]] = None
+    group_label: Optional[str] = None  # Section header this step belongs to (for flow-synced steps)


 # Base schemas
@@ -59,6 +60,8 @@ class StepLibraryResponse(StepLibraryBase):
    # Computed fields (populated by API)
    category_name: Optional[str] = None
    author_name: Optional[str] = None
+    is_flow_synced: bool = False
+    source_tree_name: Optional[str] = None

    class Config:
        from_attributes = True
@@ -79,6 +82,8 @@ class StepLibraryListResponse(BaseModel):
    created_by: UUID
    author_name: Optional[str] = None
    created_at: datetime
+    is_flow_synced: bool = False
+    source_tree_name: Optional[str] = None

    class Config:
        from_attributes = True
--- a/backend/tests/test_step_sync.py
+++ b/backend/tests/test_step_sync.py
@@ -0,0 +1,216 @@
+"""Tests for flow-to-library step sync."""
+import pytest
+from uuid import uuid4
+from app.core.step_sync import extract_steps_for_sync, resolve_step_visibility
+
+
+class TestResolveStepVisibility:
+    """Test visibility resolution logic."""
+
+    def test_public_flow_gives_public_steps(self):
+        result = resolve_step_visibility(is_public=True, account_id=None, node_override=None)
+        assert result == 'public'
+
+    def test_team_flow_gives_team_steps(self):
+        result = resolve_step_visibility(is_public=False, account_id=uuid4(), node_override=None)
+        assert result == 'team'
+
+    def test_private_flow_gives_team_steps(self):
+        result = resolve_step_visibility(is_public=False, account_id=None, node_override=None)
+        assert result == 'team'
+
+    def test_node_override_takes_precedence(self):
+        result = resolve_step_visibility(is_public=True, account_id=None, node_override='team')
+        assert result == 'team'
+
+    def test_public_override_on_team_flow(self):
+        result = resolve_step_visibility(is_public=False, account_id=uuid4(), node_override='public')
+        assert result == 'public'
+
+
+class TestExtractStepsForSync:
+    """Test step extraction from tree structures."""
+
+    def test_extracts_procedure_steps_from_procedural_flow(self):
+        tree_structure = {
+            "steps": [
+                {"id": "step_1", "type": "procedure_step", "title": "Verify prerequisites",
+                 "description": "Check all prereqs", "content_type": "action"},
+                {"id": "end_1", "type": "procedure_end", "title": "Done"},
+            ]
+        }
+        results = list(extract_steps_for_sync(tree_structure, tree_type='procedural'))
+        assert len(results) == 1
+        assert results[0]['source_node_id'] == 'step_1'
+        assert results[0]['title'] == 'Verify prerequisites'
+        assert results[0]['step_type'] == 'action'
+        assert results[0]['content']['instructions'] == 'Check all prereqs'
+
+    def test_skips_section_header_nodes(self):
+        tree_structure = {
+            "steps": [
+                {"id": "sec_1", "type": "section_header", "title": "Phase 1"},
+                {"id": "step_1", "type": "procedure_step", "title": "First step",
+                 "description": "Do this"},
+            ]
+        }
+        results = list(extract_steps_for_sync(tree_structure, tree_type='procedural'))
+        assert len(results) == 1
+        assert results[0]['source_node_id'] == 'step_1'
+
+    def test_captures_section_header_as_group_label(self):
+        tree_structure = {
+            "steps": [
+                {"id": "sec_1", "type": "section_header", "title": "Cable Checks"},
+                {"id": "step_1", "type": "procedure_step", "title": "Check cable",
+                 "description": "Verify cable is seated"},
+            ]
+        }
+        results = list(extract_steps_for_sync(tree_structure, tree_type='procedural'))
+        assert results[0]['content']['group_label'] == 'Cable Checks'
+
+    def test_normalizes_string_commands(self):
+        tree_structure = {
+            "steps": [
+                {"id": "step_1", "type": "procedure_step", "title": "Run command",
+                 "description": "Execute this", "commands": "ping 8.8.8.8"},
+            ]
+        }
+        results = list(extract_steps_for_sync(tree_structure, tree_type='procedural'))
+        assert results[0]['content']['commands'] == [{"label": "", "command": "ping 8.8.8.8", "command_type": None}]
+
+    def test_normalizes_commandblock_commands(self):
+        tree_structure = {
+            "steps": [
+                {"id": "step_1", "type": "procedure_step", "title": "Run PS",
+                 "description": "Run powershell",
+                 "commands": [{"code": "Get-Service", "language": "powershell", "label": "Check services"}]},
+            ]
+        }
+        results = list(extract_steps_for_sync(tree_structure, tree_type='procedural'))
+        cmds = results[0]['content']['commands']
+        assert len(cmds) == 1
+        assert cmds[0]['command'] == 'Get-Service'
+        assert cmds[0]['command_type'] == 'powershell'
+        assert cmds[0]['label'] == 'Check services'
+
+    def test_extracts_action_and_solution_from_troubleshooting(self):
+        tree_structure = {
+            "id": "root",
+            "type": "decision",
+            "question": "What is wrong?",
+            "options": [{"id": "o1", "label": "Thing A", "next_node_id": "act_1"}],
+            "children": [
+                {"id": "act_1", "type": "action", "title": "Fix thing A",
+                 "description": "Do the fix", "next_node_id": "sol_1",
+                 "children": [{"id": "sol_1", "type": "solution", "title": "All fixed",
+                               "description": "Problem resolved"}]},
+            ]
+        }
+        results = list(extract_steps_for_sync(tree_structure, tree_type='troubleshooting'))
+        node_ids = {r['source_node_id'] for r in results}
+        assert 'act_1' in node_ids
+        assert 'sol_1' in node_ids
+        types = {r['source_node_id']: r['step_type'] for r in results}
+        assert types['act_1'] == 'action'
+        assert types['sol_1'] == 'solution'
+
+    def test_uses_title_as_instructions_fallback(self):
+        tree_structure = {
+            "steps": [
+                {"id": "step_1", "type": "procedure_step", "title": "Do the thing"},
+            ]
+        }
+        results = list(extract_steps_for_sync(tree_structure, tree_type='procedural'))
+        assert results[0]['content']['instructions'] == 'Do the thing'
+
+    def test_empty_steps_list(self):
+        tree_structure = {"steps": []}
+        results = list(extract_steps_for_sync(tree_structure, tree_type='procedural'))
+        assert results == []
+
+    def test_maintenance_treated_same_as_procedural(self):
+        tree_structure = {
+            "steps": [
+                {"id": "step_1", "type": "procedure_step", "title": "Maintenance step",
+                 "description": "Do maintenance"},
+            ]
+        }
+        results = list(extract_steps_for_sync(tree_structure, tree_type='maintenance'))
+        assert len(results) == 1
+
+
+class TestSyncOnPublish:
+    """Integration tests — sync triggered by publishing a tree."""
+
+    @pytest.mark.asyncio
+    async def test_publishing_procedural_tree_creates_library_steps(
+        self, client, auth_headers
+    ):
+        # Create a procedural tree with two steps
+        tree_resp = await client.post("/api/v1/trees", json={
+            "name": "Test Procedure",
+            "tree_type": "procedural",
+            "status": "draft",
+            "tree_structure": {
+                "steps": [
+                    {"id": "step_1", "type": "procedure_step",
+                     "title": "First step", "description": "Do this first"},
+                    {"id": "step_2", "type": "procedure_step",
+                     "title": "Second step", "description": "Do this second"},
+                    {"id": "end_1", "type": "procedure_end", "title": "Done"},
+                ]
+            }
+        }, headers=auth_headers)
+        assert tree_resp.status_code == 201
+        tree_id = tree_resp.json()["id"]
+
+        # Publish the tree
+        pub_resp = await client.put(f"/api/v1/trees/{tree_id}", json={"status": "published"}, headers=auth_headers)
+        assert pub_resp.status_code == 200
+
+        # Check library has synced entries
+        lib_resp = await client.get("/api/v1/steps", headers=auth_headers)
+        assert lib_resp.status_code == 200
+        steps = lib_resp.json()
+        synced = [s for s in steps if s.get("is_flow_synced")]
+        assert len(synced) == 2
+        titles = {s["title"] for s in synced}
+        assert "First step" in titles
+        assert "Second step" in titles
+
+    @pytest.mark.asyncio
+    async def test_republishing_updates_existing_library_steps(
+        self, client, auth_headers
+    ):
+        # Create a draft tree first, then publish
+        tree_resp = await client.post("/api/v1/trees", json={
+            "name": "Update Test",
+            "tree_type": "procedural",
+            "status": "draft",
+            "tree_structure": {"steps": [
+                {"id": "step_1", "type": "procedure_step",
+                 "title": "Original title", "description": "Original desc"},
+                {"id": "end_1", "type": "procedure_end", "title": "Done"},
+            ]}
+        }, headers=auth_headers)
+        tree_id = tree_resp.json()["id"]
+        first_pub = await client.put(f"/api/v1/trees/{tree_id}", json={"status": "published"}, headers=auth_headers)
+        assert first_pub.status_code == 200
+
+        # Republish with updated step title
+        second_pub = await client.put(f"/api/v1/trees/{tree_id}", json={
+            "tree_structure": {"steps": [
+                {"id": "step_1", "type": "procedure_step",
+                 "title": "Updated title", "description": "Updated desc"},
+                {"id": "end_1", "type": "procedure_end", "title": "Done"},
+            ]},
+            "status": "published"
+        }, headers=auth_headers)
+        assert second_pub.status_code == 200
+
+        # Check library shows updated title (not a duplicate)
+        lib_resp = await client.get("/api/v1/steps", headers=auth_headers)
+        synced = [s for s in lib_resp.json() if s.get("is_flow_synced")]
+        assert len(synced) == 1
+        assert synced[0]["title"] == "Updated title"