Files
praxis/server/operator/failure_patterns.py
T
Praxis CI ec397f2c65 docs(milestone): complete v0.5-live-assist — v0.1.13 tagged, milestone release, merged to main
v0.5 (Live Assist — on-the-job voice companion) milestone complete.
4 phases: P0 (pre-execution, v0.1.10) → P1 (assist core + guardrail,
v0.1.11) → P2 (integration + tech-debt + NFR, v0.1.12) → P3 (final
review + ship, v0.1.13 = milestone release).

16/16 REQs covered (3 ASSIST + 4 NFR + 9 IDEATE). 4 v0.6 backlog.
469 tests passed, 0 failed. 1 P0 fixed (guardrail processor safety).
8 P1+ flagged for v0.6. 8 v0.4 P1+ tech-debt addressed.
G-049 + G-067 grill MUSTs resolved. ESCALATION-01 (PIPEDA) OPEN for
human legal review before assist surface go-live.

---ci---
project: praxis
phase: 3
milestone: v0.5
status: complete
requirements:
  covered: [REQ-ASSIST-01, REQ-ASSIST-02, REQ-ASSIST-03, REQ-NFR-ASSIST-01, REQ-NFR-ASSIST-02, REQ-NFR-ASSIST-03, REQ-NFR-ASSIST-04, REQ-IDEATE-01, REQ-IDEATE-02, REQ-IDEATE-03, REQ-IDEATE-04, REQ-IDEATE-05, REQ-IDEATE-06, REQ-IDEATE-07, REQ-IDEATE-08, REQ-IDEATE-09]
  partial: []
---/ci---
2026-08-04 22:35:56 +00:00

57 lines
2.0 KiB
Python

"""GET /api/operator/failure-patterns — failure patterns + safety signals (TASK-08-03, TASK-10-02, D-053).
Auth-gated. Returns failure pattern metrics: failure_mode frequency (cells
with metric prefix `failure_mode:`) + branch outcome distribution (cells
with metric prefix `branch:`) + the assist guardrail block rate safety signal
(TASK-10-02 — `assist_guardrail_block_rate`). Weak-spot rubric criteria
(mean < 3.0) are highlighted by the frontend. All k-anonymized.
The `assist_guardrail_block_rate` is a safety signal for operators: a sudden
spike signals either a prompt regression or learners pushing boundaries. High
block rate = flag for operator review.
"""
from __future__ import annotations
import datetime as _dt
from fastapi import APIRouter, Depends, Request
from server.auth.dependencies import current_operator
from server.auth.models import Operator
from server.operator._common import (
ViewResponse,
all_recent_aggregates,
group_by_path,
require_pg_store,
)
router = APIRouter(prefix="/api/operator", tags=["operator-failure-patterns"])
# The assist guardrail block-rate safety signal (TASK-10-02, D-060 layer 3).
ASSIST_GUARDRAIL_BLOCK_RATE = "assist_guardrail_block_rate"
def _is_failure_metric(metric: str) -> bool:
# Failure patterns (v0.4) + the assist guardrail block-rate safety signal (v0.5).
return (
metric.startswith("failure_mode:")
or metric.startswith("branch:")
or metric == ASSIST_GUARDRAIL_BLOCK_RATE
)
@router.get("/failure-patterns", response_model=ViewResponse)
async def failure_patterns_view(
request: Request,
op: Operator = Depends(current_operator),
) -> ViewResponse:
pg_store = await require_pg_store(request)
since = _dt.date.today() - _dt.timedelta(days=30)
rows = await all_recent_aggregates(pg_store, since)
failure_rows = [r for r in rows if _is_failure_metric(r.get("metric", ""))]
views, last_updated = group_by_path(failure_rows)
return ViewResponse(views=views, last_updated=last_updated)
__all__ = ["router"]