import re import uuid from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from ..db import get_session from ..models import AnalysisJob, Dump, Escalation, KBEntry, TierReport from ..schemas.escalations import EscalationDetail, EscalationSummary, ResolveEscalationRequest from ..security.api_keys import StaffActor, require_staff from ..security.audit import write_audit from ..triage.schemas import DumpType from ..triage.transcript_signals import normalize_module from ..triage.transcripts import load_transcript router = APIRouter(prefix="/api/v1/staff/escalations", tags=["staff"], dependencies=[Depends(require_staff)]) _CODE_HEX = re.compile(r"^0x([0-9A-Fa-f]+)") def _code_int(code_display: str | None) -> int | None: """Recovers the raw int Tier 1 parsed from the transcript out of its display string (e.g. "0x1E KMODE_EXCEPTION_NOT_HANDLED", ParsedTriage's `code_display`) — the KB entry wants the int, not the human-readable label, and this is denormalizing an already-computed signal, not re-parsing the transcript.""" if not code_display: return None match = _CODE_HEX.match(code_display) return int(match.group(1), 16) if match else None def _summary(escalation: Escalation, dump_id: uuid.UUID) -> EscalationSummary: return EscalationSummary( id=escalation.id, job_id=escalation.job_id, dump_id=dump_id, reason=escalation.reason, assigned_engineer=escalation.assigned_engineer, created_at=escalation.created_at, resolved_at=escalation.resolved_at, incident_io_id=escalation.incident_io_id, incident_io_url=escalation.incident_io_url, ) async def _latest_tier_reports(session: AsyncSession, job_id: uuid.UUID) -> tuple[TierReport | None, TierReport | None]: reports = ( (await session.execute(select(TierReport).where(TierReport.job_id == job_id).order_by(TierReport.created_at.desc()))) .scalars() .all() ) # newest first, so next() below picks each tier's latest run tier1_report = next((r for r in reports if r.tier == 1), None) tier2_report = next((r for r in reports if r.tier == 2), None) return tier1_report, tier2_report @router.get("", response_model=list[EscalationSummary]) async def list_escalations(session: AsyncSession = Depends(get_session)): # dump_id lives on AnalysisJob, not Escalation — one join beats an # N+1 session.get(AnalysisJob, ...) per row. result = await session.execute( select(Escalation, AnalysisJob.dump_id) .join(AnalysisJob, AnalysisJob.id == Escalation.job_id) .order_by(Escalation.created_at.desc()) ) return [_summary(escalation, dump_id) for escalation, dump_id in result.all()] @router.get("/{escalation_id}", response_model=EscalationDetail) async def get_escalation( escalation_id: uuid.UUID, session: AsyncSession = Depends(get_session), actor: StaffActor = Depends(require_staff), ): escalation = await session.get(Escalation, escalation_id) if escalation is None: raise HTTPException(status.HTTP_404_NOT_FOUND, "escalation not found") job = await session.get(AnalysisJob, escalation.job_id) dump = await session.get(Dump, job.dump_id) tier1_report, tier2_report = await _latest_tier_reports(session, job.id) tier1 = tier1_report.findings if tier1_report else None tier2 = tier2_report.findings if tier2_report else None async def _transcript(report): if report is None: return None try: loaded = await load_transcript(report.transcript_ref) except Exception: # noqa: BLE001 — expired with the dump, corrupt (storage.crypto.CorruptObject), or gone return None # the findings must stay reachable whatever happened to the transcript return [{"command": cmd, "output": output} for cmd, output in loaded] if loaded else None # Every human session-open on an escalation's full bundle is an # audit-logged event (PRD §8), same as every debugger command. await write_audit(session, actor=actor.audit_actor, action="escalation_viewed", dump_id=dump.id) return EscalationDetail( **_summary(escalation, dump.id).model_dump(), dump_type=dump.dump_type.value, tier1_findings=tier1, tier2_findings=tier2, tier1_transcript=await _transcript(tier1_report), tier2_transcript=await _transcript(tier2_report), resolution=escalation.resolution, root_cause_tag=escalation.root_cause_tag, ) @router.post("/{escalation_id}/resolve", response_model=EscalationSummary) async def resolve_escalation( escalation_id: uuid.UUID, body: ResolveEscalationRequest, session: AsyncSession = Depends(get_session), actor: StaffActor = Depends(require_staff), ): escalation = await session.get(Escalation, escalation_id) if escalation is None: raise HTTPException(status.HTTP_404_NOT_FOUND, "escalation not found") job = await session.get(AnalysisJob, escalation.job_id) dump = await session.get(Dump, job.dump_id) escalation.assigned_engineer = body.assigned_engineer escalation.resolution = body.resolution escalation.root_cause_tag = body.root_cause_tag escalation.kb_opt_in = body.kb_opt_in # V1.1's KB retrieval reads from resolved cases with this flag set — # V1 only ever writes it, per the PRD §4.3 "V1 cut" box. escalation.resolution_fed_back_to_kb = body.kb_opt_in escalation.resolved_at = datetime.now(timezone.utc) if body.kb_opt_in: # V1.1 Knowledge Base PRD §5/§6: only structured signals already # computed by Tier 1/2 plus the engineer's own text — never a # transcript reference. Tier 2's faulting_module (the more refined # conclusion) wins over Tier 1's when both are present; bugcheck/ # exception code and failure_bucket_id only ever come from Tier 1, # since parse_signals runs once against the same underlying dump. tier1_report, tier2_report = await _latest_tier_reports(session, job.id) tier1_findings = tier1_report.findings if tier1_report else {} tier2_findings = tier2_report.findings if tier2_report else {} code_int = _code_int(tier1_findings.get("bugcheck_or_exception_code")) faulting_module = tier2_findings.get("faulting_module") or tier1_findings.get("faulting_module") session.add( KBEntry( bugcheck_code=code_int if dump.dump_type == DumpType.KERNEL_MODE else None, exception_code=code_int if dump.dump_type == DumpType.USER_MODE else None, faulting_module=normalize_module(faulting_module), failure_bucket_id=tier1_findings.get("failure_bucket_id"), dump_type=dump.dump_type, root_cause_tag=body.root_cause_tag, resolution_note=body.resolution, source_escalation_id=escalation.id, ) ) await session.commit() await session.refresh(escalation) # The authenticated key's owner, not body.assigned_engineer: that field # is free text a caller could set to anyone's name. await write_audit(session, actor=actor.audit_actor, action="escalation_resolved", dump_id=job.dump_id) return _summary(escalation, dump.id)