#!/usr/bin/env python3 """ Seed Phase 9 QA fixtures: 4 ai_sessions + matching suggested_fixes that exercise the five Phase 9 components which gate on a backend-emitted `SUGGEST_FIX` action and don't fire reliably in normal local sessions. Usage: cd backend python -m scripts.seed_phase9_qa_fixtures python -m scripts.seed_phase9_qa_fixtures --reset # delete & recreate Targets the super-admin from `seed_test_users.py` (admin@resolutionflow.example.com) and their account. UUIDs are deterministic (UUID5 over a fixed namespace) so re-runs are idempotent without --reset. Sessions created: | # | Title | Phase 9 component reached when… | |---|---------------------------------|-------------------------------------------------------| | A | Phase 9 QA — no-template path | ChatTabStrip + ScriptBuilderTab + ProposalBanner | | B | Phase 9 QA — drafted-script | InlineNoTemplateDialog + ProposalBanner | | C | Phase 9 QA — template match | TemplateMatchPanel + ProposalBanner | | D | Phase 9 QA — verify state | EscalateInterceptDialog (with new "partial" choice) | Run /qa, then in the browser go to /pilot, click each session in the sidebar, and exercise its Phase 9 surface. The session URLs are printed at the end. """ import argparse import asyncio import sys import uuid from datetime import datetime, timedelta, timezone from sqlalchemy import text from sqlalchemy.ext.asyncio import create_async_engine from app.core.config import settings ADMIN_EMAIL = "admin@resolutionflow.example.com" # Deterministic UUIDs so re-running the seeder updates rather than duplicates. NS = uuid.UUID("00000000-0000-0000-0000-000000000901") SESSION_A = uuid.uuid5(NS, "session-A-no-template") SESSION_B = uuid.uuid5(NS, "session-B-drafted-script") SESSION_C = uuid.uuid5(NS, "session-C-template-match") SESSION_D = uuid.uuid5(NS, "session-D-verify-state") FIX_A = uuid.uuid5(NS, "fix-A") FIX_B = uuid.uuid5(NS, "fix-B") FIX_C = uuid.uuid5(NS, "fix-C") FIX_D = uuid.uuid5(NS, "fix-D") CATEGORY_QA = uuid.uuid5(NS, "category-qa-fixtures") TEMPLATE_QA = uuid.uuid5(NS, "template-qa-fixtures") DRAFTED_SCRIPT = """\ # Phase 9 QA fixture — AI-drafted PowerShell to flush DNS and # restart the FortiClient service. Not for production use. ipconfig /flushdns Restart-Service -Name "FortiSslvpnDaemon" -Force Get-Service -Name "FortiSslvpnDaemon" | Format-Table -AutoSize """ TEMPLATE_BODY = """\ # Phase 9 QA fixture — canned template that the AI matches against. param([string]$ServiceName = "FortiSslvpnDaemon") Restart-Service -Name $ServiceName -Force Get-Service -Name $ServiceName | Select-Object Status, Name """ async def main(reset: bool = False) -> None: db_url = ( settings.ADMIN_DATABASE_URL if hasattr(settings, "ADMIN_DATABASE_URL") and settings.ADMIN_DATABASE_URL else settings.DATABASE_URL ) engine = create_async_engine(db_url, echo=False) now = datetime.now(timezone.utc) async with engine.begin() as conn: # ─── Locate the admin user + account ─────────────────────────── row = ( await conn.execute( text( "SELECT id, account_id FROM users WHERE email = :email LIMIT 1" ), {"email": ADMIN_EMAIL}, ) ).first() if row is None: print( f"ERROR: user {ADMIN_EMAIL!r} not found. Run " "`python -m scripts.seed_test_users` first.", file=sys.stderr, ) sys.exit(2) user_id, account_id = row if reset: await conn.execute( text( "DELETE FROM session_suggested_fixes WHERE id = ANY(:ids)" ), {"ids": [FIX_A, FIX_B, FIX_C, FIX_D]}, ) await conn.execute( text("DELETE FROM ai_sessions WHERE id = ANY(:ids)"), {"ids": [SESSION_A, SESSION_B, SESSION_C, SESSION_D]}, ) await conn.execute( text("DELETE FROM script_templates WHERE id = :id"), {"id": TEMPLATE_QA}, ) await conn.execute( text("DELETE FROM script_categories WHERE id = :id"), {"id": CATEGORY_QA}, ) # ─── Script category + template (for Session C) ──────────────── await conn.execute( text( """ INSERT INTO script_categories (id, name, slug, sort_order, is_active, created_at, updated_at) VALUES (:id, 'QA Fixtures', 'qa-fixtures', 999, true, :now, :now) ON CONFLICT (id) DO NOTHING """ ), {"id": CATEGORY_QA, "now": now}, ) await conn.execute( text( """ INSERT INTO script_templates ( id, category_id, account_id, created_by, name, slug, description, script_body, language, parameters_schema, default_values, validation_rules, tags, complexity, requires_elevation, requires_modules, created_at, updated_at ) VALUES ( :id, :cat_id, :acct_id, :user_id, 'QA Fixture: Restart Forti Service', 'qa-fixture-restart-forti-service', 'Phase 9 QA fixture template for TemplateMatchPanel testing.', :body, 'powershell', '{}'::jsonb, '{}'::jsonb, '{}'::jsonb, '[]'::jsonb, 'beginner', false, '[]'::jsonb, :now, :now ) ON CONFLICT (id) DO NOTHING """ ), { "id": TEMPLATE_QA, "cat_id": CATEGORY_QA, "acct_id": account_id, "user_id": user_id, "body": TEMPLATE_BODY, "now": now, }, ) # ─── 4 sessions ──────────────────────────────────────────────── # `canAct` in the chat header gates Resolve/Escalate on # `messages.length >= 2`, so each fixture seeds two synthetic # conversation messages — enough to enable the buttons that drive # the Phase 9 surfaces. seed_messages = ( '[' '{"role":"user","content":"QA fixture: see seed_phase9_qa_fixtures.py"},' '{"role":"assistant","content":"This session is a Phase 9 QA fixture. The suggested fix below is pre-seeded — drive it from the UI."}' ']' ) sessions = [ (SESSION_A, "Phase 9 QA — no-template path"), (SESSION_B, "Phase 9 QA — drafted-script path"), (SESSION_C, "Phase 9 QA — template-match path"), (SESSION_D, "Phase 9 QA — verify state (Escalate intercept)"), ] for sid, title in sessions: await conn.execute( text( """ INSERT INTO ai_sessions ( id, user_id, account_id, session_type, title, intake_type, intake_content, status, confidence_tier, confidence_score, conversation_messages, total_input_tokens, total_output_tokens, step_count, is_branching, state_version, handoff_count, total_active_seconds, total_parked_seconds, created_at, updated_at ) VALUES ( :id, :user_id, :acct_id, 'chat', :title, 'free_text', '{"text": "QA fixture session"}'::jsonb, 'active', 'discovery', 0.0, (:msgs)::jsonb, 0, 0, 0, false, 0, 0, 0, 0, :now, :now ) ON CONFLICT (id) DO UPDATE SET title = EXCLUDED.title, status = EXCLUDED.status, conversation_messages = EXCLUDED.conversation_messages, updated_at = EXCLUDED.updated_at """ ), { "id": sid, "user_id": user_id, "acct_id": account_id, "title": title, "msgs": seed_messages, "now": now, }, ) # ─── 4 suggested fixes ───────────────────────────────────────── # Fix A — no template, no draft → ChatTabStrip + ScriptBuilderTab await _upsert_fix( conn, fix_id=FIX_A, session_id=SESSION_A, account_id=account_id, title="Restart the FortiClient daemon and flush DNS", description=( "Error -8 on FortiClient SSL VPN typically clears after a " "service restart on the endpoint. No matching template; " "no AI draft yet — engineer should choose Build Template " "or One-Off in the Script Builder tab." ), confidence_pct=72, script_template_id=None, ai_drafted_script=None, status="proposed", applied_at=None, now=now, ) # Fix B — drafted script, no template → InlineNoTemplateDialog await _upsert_fix( conn, fix_id=FIX_B, session_id=SESSION_B, account_id=account_id, title="Run AI-drafted PowerShell to recover SSL VPN", description=( "AI drafted a session-specific script because no library " "template matched. Inline dialog should offer Save-as-template, " "Run-once, or Discard." ), confidence_pct=68, script_template_id=None, ai_drafted_script=DRAFTED_SCRIPT, status="proposed", applied_at=None, now=now, ) # Fix C — template match → TemplateMatchPanel await _upsert_fix( conn, fix_id=FIX_C, session_id=SESSION_C, account_id=account_id, title="Match: QA Fixture Restart Forti Service", description=( "AI matched an existing library template. The match panel " "should render with the parameterization preview and an " "explicit 'I ran this' action." ), confidence_pct=88, script_template_id=TEMPLATE_QA, ai_drafted_script=None, status="proposed", applied_at=None, now=now, ) # Fix D — applied_at set, status='proposed' → verify state. # Hitting Escalate from this state opens EscalateInterceptDialog. await _upsert_fix( conn, fix_id=FIX_D, session_id=SESSION_D, account_id=account_id, title="Verifying: post-apply tunnel reconnect", description=( "Engineer marked the fix as Applied; we're now in the " "verify window. Clicking Escalate from here should open " "the EscalateInterceptDialog with the four outcome choices " "(worked / didn't / partial / never-applied)." ), confidence_pct=80, script_template_id=None, ai_drafted_script=DRAFTED_SCRIPT, status="proposed", applied_at=now - timedelta(minutes=2), now=now, ) await engine.dispose() print() print("=" * 64) print(" Phase 9 QA fixtures ready.") print("=" * 64) print() print(f" Sign in as : {ADMIN_EMAIL}") print(f" Then visit : http://docker-01:5173/pilot") print(f" Pick from the History sidebar:") print(f" A. Phase 9 QA — no-template path (ChatTabStrip + ScriptBuilderTab)") print(f" B. Phase 9 QA — drafted-script path (InlineNoTemplateDialog)") print(f" C. Phase 9 QA — template-match path (TemplateMatchPanel)") print(f" D. Phase 9 QA — verify state (EscalateInterceptDialog)") print() print(f" Re-run with --reset to wipe and recreate.") print() async def _upsert_fix( conn, *, fix_id: uuid.UUID, session_id: uuid.UUID, account_id: uuid.UUID, title: str, description: str, confidence_pct: int, script_template_id: uuid.UUID | None, ai_drafted_script: str | None, status: str, applied_at: datetime | None, now: datetime, ) -> None: await conn.execute( text( """ INSERT INTO session_suggested_fixes ( id, session_id, account_id, title, description, confidence_pct, script_template_id, ai_drafted_script, status, applied_at, created_at ) VALUES ( :id, :sid, :acct, :title, :desc, :conf, :tmpl, :draft, :status, :applied, :now ) ON CONFLICT (id) DO UPDATE SET title = EXCLUDED.title, description = EXCLUDED.description, confidence_pct = EXCLUDED.confidence_pct, script_template_id = EXCLUDED.script_template_id, ai_drafted_script = EXCLUDED.ai_drafted_script, status = EXCLUDED.status, applied_at = EXCLUDED.applied_at, superseded_at = NULL """ ), { "id": fix_id, "sid": session_id, "acct": account_id, "title": title, "desc": description, "conf": confidence_pct, "tmpl": script_template_id, "draft": ai_drafted_script, "status": status, "applied": applied_at, "now": now, }, ) if __name__ == "__main__": parser = argparse.ArgumentParser(description="Seed Phase 9 QA fixtures.") parser.add_argument( "--reset", action="store_true", help="Delete and recreate the fixtures.", ) args = parser.parse_args() asyncio.run(main(reset=args.reset))