feat(pilot): PATCH /suggested-fixes/:id/outcome endpoint + tests

Records engineer-reported outcome (applied_success|applied_failed|
applied_partial|dismissed). Enforces transition rules (partial → success/
failed allowed; terminal outcomes return 409) and notes requirements
(applied_partial requires notes).

Sets verified_at on success/failure, stamps applied_at if not already
set (handles the case where the AI [FIX_OUTCOME] marker fires before
the engineer clicks Apply).

Also fixes pre-existing test-infrastructure bug: network_diagram.py used
bare string server_default="'[]'" for JSONB columns, which asyncpg
rejects during test schema creation. Changed to text("'[]'::jsonb") to
match the pattern used by script_template.py.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-04-23 14:50:21 -04:00
parent 4a8e3ae954
commit 8988dbc885
5 changed files with 280 additions and 12 deletions

View File

@@ -30,6 +30,7 @@ from app.schemas.session_suggested_fix import (
ResolutionPostResponse,
SessionSuggestedFixDecisionRequest,
SessionSuggestedFixDecisionResponse,
SessionSuggestedFixOutcomeRequest,
SessionSuggestedFixResponse,
)
from app.models.draft_template import DraftTemplate
@@ -216,6 +217,70 @@ async def record_decision(
)
# ── Suggested fix: outcome ────────────────────────────────────────────────
@router.patch(
"/suggested-fixes/{fix_id}/outcome",
response_model=SessionSuggestedFixResponse,
)
async def patch_suggested_fix_outcome(
session_id: UUID,
fix_id: UUID,
body: SessionSuggestedFixOutcomeRequest,
current_user: Annotated[User, Depends(get_current_active_user)],
db: Annotated[AsyncSession, Depends(get_db)],
_: None = Depends(require_engineer_or_admin),
) -> SessionSuggestedFixResponse:
"""Record the engineer's outcome for an applied fix.
See `SessionSuggestedFixOutcomeRequest` for transition rules.
"""
await _load_session_or_404(db, session_id)
now = datetime.now(timezone.utc)
result = await db.execute(
select(SessionSuggestedFix).where(
SessionSuggestedFix.id == fix_id,
SessionSuggestedFix.session_id == session_id,
)
)
fix = result.scalar_one_or_none()
if fix is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Suggested fix not found"
)
if body.outcome == "applied_partial" and not (body.notes and body.notes.strip()):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="notes are required when outcome is applied_partial",
)
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.status = body.outcome
if body.outcome == "applied_partial":
fix.partial_notes = (body.notes or "").strip() or None
elif body.outcome == "applied_failed":
fix.failure_reason = (body.notes or "").strip() or None
fix.verified_at = now
elif body.outcome == "applied_success":
fix.verified_at = now
# dismissed: no timestamp/notes stamping
if fix.applied_at is None and body.outcome != "dismissed":
fix.applied_at = now
await db.commit()
await db.refresh(fix)
return SessionSuggestedFixResponse.model_validate(fix)
async def _summarize_session_for_extraction(
db: AsyncSession, session_id: UUID,
) -> str:

View File

@@ -3,7 +3,7 @@ import uuid
from datetime import datetime, timezone
from typing import Any, TYPE_CHECKING
from sqlalchemy import String, Text, Boolean, DateTime, ForeignKey
from sqlalchemy import String, Text, Boolean, DateTime, ForeignKey, text
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.dialects.postgresql import UUID, JSONB
@@ -30,8 +30,8 @@ class NetworkDiagram(Base):
client_name: Mapped[str | None] = mapped_column(String(255), nullable=True)
asset_name: Mapped[str | None] = mapped_column(String(255), nullable=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
nodes: Mapped[list[dict[str, Any]]] = mapped_column(JSONB, nullable=False, server_default="'[]'")
edges: Mapped[list[dict[str, Any]]] = mapped_column(JSONB, nullable=False, server_default="'[]'")
nodes: Mapped[list[dict[str, Any]]] = mapped_column(JSONB, nullable=False, server_default=text("'[]'::jsonb"))
edges: Mapped[list[dict[str, Any]]] = mapped_column(JSONB, nullable=False, server_default=text("'[]'::jsonb"))
thumbnail_url: Mapped[str | None] = mapped_column(Text, nullable=True)
is_archived: Mapped[bool] = mapped_column(
Boolean, nullable=False, default=False,

View File

@@ -103,9 +103,11 @@ class SessionSuggestedFixOutcomeRequest(BaseModel):
engineer took); outcome captures whether the fix actually worked.
Allowed transitions:
- from `proposed`: any of applied_success | applied_failed | applied_partial | dismissed
- from `applied_partial`: applied_success | applied_failed (partial is not terminal)
- from any terminal outcome: no change (server returns 409)
- from `proposed` or `applied_partial`: any outcome is valid
(partial is parked, not terminal — the engineer may update notes,
abandon via dismiss, or advance to success/failed)
- from any terminal outcome (`applied_success`, `applied_failed`,
`dismissed`): server returns 409
"""
outcome: FixOutcome
# Required for applied_partial, optional for applied_failed, ignored otherwise.