feat(pilot): [FIX_OUTCOME] marker parser + AI outcome proposal

The AI emits [FIX_OUTCOME] when the engineer indicates in chat that a
prior suggested fix worked, didn't work, or was partially applied. The
marker writes to session_suggested_fixes.ai_outcome_proposal (JSONB),
which the frontend surfaces as a "confirm outcome?" banner. The status
column is only updated when the engineer clicks confirm (via PATCH
/outcome endpoint from Task 3).

Placeholder-only system prompt wiring comes in Task 5.

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

backend/app/services/unified_chat_service.py

@@ -354,6 +354,56 @@ def _parse_suggest_fix_marker(
     return cleaned, parsed
 
 
+def _parse_fix_outcome_marker(
+    ai_content: str,
+) -> tuple[str, dict[str, Any] | None]:
+    """Extract a single [FIX_OUTCOME]...[/FIX_OUTCOME] JSON block.
+
+    Block shape:
+      {"fix_id": "<uuid>", "outcome": "success"|"failure"|"partial",
+       "reason": "<one-line>"}
+
+    Emitted by the AI when the engineer clearly indicates in chat that a
+    prior suggested fix worked, didn't work, or was partially applied.
+    The marker PROPOSES an outcome — the engineer confirms via the UI.
+    Only the last block in a response is honored.
+    """
+    blocks = list(re.finditer(
+        r"\[FIX_OUTCOME\]\s*([\s\S]*?)\s*\[/FIX_OUTCOME\]", ai_content,
+    ))
+    if not blocks:
+        return ai_content, None
+
+    last = blocks[-1]
+    raw = last.group(1).strip()
+    if raw.startswith("```"):
+        raw = re.sub(r"^```(?:json)?\s*", "", raw)
+        raw = re.sub(r"\s*```$", "", raw)
+
+    cleaned = re.sub(
+        r"\[FIX_OUTCOME\]\s*[\s\S]*?\s*\[/FIX_OUTCOME\]", "", ai_content,
+    ).strip()
+
+    try:
+        data = json.loads(raw)
+    except (json.JSONDecodeError, ValueError) as e:
+        logger.warning("Failed to parse [FIX_OUTCOME] block: %s", e)
+        return cleaned, None
+
+    if not isinstance(data, dict):
+        return cleaned, None
+
+    fix_id = str(data.get("fix_id") or "").strip()
+    outcome = str(data.get("outcome") or "").strip().lower()
+    reason = str(data.get("reason") or "").strip()
+
+    if not fix_id or outcome not in {"success", "failure", "partial"}:
+        logger.warning("[FIX_OUTCOME] missing/invalid fields, dropping")
+        return cleaned, None
+
+    return cleaned, {"fix_id": fix_id, "outcome": outcome, "reason": reason}
+
+
 async def _persist_suggested_fix(
     *,
     db: AsyncSession,
@@ -415,6 +465,39 @@ async def _persist_suggested_fix(
     await db.flush()
 
 
+async def _record_ai_outcome_proposal(
+    *,
+    db: AsyncSession,
+    session: AISession,
+    proposal: dict[str, Any],
+) -> None:
+    """Persist the AI's proposed outcome on the active fix.
+
+    Writes to session_suggested_fixes.ai_outcome_proposal. Frontend polls
+    the active fix and renders the AI-confirming banner state when this is
+    non-null. Does NOT mutate the fix's status — the engineer's confirmation
+    click via PATCH /outcome is what changes the status.
+
+    Drops silently when the fix_id isn't a valid UUID or doesn't belong to
+    this session.
+    """
+    try:
+        fix_uuid = UUID(proposal["fix_id"])
+    except (ValueError, KeyError, TypeError):
+        logger.warning("[FIX_OUTCOME] invalid fix_id, dropping")
+        return
+
+    await db.execute(
+        update(SessionSuggestedFix)
+        .where(
+            SessionSuggestedFix.id == fix_uuid,
+            SessionSuggestedFix.session_id == session.id,
+        )
+        .values(ai_outcome_proposal=proposal)
+    )
+    await db.flush()
+
+
 async def _persist_promote_items(
     *,
     db: AsyncSession,
@@ -566,6 +649,7 @@ async def send_chat_message(
             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)
+            branch_display, branch_outcome_proposal = _parse_fix_outcome_marker(branch_display)
             if branch_display != ai_content:
                 # Store stripped content in branch history
                 msgs[-1] = {"role": "assistant", "content": branch_display}
@@ -629,6 +713,12 @@ async def send_chat_message(
                     db=db, session=session, fix=branch_suggest_fix,
                 )
 
+            # Persist a [FIX_OUTCOME] proposal if the branch turn included one.
+            if branch_outcome_proposal is not None:
+                await _record_ai_outcome_proposal(
+                    db=db, session=session, proposal=branch_outcome_proposal,
+                )
+
             suggested_flows = extract_suggested_flows(
                 await rag_search(query=message, account_id=account_id, db=db, limit=8)
             )
@@ -681,11 +771,16 @@ async def send_chat_message(
     # Check for a [SUGGEST_FIX] marker — supersedes the prior active fix.
     display_content, suggest_fix_data = _parse_suggest_fix_marker(display_content)
 
+    # Check for a [FIX_OUTCOME] proposal — AI confirms a prior fix's outcome.
+    display_content, outcome_proposal = _parse_fix_outcome_marker(display_content)
+
     logger.info(
         "Marker parsing results — actions: %s, questions: %s, fork: %s, "
-        "promote: %d, suggest_fix: %s, raw_length: %d, display_length: %d",
+        "promote: %d, suggest_fix: %s, outcome_proposal: %s, "
+        "raw_length: %d, display_length: %d",
         bool(actions_data), bool(questions_data), bool(fork_data),
         len(promote_items or []), bool(suggest_fix_data),
+        bool(outcome_proposal),
         len(ai_content), len(display_content),
     )
 
@@ -774,6 +869,12 @@ async def send_chat_message(
     if suggest_fix_data:
         await _persist_suggested_fix(db=db, session=session, fix=suggest_fix_data)
 
+    # Persist a [FIX_OUTCOME] proposal if this turn included one.
+    if outcome_proposal is not None:
+        await _record_ai_outcome_proposal(
+            db=db, session=session, proposal=outcome_proposal,
+        )
+
     suggested_flows = extract_suggested_flows(rag_results)
 
     return display_content, suggested_flows, session, fork_metadata, actions_data, questions_data

backend/tests/test_fix_outcome_marker.py

@@ -0,0 +1,91 @@
+"""Unit tests for the [FIX_OUTCOME] marker parser."""
+from __future__ import annotations
+
+from app.services.unified_chat_service import _parse_fix_outcome_marker
+
+
+def test_parses_success_outcome():
+    ai = (
+        "Great news — that confirms the root cause.\n\n"
+        "[FIX_OUTCOME]\n"
+        '{"fix_id":"11111111-1111-1111-1111-111111111111",'
+        '"outcome":"success","reason":"user said the fix worked"}\n'
+        "[/FIX_OUTCOME]\n"
+    )
+    cleaned, parsed = _parse_fix_outcome_marker(ai)
+    assert "[FIX_OUTCOME]" not in cleaned
+    assert "confirms the root cause" in cleaned
+    assert parsed == {
+        "fix_id": "11111111-1111-1111-1111-111111111111",
+        "outcome": "success",
+        "reason": "user said the fix worked",
+    }
+
+
+def test_parses_failure_outcome():
+    ai = (
+        "[FIX_OUTCOME]\n"
+        '{"fix_id":"22222222-2222-2222-2222-222222222222",'
+        '"outcome":"failure","reason":"user reports still broken"}\n'
+        "[/FIX_OUTCOME]"
+    )
+    cleaned, parsed = _parse_fix_outcome_marker(ai)
+    assert "[FIX_OUTCOME]" not in cleaned
+    assert parsed["outcome"] == "failure"
+
+
+def test_missing_marker_returns_none():
+    ai = "no marker here"
+    cleaned, parsed = _parse_fix_outcome_marker(ai)
+    assert cleaned == ai
+    assert parsed is None
+
+
+def test_invalid_json_is_dropped():
+    ai = "[FIX_OUTCOME]\nnot-json\n[/FIX_OUTCOME]"
+    cleaned, parsed = _parse_fix_outcome_marker(ai)
+    assert "[FIX_OUTCOME]" not in cleaned
+    assert parsed is None
+
+
+def test_unknown_outcome_rejected():
+    ai = (
+        "[FIX_OUTCOME]\n"
+        '{"fix_id":"33333333-3333-3333-3333-333333333333",'
+        '"outcome":"maybe","reason":"x"}\n'
+        "[/FIX_OUTCOME]"
+    )
+    _, parsed = _parse_fix_outcome_marker(ai)
+    assert parsed is None
+
+
+def test_last_block_wins_when_multiple():
+    ai = (
+        "[FIX_OUTCOME]\n"
+        '{"fix_id":"44444444-4444-4444-4444-444444444444",'
+        '"outcome":"failure","reason":"first"}\n'
+        "[/FIX_OUTCOME]\n"
+        "[FIX_OUTCOME]\n"
+        '{"fix_id":"55555555-5555-5555-5555-555555555555",'
+        '"outcome":"success","reason":"second"}\n'
+        "[/FIX_OUTCOME]"
+    )
+    cleaned, parsed = _parse_fix_outcome_marker(ai)
+    assert "[FIX_OUTCOME]" not in cleaned
+    assert parsed["fix_id"] == "55555555-5555-5555-5555-555555555555"
+    assert parsed["outcome"] == "success"
+
+
+def test_parses_partial_outcome():
+    ai = (
+        "[FIX_OUTCOME]\n"
+        '{"fix_id":"66666666-6666-6666-6666-666666666666",'
+        '"outcome":"partial","reason":"user ran cred clear only"}\n'
+        "[/FIX_OUTCOME]"
+    )
+    _, parsed = _parse_fix_outcome_marker(ai)
+    assert parsed == {
+        "fix_id": "66666666-6666-6666-6666-666666666666",
+        "outcome": "partial",
+        "reason": "user ran cred clear only",
+    }