From 608a978d8db1925deb9a891175a5bb4b8e10ca99 Mon Sep 17 00:00:00 2001 From: Robin Date: Thu, 10 Sep 2026 17:43:30 -0700 Subject: [PATCH] fix tool calls data loss --- agentx/__init__.py | 4 + agentx/agentx.py | 40 ++++- agentx/evaluations/client.py | 46 +++--- agentx/evaluations/models.py | 29 +++- agentx/evaluations/runner.py | 112 +++++++++---- agentx/exceptions.py | 20 ++- agentx/export.py | 3 +- agentx/feedback.py | 3 +- agentx/integrations/autogen.py | 4 + agentx/integrations/crewai.py | 94 +++++++---- agentx/integrations/google_adk.py | 34 +++- agentx/integrations/langchain.py | 140 ++++++++++------ agentx/integrations/llamaindex.py | 131 +++++++++------ agentx/integrations/openai_agents.py | 20 ++- agentx/monitor/client.py | 40 +++-- agentx/monitor/improvement_groups.py | 3 +- agentx/monitor/judge_scorers.py | 13 +- agentx/monitor/scorer_groups.py | 4 +- agentx/monitor/scorers.py | 3 +- agentx/outcomes.py | 3 +- agentx/projects.py | 3 +- agentx/traces.py | 3 +- agentx/tracing/ingest_client.py | 36 +++- agentx/tracing/tracer.py | 64 ++++++-- agentx/util.py | 19 ++- tests/test_deep_dive_fixes.py | 45 +++++ tests/test_error_taxonomy.py | 86 ++++++++++ tests/test_integrations.py | 2 +- tests/test_judge_scorers.py | 47 ++++++ tests/test_runner_features.py | 73 +++++++++ tests/test_span_tree.py | 235 ++++++++++++++++++++++++++- tests/test_wire_models.py | 83 ++++++++++ 32 files changed, 1196 insertions(+), 246 deletions(-) create mode 100644 tests/test_error_taxonomy.py create mode 100644 tests/test_wire_models.py diff --git a/agentx/__init__.py b/agentx/__init__.py index 7c4a183..6d05909 100644 --- a/agentx/__init__.py +++ b/agentx/__init__.py @@ -5,6 +5,8 @@ from agentx.exceptions import ( AgentXError, AgentXAuthError, + AgentXValidationError, + AgentXConnectionError, AgentXAPIError, DatasetNotFound, CINotEnabled, @@ -21,6 +23,8 @@ "AgentX", "AgentXError", "AgentXAuthError", + "AgentXValidationError", + "AgentXConnectionError", "AgentXAPIError", "DatasetNotFound", "CINotEnabled", diff --git a/agentx/agentx.py b/agentx/agentx.py index fceced0..18b3e6f 100644 --- a/agentx/agentx.py +++ b/agentx/agentx.py @@ -3,7 +3,7 @@ import os import logging -from agentx.util import get_headers, api_base +from agentx.util import get_headers, api_base, normalize_base from agentx.resources.agent import Agent from agentx.resources.workforce import Workforce @@ -16,16 +16,23 @@ def __init__( base_url: Optional[str] = None, workspace_id: Optional[str] = None, ): + # The api_key is NOT written back into os.environ (it used to be): every sub-client + # below receives it explicitly, and mutating process-global state from a constructor + # re-pointed unrelated code - the same leak the base_url write below had (deep-dive + # round 3, bug #1). Static flows that still read the env (AgentX.list_workforces, + # bare get_headers()) now require the caller to set AGENTX_API_KEY themselves. self.api_key = api_key or os.getenv("AGENTX_API_KEY") - if self.api_key and not os.getenv("AGENTX_API_KEY"): - os.environ["AGENTX_API_KEY"] = self.api_key # base_url overrides AGENTX_API_BASE_URL env var (and the SDK default). It is # deliberately NOT written back into os.environ: the constructor used to do that, which # made the last-constructed client silently re-point every other client in the process # (deep-dive round 3, bug #1). Each sub-client below receives this value explicitly and - # captures it at construction instead. + # captures it at construction instead. Normalized (trailing slash and the + # /custom-agent-evaluations suffix stripped) so an evaluations-shaped URL works for + # every sub-client, not just evaluations. self.base_url = base_url or os.getenv("AGENTX_API_BASE_URL") + if self.base_url: + self.base_url = normalize_base(self.base_url) self.workspace_id = workspace_id or os.getenv("AGENTX_WORKSPACE_ID") @@ -91,6 +98,27 @@ def __init__( workspace_id=self.workspace_id, ) self.tracer = Tracer(_ingest_client) + self._ingest_client = _ingest_client + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + def close(self, timeout: float = 5.0) -> bool: + """Flush queued traces and stop the tracer's background ingest worker. + + Returns ``True`` when everything drained before ``timeout`` seconds elapsed. Optional - + an ``atexit`` hook already flushes queued traces on interpreter shutdown - but a + long-running service that tears clients down mid-process should call it (or use the + client as a context manager) so worker threads don't accumulate. + """ + return self._ingest_client.close(timeout) + + def __enter__(self) -> "AgentX": + return self + + def __exit__(self, exc_type, exc_val, tb) -> None: + self.close() @classmethod def from_env(cls) -> "AgentX": @@ -129,7 +157,9 @@ def list_agents(self) -> List[Agent]: @staticmethod def list_workforces() -> List["Workforce"]: - """List all workforces/teams.""" + """List all workforces/teams. Static, so it reads AGENTX_API_KEY from the environment + directly - the constructor no longer writes ``api_key`` into os.environ, so set the + env var yourself before calling this.""" url = f"{api_base()}/access/teams" response = requests.get(url, headers=get_headers()) if response.status_code == 200: diff --git a/agentx/evaluations/client.py b/agentx/evaluations/client.py index 5c88fd3..0dca7ea 100644 --- a/agentx/evaluations/client.py +++ b/agentx/evaluations/client.py @@ -24,7 +24,12 @@ logger = logging.getLogger(__name__) -from agentx.util import _DEFAULT_API_BASE as _UTIL_API_BASE +from agentx.util import _DEFAULT_API_BASE as _UTIL_API_BASE, normalize_base + +# The canonical error classes (agentx.exceptions) are raised - and re-exported here for +# compat with code that imported them from this module - so `except agentx.AgentXAuthError` +# works whichever client raised. +from agentx.exceptions import AgentXError, AgentXAuthError, AgentXValidationError _DEFAULT_BASE_URL = f"{_UTIL_API_BASE}/custom-agent-evaluations" SDK_NAME = "agentx-python" @@ -42,7 +47,7 @@ _SELF_HOST_SCORING_TIMEOUT = 900 -class AgentXEvaluationsError(Exception): +class AgentXEvaluationsError(AgentXError): """An evaluations API call failed. ``status_code`` carries the HTTP status when the failure came from a response rather @@ -55,14 +60,6 @@ def __init__(self, message: str, status_code: Optional[int] = None) -> None: self.status_code = status_code -class AgentXAuthError(AgentXEvaluationsError): - pass - - -class AgentXValidationError(AgentXEvaluationsError): - pass - - class EvaluationSubmissionError(AgentXEvaluationsError): """A result batch could not be submitted (after one retry). The run is left unfinalized; re-running execute() on the same context resumes past already-submitted cases.""" @@ -96,13 +93,10 @@ def __init__( # whatever workspace the API key's user defaults to, not the one the caller intended. self._workspace_id = workspace_id # Priority: constructor arg > env var > SDK default - # Always append /custom-agent-evaluations so users only need to provide /api/v1 - _api_base = ( - base_url or os.getenv("AGENTX_API_BASE_URL", _UTIL_API_BASE) - ).rstrip("/") - if not _api_base.endswith("/custom-agent-evaluations"): - _api_base = f"{_api_base}/custom-agent-evaluations" - self._base_url = _api_base + # normalize_base strips a trailing slash and any /custom-agent-evaluations suffix, + # then the suffix is appended - users only need to provide /api/v1 either way. + _api_base = normalize_base(base_url or os.getenv("AGENTX_API_BASE_URL", _UTIL_API_BASE)) + self._base_url = f"{_api_base}/custom-agent-evaluations" # None until an analysis call tells us which engine this is; see _api_root. self._analysis_on_dashboard_router: Optional[bool] = None self._session = requests.Session() @@ -184,13 +178,16 @@ def _request( continue if resp.status_code == 401: - raise AgentXAuthError("Invalid or missing API key") + raise AgentXAuthError("Invalid or missing API key", status_code=401) if resp.status_code == 422: raise AgentXValidationError(resp.text) + # Gate on the schedule itself so HTTP-status retries walk the SAME full backoff + # schedule connection errors do - the old `attempt < _MAX_RETRIES - 1` gate left + # the schedule's last entry unreachable for HTTP retries (ingest_client precedent). if ( resp.status_code in _RETRYABLE_STATUS and retry - and attempt < _MAX_RETRIES - 1 + and attempt < len(schedule) - 1 ): logger.debug( "Retryable status %d (attempt %d)", resp.status_code, attempt + 1 @@ -500,6 +497,17 @@ def get_report(self, run_id: str) -> Report: return self._report_from_dashboard(run_id) def get_missing_results(self, run_id: str) -> List[Dict[str, Any]]: + """Deprecated: on self-host the route's response body has no top-level list, so this + always returns ``[]``. Use :meth:`get_submitted_keys` - the same route's + ``submittedKeys`` - to find out what a run still needs.""" + import warnings + + warnings.warn( + "get_missing_results() always returns [] on self-host - use get_submitted_keys() " + "to resume a run instead.", + DeprecationWarning, + stacklevel=2, + ) data = self._request("GET", f"/runs/{run_id}/missing-results") return data if isinstance(data, list) else data.get("missing", []) diff --git a/agentx/evaluations/models.py b/agentx/evaluations/models.py index b2a2cad..e73c21c 100644 --- a/agentx/evaluations/models.py +++ b/agentx/evaluations/models.py @@ -1,7 +1,7 @@ from __future__ import annotations from typing import Any, Dict, List, Literal, Optional, Union -from pydantic import AliasChoices, BaseModel, Field, model_validator +from pydantic import AliasChoices, BaseModel, Field, field_validator, model_validator # --------------------------------------------------------------------------- # Observable trace @@ -78,6 +78,17 @@ class Dataset(BaseModel): # Custom code scorers attached to this dataset - [{ id, name, code, enabled }]. Retrievable, # so a fetched dataset round-trips them (import_dataset copies them to the new dataset). code_scorers: Optional[List[Dict[str, Any]]] = Field(default=None, alias="codeScorers") + # Grading config carried on the dataset itself - similarity metric toggles (each a + # {"enabled": bool, ...} object on the wire), LLM-as-judge overrides, and the raw + # sovereigntyIndex object. Modeled so a fetched Dataset round-trips them: extra="ignore" + # used to silently drop all of these on read, and import_dataset lost them on the copy. + vector_similarity: Optional[Any] = Field(default=None, alias="vectorSimilarity") + jaccard_similarity: Optional[Any] = Field(default=None, alias="jaccardSimilarity") + bleu_score: Optional[Any] = Field(default=None, alias="bleuScore") + rouge_score: Optional[Any] = Field(default=None, alias="rougeScore") + judge_prompt: Optional[str] = Field(default=None, alias="judgePrompt") + judge_model: Optional[str] = Field(default=None, alias="judgeModel") + sovereignty_index: Optional[Dict[str, Any]] = Field(default=None, alias="sovereigntyIndex") status: str = "published" version_id: Optional[str] = Field(default=None, alias="versionId") # Sovereignty & Portability - models selected to compare on this dataset. @@ -430,7 +441,12 @@ class RunResultRow(BaseModel): question_index: Optional[int] = Field(default=None, alias="questionIndex") run_number: Optional[int] = Field(default=None, alias="runNumber") question_text: Optional[str] = Field(default=None, alias="questionText") - response: Optional[str] = None + # The engine sends the agent's answer as an `output` OBJECT ({"text": ...}), not a + # `response` string - accept both spellings and lift the dict's text (see the + # validator below), so row.response actually populates on self-host. + response: Optional[str] = Field( + default=None, validation_alias=AliasChoices("response", "output") + ) trace_id: Optional[str] = Field(default=None, alias="traceId") latency_ms: Optional[float] = Field(default=None, alias="latencyMs") input_tokens: Optional[int] = Field(default=None, alias="inputTokens") @@ -452,6 +468,15 @@ class Config: populate_by_name = True extra = "ignore" + @field_validator("response", mode="before") + @classmethod + def _lift_output_text(cls, value: Any) -> Any: + # The `output` alias delivers the wire's whole output object - keep the declared + # Optional[str] by lifting its text field. + if isinstance(value, dict): + return value.get("text") + return value + @classmethod def from_wire(cls, wire: Dict[str, Any]) -> "RunResultRow": row = cls.model_validate(wire) diff --git a/agentx/evaluations/runner.py b/agentx/evaluations/runner.py index 31293e4..27bdd9d 100644 --- a/agentx/evaluations/runner.py +++ b/agentx/evaluations/runner.py @@ -11,7 +11,11 @@ from agentx.evaluations.adapters.raw import RawCallableAdapter from agentx.evaluations.adapters.precomputed import PrecomputedAdapter from agentx.evaluations.adapters.http_endpoint import HttpEndpointAdapter -from agentx.evaluations.client import EvaluationsClient, EvaluationSubmissionError +from agentx.evaluations.client import ( + AgentXEvaluationsError, + EvaluationsClient, + EvaluationSubmissionError, +) from agentx.evaluations.models import ( AnalysisStatus, Dataset, @@ -227,6 +231,7 @@ def produce(case: EvaluationCase) -> EvaluationResult: if concurrency > 1: import concurrent.futures import contextvars + from collections import deque def in_scope(case: EvaluationCase) -> EvaluationResult: # ContextVars (the eval-run scope) do not cross thread boundaries on their own - @@ -240,45 +245,68 @@ def in_scope(case: EvaluationCase) -> EvaluationResult: if _idem_key(self._run.run_id, case.case_id, case.run_number) not in already_done ] executor = concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) - # map() yields in submission order, so batching/submission below stays deterministic. - mapped = executor.map(in_scope, pending) - def ordered() -> "Iterator[EvaluationResult]": + def bounded() -> "Iterator[EvaluationResult]": + # Bounded submit loop instead of executor.map(): map() dispatches EVERY case + # up front, so a fail-fast flush failure (EvaluationSubmissionError below) + # still paid for the whole rest of the run in agent calls. Keep at most + # `concurrency` cases in flight, topping up as results are consumed; yields + # stay in submission order so batching below is deterministic. On teardown + # (an exception in the consuming loop closes this generator) whatever is + # queued but unstarted is cancelled. + import itertools + + case_iter = iter(pending) + in_flight: "deque[concurrent.futures.Future]" = deque() try: - yield from mapped + for case in itertools.islice(case_iter, concurrency): + in_flight.append(executor.submit(in_scope, case)) + while in_flight: + result = in_flight.popleft().result() + next_case = next(case_iter, None) + if next_case is not None: + in_flight.append(executor.submit(in_scope, next_case)) + yield result finally: - executor.shutdown(wait=True) + executor.shutdown(wait=False, cancel_futures=True) - results_iter = ordered() + results_iter = bounded() else: results_iter = None # sequential path below produces inline - for idx, case in enumerate(cases, start=1): - idem_key = _idem_key(self._run.run_id, case.case_id, case.run_number) - - if idem_key in already_done: - logger.debug("Skipping already-submitted case: %s", idem_key) - _print_progress(idx, total, case, skipped=True) - continue - - result = next(results_iter) if results_iter is not None else produce(case) - result.idempotency_key = idem_key - # Tag the result with the case's model so the server can group it into - # the Sovereignty & Portability matrix (the callable may also set it). - if case.model: - meta = dict(result.metadata or {}) - meta.setdefault("model", case.model) - result.metadata = meta - result = EvaluationResult( - **{**result.model_dump(), "idempotencyKey": idem_key} - ) - self._results.append(result) - batch.append(result) - _print_progress(idx, total, case, result=result) + try: + for idx, case in enumerate(cases, start=1): + idem_key = _idem_key(self._run.run_id, case.case_id, case.run_number) + + if idem_key in already_done: + logger.debug("Skipping already-submitted case: %s", idem_key) + _print_progress(idx, total, case, skipped=True) + continue + + result = next(results_iter) if results_iter is not None else produce(case) + result.idempotency_key = idem_key + # Tag the result with the case's model so the server can group it into + # the Sovereignty & Portability matrix (the callable may also set it). + if case.model: + meta = dict(result.metadata or {}) + meta.setdefault("model", case.model) + result.metadata = meta + result = EvaluationResult( + **{**result.model_dump(), "idempotencyKey": idem_key} + ) + self._results.append(result) + batch.append(result) + _print_progress(idx, total, case, result=result) - if len(batch) >= max_batch: - self._flush_batch(batch) - batch = [] + if len(batch) >= max_batch: + self._flush_batch(batch) + batch = [] + finally: + # Deterministic teardown: a flush failure mid-run must stop the in-flight agent + # dispatch NOW (bounded()'s finally cancels queued cases), not whenever the + # generator happens to be garbage-collected. + if results_iter is not None: + results_iter.close() if batch: self._flush_batch(batch) @@ -333,9 +361,21 @@ def _fetch_submitted_keys(self) -> Set[str]: so a re-execute() after a crash skips (and never re-pays for) finished cases.""" try: return set(self._client.get_submitted_keys(self._run.run_id)) - except Exception: - # Older engines without the route: no resume, identical to the historical behavior. - return set() + except AgentXEvaluationsError as exc: + if exc.status_code == 404: + # Older engines without the route: no resume, identical to the historical + # behavior. ONLY the 404 qualifies - a transient 502/timeout here used to be + # swallowed too, and an empty resume set silently re-runs (and re-bills) + # every already-finished case. + return set() + _say(f" {red('✗')} Could not fetch already-submitted keys: {dim(str(exc))}") + logger.error( + "Resume-key fetch for run %s failed (%s) - refusing to re-run the whole run " + "blind; retry execute() once the engine is reachable", + self._run.run_id, + exc, + ) + raise # ------------------------------------------------------------------ # Step 2: finalize @@ -465,6 +505,8 @@ def analyze( Args: mode: "auto" (default), "sync", or "batch" - how item scoring executes server-side. + Hosted-only: self-host runs the analysis synchronously regardless; see the + response's mode field for what actually ran. quality_mode: "quality_first" or "balanced" - how many items get a second/third judge. judges: 1-3 model ids from ``client.evaluations.list_models()``, e.g. ``["gpt-5.6-luna", "claude-opus-4-8"]``. Omit to let the engine score with its diff --git a/agentx/exceptions.py b/agentx/exceptions.py index c0e9121..f400733 100644 --- a/agentx/exceptions.py +++ b/agentx/exceptions.py @@ -12,7 +12,25 @@ class AgentXError(Exception): class AgentXAuthError(AgentXError): - """Invalid or missing API key.""" + """Invalid or missing API key. + + Canonical across every sub-client (evaluations, monitor, ...) - ``except + agentx.AgentXAuthError`` catches an auth failure no matter which client raised it. + ``status_code`` carries the HTTP status when known (401/403), ``None`` otherwise. + """ + + def __init__(self, message: str, status_code: int | None = None) -> None: + super().__init__(message) + self.status_code = status_code + + +class AgentXValidationError(AgentXError): + """The API rejected the request as invalid (HTTP 422). Canonical across every + sub-client, same as :class:`AgentXAuthError`.""" + + def __init__(self, message: str, status_code: int | None = 422) -> None: + super().__init__(message) + self.status_code = status_code class AgentXConnectionError(AgentXError): diff --git a/agentx/export.py b/agentx/export.py index 8ebc61b..2ba1194 100644 --- a/agentx/export.py +++ b/agentx/export.py @@ -8,11 +8,12 @@ import requests from agentx.util import api_base, get_headers +from agentx.exceptions import AgentXError logger = logging.getLogger(__name__) -class AgentXExportError(Exception): +class AgentXExportError(AgentXError): pass diff --git a/agentx/feedback.py b/agentx/feedback.py index a5138f3..ea409ea 100644 --- a/agentx/feedback.py +++ b/agentx/feedback.py @@ -6,11 +6,12 @@ import requests from agentx.util import api_base, get_headers +from agentx.exceptions import AgentXError logger = logging.getLogger(__name__) -class AgentXFeedbackError(Exception): +class AgentXFeedbackError(AgentXError): pass diff --git a/agentx/integrations/autogen.py b/agentx/integrations/autogen.py index 68bf233..3e5dc19 100644 --- a/agentx/integrations/autogen.py +++ b/agentx/integrations/autogen.py @@ -195,6 +195,10 @@ def _summarize_messages(self, messages: List[Any], run_start: float) -> tuple: "output": text, "inputTokenSize": input_tokens, "outputTokenSize": out_tokens, + # A named source is an agent turn in the team's trajectory, not a bare + # model call - stated so _merge_child_run doesn't stamp it "llm" (the + # crewai task-step precedent). + **({"kind": "agent"} if source else {}), }) if text: diff --git a/agentx/integrations/crewai.py b/agentx/integrations/crewai.py index 54e6c18..1c348e3 100644 --- a/agentx/integrations/crewai.py +++ b/agentx/integrations/crewai.py @@ -18,11 +18,18 @@ """ from __future__ import annotations +import logging import time from typing import Any, Dict, List, Optional from agentx.tracing.tracer import Tracer, _safe_serialize +logger = logging.getLogger(__name__) + +# Warn once per process when CrewAI's event bus can't be imported and task timings fall back +# to the evenly-divided approximation - fabricated timings shouldn't be silent. +_warned_no_event_bus = False + class AgentXCrewObserver: """ @@ -53,7 +60,7 @@ def kickoff(self, crew: Any, inputs: Optional[Dict[str, Any]] = None) -> Any: present; on older CrewAI versions that predate it, this falls back to evenly dividing the total latency across tasks. """ - task_timings, unregister = self._start_task_timing_capture() + task_timings, unregister = self._start_task_timing_capture(crew) start = time.time() error: Optional[str] = None @@ -97,7 +104,7 @@ def kickoff(self, crew: Any, inputs: Optional[Dict[str, Any]] = None) -> Any: framework="crewai", ) - def _start_task_timing_capture(self): + def _start_task_timing_capture(self, crew: Any = None): """ Register temporary, additive listeners on CrewAI's event bus to capture each task's real start/end wall-clock time, keyed by @@ -105,10 +112,16 @@ def _start_task_timing_capture(self): ``async_execution=True`` - unlike attributing the most-recently- started task, which would misattribute end times under overlap). + The event bus is a global singleton, so events from a DIFFERENT crew's + overlapping kickoff arrive here too - listeners are scoped to ``crew`` + (event/source crew identity when the event carries it, this kickoff's + task ids otherwise) so each trace only records its own kickoff's tasks. + Returns ``(task_timings, unregister)``. ``task_timings`` stays empty (and ``unregister`` is a no-op) on CrewAI versions that predate the - ``crewai.events`` module - callers should fall back to the - evenly-divided approximation in that case. + events module - callers should fall back to the evenly-divided + approximation in that case (warned once per process, since those + timings are fabricated). Uses ``crewai_event_bus.on()``/``.off()`` directly rather than ``scoped_handlers()`` - the latter temporarily disables *every* @@ -116,30 +129,60 @@ def _start_task_timing_capture(self): built-in ones) for the duration of the `with` block, which isn't what we want for a handler meant to run alongside them. """ + global _warned_no_event_bus task_timings: Dict[str, Dict[str, Any]] = {} try: - from crewai.events.event_bus import crewai_event_bus - from crewai.events.types.task_events import ( - TaskCompletedEvent, - TaskFailedEvent, - TaskStartedEvent, - ) + # Modern shape first (the crewai.events module), then the older + # crewai.utilities.events layout that shipped the same bus/events. + try: + from crewai.events.event_bus import crewai_event_bus + from crewai.events.types.task_events import ( + TaskCompletedEvent, + TaskFailedEvent, + TaskStartedEvent, + ) + except ImportError: + from crewai.utilities.events import crewai_event_bus + from crewai.utilities.events.task_events import ( + TaskCompletedEvent, + TaskFailedEvent, + TaskStartedEvent, + ) except ImportError: + if not _warned_no_event_bus: + _warned_no_event_bus = True + logger.warning( + "CrewAI's event bus is not importable (tried crewai.events and " + "crewai.utilities.events) - per-task timings will be approximated by " + "evenly dividing the kickoff's total latency across tasks" + ) return task_timings, lambda: None - # Double-instrumentation guard (bus-keyed latch, the same idea as the - # other integrations' _agentx_patched flag): the event bus is a global - # singleton, so a notebook re-run or an overlapping kickoff that - # already has AgentX listeners registered would otherwise get a second - # set and duplicate every task span. When already attached, this - # kickoff just falls back to the evenly-divided timing approximation. - if getattr(crewai_event_bus, "_agentx_attached", False): - return task_timings, lambda: None - crewai_event_bus._agentx_attached = True + # This kickoff's own task ids - the fallback scope filter when an event carries no + # crew reference to compare against. + own_task_ids = { + str(getattr(task, "id", None)) + for task in (getattr(crew, "tasks", None) or []) + if getattr(task, "id", None) is not None + } + + def is_ours(source: Any, event: Any) -> bool: + """Only record events that belong to THIS kickoff's crew - the bus is global, + so a concurrent kickoff's task events land on every registered listener.""" + if crew is None: + return True + if source is crew: + return True + event_crew = getattr(event, "crew", None) or getattr(source, "crew", None) + if event_crew is not None: + return event_crew is crew + if own_task_ids: + return str(getattr(event, "task_id", None)) in own_task_ids + return True def on_task_started(source: Any, event: Any) -> None: task_id = getattr(event, "task_id", None) - if task_id is None: + if task_id is None or not is_ours(source, event): return task_timings[task_id] = { "name": getattr(event, "task_name", None), @@ -164,14 +207,9 @@ def on_task_failed(source: Any, event: Any) -> None: crewai_event_bus.on(TaskFailedEvent)(on_task_failed) def unregister() -> None: - try: - crewai_event_bus.off(TaskStartedEvent, on_task_started) - crewai_event_bus.off(TaskCompletedEvent, on_task_completed) - crewai_event_bus.off(TaskFailedEvent, on_task_failed) - finally: - # Clear the latch even if .off() raises, so a later kickoff - # can re-attach instead of being locked out forever. - crewai_event_bus._agentx_attached = False + crewai_event_bus.off(TaskStartedEvent, on_task_started) + crewai_event_bus.off(TaskCompletedEvent, on_task_completed) + crewai_event_bus.off(TaskFailedEvent, on_task_failed) return task_timings, unregister diff --git a/agentx/integrations/google_adk.py b/agentx/integrations/google_adk.py index 4be7c74..c5a1948 100644 --- a/agentx/integrations/google_adk.py +++ b/agentx/integrations/google_adk.py @@ -22,6 +22,7 @@ import time from typing import Any, Dict, List, Optional +from uuid import uuid4 from agentx.tracing.tracer import Tracer, _safe_serialize @@ -135,7 +136,16 @@ async def before_run_callback(self, *, invocation_context: Any) -> None: root_span = self._tracer.trace( agent_name, framework="google-adk", metadata=self._metadata, session_id=self._session_id ) - root_span.__enter__() + # Deliberately NOT root_span.__enter__() - the same reasoning as openai_agents' + # on_trace_start: enter pushes onto the CALLING context's active-span stack, but ADK + # may fire after_run_callback on a different task/thread, so the pop no-ops there, the + # entry never drains, and later unrelated traces on this context get mis-filed as + # children of this dead run (and inherit its session). Start time and session are set + # by hand instead; every callback below already parents via child_span() on the held + # state["root_span"] reference, no stack involved. + root_span._start = time.time() + if root_span._session_id is None: + root_span._session_id = f"sdk_{uuid4().hex}" self._runs[inv_id] = { "root_span": root_span, "llm_call_count": 0, @@ -163,6 +173,9 @@ async def after_run_callback(self, *, invocation_context: Any) -> None: root_span._output_tokens = state["output_tokens"] if state["error"]: root_span.set_error(state["error"]) + # Close via the held reference WITHOUT touching the active-span stack (see + # before_run_callback) - __exit__'s only stack interaction is the pop, a no-op for a + # never-pushed span, so calling it directly for its send behavior is safe. root_span.__exit__(None, None, None) # ------------------------------------------------------------------ @@ -295,6 +308,16 @@ async def after_tool_callback( tool_name, start_time=start_t, end_time=end_t, input=tool_input, output=tool_output, span_kind="tool", ) + # Also mirror onto the root's flat tool_calls list (the dual-write every other + # integration does): the engine's built-in "Tool failure" check and the dashboard's + # Tool quality column read the ROOT trace's flat toolCalls, not child-span rows. + state["root_span"].tool_calls.append({ + "name": tool_name, + "input": tool_input, + "output": tool_output, + "latency_ms": int((end_t - start_t) * 1000) if start_t is not None else None, + "success": True, + }) async def on_tool_error_callback( self, @@ -317,3 +340,12 @@ async def on_tool_error_callback( tool_name, start_time=start_t, end_time=end_t, input=tool_input, output=tool_output, error=str(error), span_kind="tool", ) + # Mirror the FAILED call onto the root's flat tool_calls list - success: False is + # exactly what the engine's "Tool failure" check reads (see after_tool_callback). + state["root_span"].tool_calls.append({ + "name": tool_name, + "input": tool_input, + "output": tool_output, + "latency_ms": int((end_t - start_t) * 1000) if start_t is not None else None, + "success": False, + }) diff --git a/agentx/integrations/langchain.py b/agentx/integrations/langchain.py index dbd0425..00bc4f4 100644 --- a/agentx/integrations/langchain.py +++ b/agentx/integrations/langchain.py @@ -20,6 +20,7 @@ """ from __future__ import annotations +import logging import threading import time from typing import Any, Dict, List, Optional, Union @@ -28,6 +29,8 @@ from agentx.tracing.tracer import Tracer, _safe_serialize from agentx.integrations._traced_call import capture_tool_definitions +logger = logging.getLogger(__name__) + try: from langchain_core.callbacks.base import BaseCallbackHandler from langchain_core.outputs import LLMResult @@ -325,34 +328,48 @@ def __init__( def _prune_stale_entries(self) -> None: """Sweep out run_id entries older than max_run_age_seconds - see __init__'s comment.""" cutoff = time.time() - self._max_run_age_seconds + swept = 0 - stale_run_ids = [ - run_id - for run_id, state in self._runs.items() - if (state.get("start") or state.get("llm_start") or 0) < cutoff - ] - for run_id in stale_run_ids: - self._runs.pop(run_id, None) - self._top_level.pop(run_id, None) - self._parents.pop(run_id, None) - - stale_retrieval_ids = [ - run_id for run_id, state in self._retrieval_starts.items() if state.get("start", 0) < cutoff - ] - for run_id in stale_retrieval_ids: - self._retrieval_starts.pop(run_id, None) - self._parents.pop(run_id, None) - - # Pre-run retrieval steps waiting for a top-level chain that never came - # (e.g. retriever.invoke() called but agent.invoke() aborted before - # on_chain_start). Each step carries its own start_time, so drop the - # pre-cutoff ones just like the run_id-keyed structures above. + # The whole sweep runs under _state_lock: it iterates dicts that on_tool_start / + # on_retriever_start / _record_llm_start insert into from LangGraph's tool-node + # threads, and iterating a dict another thread mutates raises RuntimeError. with self._state_lock: + stale_run_ids = [ + run_id + for run_id, state in list(self._runs.items()) + if (state.get("start") or state.get("llm_start") or 0) < cutoff + ] + for run_id in stale_run_ids: + self._runs.pop(run_id, None) + self._top_level.pop(run_id, None) + self._parents.pop(run_id, None) + swept += len(stale_run_ids) + + stale_retrieval_ids = [ + run_id for run_id, state in list(self._retrieval_starts.items()) if state.get("start", 0) < cutoff + ] + for run_id in stale_retrieval_ids: + self._retrieval_starts.pop(run_id, None) + self._parents.pop(run_id, None) + swept += len(stale_retrieval_ids) + + # Pre-run retrieval steps waiting for a top-level chain that never came + # (e.g. retriever.invoke() called but agent.invoke() aborted before + # on_chain_start). Each step carries its own start_time, so drop the + # pre-cutoff ones just like the run_id-keyed structures above. if self._pending_retrieval_steps: self._pending_retrieval_steps[:] = [ step for step in self._pending_retrieval_steps if step.get("start_time", 0) >= cutoff ] + if swept: + logger.warning( + "AgentXCallbackHandler swept %d in-flight run record(s) older than %.0fs - " + "their end callbacks never fired, so their traces were never sent", + swept, + self._max_run_age_seconds, + ) + # ------------------------------------------------------------------ # Chain lifecycle # ------------------------------------------------------------------ @@ -372,30 +389,32 @@ def on_chain_start( if is_top: self._prune_stale_entries() # Consume any retrieval steps that ran before this chain started - # (pre-run RAG: retriever.invoke() called before agent.invoke()) + # (pre-run RAG: retriever.invoke() called before agent.invoke()). The _runs + # insert shares the lock with _prune_stale_entries' iteration. with self._state_lock: pending = self._pending_retrieval_steps[:] self._pending_retrieval_steps.clear() - self._runs[run_id] = { - "start": time.time(), - "input": _extract_input(inputs), - "tool_calls": [], - "model": None, - "execution_steps": [], - "retrieval_steps": pending, - "input_tokens": 0, - "output_tokens": 0, - # Graph structure captured under this top-level run: named nested chain runs - # (LangGraph nodes, sub-agents) keyed by run_id, and the FULL nested-chain - # parent map (noise chains included) so _emit_span_tree can walk through - # skipped plumbing runs to the nearest emitted ancestor. - "node_runs": {}, - "chain_parents": {}, - # The request's tools=[...] as seen on the first LLM call's invocation params - - # attached to the root trace's metadata so the engine's unregistered-tool - # listing can surface the REAL definition (not one inferred from arguments). - "tool_definitions": None, - } + self._runs[run_id] = { + "start": time.time(), + "input": _extract_input(inputs), + "tool_calls": [], + "model": None, + "execution_steps": [], + "retrieval_steps": pending, + "input_tokens": 0, + "output_tokens": 0, + # Graph structure captured under this top-level run: named nested chain runs + # (LangGraph nodes, sub-agents) keyed by run_id, and the FULL nested-chain + # parent map (noise chains included) so _emit_span_tree can walk through + # skipped plumbing runs to the nearest emitted ancestor. + "node_runs": {}, + "chain_parents": {}, + # The request's tools=[...] as seen on the first LLM call's invocation + # params - attached to the root trace's metadata so the engine's + # unregistered-tool listing can surface the REAL definition (not one + # inferred from arguments). + "tool_definitions": None, + } else: top = self._find_top_ancestor(parent_run_id) if top is None: @@ -507,6 +526,7 @@ def resolve_parent(parent_id: Any): duration_ms=step.get("duration_ms"), input=step.get("query"), output=step.get("output"), + error=step.get("error"), metadata={"kind": "retrieval"}, span_kind="retrieval", ) @@ -679,11 +699,14 @@ def _record_llm_start( captured = capture_tool_definitions(kwargs.get("invocation_params", {}).get("tools")) if captured: self._runs[top_for_tools]["tool_definitions"] = captured - self._runs[run_id] = { - "llm_start": time.time(), - "model": model, - "input": _extract_llm_input(prompts=prompts, messages=messages), - } + # Insert under the lock _prune_stale_entries' iteration holds - LangGraph can start + # LLM runs from worker threads while another thread's on_chain_start prunes. + with self._state_lock: + self._runs[run_id] = { + "llm_start": time.time(), + "model": model, + "input": _extract_llm_input(prompts=prompts, messages=messages), + } top = self._find_top_ancestor(parent_run_id) if top and top in self._runs and not self._runs[top].get("model") and model: self._runs[top]["model"] = model @@ -780,11 +803,13 @@ def on_tool_start( **kwargs, ) -> None: self._parents[run_id] = parent_run_id - self._runs[run_id] = { - "tool_name": serialized.get("name", "unknown"), - "tool_input": input_str, - "start": time.time(), - } + # Insert under the prune's lock - see _record_llm_start. + with self._state_lock: + self._runs[run_id] = { + "tool_name": serialized.get("name", "unknown"), + "tool_input": input_str, + "start": time.time(), + } def on_tool_end( self, @@ -862,7 +887,9 @@ def on_retriever_start( **kwargs, ) -> None: self._parents[run_id] = parent_run_id - self._retrieval_starts[run_id] = {"start": time.time(), "query": query} + # Insert under the prune's lock - see _record_llm_start. + with self._state_lock: + self._retrieval_starts[run_id] = {"start": time.time(), "query": query} def on_retriever_end( self, @@ -924,11 +951,18 @@ def on_retriever_error( start_t = state["start"] query: Optional[str] = state["query"] or None step: Dict[str, Any] = { + # parent_run_id so _emit_span_tree parents the failed retrieval under the node + # that ran it, like on_retriever_end's steps - without it the span always fell + # back to the root. + "parent_run_id": parent_run_id, "name": "Retrieval 1", # renumbered below "duration_ms": (end_t - start_t) * 1000, "start_time": start_t, "end_time": end_t, "output": f"ERROR: {error}", + # Structured error - _emit_span_tree threads it to child_span(error=...) so the + # span row is marked failed, not just an output that happens to say ERROR. + "error": str(error), } if query: step["query"] = query diff --git a/agentx/integrations/llamaindex.py b/agentx/integrations/llamaindex.py index 5422585..8810290 100644 --- a/agentx/integrations/llamaindex.py +++ b/agentx/integrations/llamaindex.py @@ -21,11 +21,15 @@ """ from __future__ import annotations +import logging +import threading import time from typing import Any, Dict, List, Optional from agentx.tracing.tracer import Tracer, _safe_serialize +logger = logging.getLogger(__name__) + try: from llama_index.core.callbacks.base_handler import BaseCallbackHandler from llama_index.core.callbacks.schema import CBEventType, EventPayload @@ -67,16 +71,24 @@ def _extract_usage_tokens(completion: Any) -> tuple: """ Best-effort: LlamaIndex's CallbackManager payload has no dedicated token fields, so this digs into the completion/response object's provider-raw - data (shape varies per LLM integration, hence the broad try/except). + data. ``raw`` is a plain dict for some LLM integrations and a typed + response object (with a ``.usage`` attribute) for others - both shapes are + read, since bailing on the non-dict form silently lost every token count + for those providers. """ raw = getattr(completion, "raw", None) - if not isinstance(raw, dict): - return None, None - usage = raw.get("usage") + if isinstance(raw, dict): + usage = raw.get("usage") + else: + usage = getattr(raw, "usage", None) if raw is not None else None if isinstance(usage, dict): input_tokens = usage.get("prompt_tokens") or usage.get("input_tokens") output_tokens = usage.get("completion_tokens") or usage.get("output_tokens") return input_tokens, output_tokens + if usage is not None: + input_tokens = getattr(usage, "prompt_tokens", None) or getattr(usage, "input_tokens", None) + output_tokens = getattr(usage, "completion_tokens", None) or getattr(usage, "output_tokens", None) + return input_tokens, output_tokens return None, None @@ -125,32 +137,49 @@ def __init__( self._roots: Dict[str, bool] = {} self._runs: Dict[str, Dict[str, Any]] = {} self._starts: Dict[str, Dict[str, Any]] = {} + # Guards the prune's iteration over the state dicts against concurrent + # on_event_start inserts from other threads (a query engine and a bare + # llm.complete() running in parallel share this singleton handler) - + # iterating a dict another thread mutates raises RuntimeError. + self._state_lock = threading.Lock() def _prune_stale_entries(self) -> None: """Sweep out event_id entries older than max_run_age_seconds - see __init__'s comment.""" cutoff = time.time() - self._max_run_age_seconds - - # Every live event_id has a _starts entry (set in on_event_start and - # popped with _parents/_roots in on_event_end), each carrying its own - # start timestamp. - stale_event_ids = [ - event_id for event_id, info in self._starts.items() if info.get("start", 0) < cutoff - ] - for event_id in stale_event_ids: - self._starts.pop(event_id, None) - self._parents.pop(event_id, None) - self._roots.pop(event_id, None) - self._runs.pop(event_id, None) - - # Root runs outlive their own _starts entry until the root's end event - # fires - sweep those by the run state's own start timestamp. - stale_run_ids = [ - event_id for event_id, state in self._runs.items() if state.get("start", 0) < cutoff - ] - for event_id in stale_run_ids: - self._runs.pop(event_id, None) - self._roots.pop(event_id, None) - self._parents.pop(event_id, None) + swept = 0 + + with self._state_lock: + # Every live event_id has a _starts entry (set in on_event_start and + # popped with _parents/_roots in on_event_end), each carrying its own + # start timestamp. + stale_event_ids = [ + event_id for event_id, info in list(self._starts.items()) if info.get("start", 0) < cutoff + ] + for event_id in stale_event_ids: + self._starts.pop(event_id, None) + self._parents.pop(event_id, None) + self._roots.pop(event_id, None) + self._runs.pop(event_id, None) + swept += len(stale_event_ids) + + # Root runs outlive their own _starts entry until the root's end event + # fires - sweep those by the run state's own start timestamp. + stale_run_ids = [ + event_id for event_id, state in list(self._runs.items()) if state.get("start", 0) < cutoff + ] + for event_id in stale_run_ids: + self._runs.pop(event_id, None) + self._roots.pop(event_id, None) + self._parents.pop(event_id, None) + swept += len(stale_run_ids) + + if swept: + logger.warning( + "AgentXLlamaIndexHandler swept %d in-flight event record(s) older than %.0fs - " + "their end callbacks never fired, so their traces were never sent", + swept, + self._max_run_age_seconds, + ) # ------------------------------------------------------------------ # BaseCallbackHandler protocol @@ -172,26 +201,28 @@ def on_event_start( ) -> str: payload = payload or {} self._prune_stale_entries() - self._parents[event_id] = parent_id - - root_id = self._find_root(parent_id) - if root_id is None and event_type in _ROOT_EVENT_TYPES: - self._roots[event_id] = True - root_id = event_id - self._runs[event_id] = { - "start": time.time(), - "input": None, - "output": None, - "model": None, - "error": None, - "execution_steps": [], - "tool_call_steps": [], - "retrieval_steps": [], - "input_tokens": 0, - "output_tokens": 0, - } - - self._starts[event_id] = {"start": time.time(), "type": event_type, "payload": payload} + # Inserts under the same lock the prune's iteration holds - see _state_lock's comment. + with self._state_lock: + self._parents[event_id] = parent_id + + root_id = self._find_root(parent_id) + if root_id is None and event_type in _ROOT_EVENT_TYPES: + self._roots[event_id] = True + root_id = event_id + self._runs[event_id] = { + "start": time.time(), + "input": None, + "output": None, + "model": None, + "error": None, + "execution_steps": [], + "tool_call_steps": [], + "retrieval_steps": [], + "input_tokens": 0, + "output_tokens": 0, + } + + self._starts[event_id] = {"start": time.time(), "type": event_type, "payload": payload} state = self._runs.get(root_id) if root_id else None if state is not None and state["input"] is None: @@ -239,7 +270,15 @@ def on_event_end( elif event_type == CBEventType.LLM: completion = payload.get(EventPayload.COMPLETION) or payload.get(EventPayload.RESPONSE) output_text = _extract_llm_text(completion) + # LLM events don't actually set MODEL_NAME (it only appears on embedding events), + # so also read it off the start payload's SERIALIZED dict - the LLM's serialized + # config, same "model"/"model_name" ladder langchain.py's _record_llm_start walks. model = payload.get(EventPayload.MODEL_NAME) + if not model and start_info: + serialized = start_info["payload"].get(EventPayload.SERIALIZED) + if isinstance(serialized, dict): + model = serialized.get("model") or serialized.get("model_name") + model = str(model) if model else None if model and not state["model"]: state["model"] = model input_tokens, output_tokens = _extract_usage_tokens(completion) diff --git a/agentx/integrations/openai_agents.py b/agentx/integrations/openai_agents.py index 9fc4f3a..92398e2 100644 --- a/agentx/integrations/openai_agents.py +++ b/agentx/integrations/openai_agents.py @@ -207,10 +207,13 @@ def on_span_end(self, span: Any) -> None: # failure lives on `span.error` (a `SpanError | None`) instead. # First error wins: one failed span is enough to flag the trace. span_error = getattr(span, "error", None) - if span_error is not None and state.get("error") is None: + span_error_text: Optional[str] = None + if span_error is not None: error_message = getattr(span_error, "message", None) or str(span_error) error_data = getattr(span_error, "data", None) - state["error"] = f"{error_message} ({error_data})" if error_data else error_message + span_error_text = f"{error_message} ({error_data})" if error_data else error_message + if state.get("error") is None: + state["error"] = span_error_text t0 = _iso_to_ts(getattr(span, "started_at", None)) t1 = _iso_to_ts(getattr(span, "ended_at", None)) @@ -315,8 +318,21 @@ def on_span_end(self, span: Any) -> None: duration_ms=latency if t0 is None or t1 is None else None, input=span_data.input, output=tool_output, + error=span_error_text, span_kind="tool", ) + # Also mirror onto the root's flat tool_calls list (the dual-write every other + # integration does via _merge_child_run): the child span above only feeds the + # trace detail's span tree, while the engine's built-in "Tool failure" check and + # the dashboard's Tool quality column read the ROOT trace's flat toolCalls - + # without this, a failed function tool would be invisible to both. + state["root_span"].tool_calls.append({ + "name": span_data.name, + "input": _safe_serialize(span_data.input) if span_data.input is not None else None, + "output": tool_output, + "latency_ms": latency, + "success": span_error is None, + }) def force_flush(self) -> None: self._tracer.flush() diff --git a/agentx/monitor/client.py b/agentx/monitor/client.py index 42f19d4..b0b0ae1 100644 --- a/agentx/monitor/client.py +++ b/agentx/monitor/client.py @@ -18,7 +18,12 @@ logger = logging.getLogger(__name__) -from agentx.util import _DEFAULT_API_BASE as _UTIL_API_BASE +from agentx.util import _DEFAULT_API_BASE as _UTIL_API_BASE, normalize_base + +# The canonical error classes (agentx.exceptions) are raised - and re-exported here for +# compat with code that imported them from this module - so `except agentx.AgentXAuthError` +# works whichever client raised. +from agentx.exceptions import AgentXError, AgentXAuthError, AgentXValidationError SDK_NAME = "agentx-python" @@ -27,7 +32,7 @@ _RETRY_BACKOFF = [1.0, 2.0, 4.0] -class AgentXMonitorError(Exception): +class AgentXMonitorError(AgentXError): """``status_code`` carries the HTTP status when the error came from a server response; it is ``None`` for transport-level failures and retry exhaustion.""" @@ -36,14 +41,6 @@ def __init__(self, message: str, status_code: Optional[int] = None) -> None: self.status_code = status_code -class AgentXAuthError(AgentXMonitorError): - pass - - -class AgentXValidationError(AgentXMonitorError): - pass - - class CalibrationSummary(dict): """Judge Calibration numbers (dict subclass, so existing key access keeps working). Properties mirror the wire's exact camelCase keys.""" @@ -109,9 +106,11 @@ def __init__( # EvaluationsClient. Without this, pattern creation silently lands in whatever # workspace the API key's user defaults to, not the one the caller intended. self._workspace_id = workspace_id - _api_base = ( + # normalize_base also strips an evaluations-shaped URL's suffix, so a base copied + # from an eval env file doesn't 404 every monitor route. + _api_base = normalize_base( base_url or os.getenv("AGENTX_API_BASE_URL", _UTIL_API_BASE) - ).rstrip("/") + ) if not _api_base.endswith("/monitor"): _api_base = f"{_api_base}/monitor" self._base_url = _api_base @@ -216,7 +215,10 @@ def _request( raise AgentXAuthError("Invalid or missing API key", status_code=401) if resp.status_code == 422: raise AgentXValidationError(resp.text, status_code=422) - if retry and resp.status_code in _RETRYABLE_STATUS and attempt < _MAX_RETRIES - 1: + # Gate on the schedule itself so HTTP-status retries walk the SAME full backoff + # schedule connection errors do - the old `attempt < _MAX_RETRIES - 1` gate left + # the schedule's last entry unreachable for HTTP retries (ingest_client precedent). + if retry and resp.status_code in _RETRYABLE_STATUS and attempt < len(schedule) - 1: logger.debug( "Retryable status %d (attempt %d)", resp.status_code, attempt + 1 ) @@ -228,6 +230,11 @@ def _request( return resp.json() except Exception: return resp.text + # A timeout keeps its type (evaluations client precedent): callers guard on + # requests.Timeout specifically for judge-billing endpoints - the server may still be + # doing the work, and wrapping the timeout made that guard unreachable. + if isinstance(last_exc, requests.Timeout): + raise last_exc raise AgentXMonitorError(f"Request failed after retries: {last_exc}") # ------------------------------------------------------------------ @@ -510,6 +517,13 @@ def publish_online_evaluator_tuning( # unless forced - see judge_scorers.publish_tuning for the full story. payload = dict(criteria) if validation is not None: + # Passed through whole; the signed `validationToken` from tune/validate is also + # lifted into the `token` key the publish route verifies, so the version history + # stamps "measured" instead of client-asserted (judge_scorers.publish_tuning + # precedent). + validation = dict(validation) + if validation.get("validationToken") and not validation.get("token"): + validation["token"] = validation["validationToken"] payload["validation"] = validation if force: payload["force"] = True diff --git a/agentx/monitor/improvement_groups.py b/agentx/monitor/improvement_groups.py index e1c87fb..3e59469 100644 --- a/agentx/monitor/improvement_groups.py +++ b/agentx/monitor/improvement_groups.py @@ -5,9 +5,10 @@ import requests from agentx.util import api_base, get_headers +from agentx.exceptions import AgentXError -class AgentXImprovementGroupsError(Exception): +class AgentXImprovementGroupsError(AgentXError): pass diff --git a/agentx/monitor/judge_scorers.py b/agentx/monitor/judge_scorers.py index d73994a..f07f447 100644 --- a/agentx/monitor/judge_scorers.py +++ b/agentx/monitor/judge_scorers.py @@ -6,13 +6,14 @@ import requests from agentx.util import api_base, get_headers +from agentx.exceptions import AgentXError logger = logging.getLogger(__name__) _SENTINEL: Any = object() -class AgentXJudgeScorersError(Exception): +class AgentXJudgeScorersError(AgentXError): pass @@ -310,10 +311,18 @@ def publish_tuning( ``force=True`` publishes without (or despite) validation - deliberate escape hatch.""" payload = dict(criteria) if validation is not None: - payload["validation"] = { + validation_payload: Dict[str, Any] = { "verdict": validation.get("verdict"), "netAgreementGain": validation.get("netAgreementGain"), } + # The signed provenance token validate_tuning's response carries as + # `validationToken` - the engine's publish route reads it as validation.token and + # stamps the version history "measured" only when it verifies. Dropping it here + # (the old projection did) downgraded every publish to client-asserted. + token = validation.get("validationToken") or validation.get("token") + if token: + validation_payload["token"] = token + payload["validation"] = validation_payload if force: payload["force"] = True return self._request("POST", f"/online-evaluators/{self._profile_id(scorer_id)}/tune/publish", json=payload) diff --git a/agentx/monitor/scorer_groups.py b/agentx/monitor/scorer_groups.py index b832c50..2ab2615 100644 --- a/agentx/monitor/scorer_groups.py +++ b/agentx/monitor/scorer_groups.py @@ -9,8 +9,10 @@ import requests +from agentx.exceptions import AgentXError -class AgentXScorerGroupsError(Exception): + +class AgentXScorerGroupsError(AgentXError): pass diff --git a/agentx/monitor/scorers.py b/agentx/monitor/scorers.py index d61f0a9..bf5a80c 100644 --- a/agentx/monitor/scorers.py +++ b/agentx/monitor/scorers.py @@ -6,11 +6,12 @@ import requests from agentx.util import api_base, get_headers +from agentx.exceptions import AgentXError logger = logging.getLogger(__name__) -class AgentXScorersError(Exception): +class AgentXScorersError(AgentXError): pass diff --git a/agentx/outcomes.py b/agentx/outcomes.py index 7410f02..559f3c8 100644 --- a/agentx/outcomes.py +++ b/agentx/outcomes.py @@ -6,11 +6,12 @@ import requests from agentx.util import api_base, get_headers +from agentx.exceptions import AgentXError logger = logging.getLogger(__name__) -class AgentXOutcomesError(Exception): +class AgentXOutcomesError(AgentXError): pass diff --git a/agentx/projects.py b/agentx/projects.py index 08ba4b1..77a79c3 100644 --- a/agentx/projects.py +++ b/agentx/projects.py @@ -6,11 +6,12 @@ import requests from agentx.util import api_base, get_headers +from agentx.exceptions import AgentXError logger = logging.getLogger(__name__) -class AgentXProjectsError(Exception): +class AgentXProjectsError(AgentXError): pass diff --git a/agentx/traces.py b/agentx/traces.py index e28eb3b..ac8a7f9 100644 --- a/agentx/traces.py +++ b/agentx/traces.py @@ -6,11 +6,12 @@ import requests from agentx.util import api_base, get_headers +from agentx.exceptions import AgentXError logger = logging.getLogger(__name__) -class AgentXTracesError(Exception): +class AgentXTracesError(AgentXError): pass diff --git a/agentx/tracing/ingest_client.py b/agentx/tracing/ingest_client.py index a0d82e0..56e4b71 100644 --- a/agentx/tracing/ingest_client.py +++ b/agentx/tracing/ingest_client.py @@ -1,5 +1,6 @@ from __future__ import annotations +import atexit import logging import os import queue @@ -9,7 +10,7 @@ import requests -from agentx.util import _DEFAULT_API_BASE as _UTIL_API_BASE +from agentx.util import _DEFAULT_API_BASE as _UTIL_API_BASE, normalize_base from agentx.exceptions import AgentXAPIError, CINotEnabled, DatasetNotFound from agentx.tracing.ci_types import ( CIRun, @@ -51,10 +52,9 @@ def __init__( self._api_key = api_key self._sdk_version = sdk_version - _base = (base_url or os.getenv("AGENTX_API_BASE_URL", _UTIL_API_BASE)).rstrip("/") - # Strip the custom-agent-evaluations suffix if someone passes the eval base URL - if _base.endswith("/custom-agent-evaluations"): - _base = _base[: -len("/custom-agent-evaluations")] + # normalize_base strips the trailing slash and the custom-agent-evaluations suffix if + # someone passes the eval base URL - shared with util.api_base()/AgentX.__init__. + _base = normalize_base(base_url or os.getenv("AGENTX_API_BASE_URL", _UTIL_API_BASE)) self._endpoint = f"{_base}/ingest/traces" self._session = requests.Session() @@ -81,6 +81,11 @@ def __init__( self._worker = threading.Thread(target=self._drain, daemon=True, name="agentx-ingest") self._worker.start() + # Flush (not close) at interpreter shutdown so traces enqueued moments before exit + # still get a bounded delivery attempt - the worker is a daemon thread, so without + # this they'd silently die with the process. close() unregisters it. + atexit.register(self.flush) + # Base URL (without the /ingest/traces suffix) for synchronous calls like evaluate_trace self._base_url = _base @@ -121,6 +126,27 @@ def flush(self, timeout: float = 5.0) -> bool: self._queue.all_tasks_done.wait(remaining) return True + def close(self, timeout: float = 5.0) -> bool: + """Drain what the deadline allows, then stop the background worker for good. + + Enqueues the ``None`` sentinel the worker loop exits on and joins the thread, both + bounded by ``timeout`` seconds of total wall-clock time. Returns ``True`` when the + queue fully drained AND the worker stopped. Idempotent; a closed client must not be + used to send further traces (they would queue with no worker to deliver them). + """ + deadline = time.time() + timeout + if not self._worker.is_alive(): + return True + drained = self.flush(max(0.0, deadline - time.time())) + try: + self._queue.put_nowait(None) + except queue.Full: + # Worker wedged behind a full queue - the bounded join below still applies. + pass + self._worker.join(max(0.1, deadline - time.time())) + atexit.unregister(self.flush) + return drained and not self._worker.is_alive() + def _record_drop(self, reason: str) -> None: self._dropped += 1 if self._dropped == 1 or self._dropped % 50 == 0: diff --git a/agentx/tracing/tracer.py b/agentx/tracing/tracer.py index 4071fbd..2bd1024 100644 --- a/agentx/tracing/tracer.py +++ b/agentx/tracing/tracer.py @@ -825,6 +825,7 @@ def record_retrieval( duration_ms: Optional[float] = None, start_time: Optional[float] = None, end_time: Optional[float] = None, + error: Optional[str] = None, ) -> None: """ Manually record a knowledge-base / vector-store retrieval that an @@ -856,6 +857,7 @@ def record_retrieval( "output": _safe_serialize(output) if output is not None else None, "duration_ms": latency_ms, **({"doc_count": doc_count} if doc_count is not None else {}), + **({"error": error} if error is not None else {}), }) return # The kind marker is what tells the engine (retrieval-context extraction for RAG @@ -869,6 +871,7 @@ def record_retrieval( duration_ms=duration_ms, input=query, output=output, + error=error, metadata={"kind": "retrieval", **({"doc_count": doc_count} if doc_count is not None else {})}, span_kind="retrieval", ) @@ -883,6 +886,7 @@ def record_memory( duration_ms: Optional[float] = None, start_time: Optional[float] = None, end_time: Optional[float] = None, + error: Optional[str] = None, ) -> None: """ Manually record a long-term-memory operation (a Mem0/Zep/Letta-style recall or store) @@ -916,6 +920,7 @@ def record_memory( duration_ms=duration_ms, input=query, output=output, + error=error, metadata={"kind": "memory", **({"operation": operation} if operation else {})}, span_kind="memory", ) @@ -945,6 +950,8 @@ def trace_memory( finally: end_t = time.time() if error is not None and recorder.output is None: + # The ERROR: prefix stays in the output for at-a-glance visibility; the + # structured error field below is what marks the span failed. recorder.output = f"ERROR: {error}" self.record_memory( name, @@ -954,6 +961,7 @@ def trace_memory( duration_ms=(end_t - start_t) * 1000, start_time=start_t, end_time=end_t, + error=error, ) @contextmanager @@ -966,13 +974,26 @@ def trace_retrieval(self, name: str = "Retrieval", *, query: Optional[str] = Non docs = retrieve(question) r.doc_count = len(docs) r.output = docs + + An exception escaping the block records the retrieval as failed (error set, output + ``ERROR: ...``) instead of as a clean empty retrieval, then propagates unchanged - + same posture as :meth:`trace_tool_call`. """ start_t = time.time() recorder = _RetrievalRecorder() + error: Optional[str] = None try: yield recorder + except BaseException as exc: + # A retrieval that raised must not be recorded as a clean empty span + # (trace_tool_call precedent, which also catches BaseException) - fold the error + # into the output and re-raise. + error = str(exc) + raise finally: end_t = time.time() + if error is not None and recorder.output is None: + recorder.output = f"ERROR: {error}" self.record_retrieval( name, query=query, @@ -981,6 +1002,7 @@ def trace_retrieval(self, name: str = "Retrieval", *, query: Optional[str] = Non duration_ms=(end_t - start_t) * 1000, start_time=start_t, end_time=end_t, + error=error, ) def trace( @@ -1323,23 +1345,31 @@ def _send(self, sync: bool = False, **kwargs) -> Optional[str]: if "span_kind" in payload: wire["span_kind"] = payload["span_kind"] - pending_tool_calls, self._pending_tool_calls = self._pending_tool_calls, [] - if pending_tool_calls: - # Passed through whole rather than re-projected field by field: record_tool_call may - # have attached success/error (the fields the engine's tool-failure check reads), and - # a projection that predates them would silently strip exactly the failure evidence. - wire["tool_calls"] = list(wire.get("tool_calls") or []) + [dict(t) for t in pending_tool_calls] - - # record_retrieval entries queued with no active span ride the root's - # performance_summary.retrieval_steps - the same shape older flat traces used, which the - # engine's retrieval-context extraction and the dashboard's references panel both read. - pending_retrievals, self._pending_retrievals = self._pending_retrievals, [] - if pending_retrievals: - summary = dict(wire.get("performance_summary") or {}) - summary["retrieval_steps"] = list(summary.get("retrieval_steps") or []) + [ - dict(r) for r in pending_retrievals - ] - wire["performance_summary"] = summary + # Drain the no-active-span pending queues into ROOT sends only. A nested + # `with tracer.trace(...)` block exits (and _send()s) before its parent, and a pending + # record drained into that child row is invisible to the engine's monitor pipeline - + # routes/ingest.ts skips every parent_span_id row - so a queued failed-tool signal + # would silently vanish into a span nobody checks. + if "parent_span_id" not in wire: + pending_tool_calls, self._pending_tool_calls = self._pending_tool_calls, [] + if pending_tool_calls: + # Passed through whole rather than re-projected field by field: record_tool_call + # may have attached success/error (the fields the engine's tool-failure check + # reads), and a projection that predates them would silently strip exactly the + # failure evidence. + wire["tool_calls"] = list(wire.get("tool_calls") or []) + [dict(t) for t in pending_tool_calls] + + # record_retrieval entries queued with no active span ride the root's + # performance_summary.retrieval_steps - the same shape older flat traces used, which + # the engine's retrieval-context extraction and the dashboard's references panel + # both read. + pending_retrievals, self._pending_retrievals = self._pending_retrievals, [] + if pending_retrievals: + summary = dict(wire.get("performance_summary") or {}) + summary["retrieval_steps"] = list(summary.get("retrieval_steps") or []) + [ + dict(r) for r in pending_retrievals + ] + wire["performance_summary"] = summary return self._dispatch(wire, sync=sync) diff --git a/agentx/util.py b/agentx/util.py index d95d565..40eddae 100644 --- a/agentx/util.py +++ b/agentx/util.py @@ -3,15 +3,24 @@ _DEFAULT_API_BASE = "https://api.agentx.so/api/v1" +_EVALUATIONS_SUFFIX = "/custom-agent-evaluations" + + +def normalize_base(base: str) -> str: + """Normalize a user-supplied API base URL so it works for ALL routes: strip a trailing + slash, and strip the evaluations-specific ``/custom-agent-evaluations`` suffix (users + copying the eval endpoint out of a dashboard/env file otherwise 404 every non-eval + sub-client - monitor, outcomes, traces, ingest, ...).""" + base = base.rstrip("/") + if base.endswith(_EVALUATIONS_SUFFIX): + base = base[: -len(_EVALUATIONS_SUFFIX)] + return base + def api_base() -> str: """Return the base URL for all AgentX API calls, respecting AGENTX_API_BASE_URL if set.""" - override = os.getenv("AGENTX_API_BASE_URL", "").rstrip("/") + override = normalize_base(os.getenv("AGENTX_API_BASE_URL", "")) if override: - # Strip the evaluations-specific suffix if present so the override works for all routes - suffix = "/custom-agent-evaluations" - if override.endswith(suffix): - override = override[: -len(suffix)] return override return _DEFAULT_API_BASE diff --git a/tests/test_deep_dive_fixes.py b/tests/test_deep_dive_fixes.py index 3777b98..2b1f54e 100644 --- a/tests/test_deep_dive_fixes.py +++ b/tests/test_deep_dive_fixes.py @@ -83,3 +83,48 @@ def refuse(*args, **kwargs): assert client._dropped >= 1 assert any("dropped a trace" in rec.message for rec in caplog.records) + + +def test_evaluations_shaped_base_url_works_for_every_subclient(): + """P0 regression: a base_url carrying the /custom-agent-evaluations suffix (what users copy + out of eval env files) used to 404 every non-eval sub-client - AgentX.__init__ now + normalizes it (trailing slash + suffix stripped) before handing it out.""" + c = AgentX(api_key="k", base_url="http://engine:4700/api/v1/custom-agent-evaluations/") + + assert c.base_url == "http://engine:4700/api/v1" + assert c.tracer._client._endpoint == "http://engine:4700/api/v1/ingest/traces" + assert c.monitor._base_url == "http://engine:4700/api/v1/monitor" + for sub in (c.projects, c.traces, c.export, c.feedback, c.outcomes): + assert sub._base_url == "http://engine:4700/api/v1", type(sub).__name__ + # The evaluations client re-appends its own suffix exactly once. + assert c.evaluations._client._base_url == "http://engine:4700/api/v1/custom-agent-evaluations" + + +def test_env_var_with_evaluations_suffix_is_normalized_too(monkeypatch): + monkeypatch.setenv("AGENTX_API_BASE_URL", "http://from-env:3333/api/v1/custom-agent-evaluations") + c = AgentX(api_key="k") + assert c.base_url == "http://from-env:3333/api/v1" + assert c.tracer._client._endpoint == "http://from-env:3333/api/v1/ingest/traces" + + +def test_ingest_client_close_stops_the_worker(): + """P2: close() enqueues the sentinel and joins the worker so a torn-down client leaves no + thread behind; AgentX exposes it (and context-manager form) on top.""" + client = IngestClient(api_key="k", sdk_version="test", base_url="http://localhost:9/api/v1") + assert client.close(timeout=2.0) is True + assert not client._worker.is_alive() + # Idempotent. + assert client.close(timeout=1.0) is True + + +def test_agentx_context_manager_closes_the_ingest_worker(): + with AgentX(api_key="k", base_url="http://localhost:9/api/v1") as c: + worker = c._ingest_client._worker + assert worker.is_alive() + assert not worker.is_alive() + + +def test_constructor_no_longer_writes_the_api_key_into_the_environment(monkeypatch): + monkeypatch.delenv("AGENTX_API_KEY", raising=False) + AgentX(api_key="k-secret", base_url="http://localhost:9/api/v1") + assert os.environ.get("AGENTX_API_KEY") is None diff --git a/tests/test_error_taxonomy.py b/tests/test_error_taxonomy.py new file mode 100644 index 0000000..4ee4202 --- /dev/null +++ b/tests/test_error_taxonomy.py @@ -0,0 +1,86 @@ +"""P1 regression: one error taxonomy. Every per-client error subclasses +agentx.exceptions.AgentXError, and the auth/validation classes are the SAME canonical classes +everywhere - so `except agentx.AgentXError` / `except agentx.AgentXAuthError` work no matter +which sub-client raised.""" + +import pytest + +import agentx +from agentx.exceptions import AgentXError + + +def test_every_client_error_subclasses_agentx_error(): + from agentx.evaluations.client import ( + AgentXEvaluationsError, + EvaluationSubmissionError, + ) + from agentx.monitor.client import AgentXMonitorError + from agentx.monitor.judge_scorers import AgentXJudgeScorersError + from agentx.monitor.scorer_groups import AgentXScorerGroupsError + from agentx.monitor.scorers import AgentXScorersError + from agentx.monitor.improvement_groups import AgentXImprovementGroupsError + from agentx.outcomes import AgentXOutcomesError + from agentx.feedback import AgentXFeedbackError + from agentx.export import AgentXExportError + from agentx.traces import AgentXTracesError + from agentx.projects import AgentXProjectsError + + for cls in ( + AgentXEvaluationsError, + EvaluationSubmissionError, + AgentXMonitorError, + AgentXJudgeScorersError, + AgentXScorerGroupsError, + AgentXScorersError, + AgentXImprovementGroupsError, + AgentXOutcomesError, + AgentXFeedbackError, + AgentXExportError, + AgentXTracesError, + AgentXProjectsError, + ): + assert issubclass(cls, AgentXError), cls.__name__ + assert issubclass(cls, agentx.AgentXError), cls.__name__ + + +def test_auth_and_validation_errors_are_the_canonical_classes(): + from agentx.evaluations import client as eval_client + from agentx.monitor import client as monitor_client + + # The shadow definitions are gone - both modules re-export the canonical classes, so the + # names still import from where they always did. + assert eval_client.AgentXAuthError is agentx.AgentXAuthError + assert eval_client.AgentXValidationError is agentx.AgentXValidationError + assert monitor_client.AgentXAuthError is agentx.AgentXAuthError + assert monitor_client.AgentXValidationError is agentx.AgentXValidationError + + +def test_evaluations_401_is_catchable_as_top_level_auth_error(monkeypatch): + from agentx.evaluations.client import EvaluationsClient + + client = EvaluationsClient(api_key="k", base_url="http://engine:1/api/v1") + + class FakeResponse: + status_code = 401 + ok = False + text = "nope" + + monkeypatch.setattr(client._session, "request", lambda *a, **kw: FakeResponse()) + with pytest.raises(agentx.AgentXAuthError): + client.get_run("run-1") + + +def test_monitor_422_is_catchable_as_top_level_validation_error(monkeypatch): + from agentx.monitor.client import MonitorClient + + client = MonitorClient(api_key="k", base_url="http://engine:1/api/v1") + + class FakeResponse: + status_code = 422 + ok = False + text = "bad payload" + + monkeypatch.setattr(client._session, "request", lambda *a, **kw: FakeResponse()) + with pytest.raises(agentx.AgentXValidationError) as caught: + client.kpis() + assert caught.value.status_code == 422 diff --git a/tests/test_integrations.py b/tests/test_integrations.py index 0f8ed50..8005c7f 100644 --- a/tests/test_integrations.py +++ b/tests/test_integrations.py @@ -696,7 +696,7 @@ def kickoff(self, inputs=None): # Force the "no events module" path regardless of whether crewai is installed. original = observer._start_task_timing_capture - observer._start_task_timing_capture = lambda: ({}, lambda: None) + observer._start_task_timing_capture = lambda crew=None: ({}, lambda: None) try: result = observer.kickoff(FakeCrew(), inputs={"topic": "AI"}) finally: diff --git a/tests/test_judge_scorers.py b/tests/test_judge_scorers.py index 692db0c..215bdab 100644 --- a/tests/test_judge_scorers.py +++ b/tests/test_judge_scorers.py @@ -299,3 +299,50 @@ def test_dataset_model_round_trips_code_scorers(): parsed = Dataset(**wire) assert parsed.code_scorers == wire["codeScorers"] assert parsed.model_dump(by_alias=True)["codeScorers"] == wire["codeScorers"] + + +def test_publish_tuning_preserves_the_validation_token(recorded): + """P1 regression: publish_tuning re-projected the validation dict and dropped the signed + validationToken from validate_tuning's response, so the engine's version history stamped + every publish client-asserted instead of measured.""" + calls, responses = recorded + responses.append( + FakeResponse({"judgeScorer": {"_id": "s1", "name": "n", "online": {"profileId": "prof-9"}}}) + ) + responses.append(FakeResponse({"published": True})) + make_client().publish_tuning( + "s1", + {"acceptanceCriteria": "a", "rejectionCriteria": "r", "evaluationCriteria": "e"}, + validation={ + "verdict": "improved", + "netAgreementGain": 0.12, + "validationToken": "tok-123", + "fixed": 3, + }, + ) + body = calls[1]["json"] + assert body["validation"] == {"verdict": "improved", "netAgreementGain": 0.12, "token": "tok-123"} + + +def test_legacy_monitor_publish_lifts_validation_token(monkeypatch): + """Same fix on the legacy MonitorClient path: the dict passes through whole AND the + validationToken is lifted into the `token` key the engine's publish route verifies.""" + from agentx.monitor.client import MonitorClient + + client = MonitorClient(api_key="k", base_url="http://engine:1/api/v1") + captured = {} + + def fake_request(method, path, base=None, json=None, **kwargs): + captured.update({"method": method, "path": path, "json": json}) + return {} + + monkeypatch.setattr(client, "_request", fake_request) + client.publish_online_evaluator_tuning( + "ev-1", + {"acceptanceCriteria": "a"}, + validation={"verdict": "improved", "validationToken": "tok-9", "fixed": 2}, + ) + validation = captured["json"]["validation"] + assert validation["token"] == "tok-9" + assert validation["validationToken"] == "tok-9" # passed through whole + assert validation["fixed"] == 2 diff --git a/tests/test_runner_features.py b/tests/test_runner_features.py index efc89e1..4bca8ad 100644 --- a/tests/test_runner_features.py +++ b/tests/test_runner_features.py @@ -160,3 +160,76 @@ def failing_finalize(run_id): with pytest.raises(RuntimeError, match="engine down"): ctx.finalize() + + +def test_fail_fast_stops_dispatching_agent_calls_at_concurrency(monkeypatch): + """P1 regression: executor.map() dispatched EVERY case up front, so a flush failure's + fail-fast still paid for the whole run in agent calls. The bounded submit loop keeps at + most `concurrency` cases in flight, so a failure after the first flush leaves the rest + of the dataset un-run.""" + monkeypatch.setenv("AGENTX_EVAL_QUIET", "1") + n_cases = 8 + concurrency = 2 + dataset = make_dataset( + questions=[{"main_question": {"query": f"q{i}"}} for i in range(n_cases)] + ) + # maxBatchSize=1 so the first result flushes (and fails) while cases remain. + run = EvaluationRun(runId="run-1", datasetId="ds-1", limits={"maxBatchSize": 1}) + client = FakeClient(fail_batches=2) # first flush attempt + its retry both fail + ctx = EvaluationRunContext(client, dataset, run, EvaluationSubject()) # type: ignore[arg-type] + + started: List[str] = [] + lock = threading.Lock() + + def agent(case): + with lock: + started.append(case.query) + time.sleep(0.02) + return f"answer to {case.query}" + + with pytest.raises(EvaluationSubmissionError): + ctx.execute(agent, concurrency=concurrency) + + # At the moment of failure at most 1 consumed + `concurrency` topped-up cases were ever + # dispatched - executor.map would have started all 8 (shutdown(wait=True) even ran them + # to completion). + assert len(started) <= 1 + concurrency, started + + +def test_resume_key_fetch_transient_failure_raises_instead_of_rebilling(monkeypatch): + """P2 regression: a transient 502 on the resume-key fetch used to be swallowed into an + empty resume set, silently re-running (and re-billing) every already-finished case.""" + monkeypatch.setenv("AGENTX_EVAL_QUIET", "1") + from agentx.evaluations.client import AgentXEvaluationsError + + client = FakeClient() + + def boom(run_id): + raise AgentXEvaluationsError("HTTP 502: bad gateway", status_code=502) + + client.get_submitted_keys = boom + ctx = make_context(client) + ran: List[str] = [] + + with pytest.raises(AgentXEvaluationsError): + ctx.execute(lambda case: ran.append(case.query) or "x") + assert ran == [] # no agent spend before the failure surfaced + + +def test_resume_key_fetch_404_still_means_no_resume(monkeypatch): + """Older engines without the route return 404 - that (and only that) keeps the historical + no-resume behavior.""" + monkeypatch.setenv("AGENTX_EVAL_QUIET", "1") + from agentx.evaluations.client import AgentXEvaluationsError + + client = FakeClient() + + def missing(run_id): + raise AgentXEvaluationsError("HTTP 404: not found", status_code=404) + + client.get_submitted_keys = missing + ctx = make_context(client) + ran: List[str] = [] + + ctx.execute(lambda case: ran.append(case.query) or "x") + assert ran == ["q0", "q1", "q2"] diff --git a/tests/test_span_tree.py b/tests/test_span_tree.py index 2fa2afc..f7c5b5e 100644 --- a/tests/test_span_tree.py +++ b/tests/test_span_tree.py @@ -356,7 +356,7 @@ def kickoff(self, inputs=None): "task-1": {"name": "Research topic", "start": now, "end": now + 0.03, "error": None}, "task-2": {"name": "Write summary", "start": now + 0.03, "end": now + 0.04, "error": None}, } - observer._start_task_timing_capture = lambda: (fake_timings, lambda: None) + observer._start_task_timing_capture = lambda crew=None: (fake_timings, lambda: None) result = observer.kickoff(FakeCrew(), inputs={"topic": "AI"}) @@ -380,7 +380,7 @@ def test_crewai_falls_back_to_evenly_divided_children_without_event_bus(): tracer = make_tracer() observer = AgentXCrewObserver(tracer, name="my-crew") - observer._start_task_timing_capture = lambda: ({}, lambda: None) + observer._start_task_timing_capture = lambda crew=None: ({}, lambda: None) class FakeTaskOutput: def __init__(self, description, raw): @@ -423,7 +423,7 @@ def kickoff(self, inputs=None): now = 1_700_000_000.0 fake_timings = {"task-1": {"name": "Research topic", "start": now, "end": now + 0.03, "error": None}} - observer._start_task_timing_capture = lambda: (fake_timings, lambda: None) + observer._start_task_timing_capture = lambda crew=None: (fake_timings, lambda: None) with pytest.raises(RuntimeError, match="boom"): observer.kickoff(FakeCrew(), inputs={"topic": "AI"}) @@ -737,3 +737,232 @@ def test_record_tool_call_with_no_active_span_still_queues(): wires = enqueued_wires(tracer) assert len(wires) == 1 assert wires[0]["tool_calls"][0]["name"] == "orphan_call" + + +def test_pending_records_drain_into_root_sends_only_never_nested_children(): + """P0 regression: a record_tool_call/record_retrieval queued with no active span must ride + the next ROOT trace's wire. The old drain ran on every _send(), and a nested `with + tracer.trace(...)` exits (and sends) before its parent - so the queue drained into a CHILD + row, which the engine's monitor pipeline ignores entirely (routes/ingest.ts skips + parent_span_id rows), silently losing the failed-tool signal.""" + tracer = make_tracer() + tracer.record_tool_call("orphan_call", input="x", output="boom", success=False, latency_ms=5) + tracer.record_retrieval("orphan_search", query="q", output="docs", doc_count=2) + with tracer.trace("outer"): + with tracer.trace("inner"): + pass + wires = enqueued_wires(tracer) + assert len(wires) == 2 + inner, outer = wires # inner exits (and sends) first + assert inner["parent_span_id"] == outer["span_id"] + assert "tool_calls" not in inner + assert "performance_summary" not in inner + assert [tc["name"] for tc in outer["tool_calls"]] == ["orphan_call"] + assert outer["tool_calls"][0]["success"] is False + assert [s["name"] for s in outer["performance_summary"]["retrieval_steps"]] == ["orphan_search"] + + +def test_trace_retrieval_exception_records_failed_retrieval(): + """P1 regression: an exception escaping the trace_retrieval block used to record a CLEAN + empty retrieval - now it records error + ERROR: output (trace_tool_call posture) and + re-raises unchanged.""" + tracer = make_tracer() + with pytest.raises(RuntimeError, match="index down"): + with tracer.trace("agent"): + with tracer.trace_retrieval("kb_search", query="refunds"): + raise RuntimeError("index down") + wires = enqueued_wires(tracer) + child = next(w for w in wires if w["name"] == "kb_search") + assert child["error"] == "index down" + assert child["output"] == "ERROR: index down" + assert child["span_kind"] == "retrieval" + + +def test_trace_memory_exception_records_failed_memory_op(): + """P1 regression: same posture for trace_memory - the captured exception is forwarded as + the span's structured error (not only folded into the output text) and re-raised.""" + tracer = make_tracer() + with pytest.raises(RuntimeError, match="store down"): + with tracer.trace("agent"): + with tracer.trace_memory("user prefs", operation="read", query="u-1"): + raise RuntimeError("store down") + wires = enqueued_wires(tracer) + child = next(w for w in wires if w["name"] == "user prefs") + assert child["error"] == "store down" + assert child["output"] == "ERROR: store down" + assert child["span_kind"] == "memory" + + +def test_openai_agents_mirrors_function_tools_onto_root_flat_tool_calls(): + """P0 regression: function (tool) spans emitted only a child-span row - the engine's + built-in "Tool failure" check and the dashboard's Tool quality column read the ROOT's + flat toolCalls, so a failed tool was invisible to both. The mirror carries + success: False for a span with an error.""" + import types + from agentx.integrations.openai_agents import AgentXTracingProcessor + + tracer = make_tracer() + processor = AgentXTracingProcessor(tracer) + + trace = types.SimpleNamespace(trace_id="trace-3", name="my-agent") + processor.on_trace_start(trace) + + ok_data = types.SimpleNamespace(type="function", name="lookup", input="q", output="r") + ok_span = types.SimpleNamespace( + trace_id="trace-3", span_data=ok_data, + started_at="2026-01-01T00:00:00Z", ended_at="2026-01-01T00:00:01Z", error=None, + ) + processor.on_span_end(ok_span) + + failed_data = types.SimpleNamespace(type="function", name="charge_card", input="{}", output=None) + failed_span = types.SimpleNamespace( + trace_id="trace-3", span_data=failed_data, + started_at="2026-01-01T00:00:01Z", ended_at="2026-01-01T00:00:02Z", + error=types.SimpleNamespace(message="card declined", data=None), + ) + processor.on_span_end(failed_span) + + processor.on_trace_end(trace) + + wires = enqueued_wires(tracer) + root = next(w for w in wires if w["name"] == "my-agent") + assert [tc["name"] for tc in root["tool_calls"]] == ["lookup", "charge_card"] + assert root["tool_calls"][0]["success"] is True + assert root["tool_calls"][1]["success"] is False + # The failed function's child span row is also marked failed. + failed_child = next(w for w in wires if w["name"] == "charge_card") + assert failed_child["error"] == "card declined" + + +def test_google_adk_mirrors_tools_onto_root_flat_tool_calls(): + """P0 regression: same mirror for the ADK plugin's tool callbacks - see the openai_agents + test above for why the root's flat list is load-bearing.""" + import asyncio + import types + + pytest.importorskip("google.adk") + from agentx.integrations.google_adk import AgentXADKPlugin + + tracer = make_tracer() + plugin = AgentXADKPlugin(tracer, name="adk-agent") + + invocation_context = types.SimpleNamespace(invocation_id="inv-3", agent=types.SimpleNamespace(name="adk-agent")) + tool_context_ok = types.SimpleNamespace(get_invocation_context=lambda: invocation_context) + tool_context_bad = types.SimpleNamespace(get_invocation_context=lambda: invocation_context) + lookup = types.SimpleNamespace(name="lookup") + charge = types.SimpleNamespace(name="charge_card") + + async def run(): + await plugin.before_run_callback(invocation_context=invocation_context) + await plugin.before_tool_callback(tool=lookup, tool_args={"q": "x"}, tool_context=tool_context_ok) + await plugin.after_tool_callback( + tool=lookup, tool_args={"q": "x"}, tool_context=tool_context_ok, result={"ok": True} + ) + await plugin.before_tool_callback(tool=charge, tool_args={}, tool_context=tool_context_bad) + await plugin.on_tool_error_callback( + tool=charge, tool_args={}, tool_context=tool_context_bad, error=RuntimeError("card declined") + ) + await plugin.after_run_callback(invocation_context=invocation_context) + + asyncio.run(run()) + + wires = enqueued_wires(tracer) + root = next(w for w in wires if w["name"] == "adk-agent") + assert [tc["name"] for tc in root["tool_calls"]] == ["lookup", "charge_card"] + assert root["tool_calls"][0]["success"] is True + assert root["tool_calls"][1]["success"] is False + assert root["tool_calls"][1]["output"] == "ERROR: card declined" + + +def test_google_adk_root_never_touches_the_active_span_stack(): + """P2 regression: before_run_callback used to __enter__ the root span, pushing it onto the + calling context's stack - ADK may end the run elsewhere, so the entry never drained and + later unrelated traces were mis-filed as its children.""" + import asyncio + import types + + pytest.importorskip("google.adk") + from agentx.integrations.google_adk import AgentXADKPlugin + + tracer = make_tracer() + plugin = AgentXADKPlugin(tracer, name="adk-agent") + invocation_context = types.SimpleNamespace(invocation_id="inv-4", agent=types.SimpleNamespace(name="adk-agent")) + + async def run(): + await plugin.before_run_callback(invocation_context=invocation_context) + assert tracer.current_span is None # never pushed + await plugin.after_run_callback(invocation_context=invocation_context) + + asyncio.run(run()) + wires = enqueued_wires(tracer) + assert len(wires) == 1 + assert "parent_span_id" not in wires[0] + assert wires[0]["session_id"].startswith("sdk_") + + +def test_llamaindex_resolves_model_from_serialized_start_payload_and_object_raw_usage(): + """P1 regression: LLM events never set MODEL_NAME (the old lookup left model None forever) + and token extraction bailed when completion.raw was a typed object instead of a dict.""" + import types + + pytest.importorskip("llama_index.core") + from llama_index.core.callbacks.schema import CBEventType, EventPayload + + from agentx.integrations.llamaindex import AgentXLlamaIndexHandler + + tracer = make_tracer() + handler = AgentXLlamaIndexHandler(tracer, name="llm-agent") + + completion = types.SimpleNamespace( + text="hello", + raw=types.SimpleNamespace(usage=types.SimpleNamespace(prompt_tokens=7, completion_tokens=3)), + ) + handler.on_event_start( + CBEventType.LLM, + payload={EventPayload.SERIALIZED: {"model": "gpt-4o-mini"}, EventPayload.PROMPT: "hi"}, + event_id="e1", + parent_id="", + ) + handler.on_event_end(CBEventType.LLM, payload={EventPayload.COMPLETION: completion}, event_id="e1") + + wires = enqueued_wires(tracer) + root = next(w for w in wires if w["name"] == "llm-agent") + assert root["model"] == "gpt-4o-mini" + assert root["input_tokens"] == 7 + assert root["output_tokens"] == 3 + child = next(w for w in wires if w is not root) + assert child["model"] == "gpt-4o-mini" + + +def test_autogen_agent_turns_carry_agent_kind(): + """P2 regression: a message with a named source is an agent turn in the team trajectory - + the step states kind "agent" (crewai task-step precedent) instead of defaulting to llm.""" + import asyncio + from datetime import datetime, timezone + import types + + from agentx.integrations.autogen import AgentXAutoGenObserver + + t0 = datetime(2026, 1, 1, tzinfo=timezone.utc) + t1 = datetime(2026, 1, 1, 0, 0, 1, tzinfo=timezone.utc) + task_echo = types.SimpleNamespace(type="TextMessage", content="Plan it", created_at=t0, models_usage=None) + turn = types.SimpleNamespace( + type="TextMessage", + content="Here is the plan.", + created_at=t1, + source="planner", + models_usage=types.SimpleNamespace(prompt_tokens=5, completion_tokens=2), + ) + task_result = types.SimpleNamespace(messages=[task_echo, turn]) + + class FakeTeam: + async def run(self, task=None, **kwargs): + return task_result + + tracer = make_tracer() + observer = AgentXAutoGenObserver(tracer, name="my-team") + asyncio.run(observer.run(FakeTeam(), task="Plan it")) + + wires = enqueued_wires(tracer) + child = next(w for w in wires if w["name"] == "planner") + assert child["span_kind"] == "agent" diff --git a/tests/test_wire_models.py b/tests/test_wire_models.py new file mode 100644 index 0000000..10e854f --- /dev/null +++ b/tests/test_wire_models.py @@ -0,0 +1,83 @@ +"""Wire-model regressions: Dataset round-tripping its grading config (import_dataset used to +silently drop it) and RunResultRow.response populating from the engine's `output` object.""" + +from typing import Any, Dict + +from agentx.evaluations.models import Dataset, RunResultRow + + +DATASET_WIRE: Dict[str, Any] = { + "_id": "ds-1", + "name": "support quality", + "description": "d", + "numberOfRequests": 2, + "acceptanceCriteria": "acc", + "questions": [{"main_question": {"query": "q0"}}], + "vectorSimilarity": {"enabled": True, "model": "text-embedding-3-small"}, + "jaccardSimilarity": {"enabled": True}, + "bleuScore": {"enabled": True}, + "rougeScore": {"enabled": True}, + "judgePrompt": "Grade strictly.", + "judgeModel": "gpt-5.6-luna", + "sovereigntyIndex": {"enabled": True, "models": ["m1", "m2"]}, + "codeScorers": [{"id": "cs1", "name": "n", "code": "c", "enabled": True}], +} + + +def test_dataset_round_trips_metrics_judge_and_sovereignty_config(): + """P2 regression: extra="ignore" dropped these fields on read, so a typed Dataset fed to + import_dataset produced a copy with no metrics/judge/sovereignty config.""" + ds = Dataset(**DATASET_WIRE) + dumped = ds.model_dump(by_alias=True) + for key in ( + "vectorSimilarity", + "jaccardSimilarity", + "bleuScore", + "rougeScore", + "judgePrompt", + "judgeModel", + "sovereigntyIndex", + "codeScorers", + ): + assert dumped[key] == DATASET_WIRE[key], key + # The hoisted convenience list still works alongside the raw object. + assert ds.sovereignty_models == ["m1", "m2"] + + +def test_import_dataset_copies_the_full_grading_config(): + from agentx.evaluations.datasets import DatasetClient + + class FakeEvalClient: + def create_dataset(self, payload): + self.payload = payload + return payload + + fake = FakeEvalClient() + DatasetClient(fake).import_dataset(Dataset(**DATASET_WIRE), name="copy") + payload = fake.payload + assert payload["name"] == "copy" + assert payload["vectorSimilarity"] == DATASET_WIRE["vectorSimilarity"] + assert payload["jaccardSimilarity"] == DATASET_WIRE["jaccardSimilarity"] + assert payload["bleuScore"] == DATASET_WIRE["bleuScore"] + assert payload["rougeScore"] == DATASET_WIRE["rougeScore"] + assert payload["judgePrompt"] == DATASET_WIRE["judgePrompt"] + assert payload["judgeModel"] == DATASET_WIRE["judgeModel"] + assert payload["sovereigntyIndex"] == DATASET_WIRE["sovereigntyIndex"] + assert payload["codeScorers"] == DATASET_WIRE["codeScorers"] + + +def test_run_result_row_response_populates_from_output_object(): + """P2 regression: the engine sends the agent's answer as an `output` object, so + row.response was permanently None.""" + row = RunResultRow.from_wire({ + "rating": 8, + "output": {"text": "the answer", "tokens": 12}, + "questionIndex": 0, + }) + assert row.response == "the answer" + assert row.rating == 8 + + +def test_run_result_row_response_still_accepts_plain_string(): + row = RunResultRow.from_wire({"response": "plain answer"}) + assert row.response == "plain answer"