Spaces:
Sleeping
Sleeping
| """Authenticated debug routes for coach traces and prompt previews. | |
| Paste bundles are self-contained for an external AI with no user history. | |
| """ | |
| from fastapi import APIRouter, Depends, Query, Request, status | |
| from fastapi.responses import PlainTextResponse | |
| from app.deps import require_login | |
| from app.models import ApiEnvelope, CoachRequest, err, ok | |
| from app.paste_bundle import render_paste_bundle | |
| router = APIRouter( | |
| prefix="/api/debug", | |
| tags=["debug"], | |
| dependencies=[Depends(require_login)], | |
| ) | |
| def _settings(request: Request): | |
| return request.app.state.settings | |
| def _paste(request: Request, trace: dict) -> PlainTextResponse: | |
| settings = _settings(request) | |
| body = render_paste_bundle( | |
| trace, | |
| app_name=settings.app_name, | |
| shrink_k=settings.stats_shrink_k, | |
| match_alpha=settings.match_alpha, | |
| min_n=settings.min_stats_n, | |
| ) | |
| return PlainTextResponse(content=body, media_type="text/markdown") | |
| def list_traces( | |
| request: Request, | |
| limit: int = Query(default=20, ge=1, le=200), | |
| ) -> dict[str, object]: | |
| """Return newest-first trace summaries.""" | |
| records = request.app.state.trace_store.list_traces(limit=limit) | |
| summaries = [ | |
| { | |
| "trace_id": r.get("trace_id"), | |
| "ts": r.get("ts"), | |
| "source": r.get("source"), | |
| "flags": r.get("flags"), | |
| "model": r.get("model_id"), | |
| } | |
| for r in records | |
| ] | |
| return ok({"items": summaries}) | |
| def get_trace(request: Request, trace_id: str) -> object: | |
| """Return one full JSON trace.""" | |
| trace = request.app.state.trace_store.get_trace(trace_id) | |
| if trace is None: | |
| return err("not_found", "Trace not found", status.HTTP_404_NOT_FOUND) | |
| return ok(trace) | |
| def paste_trace(request: Request, trace_id: str) -> object: | |
| """Return the context-free markdown paste bundle for one trace.""" | |
| trace = request.app.state.trace_store.get_trace(trace_id) | |
| if trace is None: | |
| return err("not_found", "Trace not found", status.HTTP_404_NOT_FOUND) | |
| return _paste(request, trace) | |
| def paste_last(request: Request) -> object: | |
| """Return the newest trace paste bundle.""" | |
| trace = request.app.state.trace_store.latest_trace() | |
| if trace is None: | |
| return err("not_found", "No traces yet", status.HTTP_404_NOT_FOUND) | |
| return _paste(request, trace) | |
| def prompt_preview(request: Request, body: CoachRequest) -> object: | |
| """Build evidence and messages without calling the model.""" | |
| if not body.text and not body.entry_id: | |
| return err( | |
| "validation_error", | |
| "Provide text or entry_id", | |
| status.HTTP_422_UNPROCESSABLE_ENTITY, | |
| ) | |
| try: | |
| preview = request.app.state.coach_service.prompt_preview( | |
| text=body.text, | |
| entry_id=body.entry_id, | |
| include_history=body.include_history, | |
| ) | |
| except KeyError: | |
| return err("not_found", "Entry not found", status.HTTP_404_NOT_FOUND) | |
| except ValueError as exc: | |
| return err("validation_error", str(exc), status.HTTP_422_UNPROCESSABLE_ENTITY) | |
| return ok(preview) | |
| def coach_test(request: Request, body: CoachRequest) -> object: | |
| """Run coach and force a trace; optional force_backup.""" | |
| if not body.text and not body.entry_id: | |
| return err( | |
| "validation_error", | |
| "Provide text or entry_id", | |
| status.HTTP_422_UNPROCESSABLE_ENTITY, | |
| ) | |
| try: | |
| result = request.app.state.coach_service.coach( | |
| text=body.text, | |
| entry_id=body.entry_id, | |
| include_history=body.include_history, | |
| persist=body.persist, | |
| force_backup=body.force_backup, | |
| ) | |
| except KeyError: | |
| return err("not_found", "Entry not found", status.HTTP_404_NOT_FOUND) | |
| except ValueError as exc: | |
| return err("validation_error", str(exc), status.HTTP_422_UNPROCESSABLE_ENTITY) | |
| return ok(result) | |