From 35126857281fd609f8463f6bdb9152af706edc7a Mon Sep 17 00:00:00 2001 From: arc53-machine <232052973+arc53-machine@users.noreply.github.com> Date: Wed, 23 Sep 2026 17:50:19 +0100 Subject: [PATCH] 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. --- docsgpt/api/answer/routes/base.py | 108 +++++++++++++++--------------- docsgpt/api/pat/rules.py | 1 + 2 files changed, 56 insertions(+), 53 deletions(-) diff --git a/docsgpt/api/answer/routes/base.py b/docsgpt/api/answer/routes/base.py index 019ab261..469271b7 100644 --- a/docsgpt/api/answer/routes/base.py +++ b/docsgpt/api/answer/routes/base.py @@ -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 diff --git a/docsgpt/api/pat/rules.py b/docsgpt/api/pat/rules.py index 227fe11a..c1d7683d 100644 --- a/docsgpt/api/pat/rules.py +++ b/docsgpt/api/pat/rules.py @@ -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/", "GET"): _rule("teams:read"),