"""A bounded, JSON-safe record of every tool call an LLM agent attempted. **Why this exists.** The DSPy agent runs inside the Deno sandbox; the tools it calls run outside it, host-side, behind the broker. Everything interesting about a tool call therefore happens across a process boundary the agent's own trajectory cannot see. Without an explicit record, a denied call, a timed-out call and a call that never happened are indistinguishable in the run log. Three rules shape the design. 1. **A denial is a fact about the run, not an absence of one.** Every attempt is recorded, including attempts the broker refused before reaching a handler. A user who selected two tools and got an answer that silently skipped one deserves to see why. 2. **Everything here is worker-supplied or tool-supplied, so everything here is untrusted.** Tool ids, error messages and argument values all originate outside the server. Every string is bounded at construction and every value is a JSON scalar, so a consumer can escape once at the render boundary and be done. See ``CONTRIBUTING.md`` honesty rule 4. 3. **Bounded at construction, never at render.** A trace that is only trimmed when displayed is a memory leak with a nice appearance. """ from __future__ import annotations import json import threading from collections.abc import Mapping, Sequence from dataclasses import dataclass, field from types import MappingProxyType from typing import Any #: Longest recorded rendering of a call's arguments. Large enough to read a #: realistic call, small enough that a runaway argument cannot dominate a #: result payload. MAX_ARGUMENTS_CHARACTERS = 2_048 #: Longest recorded rendering of a successful result. Deliberately shorter #: than ``JobPolicy.max_output_bytes``: the trace is for diagnosis, and the #: full result already travelled to the model. MAX_RESULT_CHARACTERS = 1_024 MAX_MESSAGE_CHARACTERS = 500 MAX_TOOL_ID_CHARACTERS = 128 MAX_VERSION_CHARACTERS = 32 MAX_CODE_CHARACTERS = 80 #: Hard cap on recorded calls. A job's call quota is far lower, so reaching #: this means something is wrong; the cap stops it becoming a memory problem. MAX_RECORDED_CALLS = 64 #: Where a refusal was decided. Kept explicit because "the tool did not run" #: has several very different causes and the operator needs to tell them apart. OUTCOME_OK = "ok" OUTCOME_TOOL_ERROR = "tool_error" OUTCOME_DENIED_RUN = "denied_run_allowlist" OUTCOME_DENIED_OPERATOR = "denied_operator_policy" OUTCOME_QUOTA = "quota_exceeded" OUTCOME_TIMEOUT = "timeout" OUTCOME_INVALID = "invalid_input" _OUTCOMES = frozenset( { OUTCOME_OK, OUTCOME_TOOL_ERROR, OUTCOME_DENIED_RUN, OUTCOME_DENIED_OPERATOR, OUTCOME_QUOTA, OUTCOME_TIMEOUT, OUTCOME_INVALID, } ) #: Broker error codes mapped onto trace outcomes. Anything unmapped becomes #: ``OUTCOME_TOOL_ERROR``, which is the conservative reading: an unrecognised #: failure is still a failure. _CODE_TO_OUTCOME = MappingProxyType( { "tool_not_allowed": OUTCOME_DENIED_RUN, "operator_not_approved": OUTCOME_DENIED_OPERATOR, "quota_exceeded": OUTCOME_QUOTA, "timeout": OUTCOME_TIMEOUT, "invalid_input": OUTCOME_INVALID, "invalid_tool_ref": OUTCOME_INVALID, "invalid_output": OUTCOME_TOOL_ERROR, "output_too_large": OUTCOME_TOOL_ERROR, "host_not_allowed": OUTCOME_DENIED_OPERATOR, "tool_error": OUTCOME_TOOL_ERROR, } ) def _clip(value: object, limit: int) -> str: """Return at most ``limit`` characters of ``value`` as plain text. Control characters are stripped rather than escaped. The consumer still has to escape for its own output format; removing C0 here means a tool cannot smuggle a newline or a terminal control sequence into a log line. """ text = "" if value is None else str(value) cleaned = "".join(character for character in text if ord(character) >= 32 or character == " ") if len(cleaned) <= limit: return cleaned return cleaned[:limit] + "...[truncated]" def _render_json(value: object, limit: int) -> str: try: encoded = json.dumps( value, ensure_ascii=False, allow_nan=False, separators=(",", ":"), default=str ) except (TypeError, ValueError): encoded = "" return _clip(encoded, limit) def _argument_keys(arguments: object) -> tuple[str, ...]: if not isinstance(arguments, Mapping): return () keys = sorted(_clip(key, 64) for key in list(arguments)[:64]) return tuple(keys) @dataclass(frozen=True) class ToolCallRecord: """One attempted tool call, whatever became of it. ``outcome`` is the field to branch on. ``error_code`` preserves the broker's own vocabulary so a consumer that knows it loses nothing, but a consumer that does not can still tell a denial from a timeout. """ sequence: int tool_id: str version: str outcome: str arguments_json: str = "{}" argument_keys: tuple[str, ...] = () result_json: str = "" error_code: str | None = None error_message: str | None = None elapsed_ms: int = 0 def __post_init__(self) -> None: if isinstance(self.sequence, bool) or not isinstance(self.sequence, int): raise ValueError("sequence must be an integer") if self.sequence < 1: raise ValueError("sequence must start at 1") if self.outcome not in _OUTCOMES: raise ValueError(f"unknown trace outcome {self.outcome!r}") object.__setattr__(self, "tool_id", _clip(self.tool_id, MAX_TOOL_ID_CHARACTERS)) object.__setattr__(self, "version", _clip(self.version, MAX_VERSION_CHARACTERS)) object.__setattr__( self, "arguments_json", _clip(self.arguments_json, MAX_ARGUMENTS_CHARACTERS) ) object.__setattr__(self, "result_json", _clip(self.result_json, MAX_RESULT_CHARACTERS)) object.__setattr__( self, "argument_keys", tuple(_clip(key, 64) for key in self.argument_keys[:64]), ) if self.error_code is not None: object.__setattr__(self, "error_code", _clip(self.error_code, MAX_CODE_CHARACTERS)) if self.error_message is not None: object.__setattr__( self, "error_message", _clip(self.error_message, MAX_MESSAGE_CHARACTERS) ) if isinstance(self.elapsed_ms, bool) or not isinstance(self.elapsed_ms, int): raise ValueError("elapsed_ms must be an integer") if self.elapsed_ms < 0: raise ValueError("elapsed_ms must not be negative") @property def ref(self) -> str: return f"{self.tool_id}@{self.version}" @property def ok(self) -> bool: return self.outcome == OUTCOME_OK @property def denied(self) -> bool: return self.outcome in (OUTCOME_DENIED_RUN, OUTCOME_DENIED_OPERATOR) def to_dict(self) -> dict[str, Any]: """Return a flat, JSON-ready record with no nested containers. Flatness is deliberate. A renderer that walks arbitrary nesting has to decide what to escape at every level; a flat record of bounded strings and integers has exactly one escaping rule. """ value: dict[str, Any] = { "sequence": self.sequence, "tool_id": self.tool_id, "version": self.version, "ref": self.ref, "outcome": self.outcome, "ok": self.ok, "arguments_json": self.arguments_json, "argument_keys": list(self.argument_keys), "elapsed_ms": self.elapsed_ms, } if self.result_json: value["result_json"] = self.result_json if self.error_code is not None: value["error_code"] = self.error_code if self.error_message is not None: value["error_message"] = self.error_message return value def to_event(self) -> dict[str, Any]: """Return the narrower shape the existing ``tool_events`` field uses. Kept compatible on purpose: ``JobResult.tool_events`` already has consumers, and widening it is not this module's decision to make. """ event: dict[str, Any] = { "tool_id": self.tool_id, "version": self.version, "ok": self.ok, "elapsed_ms": self.elapsed_ms, } if self.error_code is not None: event["error_code"] = self.error_code return event @dataclass class ToolTrace: """Ordered, bounded, thread-safe record of one job's tool calls. Thread safety matters: ``ToolBroker.call`` runs each handler on its own thread, and a DSPy agent may issue calls from more than one place. """ job_id: str = "" _records: list[ToolCallRecord] = field(default_factory=list) _dropped: int = 0 _lock: threading.Lock = field(default_factory=threading.Lock, repr=False) def __post_init__(self) -> None: self.job_id = _clip(self.job_id, MAX_TOOL_ID_CHARACTERS) def record( self, *, tool_id: str, version: str, outcome: str, arguments: object = None, result: object = None, error_code: str | None = None, error_message: str | None = None, elapsed_ms: int = 0, ) -> ToolCallRecord | None: """Append one attempt. Returns ``None`` once the cap is reached.""" with self._lock: if len(self._records) >= MAX_RECORDED_CALLS: self._dropped += 1 return None sequence = len(self._records) + 1 record = ToolCallRecord( sequence=sequence, tool_id=tool_id, version=version, outcome=outcome, arguments_json=_render_json(arguments if arguments is not None else {}, MAX_ARGUMENTS_CHARACTERS), argument_keys=_argument_keys(arguments), result_json="" if result is None else _render_json(result, MAX_RESULT_CHARACTERS), error_code=error_code, error_message=error_message, elapsed_ms=max(0, int(elapsed_ms)) if not isinstance(elapsed_ms, bool) else 0, ) self._records.append(record) return record def record_result( self, *, tool_id: str, version: str, arguments: object, result_envelope: Mapping[str, Any], ) -> ToolCallRecord | None: """Record from a ``ToolResult.to_dict()`` envelope. This is the path the broker uses, so the mapping from broker error codes to trace outcomes lives in exactly one place. """ ok = result_envelope.get("ok") is True error = result_envelope.get("error") code = None message = None if isinstance(error, Mapping): code = error.get("code") message = error.get("message") outcome = OUTCOME_OK if ok else _CODE_TO_OUTCOME.get(str(code), OUTCOME_TOOL_ERROR) elapsed = result_envelope.get("elapsed_ms", 0) return self.record( tool_id=tool_id, version=version, outcome=outcome, arguments=arguments, result=result_envelope.get("output") if ok else None, error_code=None if code is None else str(code), error_message=None if message is None else str(message), elapsed_ms=elapsed if isinstance(elapsed, int) and not isinstance(elapsed, bool) else 0, ) @property def records(self) -> tuple[ToolCallRecord, ...]: with self._lock: return tuple(self._records) @property def dropped(self) -> int: """Attempts not recorded because the cap was reached. Never hidden.""" with self._lock: return self._dropped def denials(self) -> tuple[ToolCallRecord, ...]: return tuple(record for record in self.records if record.denied) def summary(self) -> dict[str, Any]: """Counts an operator can read at a glance, with no zero standing in for a missing measurement.""" records = self.records by_outcome: dict[str, int] = {} for record in records: by_outcome[record.outcome] = by_outcome.get(record.outcome, 0) + 1 return { "job_id": self.job_id, "attempted": len(records), "succeeded": sum(1 for record in records if record.ok), "denied": sum(1 for record in records if record.denied), "by_outcome": dict(sorted(by_outcome.items())), "dropped": self.dropped, } def to_dicts(self) -> tuple[dict[str, Any], ...]: return tuple(record.to_dict() for record in self.records) def to_events(self) -> tuple[dict[str, Any], ...]: """Return ``JobResult.tool_events``-shaped entries, denials included.""" return tuple(record.to_event() for record in self.records) def describe_for_agent(records: Sequence[ToolCallRecord]) -> str: """Render a trace back for the LLM agent to read. Used when a denial should change what the agent does next. A model that is told plainly that a tool was refused can stop asking for it; a model given silence will try again and spend the budget. The wording states the refusal as settled, because it is: the allowlist is immutable for the run. """ if not records: return "No tools were called." lines = [] for record in records: if record.ok: lines.append(f"{record.sequence}. {record.ref} succeeded.") elif record.outcome == OUTCOME_DENIED_RUN: lines.append( f"{record.sequence}. {record.ref} was refused: it is not in this run's " "allowed tools. This cannot change during the run; do not call it again." ) elif record.outcome == OUTCOME_DENIED_OPERATOR: lines.append( f"{record.sequence}. {record.ref} was refused: this worker's operator has not " "approved it. This cannot change during the run; do not call it again." ) elif record.outcome == OUTCOME_QUOTA: lines.append( f"{record.sequence}. {record.ref} was refused: the call budget is spent. " "Answer with what you already have." ) else: code = record.error_code or record.outcome lines.append(f"{record.sequence}. {record.ref} failed ({code}).") return "\n".join(lines) __all__ = [ "MAX_ARGUMENTS_CHARACTERS", "MAX_RECORDED_CALLS", "MAX_RESULT_CHARACTERS", "OUTCOME_DENIED_OPERATOR", "OUTCOME_DENIED_RUN", "OUTCOME_INVALID", "OUTCOME_OK", "OUTCOME_QUOTA", "OUTCOME_TIMEOUT", "OUTCOME_TOOL_ERROR", "ToolCallRecord", "ToolTrace", "describe_for_agent", ]