Keep complete_stream's signature and classify /api/traces for tokens

The trace lifecycle moves into a decorator so complete_stream keeps its
real parameters for introspection. GET /api/traces is readable with the
analytics:read scope, like the Logs endpoint it complements.
This commit is contained in:
arc53-machine committed 2026-09-23 17:50:19 +01:00
1 parent 4e1002f5e0
commit 3512685728
2 files changed
+56 -53

No files matched your search

+55 -53
View File
@@ -1,4 +1,5 @@
import datetime
import functools
import inspect
import json
import logging
@@ -68,6 +69,50 @@ class StreamSuperseded(Exception):
answer_ns = Namespace("answer", description="Answer related operations", path="/")
def _traced_stream(method):
"""Record ``complete_stream`` as the request's execution trace.
The stream runs in the SSE pump thread, not the request thread, so the
trace the route started is activated here for the whole stream and written
exactly once when the stream ends -- normally, paused, failed or
abandoned. The request id is resolved here too so the trace and the
stream agree on it.
"""
signature = inspect.signature(method)
@functools.wraps(method)
def wrapper(*args, **kwargs):
bound = signature.bind(*args, **kwargs)
arguments = bound.arguments
continuation = arguments.get("_continuation")
request_id = (
continuation.get("request_id") if continuation else None
) or arguments.get("request_id") or str(uuid.uuid4())
arguments["request_id"] = request_id
trace = arguments.get("trace")
if not isinstance(trace, tracing.Trace):
agent = arguments.get("agent")
trace = tracing.start_trace(
source=str(getattr(agent, "endpoint", None) or "stream"),
capture_otel_context=False,
)
arguments["trace"] = trace
if trace is not None:
decoded_token = arguments.get("decoded_token")
trace.bind(
request_id=request_id,
user_id=decoded_token.get("sub") if isinstance(decoded_token, dict) else None,
agent_id=arguments.get("agent_id"),
)
with tracing.activate(trace):
try:
yield from method(*bound.args, **bound.kwargs)
finally:
tracing.flush(trace)
return wrapper
class BaseAnswerResource:
"""Shared base class for answer endpoints"""
@@ -246,60 +291,8 @@ class BaseAnswerResource:
logger.exception("Failed to release resume claim after a usage refusal")
return error
@_traced_stream
def complete_stream(
self,
*args: Any,
request_id: Optional[str] = None,
trace: Optional["tracing.Trace"] = None,
**kwargs: Any,
) -> Generator[str, None, None]:
"""Stream the conversation response and record its execution trace.
Accepts every argument of :meth:`_complete_stream_body`. The trace is
activated for the whole stream -- which runs in the SSE pump thread,
not the request thread -- and written exactly once when the stream
ends, however it ends.
Args:
request_id: The request's id, generated by the route before the
agent was built so pre-fetch retrieval shares it. A
continuation's saved id takes precedence; absent both, a new
one is minted.
trace: The trace the route started (it already holds the
pre-fetch retrieval). A new one is started when omitted.
Yields:
Server-sent event strings.
"""
bound = inspect.signature(self._complete_stream_body).bind(*args, **kwargs)
continuation = bound.arguments.get("_continuation")
request_id = (
continuation.get("request_id") if continuation else None
) or request_id or str(uuid.uuid4())
decoded_token = bound.arguments.get("decoded_token") or {}
if not isinstance(trace, tracing.Trace):
trace = None
if trace is None:
agent = bound.arguments.get("agent")
trace = tracing.start_trace(
source=str(getattr(agent, "endpoint", None) or "stream"),
capture_otel_context=False,
)
if trace is not None:
trace.bind(
request_id=request_id,
user_id=decoded_token.get("sub") if isinstance(decoded_token, dict) else None,
agent_id=bound.arguments.get("agent_id"),
)
with tracing.activate(trace):
try:
yield from self._complete_stream_body(
*bound.args, request_id=request_id, **bound.kwargs
)
finally:
tracing.flush(trace)
def _complete_stream_body(
self,
question: str,
agent: Any,
@@ -319,6 +312,7 @@ class BaseAnswerResource:
_continuation: Optional[Dict] = None,
finalize_tool_pause_as_complete: bool = False,
request_id: Optional[str] = None,
trace: Optional["tracing.Trace"] = None,
) -> Generator[str, None, None]:
"""
Generator function that streams the complete conversation response.
@@ -342,6 +336,14 @@ class BaseAnswerResource:
shared_token: Token for shared agent
model_id: Model ID used for the request
retrieved_docs: Pre-fetched documents for sources (optional)
request_id: The request's id, minted by the route before the
agent was built so pre-fetch retrieval shares it. A
continuation's saved id takes precedence; absent both, a new
one is minted.
trace: The execution trace the route started (it already holds
the pre-fetch retrieval); :func:`_traced_stream` activates it
and writes it when the stream ends. A new one is started when
omitted.
finalize_tool_pause_as_complete: Stateless-tool-round mode for
the OpenAI-compatible ``/v1/chat/completions`` endpoint.
OpenAI clients resume a tool call by re-POSTing the full
+1
View File
@@ -298,6 +298,7 @@ RULES: dict[tuple[str, str], Rule] = {
("/api/get_tool_analytics", "POST"): _rule("analytics:read", blocked_by=_ALL_FAMILIES),
("/api/get_schedule_analytics", "POST"): _rule("analytics:read", blocked_by=_ALL_FAMILIES),
("/api/get_user_logs", "POST"): _rule("analytics:read", blocked_by=_ALL_FAMILIES),
("/api/traces", "GET"): _rule("analytics:read", blocked_by=_ALL_FAMILIES),
# Teams (read only)
("/api/teams", "GET"): _rule("teams:read"),
("/api/teams/<string:team_id>", "GET"): _rule("teams:read"),