feat(l1): L1 endpoint surface (intake/queue/step/notes/resolve/escalate)

Mounts /api/v1/l1/* with require_l1_or_coverage on every route. Intake
creates an internal ticket and starts a flow OR adhoc session (PSA queue
merge follows in Phase 2). Step/notes/resolve/escalate delegate to
l1_session_service. escalate-without-walk creates an immediately-
escalated session for the BuildAbortedNoKB path.

ValueError from services → 400. Cross-account session access → 404.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-28 13:33:18 -04:00
parent 054e9da49b
commit 96973c7968
4 changed files with 704 additions and 0 deletions

72
backend/app/schemas/l1.py Normal file
View File

@@ -0,0 +1,72 @@
"""Pydantic schemas for the /l1/* endpoint surface."""
from datetime import datetime
from typing import Any, Literal, Optional
from uuid import UUID
from pydantic import BaseModel, Field
class IntakeRequest(BaseModel):
problem_statement: str = Field(..., min_length=1)
customer_name: Optional[str] = None
customer_contact: Optional[str] = None
flow_id: Optional[UUID] = None
class IntakeResponse(BaseModel):
session_id: UUID
session_kind: Literal["flow", "proposal", "adhoc"]
ticket_id: str
ticket_kind: Literal["psa", "internal"]
class StepRequest(BaseModel):
node_id: str
question: str
answer: str
note: Optional[str] = None
class NotesRequest(BaseModel):
notes: list[dict[str, Any]]
class ResolveRequest(BaseModel):
helpful: bool
resolution_notes: str
class EscalateRequest(BaseModel):
reason: Optional[str] = None
reason_category: str = Field(..., min_length=1)
class EscalateWithoutWalkRequest(BaseModel):
problem_statement: str = Field(..., min_length=1)
customer_name: Optional[str] = None
customer_contact: Optional[str] = None
reason_category: str = Field(..., min_length=1)
reason: Optional[str] = None
class WalkSessionResponse(BaseModel):
id: UUID
session_kind: str
flow_id: Optional[UUID]
flow_proposal_id: Optional[UUID]
current_node_id: Optional[str]
walked_path: list[dict[str, Any]]
walk_notes: list[dict[str, Any]]
status: str
started_at: datetime
last_step_at: datetime
resolved_at: Optional[datetime]
class QueueRow(BaseModel):
ticket_id: str
ticket_kind: Literal["psa", "internal"]
problem_statement: Optional[str] = None
customer_name: Optional[str] = None
status: str
created_at: Optional[datetime] = None