feat(pilot): PATCH /suggested-fixes/:id/script endpoint

Called by the inline Script Builder tab on Submit. Writes
ai_drafted_script + ai_drafted_parameters to the fix without stamping
applied_at (a draft is not an application — that's §5 of the Phase 9
spec). Bumps state_version so Resolve/Escalate preview bundles
regenerate.

409 on terminal fix status. 404 on wrong session. 422 on empty script.

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

backend/app/api/endpoints/session_suggested_fixes.py

@@ -32,6 +32,7 @@ from app.schemas.session_suggested_fix import (
     SessionSuggestedFixDecisionResponse,
     SessionSuggestedFixOutcomeRequest,
     SessionSuggestedFixResponse,
+    SessionSuggestedFixScriptRequest,
 )
 from app.models.draft_template import DraftTemplate
 from app.models.session_fact import SessionFact
@@ -355,6 +356,60 @@ async def patch_suggested_fix_outcome(
     return SessionSuggestedFixResponse.model_validate(fix)
 
 
+# ── Suggested fix: attach drafted script ─────────────────────────────────────
+
+@router.patch(
+    "/suggested-fixes/{fix_id}/script",
+    response_model=SessionSuggestedFixResponse,
+)
+async def patch_suggested_fix_script(
+    session_id: UUID,
+    fix_id: UUID,
+    body: SessionSuggestedFixScriptRequest,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+    _: None = Depends(require_engineer_or_admin),
+) -> SessionSuggestedFixResponse:
+    """Attach an engineer-drafted script to a suggested fix.
+
+    Called by the inline Script Builder tab on Submit. Does NOT stamp
+    applied_at — a draft is not an application. Bumps state_version so
+    the Resolve/Escalate preview bundles regenerate.
+    """
+    await _load_session_or_404(db, session_id)
+
+    fix = await db.scalar(
+        select(SessionSuggestedFix).where(
+            SessionSuggestedFix.id == fix_id,
+            SessionSuggestedFix.session_id == session_id,
+        )
+    )
+    if fix is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Suggested fix not found")
+
+    TERMINAL = {"applied_success", "applied_failed", "dismissed"}
+    if fix.status in TERMINAL:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=f"Fix is already in terminal status {fix.status!r}",
+        )
+
+    fix.ai_drafted_script = body.ai_drafted_script
+    fix.ai_drafted_parameters = body.ai_drafted_parameters
+
+    # Bump state_version on the parent session — previews cached by
+    # (session_id, state_version) must regenerate to reflect the new draft.
+    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 SessionSuggestedFixResponse.model_validate(fix)
+
+
 # ── Suggested fix: clear AI outcome proposal ("Not yet") ─────────────────────
 
 @router.delete(

backend/tests/test_fix_script_endpoint.py

@@ -0,0 +1,120 @@
+"""Integration tests for PATCH /ai-sessions/{sid}/suggested-fixes/{fid}/script."""
+from __future__ import annotations
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+from uuid import UUID, uuid4
+
+from app.models.ai_session import AISession
+from app.models.session_suggested_fix import SessionSuggestedFix
+
+
+async def _make_session_with_fix(
+    test_db, user, *, status: str = "proposed", with_script: bool = False,
+) -> tuple[str, str]:
+    """Create a pilot session + suggested fix for tests. Returns (sid, fid)."""
+    session = AISession(
+        id=uuid4(),
+        user_id=user["user_data"]["id"],
+        account_id=user["user_data"]["account_id"],
+        session_type="tshoot",
+        intake_type="psa_ticket",
+        intake_content={},
+        title="QA",
+        status="active",
+        confidence_tier="exploring",
+        confidence_score=0.0,
+    )
+    test_db.add(session)
+    await test_db.flush()
+    fix = SessionSuggestedFix(
+        id=uuid4(),
+        session_id=session.id,
+        account_id=user["user_data"]["account_id"],
+        title="QA: test fix",
+        description="desc",
+        confidence_pct=80,
+        status=status,
+        ai_drafted_script="pre-existing" if with_script else None,
+    )
+    test_db.add(fix)
+    await test_db.commit()
+    return str(session.id), str(fix.id)
+
+
+@pytest.mark.asyncio
+async def test_patch_script_happy_path(
+    client: AsyncClient, test_user, auth_headers, test_db
+):
+    sid, fid = await _make_session_with_fix(test_db, test_user)
+    r = await client.patch(
+        f"/api/v1/ai-sessions/{sid}/suggested-fixes/{fid}/script",
+        json={"ai_drafted_script": "Write-Host 'hello'"},
+        headers=auth_headers,
+    )
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["ai_drafted_script"] == "Write-Host 'hello'"
+    assert body["applied_at"] is None  # draft != apply
+    assert body["status"] == "proposed"
+
+
+@pytest.mark.asyncio
+async def test_patch_script_bumps_state_version(
+    client: AsyncClient, test_user, auth_headers, test_db
+):
+    sid, fid = await _make_session_with_fix(test_db, test_user)
+    before = await test_db.scalar(
+        select(AISession.state_version).where(AISession.id == UUID(sid))
+    )
+    r = await client.patch(
+        f"/api/v1/ai-sessions/{sid}/suggested-fixes/{fid}/script",
+        json={"ai_drafted_script": "echo hi"},
+        headers=auth_headers,
+    )
+    assert r.status_code == 200
+    after = await test_db.scalar(
+        select(AISession.state_version).where(AISession.id == UUID(sid))
+    )
+    assert after == (before or 0) + 1
+
+
+@pytest.mark.asyncio
+async def test_patch_script_rejects_terminal_fix(
+    client: AsyncClient, test_user, auth_headers, test_db
+):
+    sid, fid = await _make_session_with_fix(test_db, test_user, status="applied_success")
+    r = await client.patch(
+        f"/api/v1/ai-sessions/{sid}/suggested-fixes/{fid}/script",
+        json={"ai_drafted_script": "echo hi"},
+        headers=auth_headers,
+    )
+    assert r.status_code == 409
+
+
+@pytest.mark.asyncio
+async def test_patch_script_rejects_empty_body(
+    client: AsyncClient, test_user, auth_headers, test_db
+):
+    sid, fid = await _make_session_with_fix(test_db, test_user)
+    r = await client.patch(
+        f"/api/v1/ai-sessions/{sid}/suggested-fixes/{fid}/script",
+        json={"ai_drafted_script": ""},
+        headers=auth_headers,
+    )
+    assert r.status_code == 422  # pydantic min_length=1
+
+
+@pytest.mark.asyncio
+async def test_patch_script_404_on_wrong_session(
+    client: AsyncClient, test_user, auth_headers, test_db
+):
+    _, fid = await _make_session_with_fix(test_db, test_user)
+    wrong_sid = str(uuid4())
+    r = await client.patch(
+        f"/api/v1/ai-sessions/{wrong_sid}/suggested-fixes/{fid}/script",
+        json={"ai_drafted_script": "echo hi"},
+        headers=auth_headers,
+    )
+    assert r.status_code == 404