deerflow-code/offline-backend-20260512/backend/app/gateway/routers/stream_truncations.py

"""Admin-only model-stream truncation event log (模型截断记录).

Companion to the per-model ``strict_stream_integrity`` switch on the model
management page: the patched OpenAI-compatible streaming layer writes one row
per abnormally ended provider stream (silently committed half answers, strict
integrity detections, output-length caps, mid-stream transport errors). This
router serves the admin viewing page; rows are retained for at most one month
by the SQL repository.
"""

from __future__ import annotations

import logging
from datetime import datetime
from typing import Any

from fastapi import APIRouter, HTTPException, Query, Request
from pydantic import BaseModel, Field

from app.gateway.deps import get_optional_user_from_request, get_stream_truncation_store
from deerflow.persistence.stream_truncation.base import StreamTruncationQuery, StreamTruncationStore

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/admin/stream-truncations", tags=["stream-truncations"])

_EVENT_TYPES = (
    "silent_no_terminal",
    "incomplete_stream_error",
    "length_truncated",
    "content_filter",
    "midstream_error",
)


class StreamTruncationEventResponse(BaseModel):
    id: str
    created_at: str | None = None
    model_name: str | None = None
    provider_completion_id: str | None = None
    run_id: str | None = None
    event_type: str
    strict_mode: bool = False
    finish_reason: str = "none"
    status: str = "success"
    error_type: str | None = None
    chunk_count: int = 0
    duration_ms: int = 0
    first_chunk_ms: int = -1
    detail: str | None = None


class StreamTruncationListResponse(BaseModel):
    items: list[StreamTruncationEventResponse]
    total: int = Field(..., description="Total rows matching the filter, ignoring pagination")
    limit: int
    offset: int


class StreamTruncationBreakdownItem(BaseModel):
    model_name: str
    total: int = 0
    last_at: str | None = None
    silent_no_terminal: int = 0
    incomplete_stream_error: int = 0
    length_truncated: int = 0
    content_filter: int = 0
    midstream_error: int = 0


class StreamTruncationBreakdownResponse(BaseModel):
    items: list[StreamTruncationBreakdownItem]


async def _require_admin(request: Request) -> None:
    user = await get_optional_user_from_request(request)
    if user is None:
        raise HTTPException(status_code=401, detail="Authentication is required")
    if getattr(user, "system_role", None) != "admin":
        raise HTTPException(status_code=403, detail="Stream truncation log is admin-only")


def _require_store(request: Request) -> StreamTruncationStore:
    store = get_stream_truncation_store(request)
    if store is None:
        raise HTTPException(status_code=503, detail="Stream truncation store is not configured")
    return store


def _parse_dt(value: str | datetime | None) -> datetime | None:
    if not value:
        return None
    if isinstance(value, datetime):
        return value
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=f"Invalid datetime: {value!r} ({exc})") from exc


def _build_query(
    *,
    model_name: str | None,
    event_type: str | None,
    since: str | datetime | None,
    until: str | datetime | None,
    limit: int,
    offset: int,
) -> StreamTruncationQuery:
    if event_type and event_type not in _EVENT_TYPES:
        raise HTTPException(status_code=422, detail=f"Unknown event type: {event_type}")
    return StreamTruncationQuery(
        model_name=model_name or None,
        event_type=event_type or None,
        since=_parse_dt(since),
        until=_parse_dt(until),
        limit=limit,
        offset=offset,
    )


def _fill_breakdown_zeros(item: dict[str, Any]) -> dict[str, Any]:
    for event_type in _EVENT_TYPES:
        item.setdefault(event_type, 0)
    return item


@router.get("", response_model=StreamTruncationListResponse)
async def list_stream_truncations(
    request: Request,
    model_name: str | None = Query(default=None, max_length=128),
    event_type: str | None = Query(default=None, max_length=32),
    since: str | None = Query(default=None),
    until: str | None = Query(default=None),
    limit: int = Query(default=50, ge=1, le=200),
    offset: int = Query(default=0, ge=0),
) -> StreamTruncationListResponse:
    """List truncation events, newest first."""
    await _require_admin(request)
    store = _require_store(request)
    query = _build_query(
        model_name=model_name,
        event_type=event_type,
        since=since,
        until=until,
        limit=limit,
        offset=offset,
    )
    items = await store.list(query)
    total = await store.count(query)
    return StreamTruncationListResponse(
        items=[StreamTruncationEventResponse(**item) for item in items],
        total=total,
        limit=limit,
        offset=offset,
    )


@router.get("/breakdown", response_model=StreamTruncationBreakdownResponse)
async def stream_truncation_breakdown(
    request: Request,
    since: str | None = Query(default=None),
    until: str | None = Query(default=None),
) -> StreamTruncationBreakdownResponse:
    """Per-model aggregate of truncation events over the query window."""
    await _require_admin(request)
    store = _require_store(request)
    query = _build_query(
        model_name=None,
        event_type=None,
        since=since,
        until=until,
        limit=1,
        offset=0,
    )
    rows = await store.model_breakdown(query)
    return StreamTruncationBreakdownResponse(
        items=[
            StreamTruncationBreakdownItem(**_fill_breakdown_zeros(item))
            for item in rows
        ]
    )