From 7607db6239b8f456c67659cab571b94539c4c893 Mon Sep 17 00:00:00 2001 From: Robin Date: Wed, 9 Sep 2026 23:55:54 -0700 Subject: [PATCH] full bug sweep fix --- EVALUATIONS.md | 4 +- TRACING.md | 17 ++++++- agentx/__init__.py | 9 ++-- agentx/evaluations/client.py | 40 +++++++++++----- agentx/evaluations/datasets.py | 16 ++++++- agentx/evaluations/reporting.py | 6 ++- agentx/evaluations/runner.py | 8 ++++ agentx/integrations/anthropic.py | 2 +- agentx/integrations/autogen.py | 3 ++ agentx/integrations/crewai.py | 21 ++++++-- agentx/integrations/langchain.py | 55 +++++++++++++-------- agentx/integrations/llamaindex.py | 48 +++++++++++++++++-- agentx/integrations/openai_agents.py | 15 +++++- agentx/monitor/client.py | 21 ++++---- agentx/monitor/scorer_groups.py | 3 +- agentx/monitor/scorers.py | 11 ++++- agentx/monitor/sessions.py | 18 ++++++- agentx/tracing/ci_types.py | 28 ++++++----- agentx/tracing/ingest_client.py | 69 ++++++++++++++++++-------- agentx/tracing/tracer.py | 72 ++++++++++++++++++++++------ tests/test_span_tree.py | 41 ++++++++++++++++ 21 files changed, 392 insertions(+), 115 deletions(-) diff --git a/EVALUATIONS.md b/EVALUATIONS.md index 057aac2..81d9ab9 100644 --- a/EVALUATIONS.md +++ b/EVALUATIONS.md @@ -470,7 +470,7 @@ client.evaluations.run( ).execute(my_agent_fn) ``` -or on a live trace from real production traffic, scored continuously by a self-host Online Evaluator: +or on a live trace from real production traffic, scored continuously by a judge scorer's online profile (self-host): ```python with client.tracer.trace("support-agent", metadata={"promptName": prompt.name}) as span: @@ -480,7 +480,7 @@ with client.tracer.trace("support-agent", metadata={"promptName": prompt.name}) From the self-host dashboard: Governance > Manage > **Prompts** > a prompt's row menu > **Suggest improvement**. It merges both kinds of evidence - deliberate eval runs (defaulting to the *current published version only*, auto-widening to every version if there isn't enough recent evidence yet) -and worst-scoring Online Evaluator ratings from a recent time window - feeds the worst-rated +and the worst-scoring ratings a judge scorer's online profile produced in a recent time window - feeds the worst-rated examples to a judge, and shows a full rewrite plus reasoning. **Nothing is saved until a human approves it as a new version.** The same propose loop is scriptable: `prompts.examples(prompt.id)` returns the evidence, `prompts.propose(prompt.id)` asks the judge for a rewrite (returns diff --git a/TRACING.md b/TRACING.md index 0f37af2..64e6a74 100644 --- a/TRACING.md +++ b/TRACING.md @@ -112,7 +112,7 @@ All parameters work in both decorator and context-manager form - the decorator f | `monitor` | `bool` | - | `True` checks this trace against Monitor patterns immediately; `False` opts out of every ingest-time check. Default (`None`) leaves the server's standard behavior. See [Monitor](#monitor) | | `pattern_ids` | `list[str]` | - | With `monitor=True`: restrict detection to exactly these pattern ids | | `agent_id` | `str` | - | Pin this trace to a known agent id instead of resolving by `name` - a disambiguator for when the name alone isn't enough | -| `span_kind` | `str` | - | What kind of step this span is (`"agent"`, `"llm"`, `"tool"`, `"retrieval"`, ...), stated instead of left to the backend's classification fallback | +| `span_kind` | `str` | - | What kind of step this span is (`"agent"`, `"llm"`, `"tool"`, `"retrieval"`, `"memory"`, ...), stated instead of left to the backend's classification fallback | ### `_TraceSpan` methods and attributes (context manager form) @@ -369,6 +369,19 @@ with tracer.trace("rag-agent") as span: `tracer.record_retrieval(name, query=..., output=..., duration_ms=...)` is the after-the-fact form. Custom names like `"kb_search"` work - the span carries an explicit retrieval marker, not a name heuristic. +### Memory operations + +The memory twins mark a long-term-memory operation (a Mem0/Zep/Letta-style recall or store) as a `span_kind="memory"` child span of the active span. Memory is deliberately NOT retrieval: retrieval spans feed the RAG judges' `{context}` (knowledge grounding), while memory is recalled state. + +```python +with tracer.trace("support-agent") as span: + with tracer.trace_memory("user prefs", operation="read", query=user_id) as m: + m.output = memory.search(user_id, question) + span.output = answer +``` + +`tracer.record_memory(name, operation=..., query=..., output=..., duration_ms=...)` is the after-the-fact form. `operation` is free text - conventionally `"read"` or `"write"` - carried in the span's metadata, while the kind itself stays one value so dashboards and scorers can select all memory activity at once. With no active span, both forms queue the record and merge it into the next trace this tracer sends (the patched-client flow where the memory op runs just before a standalone completions call) instead of silently dropping it. + --- ## Session grouping @@ -672,6 +685,6 @@ Constructing the client makes no network call; `client.ping()` is the fail-fast ## Delivery behavior and limits - **Queueing** - traces are enqueued (up to 500 in flight) and drained by a background daemon thread. On overflow, or when retries are exhausted, the trace is dropped **with a logged warning** (first drop, then every 50th, with a cumulative count) - never silently. -- **Retries** - each queued trace is retried up to 3 times with backoff on connection errors, 429, and 5xx responses; a 429's `Retry-After` header is honored. `sync=True` sends block once with a 10s timeout and do not retry - a failed sync send just means `span.trace_id` stays `None`. +- **Retries** - each queued trace walks the full backoff schedule (up to 3 retries after the first attempt) on connection errors, 429, and 5xx responses alike; a 429's `Retry-After` header is honored in place of the schedule's next wait. `sync=True` sends block with a 10s timeout and retry only briefly - up to 2 bounded retries on 429/503, honoring `Retry-After` (capped at 5s per wait); span ids make redelivery idempotent server-side. A sync send that still fails means the trace was **not stored** (nothing is persisted locally or retried in the background), so `span.trace_id` stays `None`. - **Payload truncation** - `input`, `output`, and `metadata` are serialized best-effort before sending: nesting deeper than 4 levels, dicts/lists beyond 30 entries, and unserializable objects are truncated/stringified (long fallback strings cut to 200 chars) to keep payloads bounded. - **First failure warns** - the first delivery failure per client logs at WARNING with a hint (bad key vs. bad URL); repeats log at DEBUG. `client.ping()` at startup fails fast instead. diff --git a/agentx/__init__.py b/agentx/__init__.py index 166bc5e..7c4a183 100644 --- a/agentx/__init__.py +++ b/agentx/__init__.py @@ -12,11 +12,10 @@ CIGateFailure, ) -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", - datefmt="%Y-%m-%d %H:%M:%S %Z", -) +# Library logging hygiene: a library must never call logging.basicConfig - it hijacks the +# host application's root logger (format AND level) and turns the app's own later basicConfig +# into a no-op. Consumers opt into our logs with logging.getLogger("agentx").setLevel(...). +logging.getLogger("agentx").addHandler(logging.NullHandler()) __all__ = [ "AgentX", diff --git a/agentx/evaluations/client.py b/agentx/evaluations/client.py index 6081fa4..03f1a74 100644 --- a/agentx/evaluations/client.py +++ b/agentx/evaluations/client.py @@ -348,7 +348,9 @@ def init_run( payload["scorerGroupId"] = scorer_group_id if split: payload["split"] = split - data = self._request("POST", "/runs", json=self._with_workspace(payload)) + # Server-side write: a timeout after the run row was created would be + # retried into a duplicate run, so no transport retry. + data = self._request("POST", "/runs", json=self._with_workspace(payload), retry=False) return EvaluationRun(**data) def append_results( @@ -432,8 +434,12 @@ def analyze_run( if not self._analysis_on_dashboard_router: try: + # The self-host route runs the analysis SYNCHRONOUSLY (engine + # routes/evaluations.ts) - a short timeout with retries re-billed the whole + # multi-judge analysis up to 4x while the first was still running. Full + # analysis timeout, no transport retry. return self._request( - "POST", f"/runs/{run_id}/analyze", json=payload, timeout=30 + "POST", f"/runs/{run_id}/analyze", json=payload, timeout=1800, retry=False ) except AgentXEvaluationsError as exc: if not self._note_missing_analysis_route(exc, "analyze"): @@ -554,13 +560,17 @@ def _report_from_dashboard(self, run_id: str) -> Report: if isinstance(dataset_id, dict): # populated reference, not a bare id dataset_id = dataset_id.get("_id") or dataset_id.get("id") - return Report( - runId=run_id, - datasetId=dataset_id or "", - status=envelope.get("status") or "completed", - statistics=envelope.get("statistics"), + # Built as one merged dict (explicit keys last, so they win) - passing the + # explicit keys as keyword arguments alongside **body raises "got multiple + # values" whenever the analysis body itself carries runId/datasetId/status/ + # statistics. + return Report(**{ **body, - ) + "runId": run_id, + "datasetId": dataset_id or "", + "status": envelope.get("status") or "completed", + "statistics": envelope.get("statistics"), + }) # ------------------------------------------------------------------ # Prompt improvement loop (examples -> propose -> publish). These ride the engine's @@ -589,8 +599,11 @@ def publish_prompt_version( payload["reasoning"] = reasoning if based_on_version is not None: payload["basedOnVersion"] = based_on_version + # Server-side write: a timeout after the version was stored would be + # retried into a duplicate version, so no transport retry. return self._request( - "POST", f"/evaluate/prompts/{prompt_id}/versions", base=self._api_root, json=payload + "POST", f"/evaluate/prompts/{prompt_id}/versions", base=self._api_root, json=payload, + retry=False, ) # ------------------------------------------------------------------ @@ -628,7 +641,7 @@ def compare_pairwise( payload["judgeModel"] = judge_model if both_orders: payload["bothOrders"] = True - response = self._request("POST", "/evaluate/runs/pairwise", json=payload, base=self._api_root) + response = self._request("POST", "/evaluate/runs/pairwise", json=payload, base=self._api_root, timeout=900, retry=False,) return PairwiseComparison(**response["comparison"]) def get_pairwise(self, batch_id: str) -> PairwiseComparison: @@ -662,7 +675,8 @@ def create_tool_schema(self, *, name: str, definition: str, description: Optiona payload: dict = {"name": name, "definition": definition} if description is not None: payload["description"] = description - return self._request("POST", "/evaluate/tool-schemas", base=self._api_root, json=payload) + # Server-side write - no transport retry (see init_run's comment). + return self._request("POST", "/evaluate/tool-schemas", base=self._api_root, json=payload, retry=False) def get_tool_schema_examples(self, tool_schema_id: str, window: Optional[str] = None) -> dict: params = {"window": window} if window else None @@ -686,8 +700,10 @@ def publish_tool_schema_version( payload["reasoning"] = reasoning if based_on_version is not None: payload["basedOnVersion"] = based_on_version + # Server-side write - no transport retry (see init_run's comment). return self._request( - "POST", f"/evaluate/tool-schemas/{tool_schema_id}/versions", base=self._api_root, json=payload + "POST", f"/evaluate/tool-schemas/{tool_schema_id}/versions", base=self._api_root, json=payload, + retry=False, ) # ------------------------------------------------------------------ diff --git a/agentx/evaluations/datasets.py b/agentx/evaluations/datasets.py index d650700..20c0fcd 100644 --- a/agentx/evaluations/datasets.py +++ b/agentx/evaluations/datasets.py @@ -16,7 +16,13 @@ class DatasetBuilder: - """Fluent builder for creating a Custom Agent Evaluations dataset.""" + """Fluent builder for creating a Custom Agent Evaluations dataset. + + ``judge_prompt``/``judge_model`` are LLM-as-judge overrides for the dataset's grading + config. NOTE (self-host): the engine's dataset-create route currently ignores both - + set them on a judge scorer / the evaluation settings instead. ``sovereignty_models`` + is accepted on the wire but not acted on by the self-host engine. + """ def __init__( self, @@ -50,6 +56,8 @@ def __init__( # LLM-as-judge overrides for this dataset's own grading config. Omit either to keep the # server default (raw prompt template / gpt-5.6-luna, see EVALUATIONS.md). judge_model # must be one of client.evaluations.list_models() (OpenAI or Anthropic). + # Self-host: the dataset-create route currently IGNORES judgePrompt/judgeModel - set + # them on a judge scorer / the evaluation settings instead (see class docstring). if judge_prompt is not None: self._payload["judgePrompt"] = judge_prompt if judge_model is not None: @@ -84,7 +92,8 @@ def __init__( if rouge_score: self._payload["rougeScore"] = {"enabled": True} # Sovereignty & Portability - the models to compare on this dataset (use - # client.evaluations.list_models() to discover valid ids). + # client.evaluations.list_models() to discover valid ids). Self-host: accepted on + # the wire but not acted on by the engine (see class docstring). if sovereignty_models: self._payload["sovereigntyIndex"] = { "enabled": True, @@ -354,10 +363,13 @@ def import_dataset(self, source: Any, name: Optional[str] = None) -> Dataset: "acceptanceCriteria", "rejectionCriteria", "evaluationCriteria", + "judgePrompt", + "judgeModel", "vectorSimilarity", "jaccardSimilarity", "bleuScore", "rougeScore", + "sovereigntyIndex", "codeScorers", ): if wire.get(key) is not None: diff --git a/agentx/evaluations/reporting.py b/agentx/evaluations/reporting.py index 8a0959a..cb49e46 100644 --- a/agentx/evaluations/reporting.py +++ b/agentx/evaluations/reporting.py @@ -1,5 +1,7 @@ from __future__ import annotations +from typing import Optional + from agentx.evaluations.models import Report from agentx.evaluations._term import ( bold, @@ -21,14 +23,14 @@ _PRI_COLORS = {"high": red, "medium": yellow, "low": dim} -def _rating_badge(rating: str | None) -> str: +def _rating_badge(rating: Optional[str]) -> str: icon = _RATING_ICONS.get(rating or "", "·") color = _RATING_COLORS.get(rating or "", dim) label = (rating or "").upper() return color(f"{icon} {label}") if label else dim(icon) -def _section(title: str, rating: str | None = None) -> None: +def _section(title: str, rating: Optional[str] = None) -> None: badge = f" {_rating_badge(rating)}" if rating else "" print(f"\n{bold(title)}{badge}") print(dim(_THIN)) diff --git a/agentx/evaluations/runner.py b/agentx/evaluations/runner.py index 26d2a23..ce3d6c8 100644 --- a/agentx/evaluations/runner.py +++ b/agentx/evaluations/runner.py @@ -3,6 +3,8 @@ import logging import os import time + +import requests import uuid from typing import Any, Callable, Dict, Iterator, List, Optional, Set, Union @@ -304,6 +306,12 @@ def _flush_batch(self, batch: List[EvaluationResult]) -> None: resp.failed_validation, ) return + except requests.Timeout as exc: + # A read timeout means the engine may STILL be scoring this batch - a + # retry re-POSTs it and double-bills every judge call (idempotency keys + # protect rows already inserted, not judge work mid-flight). Fail loud. + last_exc = exc + break except Exception as exc: last_exc = exc if attempt == 1: diff --git a/agentx/integrations/anthropic.py b/agentx/integrations/anthropic.py index 2d54ea2..002cd2f 100644 --- a/agentx/integrations/anthropic.py +++ b/agentx/integrations/anthropic.py @@ -274,7 +274,7 @@ def __iter__(self_inner): return iter(ctx) def __aiter__(self_inner): - return aiter(ctx) + return ctx.__aiter__() # aiter() builtin is 3.10+; python_requires is >=3.9 def __getattr__(self_inner, item): return getattr(ctx, item) diff --git a/agentx/integrations/autogen.py b/agentx/integrations/autogen.py index b0d4f2a..68bf233 100644 --- a/agentx/integrations/autogen.py +++ b/agentx/integrations/autogen.py @@ -167,6 +167,9 @@ def _summarize_messages(self, messages: List[Any], run_start: float) -> tuple: "end_time": end_t, "input": pending["input"] if pending else None, "output": f"ERROR: {output}" if is_error else (str(output) if output is not None else None), + # The engine's failure test is success === false; without + # this a failed tool call would read as passing. + "success": not is_error, }) continue diff --git a/agentx/integrations/crewai.py b/agentx/integrations/crewai.py index 7ca40b5..1f3b7b5 100644 --- a/agentx/integrations/crewai.py +++ b/agentx/integrations/crewai.py @@ -127,6 +127,16 @@ def _start_task_timing_capture(self): except ImportError: 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 + def on_task_started(source: Any, event: Any) -> None: task_id = getattr(event, "task_id", None) if task_id is None: @@ -154,9 +164,14 @@ def on_task_failed(source: Any, event: Any) -> None: crewai_event_bus.on(TaskFailedEvent)(on_task_failed) def unregister() -> None: - crewai_event_bus.off(TaskStartedEvent, on_task_started) - crewai_event_bus.off(TaskCompletedEvent, on_task_completed) - crewai_event_bus.off(TaskFailedEvent, on_task_failed) + 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 return task_timings, unregister diff --git a/agentx/integrations/langchain.py b/agentx/integrations/langchain.py index c8fce1f..7d6c353 100644 --- a/agentx/integrations/langchain.py +++ b/agentx/integrations/langchain.py @@ -339,6 +339,16 @@ def _prune_stale_entries(self) -> None: 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. + with self._state_lock: + if self._pending_retrieval_steps: + self._pending_retrieval_steps[:] = [ + step for step in self._pending_retrieval_steps if step.get("start_time", 0) >= cutoff + ] + # ------------------------------------------------------------------ # Chain lifecycle # ------------------------------------------------------------------ @@ -359,8 +369,9 @@ def on_chain_start( self._prune_stale_entries() # Consume any retrieval steps that ran before this chain started # (pre-run RAG: retriever.invoke() called before agent.invoke()) - pending = self._pending_retrieval_steps[:] - self._pending_retrieval_steps.clear() + 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), @@ -670,7 +681,7 @@ def _record_llm_start( "input": _extract_llm_input(prompts=prompts, messages=messages), } top = self._find_top_ancestor(parent_run_id) - if top and not self._runs[top].get("model") and model: + if top and top in self._runs and not self._runs[top].get("model") and model: self._runs[top]["model"] = model def on_llm_start( @@ -882,15 +893,16 @@ def on_retriever_end( step["output"] = "\n\n---\n\n".join(contents) top = self._find_top_ancestor(parent_run_id) - if top and top in self._runs: - # Retriever ran inside an active chain - attach directly - retrievals = self._runs[top]["retrieval_steps"] - step["name"] = f"Retrieval {len(retrievals) + 1}" - retrievals.append(step) - else: - # Retriever ran before the chain started (pre-run RAG pattern) - step["name"] = f"Retrieval {len(self._pending_retrieval_steps) + 1}" - self._pending_retrieval_steps.append(step) + with self._state_lock: + if top and top in self._runs: + # Retriever ran inside an active chain - attach directly + retrievals = self._runs[top]["retrieval_steps"] + step["name"] = f"Retrieval {len(retrievals) + 1}" + retrievals.append(step) + else: + # Retriever ran before the chain started (pre-run RAG pattern) + step["name"] = f"Retrieval {len(self._pending_retrieval_steps) + 1}" + self._pending_retrieval_steps.append(step) def on_retriever_error( self, @@ -918,15 +930,16 @@ def on_retriever_error( step["query"] = query top = self._find_top_ancestor(parent_run_id) - if top and top in self._runs: - # Retriever ran inside an active chain - attach directly - retrievals = self._runs[top]["retrieval_steps"] - step["name"] = f"Retrieval {len(retrievals) + 1}" - retrievals.append(step) - else: - # Retriever ran before the chain started (pre-run RAG pattern) - step["name"] = f"Retrieval {len(self._pending_retrieval_steps) + 1}" - self._pending_retrieval_steps.append(step) + with self._state_lock: + if top and top in self._runs: + # Retriever ran inside an active chain - attach directly + retrievals = self._runs[top]["retrieval_steps"] + step["name"] = f"Retrieval {len(retrievals) + 1}" + retrievals.append(step) + else: + # Retriever ran before the chain started (pre-run RAG pattern) + step["name"] = f"Retrieval {len(self._pending_retrieval_steps) + 1}" + self._pending_retrieval_steps.append(step) # ------------------------------------------------------------------ # Helpers diff --git a/agentx/integrations/llamaindex.py b/agentx/integrations/llamaindex.py index 2efc579..5422585 100644 --- a/agentx/integrations/llamaindex.py +++ b/agentx/integrations/llamaindex.py @@ -107,18 +107,51 @@ def __init__( name: str = "llamaindex-agent", metadata: Optional[Dict[str, Any]] = None, session_id: Optional[str] = None, + max_run_age_seconds: float = 900.0, ) -> None: super().__init__(event_starts_to_ignore=[], event_ends_to_ignore=[]) self._tracer = tracer self._name = name self._metadata = metadata self._session_id = session_id + # Safety net mirroring langchain.py's _prune_stale_entries: state is + # normally popped in on_event_end, but an event whose end callback never + # fires (hard crash, integration bug) would leak forever in this + # long-lived singleton handler. Entries older than this are swept out + # at the top of on_event_start. + self._max_run_age_seconds = max_run_age_seconds self._parents: Dict[str, Optional[str]] = {} self._roots: Dict[str, bool] = {} self._runs: Dict[str, Dict[str, Any]] = {} self._starts: Dict[str, Dict[str, Any]] = {} + 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) + # ------------------------------------------------------------------ # BaseCallbackHandler protocol # ------------------------------------------------------------------ @@ -138,6 +171,7 @@ def on_event_start( **kwargs: Any, ) -> str: payload = payload or {} + self._prune_stale_entries() self._parents[event_id] = parent_id root_id = self._find_root(parent_id) @@ -251,11 +285,16 @@ def on_event_end( tool_output = payload.get(EventPayload.FUNCTION_OUTPUT) state["tool_call_steps"].append({ "name": tool_name, - "duration_ms": (end_t - start_t) * 1000, + # tracer._merge_child_run's tool_calls loop reads "latency_ms" + # (not "duration_ms" like execution/retrieval steps). + "latency_ms": int((end_t - start_t) * 1000), "start_time": start_t, "end_time": end_t, "input": _safe_serialize(tool_input) if tool_input is not None else None, "output": f"ERROR: {exception}" if exception else (str(tool_output) if tool_output is not None else None), + # The engine's failure test is success === false; without this a + # failed tool call would read as passing. + "success": exception is None, }) if is_root: @@ -277,10 +316,9 @@ def _find_root(self, parent_id: Optional[str]) -> Optional[str]: return None def _send_trace(self, state: Dict[str, Any]) -> None: - # tool_call_steps entries carry start_time/end_time (unlike langchain.py's leaner - # wire-shaped tool_calls list) - _merge_child_run's tool_calls loop falls back to - # computing duration from those when no explicit latency_ms is present, so each tool - # call still positions correctly in the tree panel instead of defaulting to offset 0. + # tool_call_steps entries carry latency_ms AND start_time/end_time - _merge_child_run's + # tool_calls loop reads latency_ms for duration and the timestamps for position, so each + # tool call lands correctly in the tree panel instead of defaulting to offset 0. with self._tracer.trace( self._name, metadata=self._metadata, session_id=self._session_id, framework="llamaindex" ) as span: diff --git a/agentx/integrations/openai_agents.py b/agentx/integrations/openai_agents.py index 7e973c5..ae9f22f 100644 --- a/agentx/integrations/openai_agents.py +++ b/agentx/integrations/openai_agents.py @@ -16,6 +16,8 @@ from __future__ import annotations from datetime import datetime, timezone +import time +from uuid import uuid4 from typing import Any, Dict, List, Optional from agentx.tracing.tracer import Tracer, _safe_serialize @@ -145,7 +147,15 @@ def on_trace_start(self, trace: Any) -> None: metadata=self._metadata, session_id=self._session_id, ) - root_span.__enter__() + # Deliberately NOT root_span.__enter__(): enter pushes onto the CALLING thread's + # active-span stack, but the Agents SDK fires on_trace_end on whatever thread it + # likes - the pop then no-ops there, the entry never drains, and every later + # unrelated trace on this thread is mis-filed as a child of this dead run (and + # inherits its session). Start time and session are set by hand instead; on_span_end + # already parents via child_span() on this exact 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._spans[trace_id] = { "root_span": root_span, "llm_call_count": 0, @@ -173,6 +183,9 @@ def on_trace_end(self, trace: Any) -> None: root_span._output_tokens = state["output_tokens"] if state.get("error"): root_span.set_error(state["error"]) + # Close WITHOUT touching the thread-local stack (see on_trace_start). __exit__'s only + # stack interaction is the pop, which is a no-op for a never-pushed span - safe to call + # directly for its send/flush behavior. root_span.__exit__(None, None, None) def on_span_start(self, span: Any) -> None: diff --git a/agentx/monitor/client.py b/agentx/monitor/client.py index 96a4ffa..5cda32c 100644 --- a/agentx/monitor/client.py +++ b/agentx/monitor/client.py @@ -28,7 +28,12 @@ class AgentXMonitorError(Exception): - pass + """``status_code`` carries the HTTP status when the error came from a server + response; it is ``None`` for transport-level failures and retry exhaustion.""" + + def __init__(self, message: str, status_code: Optional[int] = None) -> None: + super().__init__(message) + self.status_code = status_code class AgentXAuthError(AgentXMonitorError): @@ -208,17 +213,17 @@ 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) + raise AgentXValidationError(resp.text, status_code=422) if retry and resp.status_code in _RETRYABLE_STATUS and attempt < _MAX_RETRIES - 1: logger.debug( "Retryable status %d (attempt %d)", resp.status_code, attempt + 1 ) - last_exc = AgentXMonitorError(f"HTTP {resp.status_code}") + last_exc = AgentXMonitorError(f"HTTP {resp.status_code}", status_code=resp.status_code) continue if not resp.ok: - raise AgentXMonitorError(f"HTTP {resp.status_code}: {resp.text}") + raise AgentXMonitorError(f"HTTP {resp.status_code}: {resp.text}", status_code=resp.status_code) try: return resp.json() except Exception: @@ -481,7 +486,7 @@ def get_online_evaluator_calibration(self, evaluator_id: str, window: str = "7d" def propose_online_evaluator_tuning(self, evaluator_id: str, window: str = "7d") -> dict: data = self._request( "POST", f"/agent-monitoring/online-evaluators/{evaluator_id}/tune", - base=self._api_root(), json={"window": window}, timeout=300, + base=self._api_root(), json={"window": window}, timeout=300, retry=False, ) return data.get("proposal", data) if isinstance(data, dict) else data @@ -490,7 +495,7 @@ def validate_online_evaluator_tuning( ) -> dict: return self._request( "POST", f"/agent-monitoring/online-evaluators/{evaluator_id}/tune/validate", - base=self._api_root(), json={**criteria, "window": window}, timeout=600, + base=self._api_root(), json={**criteria, "window": window}, timeout=600, retry=False, ) def publish_online_evaluator_tuning( @@ -505,7 +510,7 @@ def publish_online_evaluator_tuning( payload["force"] = True return self._request( "POST", f"/agent-monitoring/online-evaluators/{evaluator_id}/tune/publish", - base=self._api_root(), json=payload, timeout=60, + base=self._api_root(), json=payload, timeout=60, retry=False, ) def update_profile(self, agent_id: str, payload: dict) -> MonitorProfile: diff --git a/agentx/monitor/scorer_groups.py b/agentx/monitor/scorer_groups.py index a712455..b832c50 100644 --- a/agentx/monitor/scorer_groups.py +++ b/agentx/monitor/scorer_groups.py @@ -49,7 +49,8 @@ def _request(self, method: str, url: str, json: Optional[Dict[str, Any]] = None) ) if response.status_code >= 400: raise AgentXScorerGroupsError(f"HTTP {response.status_code}: {response.text}") - return response.json() + # DELETE (and any other empty 2xx) has no body - .json() on it raises. + return response.json() if response.text else {} def list(self) -> List[ScorerGroup]: return [ScorerGroup(g) for g in self._request("GET", self._base).get("scorerGroups", [])] diff --git a/agentx/monitor/scorers.py b/agentx/monitor/scorers.py index a3293b0..d61f0a9 100644 --- a/agentx/monitor/scorers.py +++ b/agentx/monitor/scorers.py @@ -65,11 +65,18 @@ def templates(self) -> List[Dict[str, Any]]: return [p for p in patterns if p.get("source") == "builtIn"] def _enabled_template_keys(self) -> List[str]: - return [p["key"] for p in self.templates() if p.get("enabled")] + # .get("key"): defensive against a template row missing its key (the wire owns this + # shape, not the SDK) - a keyless row is skipped rather than KeyError-ing the sweep. + return [p.get("key") for p in self.templates() if p.get("enabled") and p.get("key")] def enable(self, keys: Sequence[str]) -> List[str]: """Enable template scorers by key (e.g. ``["pii-in-response"]``), preserving what is - already on. Returns the resulting enabled-key list.""" + already on. Returns the resulting enabled-key list. + + Note: enable()/disable() are a read-modify-write over the project's single + enabledBuiltinPatterns list - two concurrent callers (or a dashboard edit racing an + SDK call) can lose one side's change. There is no engine-side merge; serialize + catalog edits if that matters.""" merged = sorted(set(self._enabled_template_keys()) | set(keys)) self._request("PUT", "/settings/monitoring-defaults", json={"enabledBuiltinPatterns": merged}) return merged diff --git a/agentx/monitor/sessions.py b/agentx/monitor/sessions.py index 562c0b4..acfde9c 100644 --- a/agentx/monitor/sessions.py +++ b/agentx/monitor/sessions.py @@ -27,8 +27,24 @@ def scores(self, session_id: str) -> List[dict]: Baseline Judge existed - branch defensively on unknown kinds.""" return self._client.list_session_scores(session_id) + def judge(self, session_id: str, evaluator_id: str, *, if_stale: bool = False) -> dict: + """Judge one session with one session-scoped evaluator, now (one judge call). + ``if_stale=True`` skips re-judging a session that was already scored since its + last activity - the engine then answers ``{"skipped": True}`` without spending + another judge call. Returns the score row (or that skip marker).""" + data = self._client._request( + "POST", + f"/agent-monitoring/sessions/{session_id}/judge/{evaluator_id}", + base=self._client._api_root(), + timeout=120, + retry=False, + params={"ifStale": "true"} if if_stale else None, + ) + return data.get("score", data) if isinstance(data, dict) else data + def run_sweep(self) -> dict: """Trigger the idle-session sweep once (normally automatic, every minute) - scores idle multi-turn sessions with every enabled session-scoped evaluator and scorer - group. Returns ``{"judged": n}``.""" + group. Returns ``{"judged": n}``; the response may instead carry ``skipped: true`` + when another sweep is already in flight.""" return self._client.run_session_sweep() diff --git a/agentx/tracing/ci_types.py b/agentx/tracing/ci_types.py index c675962..92c471a 100644 --- a/agentx/tracing/ci_types.py +++ b/agentx/tracing/ci_types.py @@ -1,14 +1,20 @@ -"""Dataclasses for CI/CD evaluation run responses.""" +"""Dataclasses for CI/CD evaluation run responses. + +Annotations deliberately use typing.Optional/List (not PEP 604/585 syntax): +``from __future__ import annotations`` only defers evaluation, and +``typing.get_type_hints`` on these classes still has to resolve the strings +at runtime, which fails for ``str | None`` / ``list[...]`` on Python 3.9. +""" from __future__ import annotations from dataclasses import dataclass, field -from typing import Any, Literal +from typing import Any, Dict, List, Literal, Optional @dataclass class CITestCase: index: int - query: str | None = None # None when ci.exposeTestInputs is false + query: Optional[str] = None # None when ci.exposeTestInputs is false @dataclass @@ -16,7 +22,7 @@ class CIRun: run_id: str dataset_id: str total_questions: int - test_cases: list[CITestCase] + test_cases: List[CITestCase] expires_at: str @@ -47,20 +53,20 @@ class CIRunResult: pass_rate: float total_questions: int passed_questions: int - scores: list[CIQuestionScore] = field(default_factory=list) - violations: list[ThresholdViolation] = field(default_factory=list) - git_context: dict | None = None - finalized_at: str | None = None + scores: List[CIQuestionScore] = field(default_factory=list) + violations: List[ThresholdViolation] = field(default_factory=list) + git_context: Optional[Dict[str, Any]] = None + finalized_at: Optional[str] = None @dataclass class CIRunStatus: run_id: str status: Literal["in_progress", "completed", "failed"] - gate: Literal["pass", "fail"] | None + gate: Optional[Literal["pass", "fail"]] results_submitted: int total_questions: int created_at: str expires_at: str - finalized_at: str | None = None - git_context: dict | None = None + finalized_at: Optional[str] = None + git_context: Optional[Dict[str, Any]] = None diff --git a/agentx/tracing/ingest_client.py b/agentx/tracing/ingest_client.py index 8568064..a0d82e0 100644 --- a/agentx/tracing/ingest_client.py +++ b/agentx/tracing/ingest_client.py @@ -138,27 +138,42 @@ def send_trace_sync(self, payload: Dict[str, Any]) -> Optional[str]: """ Send a trace payload synchronously and return the ingested trace's id, or ``None`` on failure. Used by ``Tracer.trace(..., sync=True)`` when the caller needs the trace_id back - immediately (e.g. to attach it to an evaluation result) - unlike ``enqueue()``, this blocks - and does not retry, trading the tracer's usual fire-and-forget guarantee for a same-call - result. Never raises; a failed send just means no trace_id (never blocks the caller's eval - run over a tracing hiccup). + immediately (e.g. to attach it to an evaluation result) - unlike ``enqueue()``, this + blocks, trading the tracer's usual fire-and-forget guarantee for a same-call result. + A 429/503 (engine shedding load or briefly unavailable) is retried up to 2 times, + honoring the server's Retry-After (capped at 5s per wait) - span ids make redelivery + idempotent server-side, so a retry can never double-ingest. Never raises; ``None`` + means the trace was NOT stored (there is no local persistence or background retry + beyond those brief attempts), so no trace_id exists for it. """ if self._workspace_id: payload = {**payload, "workspaceId": self._workspace_id} - try: - resp = self._session.post(self._endpoint, json=payload, timeout=10) - except requests.RequestException as exc: - self._warn_delivery(f"{exc.__class__.__name__}: {exc}") - logger.debug("agentx ingest sync send error: %s", exc) - return None - if not resp.ok: - self._warn_delivery(f"HTTP {resp.status_code}", status=resp.status_code) - logger.debug("agentx ingest sync HTTP %d: %s", resp.status_code, resp.text[:200]) - return None - try: - return resp.json().get("trace_id") - except Exception: - return None + for attempt in range(3): # 1 try + up to 2 bounded retries on 429/503 + try: + resp = self._session.post(self._endpoint, json=payload, timeout=10) + except requests.RequestException as exc: + self._warn_delivery(f"{exc.__class__.__name__}: {exc}") + logger.debug("agentx ingest sync send error: %s", exc) + return None + if resp.status_code in (429, 503) and attempt < 2: + retry_after = resp.headers.get("Retry-After") + wait = 1.0 + if retry_after: + try: + wait = min(5.0, float(retry_after)) + except ValueError: + pass + time.sleep(wait) + continue + if not resp.ok: + self._warn_delivery(f"HTTP {resp.status_code}", status=resp.status_code) + logger.debug("agentx ingest sync HTTP %d: %s", resp.status_code, resp.text[:200]) + return None + try: + return resp.json().get("trace_id") + except Exception: + return None + return None # pragma: no cover - loop always returns def send_trace_sync_detailed(self, payload: Dict[str, Any]) -> Optional[Dict[str, Any]]: """``send_trace_sync`` returning the full response body instead of just the id - the @@ -418,16 +433,22 @@ def _drain(self) -> None: def _send(self, payload: Dict[str, Any]) -> None: last_exc: Optional[Exception] = None - for attempt, wait in enumerate([0.0] + _RETRY_BACKOFF): - if wait: + schedule = [0.0] + _RETRY_BACKOFF + skip_next_wait = False + for attempt, wait in enumerate(schedule): + if wait and not skip_next_wait: time.sleep(wait) + skip_next_wait = False try: resp = self._session.post(self._endpoint, json=payload, timeout=10) except requests.RequestException as exc: last_exc = exc continue - if 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 backoff entry unreachable for HTTP retries). + if resp.status_code in _RETRYABLE_STATUS and attempt < len(schedule) - 1: # 429 = the engine's bounded ingest queue shedding load (its ADR-0005): honor # Retry-After exactly instead of the generic backoff schedule, so the SDK backs # off in step with the server's own flush cadence. @@ -435,6 +456,9 @@ def _send(self, payload: Dict[str, Any]) -> None: if resp.status_code == 429 and retry_after: try: time.sleep(min(30.0, float(retry_after))) + # Retry-After REPLACES the schedule's next wait - sleeping both would + # back off longer than either the server or the schedule asked for. + skip_next_wait = True except ValueError: pass last_exc = Exception(f"HTTP {resp.status_code}") @@ -442,6 +466,9 @@ def _send(self, payload: Dict[str, Any]) -> None: if not resp.ok: self._warn_delivery(f"HTTP {resp.status_code}", status=resp.status_code) logger.debug("agentx ingest HTTP %d: %s", resp.status_code, resp.text[:200]) + # Non-retryable failure still cost us this payload - never drop silently + # (same rule _drain and the retries-exhausted path below follow). + self._record_drop(f"HTTP {resp.status_code}") return return diff --git a/agentx/tracing/tracer.py b/agentx/tracing/tracer.py index b4e389d..321c62a 100644 --- a/agentx/tracing/tracer.py +++ b/agentx/tracing/tracer.py @@ -4,6 +4,7 @@ import concurrent.futures import functools import inspect +import contextvars import threading import time from contextlib import contextmanager @@ -70,7 +71,7 @@ def __init__( model: Optional[str] = None, session_id: Optional[str] = None, sync: bool = False, - monitor: bool = False, + monitor: Optional[bool] = None, pattern_ids: Optional[List[str]] = None, agent_id: Optional[str] = None, span_kind: Optional[str] = None, @@ -104,6 +105,11 @@ def __init__( # __enter__/_merge_child_run/child_span. self._span_id = uuid4().hex self._parent_span_id: Optional[str] = None + # Set on first __enter__ - a re-entered span object regenerates its + # span_id there so each `with span:` re-use sends a fresh identity + # (the server dedupes on span_id, so a reused id would silently + # collapse the second run into the first). + self._entered = False # Numbers auto-named "LLM Call N"/"Retrieval N" child spans - see _merge_child_run. self._child_span_count = 0 # Monitor: True checks this trace against patterns immediately on ingest, no dashboard @@ -145,6 +151,13 @@ def __init__( # ------------------------------------------------------------------ def __enter__(self) -> "_TraceSpan": + if self._entered: + # Re-using one span object for another `with` block: regenerate the + # identity so this run sends its own span row instead of being + # deduped server-side against the first entry's span_id. + self._span_id = uuid4().hex + self._trace_id = None + self._entered = True self._start = time.time() # Resolve real span hierarchy against whatever's currently active on this thread, before # pushing self (so `parent` here is the actual enclosing span, not self). @@ -627,36 +640,39 @@ def __init__(self, ingest_client: IngestClient) -> None: self._client = ingest_client self._pending_tool_calls: List[Dict[str, Any]] = [] self._pending_retrievals: List[Dict[str, Any]] = [] - self._local = threading.local() + # Context-local, not thread-local: two coroutines interleaving on one event loop each + # get their own asyncio task Context, so concurrent `async def` agents no longer + # mis-parent each other's spans (a thread-local stack merged them into one fabricated + # tree). Bare threads keep the old behavior - each starts an empty Context. Stored + # immutably (tuple, copy-on-write) so a child task's pushes never leak into siblings. + self._span_stack_var: "contextvars.ContextVar[tuple]" = contextvars.ContextVar( + f"agentx_span_stack_{id(self)}", default=() + ) # ------------------------------------------------------------------ - # Active-span stack (per thread) - lets auto-instrumented integrations + # Active-span stack (per context) - lets auto-instrumented integrations # (e.g. patch_anthropic_client) detect they're running inside a # `with tracer.trace(...)` block and attach to it as an LLM-call step # instead of sending their own independent trace. # ------------------------------------------------------------------ - def _get_span_stack(self) -> List["_TraceSpan"]: - stack = getattr(self._local, "span_stack", None) - if stack is None: - stack = [] - self._local.span_stack = stack - return stack + def _get_span_stack(self) -> tuple: + return self._span_stack_var.get() def _push_active_span(self, span: "_TraceSpan") -> None: - self._get_span_stack().append(span) + self._span_stack_var.set(self._span_stack_var.get() + (span,)) def _pop_active_span(self, span: "_TraceSpan") -> None: - stack = self._get_span_stack() + stack = self._span_stack_var.get() if stack and stack[-1] is span: - stack.pop() + self._span_stack_var.set(stack[:-1]) elif span in stack: - stack.remove(span) + self._span_stack_var.set(tuple(item for item in stack if item is not span)) @property def current_span(self) -> Optional["_TraceSpan"]: - """The innermost ``with tracer.trace(...)`` span active on this thread, if any.""" - stack = self._get_span_stack() + """The innermost ``with tracer.trace(...)`` span active in this context, if any.""" + stack = self._span_stack_var.get() return stack[-1] if stack else None @contextmanager @@ -861,6 +877,24 @@ def record_memory( """ active_span = self.current_span if active_span is None: + # Same posture as record_retrieval: queue and merge into the next trace this + # tracer sends (the patched-client flow where the memory op runs just before a + # standalone completions call) instead of silently dropping the record. + latency_ms = ( + int(duration_ms) + if duration_ms is not None + else int((end_time - start_time) * 1000) + if start_time is not None and end_time is not None + else None + ) + self._pending_retrievals.append({ + "name": name, + "query": _safe_serialize(query) if query is not None else None, + "output": _safe_serialize(output) if output is not None else None, + "duration_ms": latency_ms, + "kind": "memory", + **({"operation": operation} if operation else {}), + }) return active_span.child_span( name, @@ -886,10 +920,18 @@ def trace_memory( """ start_t = time.time() recorder = _MemoryOpRecorder() + error: Optional[str] = None try: yield recorder + except Exception as exc: + # A memory op that raised must not be recorded as a clean span (trace_tool_call + # precedent) - 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_memory( name, operation=operation, diff --git a/tests/test_span_tree.py b/tests/test_span_tree.py index fc8b424..d127dc2 100644 --- a/tests/test_span_tree.py +++ b/tests/test_span_tree.py @@ -139,6 +139,23 @@ def run(x: int) -> int: assert_no_performance_summary(wires) +def test_reentering_same_span_object_sends_fresh_span_identity(): + """Re-using ONE span object for two sequential `with` blocks must send two distinct + span_ids - the server dedupes on span_id, so a reused identity would silently collapse + the second run into the first.""" + tracer = make_tracer() + span = tracer.trace("agent") + with span: + pass + first_span_id = span.span_id + with span: + pass + wires = enqueued_wires(tracer) + assert len(wires) == 2 + assert wires[0]["span_id"] == first_span_id + assert wires[1]["span_id"] != wires[0]["span_id"] + + # --------------------------------------------------------------------------- # langchain.py — AgentXCallbackHandler opens a real root span per top-level chain invocation (or # folds into an already-active enclosing span) and lets _merge_child_run explode it into real @@ -594,6 +611,30 @@ def test_trace_tool_call_emits_real_child_span(): assert root["tool_calls"][0]["output"] == "digital purchases are final" +def test_concurrent_async_agents_do_not_mis_parent(): + """Two coroutines interleaving on one event loop must stay two independent trees - + the thread-local stack used to merge them (fabricated parent edge, shared session).""" + import asyncio + + tracer = make_tracer() + + @tracer.trace("agent-a") + async def a(): + await asyncio.sleep(0.02) + + @tracer.trace("agent-b") + async def b(): + await asyncio.sleep(0.01) + + async def main(): + await asyncio.gather(a(), b()) + + asyncio.run(main()) + wires = enqueued_wires(tracer) + assert all(w.get("parent_span_id") is None for w in wires) + assert len({w.get("session_id") for w in wires}) == 2 + + def test_trace_memory_emits_a_memory_kind_child_span(): tracer = make_tracer() with tracer.trace("agent") as span: