Trace agent runs and tool calls

@log_activity opens the invoke_agent span for every agent run and names
the trace's activity; gen_continuation opens its own. ToolExecutor.execute
records each executed call with redacted argument and result previews, and
paused, denied, skipped and client-executed calls are recorded where they
are decided.
This commit is contained in:
arc53-machine committed 2026-09-23 17:25:41 +01:00
1 parent e8166e6c29
commit a15940c281
6 files changed
+557 -35

No files matched your search

+62 -19
View File
@@ -6,8 +6,11 @@ from abc import ABC, abstractmethod
from datetime import datetime, timezone
from typing import Any, Dict, Generator, List, Optional
from docsgpt import tracing
from docsgpt.agents.tool_executor import (
ToolExecutor,
finish_tool_span,
record_tool_span_start,
result_status,
truncate_tool_result,
)
@@ -492,12 +495,51 @@ class BaseAgent(ABC):
or client-side results), appends the resulting messages, then
hands back to the LLM to continue the conversation.
Unlike :meth:`gen` this is not wrapped by ``@log_activity``, so the
continuation's ``invoke_agent`` trace span is opened here.
Args:
messages: The saved messages array from the pause point.
tools_dict: The saved tools dictionary.
pending_tool_calls: The pending tool call descriptors from the pause.
tool_actions: Client-provided actions resolving the pending calls.
"""
label = type(self).__name__
span = tracing.start_span(
tracing.KIND_AGENT,
f"invoke_agent {label}",
attributes={
k: v
for k, v in {
"gen_ai.operation.name": "invoke_agent",
"gen_ai.agent.id": str(self.agent_id) if getattr(self, "agent_id", None) else None,
"gen_ai.request.model": getattr(self, "model_id", None),
"docsgpt.agent_type": label,
"docsgpt.continuation": True,
}.items()
if v is not None
},
)
completed = False
try:
yield from self._gen_continuation_inner(
messages, tools_dict, pending_tool_calls, tool_actions, reasoning_content
)
completed = True
except Exception as exc:
span.end(error=exc)
raise
finally:
span.end(None if completed else tracing.STATUS_CANCELLED)
def _gen_continuation_inner(
self,
messages: List[Dict],
tools_dict: Dict,
pending_tool_calls: List[Dict],
tool_actions: List[Dict],
reasoning_content: str = "",
) -> Generator[Dict, None, None]:
self._prepare_tools(tools_dict)
actions_by_id = {a["call_id"]: a for a in tool_actions}
@@ -583,16 +625,15 @@ class BaseAgent(ABC):
messages.append(
self.llm_handler.create_tool_message(tc, denial)
)
yield {
"type": "tool_call",
"data": {
"tool_name": pending.get("tool_name", "unknown"),
"call_id": call_id,
"action_name": pending.get("llm_name", pending["name"]),
"arguments": args,
"status": "denied",
},
denied_data = {
"tool_name": pending.get("tool_name", "unknown"),
"call_id": call_id,
"action_name": pending.get("llm_name", pending["name"]),
"arguments": args,
"status": "denied",
}
finish_tool_span(record_tool_span_start(tc), {**denied_data, "error": comment or None})
yield {"type": "tool_call", "data": denied_data}
elif "result" in action:
result = action["result"]
@@ -616,17 +657,19 @@ class BaseAgent(ABC):
tc, _bound_tool_response_for_llm(result_str)
)
)
yield {
"type": "tool_call",
"data": {
"tool_name": pending.get("tool_name", "unknown"),
"call_id": call_id,
"action_name": pending.get("llm_name", pending["name"]),
"arguments": args,
"result": truncate_tool_result(result_str),
"status": result_status(result),
},
client_data = {
"tool_name": pending.get("tool_name", "unknown"),
"call_id": call_id,
"action_name": pending.get("llm_name", pending["name"]),
"arguments": args,
"result": truncate_tool_result(result_str),
"status": result_status(result),
}
finish_tool_span(
record_tool_span_start(tc, **{"docsgpt.client_executed": True}),
client_data,
)
yield {"type": "tool_call", "data": client_data}
# Resume the LLM loop with the updated messages
llm_response = self._llm_gen(messages, preserve_responses_state=True)
+81 -1
View File
@@ -13,6 +13,7 @@ from docsgpt.agents.default_tools import (
resolve_tool_by_id,
synthesized_default_tools,
)
from docsgpt import tracing
from docsgpt.agents.tools.tool_action_parser import ToolActionParser
from docsgpt.agents.tools.tool_manager import ToolManager
from docsgpt.guardrails.types import Stage as GuardrailStage, resolve_tool_result
@@ -29,6 +30,65 @@ from docsgpt.storage.db.session import db_readonly, db_session
logger = logging.getLogger(__name__)
def record_tool_span_start(call: Any, **attributes: Any):
"""Open an ``execute_tool`` span for ``call`` (no-op without an active trace)."""
llm_name = getattr(call, "name", None) or "unknown"
base = {
"gen_ai.operation.name": "execute_tool",
"gen_ai.tool.name": llm_name,
"gen_ai.tool.call.id": getattr(call, "id", None),
"gen_ai.tool.type": "function",
}
base.update(attributes)
return tracing.start_span(
tracing.KIND_TOOL,
f"execute_tool {llm_name}",
attributes={k: v for k, v in base.items() if v is not None},
)
#: ``tool_calls`` statuses mapped to trace span statuses.
_TOOL_SPAN_STATUS = {
"error": tracing.STATUS_ERROR,
"denied": tracing.STATUS_DENIED,
"skipped": tracing.STATUS_SKIPPED,
"awaiting_approval": tracing.STATUS_PENDING,
"requires_client_execution": tracing.STATUS_PENDING,
}
def finish_tool_span(span: Any, data: Dict[str, Any]) -> None:
"""Close a tool span from the ``tool_calls`` entry the call produced.
Args:
span: The span from :func:`record_tool_span_start`.
data: The tool-call record (``tool_name``, ``arguments``, ``result``,
``status``, ...); results in it are already guardrail-scanned.
"""
if not span:
return
status = str(data.get("status") or "completed")
span.set(
**{
"docsgpt.tool": data.get("tool_name"),
"docsgpt.action": data.get("action_name"),
"docsgpt.tool_status": status,
"docsgpt.artifact_id": data.get("artifact_id"),
}
)
if data.get("arguments") is not None:
span.preview("arguments", data.get("arguments"))
if data.get("result") is not None:
span.preview("result", data.get("result"))
span_status = _TOOL_SPAN_STATUS.get(status, tracing.STATUS_OK)
if span_status == tracing.STATUS_ERROR:
span.error = str(data.get("error") or data.get("result") or "tool error")[:500]
span.set(**{"error.type": str(data.get("error_type") or "ToolError")})
elif data.get("error"):
span.error = str(data["error"])[:500]
span.end(span_status)
def _is_foreign_key_violation(exc: BaseException) -> bool:
"""Whether ``exc`` is a Postgres FK violation (SQLSTATE 23503)."""
if not isinstance(exc, IntegrityError):
@@ -950,7 +1010,27 @@ class ToolExecutor:
MAX_ADVERTISED_TOOL_NAMES = 30
def execute(self, tools_dict: Dict, call, llm_class_name: str):
"""Execute a tool call. Yields status events, returns (result, call_id)."""
"""Execute a tool call. Yields status events, returns (result, call_id).
Every call is recorded as an ``execute_tool`` trace span; its outcome
is read from the ``tool_calls`` entry each branch of :meth:`_execute`
appends, so the span matches what the conversation stores.
"""
span = record_tool_span_start(call)
recorded = len(self.tool_calls)
try:
outcome = yield from self._execute(tools_dict, call, llm_class_name)
except Exception as exc:
span.end(error=exc)
raise
except GeneratorExit:
span.end(tracing.STATUS_CANCELLED)
raise
data = self.tool_calls[-1] if len(self.tool_calls) > recorded else {}
finish_tool_span(span, data)
return outcome
def _execute(self, tools_dict: Dict, call, llm_class_name: str):
parser = ToolActionParser(llm_class_name, name_mapping=self._name_to_tool)
tool_id, action_name, call_args = parser.parse_args(call)
llm_name = getattr(call, "name", "unknown")
+30 -15
View File
@@ -97,6 +97,17 @@ class LLMResponse:
return bool(self.tool_calls) and self.finish_reason == "tool_calls"
def _trace_unexecuted_tool(call: Any, data: Dict[str, Any]) -> None:
"""Record a tool call that was paused, denied or skipped instead of run.
Those calls never reach ``ToolExecutor.execute`` (which traces executed
calls), so without this the trace would not show them at all.
"""
from docsgpt.agents.tool_executor import finish_tool_span, record_tool_span_start
finish_tool_span(record_tool_span_start(call), data)
def _is_restated_payload(existing: str, incoming: str) -> bool:
"""True when ``incoming`` restates an already-complete ``existing`` payload.
@@ -1127,6 +1138,10 @@ class LLMHandler(ABC):
# Mark remaining tools as skipped
for remaining_call in tool_calls[i:]:
_trace_unexecuted_tool(
remaining_call,
{"tool_name": "system", "action_name": remaining_call.name, "status": "skipped"},
)
skip_message = {
"type": "tool_call",
"data": {
@@ -1207,22 +1222,21 @@ class LLMHandler(ABC):
message_id=agent.tool_executor.message_id,
user_id=agent.tool_executor.user,
)
yield {
"type": "tool_call",
"data": {
"tool_name": pause_info["tool_name"],
"call_id": pause_info["call_id"],
"action_name": pause_info.get(
"llm_name", pause_info["name"]
),
"arguments": pause_info["arguments"],
"status": "denied",
"error": deny_reason,
"error_type": pause_info.get(
"error_type", "tool_not_allowed"
),
},
denied_data = {
"tool_name": pause_info["tool_name"],
"call_id": pause_info["call_id"],
"action_name": pause_info.get(
"llm_name", pause_info["name"]
),
"arguments": pause_info["arguments"],
"status": "denied",
"error": deny_reason,
"error_type": pause_info.get(
"error_type", "tool_not_allowed"
),
}
_trace_unexecuted_tool(call, denied_data)
yield {"type": "tool_call", "data": denied_data}
continue
# Yield pause event so the client knows this tool is waiting
pause_data = {
@@ -1236,6 +1250,7 @@ class LLMHandler(ABC):
# can wire the sticky "don't ask again" button.
if pause_info.get("device_id"):
pause_data["device_id"] = pause_info["device_id"]
_trace_unexecuted_tool(call, pause_data)
yield {"type": "tool_call", "data": pause_data}
pending_actions.append(pause_info)
# Do NOT add messages for pending tools here.
+48
View File
@@ -7,6 +7,7 @@ import logging
import uuid
from typing import Any, Callable, Dict, Generator, List, Optional
from docsgpt import tracing
from docsgpt.core import log_context
from docsgpt.storage.db.repositories.stack_logs import StackLogsRepository
from docsgpt.storage.db.session import db_session
@@ -136,10 +137,17 @@ def log_activity() -> Callable:
},
)
# The outermost agent run names the trace's activity, which is how
# a webhook/system Logs row (keyed by activity_id) finds its trace.
tracing.bind_if_unset(activity_id=activity_id)
span = _start_agent_span(args[0], agent_id=agent_id, model=model, endpoint=endpoint)
error: BaseException | None = None
completed = False
try:
generator = func(*args, **kwargs)
yield from _consume_and_log(generator, context)
completed = True
except Exception as exc:
# Only ``Exception`` counts as an activity error; ``GeneratorExit``
# (consumer disconnected mid-stream) and ``KeyboardInterrupt``
@@ -154,6 +162,7 @@ def log_activity() -> Callable:
started_at=started_at,
error=error,
)
_end_agent_span(span, context, error=error, completed=completed)
log_context.reset(ctx_token)
return wrapper
@@ -161,6 +170,45 @@ def log_activity() -> Callable:
return decorator
def _start_agent_span(agent: Any, *, agent_id: Any, model: Any, endpoint: str):
"""Open the ``invoke_agent`` span for one agent run (no-op without a trace)."""
label = type(agent).__name__
return tracing.start_span(
tracing.KIND_AGENT,
f"invoke_agent {label}",
attributes={
k: v
for k, v in {
"gen_ai.operation.name": "invoke_agent",
"gen_ai.agent.id": str(agent_id) if agent_id else None,
"gen_ai.request.model": str(model) if model else None,
"docsgpt.agent_type": label,
"docsgpt.endpoint": endpoint or None,
}.items()
if v is not None
},
)
def _end_agent_span(span: Any, context: "LogContext", *, error: BaseException | None, completed: bool) -> None:
"""Close the agent span with the run's response aggregates and outcome."""
if not span:
return
attributes = {
"docsgpt.answer_chars": context.answer_length,
"docsgpt.source_count": context.source_count,
"docsgpt.tool_call_count": context.tool_call_count,
}
if error is not None:
span.end(error=error, attributes=attributes)
elif context.stream_error:
span.set(**attributes, **{"error.type": "StreamError"})
span.error = context.stream_error
span.end(tracing.STATUS_ERROR)
else:
span.end(None if completed else tracing.STATUS_CANCELLED, attributes=attributes)
def _emit_activity_finished(
*,
context: "LogContext",
+92
View File
@@ -398,3 +398,95 @@ class TestAccumulateResponseSummary:
assert ctx.tool_call_count == 2
class TestLogActivityTraceSpan:
"""``@log_activity`` opens the ``invoke_agent`` span for every agent run."""
@pytest.fixture(autouse=True)
def _tracing_on(self, monkeypatch):
from docsgpt.core.settings import settings
monkeypatch.setattr(settings, "TRACES_ENABLED", True)
class _Agent:
endpoint = "stream"
user = "user1"
user_api_key = ""
agent_id = "agent-1"
model_id = "gpt-4o"
def test_agent_span_wraps_the_run_and_binds_activity_id(self):
from docsgpt import tracing
from docsgpt.logging import log_activity
@log_activity()
def gen(agent, log_context=None):
with tracing.span(tracing.KIND_LLM, "chat m"):
pass
yield {"answer": "hi"}
yield {"sources": [{"title": "a"}]}
trace = tracing.start_trace(source="stream", capture_otel_context=False)
with patch("docsgpt.logging._log_activity_to_db"), tracing.activate(trace):
list(gen(self._Agent()))
agent_span, llm_span = trace.spans
assert agent_span.kind == tracing.KIND_AGENT
assert agent_span.name == "invoke_agent _Agent"
assert agent_span.status == "ok"
assert agent_span.attributes["gen_ai.operation.name"] == "invoke_agent"
assert agent_span.attributes["gen_ai.agent.id"] == "agent-1"
assert agent_span.attributes["docsgpt.answer_chars"] == 2
assert agent_span.attributes["docsgpt.source_count"] == 1
assert llm_span.parent_id == agent_span.id
assert trace.activity_id is not None
def test_nested_agent_keeps_outer_activity_id(self):
from docsgpt import tracing
from docsgpt.logging import log_activity
@log_activity()
def inner(agent, log_context=None):
yield "x"
@log_activity()
def outer(agent, log_context=None):
yield log_context.activity_id
yield from inner(agent)
trace = tracing.start_trace(source="stream", capture_otel_context=False)
with patch("docsgpt.logging._log_activity_to_db"), tracing.activate(trace):
outer_activity = list(outer(self._Agent()))[0]
assert trace.activity_id == outer_activity
outer_span, inner_span = trace.spans
assert inner_span.parent_id == outer_span.id
def test_yielded_error_marks_span_error(self):
from docsgpt import tracing
from docsgpt.logging import log_activity
@log_activity()
def gen(agent, log_context=None):
yield {"type": "error", "error": "node failed"}
trace = tracing.start_trace(source="stream", capture_otel_context=False)
with patch("docsgpt.logging._log_activity_to_db"), tracing.activate(trace):
list(gen(self._Agent()))
assert trace.spans[0].status == "error"
assert trace.spans[0].error == "node failed"
def test_raised_error_marks_span_error(self):
from docsgpt import tracing
from docsgpt.logging import log_activity
@log_activity()
def gen(agent, log_context=None):
yield "a"
raise RuntimeError("boom")
trace = tracing.start_trace(source="stream", capture_otel_context=False)
with patch("docsgpt.logging._log_activity_to_db"), tracing.activate(trace), \
pytest.raises(RuntimeError):
list(gen(self._Agent()))
assert trace.spans[0].status == "error"
assert trace.spans[0].attributes["error.type"] == "RuntimeError"
+244
View File
@@ -0,0 +1,244 @@
"""Tool calls become ``execute_tool`` spans: executed, paused, denied, skipped."""
from __future__ import annotations
from unittest.mock import Mock
import pytest
from docsgpt import tracing
from docsgpt.agents.tool_executor import ToolExecutor
from docsgpt.core.settings import settings
from docsgpt.llm.handlers.base import LLMHandler, ToolCall
@pytest.fixture(autouse=True)
def _env(monkeypatch):
monkeypatch.setattr(settings, "TRACES_ENABLED", True)
monkeypatch.setattr(settings, "TRACES_CAPTURE_CONTENT", True)
@pytest.fixture()
def trace():
t = tracing.start_trace(source="stream", capture_otel_context=False)
with tracing.activate(t):
yield t
def _executor_with(inner):
executor = ToolExecutor.__new__(ToolExecutor)
executor.tool_calls = []
executor._execute = lambda tools_dict, call, llm_class_name: inner(executor, call)
return executor
def _drain(gen):
events = []
while True:
try:
events.append(next(gen))
except StopIteration as stop:
return events, stop.value
class TestExecute:
def test_completed_call(self, trace):
def inner(executor, call):
yield {"type": "tool_call", "data": {"status": "pending"}}
executor.tool_calls.append(
{
"tool_name": "brave",
"action_name": "search",
"arguments": {"q": "x", "api_key": "k"},
"result": "3 results",
"status": "completed",
}
)
return "3 results", call.id
executor = _executor_with(inner)
call = ToolCall(id="c1", name="search_1", arguments="{}")
events, outcome = _drain(executor.execute({}, call, "OpenAILLM"))
assert outcome == ("3 results", "c1")
assert len(events) == 1
(span,) = trace.spans
assert span.kind == tracing.KIND_TOOL
assert span.name == "execute_tool search_1"
assert span.status == "ok"
assert span.attributes["gen_ai.tool.call.id"] == "c1"
assert span.attributes["docsgpt.tool"] == "brave"
assert span.previews["arguments"]["api_key"] == "[REDACTED]"
assert span.previews["result"] == "3 results"
def test_in_band_error(self, trace):
def inner(executor, call):
executor.tool_calls.append(
{"tool_name": "unknown", "result": "no such tool", "status": "error"}
)
return "no such tool", call.id
yield # pragma: no cover
executor = _executor_with(inner)
_drain(executor.execute({}, ToolCall(id="c", name="x", arguments="{}"), "L"))
span = trace.spans[0]
assert span.status == "error"
assert span.error == "no such tool"
def test_raised_error(self, trace):
def inner(executor, call):
yield {"type": "tool_call", "data": {"status": "pending"}}
raise RuntimeError("api down")
executor = _executor_with(inner)
with pytest.raises(RuntimeError):
_drain(executor.execute({}, ToolCall(id="c", name="x", arguments="{}"), "L"))
assert trace.spans[0].status == "error"
assert trace.spans[0].attributes["error.type"] == "RuntimeError"
def test_nested_retrieval_parents_to_tool(self, trace):
def inner(executor, call):
with tracing.span(tracing.KIND_RETRIEVAL, "retrieval"):
pass
executor.tool_calls.append({"status": "completed"})
return "", call.id
yield # pragma: no cover
executor = _executor_with(inner)
_drain(executor.execute({}, ToolCall(id="c", name="internal_search", arguments="{}"), "L"))
tool, retrieval = trace.spans
assert retrieval.parent_id == tool.id
class _Handler(LLMHandler):
def parse_response(self, response): # pragma: no cover - unused
raise NotImplementedError
def create_tool_message(self, tool_call, result):
return {"role": "tool", "content": str(result)}
def _iterate_stream(self, response): # pragma: no cover - unused
return iter(())
class TestPausedCalls:
def _agent(self, pause):
agent = Mock()
agent._check_context_limit = Mock(return_value=False)
agent.context_limit_reached = False
agent.llm.__class__.__name__ = "MockLLM"
agent.tool_executor.check_pause = Mock(return_value=pause)
return agent
def test_awaiting_approval_is_pending(self, trace):
agent = self._agent(
{
"call_id": "c1",
"name": "send_0",
"tool_name": "telegram",
"tool_id": "0",
"action_name": "send",
"arguments": {"text": "hi"},
"pause_type": "awaiting_approval",
}
)
call = ToolCall(id="c1", name="send_0", arguments='{"text": "hi"}')
_drain(_Handler().handle_tool_calls(agent, [call], {"0": {"name": "telegram"}}, []))
(span,) = trace.spans
assert span.status == "pending"
assert span.attributes["docsgpt.tool_status"] == "awaiting_approval"
def test_headless_denial_is_denied(self, trace):
agent = self._agent(
{
"call_id": "c1",
"name": "send_0",
"tool_name": "telegram",
"tool_id": "0",
"action_name": "send",
"arguments": {},
"pause_type": "headless_denied",
"deny_reason": "not allowed",
}
)
agent.tool_executor.message_id = None
agent.tool_executor.user = "u"
agent.tool_executor.agent_id = None
call = ToolCall(id="c1", name="send_0", arguments="{}")
with pytest.MonkeyPatch.context() as mp:
mp.setattr("docsgpt.agents.tool_executor._record_proposed", lambda *a, **k: False)
_drain(_Handler().handle_tool_calls(agent, [call], {"0": {"name": "telegram"}}, []))
(span,) = trace.spans
assert span.status == "denied"
assert span.error == "not allowed"
class TestContinuation:
def _agent(self):
from docsgpt.agents.classic_agent import ClassicAgent
llm = Mock()
llm._supports_tools = True
llm.gen_stream = Mock(return_value=iter(["Answer"]))
llm._supports_structured_output = Mock(return_value=False)
handler = Mock()
handler.process_message_flow = Mock(return_value=iter([]))
handler.create_tool_message = Mock(return_value={"role": "tool", "content": "x"})
executor = Mock()
executor.tool_calls = []
executor.prepare_tools_for_llm = Mock(return_value=[])
executor.get_truncated_tool_calls = Mock(return_value=[])
return ClassicAgent(
endpoint="stream",
llm_name="openai",
model_id="gpt-4",
api_key="test",
llm=llm,
llm_handler=handler,
tool_executor=executor,
)
def _pending(self):
return [
{
"call_id": "c1",
"name": "danger_0",
"tool_name": "danger",
"tool_id": "0",
"action_name": "danger",
"arguments": {},
"pause_type": "awaiting_approval",
}
]
def test_continuation_opens_agent_span_with_denied_tool(self, trace):
agent = self._agent()
list(
agent.gen_continuation(
[{"role": "system", "content": "s"}],
{"0": {"name": "danger"}},
self._pending(),
[{"call_id": "c1", "decision": "denied", "comment": "too risky"}],
)
)
agent_span = trace.spans[0]
tool_span = trace.spans[1]
assert agent_span.kind == tracing.KIND_AGENT
assert agent_span.attributes["docsgpt.continuation"] is True
assert agent_span.status == "ok"
assert tool_span.parent_id == agent_span.id
assert tool_span.status == "denied"
assert tool_span.error == "too risky"
def test_client_result_is_recorded(self, trace):
agent = self._agent()
list(
agent.gen_continuation(
[{"role": "system", "content": "s"}],
{"0": {"name": "danger"}},
self._pending(),
[{"call_id": "c1", "result": {"ok": True}}],
)
)
tool_span = trace.spans[1]
assert tool_span.status == "ok"
assert tool_span.attributes["docsgpt.client_executed"] is True