diff --git a/EVALUATIONS.md b/EVALUATIONS.md index 81d9ab9..77a3f97 100644 --- a/EVALUATIONS.md +++ b/EVALUATIONS.md @@ -387,7 +387,7 @@ except Exception as exc: # the legacy views below work on every engine ``` -The legacy views are not affected - they call the long-standing `/custom-agent-evaluations/evaluation-settings` and `/agent-monitoring/online-evaluators` routes - so they remain the portable choice for code that must run against engines you do not control. This does not affect the `scorer_id` kwarg on `.run()`, which is client-side naming over a field the wire has always had. +The legacy views are not affected - they call the long-standing `/custom-agent-evaluations/evaluation-settings` and `/monitor/online-evaluators` routes (only the calibration/tuning calls ride `/agent-monitoring`) - so they remain the portable choice for code that must run against engines you do not control. This does not affect the `scorer_id` kwarg on `.run()`, which is client-side naming over a field the wire has always had. #### Legacy views @@ -896,7 +896,7 @@ This run + gate flow is the **self-host CI path**. The separate CI-runs API in [ ### AI analysis report -`.analyze()` is the last step in the chain. It runs the same durable, multi-stage pipeline as the dashboard's "Analyze" button: each response is scored by 1-3 LLM judges, then reduced through question- and cluster-level summaries into one final qualitative report, returned as the `Report` object. Because of this, `.analyze()` polls until the job finishes rather than returning instantly, and can take noticeably longer than a single LLM call for larger runs (progress is shown in the terminal while it waits). +`.analyze()` is the last step in the chain. It triggers the same analysis as the dashboard's "Analyze" button: each response is scored by 1-3 LLM judges and reduced into one final qualitative report, returned as the `Report` object. On self-host this is a single synchronous pass - the request returns when the whole analysis is done, so the poll loop sees a terminal status on its first check and the per-level progress percentages never populate. Either way, `.analyze()` can take noticeably longer than a single LLM call for larger runs (a spinner is shown in the terminal while it waits). ```python report = client.evaluations.run(...).execute(my_agent).finalize().analyze( diff --git a/TRACING.md b/TRACING.md index 64e6a74..a481dda 100644 --- a/TRACING.md +++ b/TRACING.md @@ -140,7 +140,7 @@ with tracer.trace("orchestrator") as root: root.output = reply ``` -The active-span stack is **thread-local**. Work submitted to a `ThreadPoolExecutor` (or any other thread) doesn't see a span opened on the calling thread - wrap the worker body in `tracer.use_span(span)` to attach it: +The active-span stack is **context-local** (a `ContextVar`): bare threads start with an empty stack, while asyncio tasks inherit a copy of their creator's. Work submitted to a `ThreadPoolExecutor` (or any other thread) doesn't see a span opened on the calling thread - wrap the worker body in `tracer.use_span(span)` to attach it: ```python with tracer.trace("orchestrator") as span: @@ -380,7 +380,7 @@ with tracer.trace("support-agent") as span: 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. +`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 the record is dropped (with a debug log) - the pending queue rides the next trace's retrieval steps, and memory content must never feed the RAG judges' retrieval context - so wrap the call in `tracer.trace()`. --- diff --git a/agentx/evaluations/client.py b/agentx/evaluations/client.py index 03f1a74..5c88fd3 100644 --- a/agentx/evaluations/client.py +++ b/agentx/evaluations/client.py @@ -207,6 +207,12 @@ def _request( return resp.json() except Exception: return resp.text + # A timeout keeps its type: runner._flush_batch catches requests.Timeout specifically + # (the engine may still be scoring the batch - a blind retry double-bills every judge + # call), and wrapping it in AgentXEvaluationsError here made that guard unreachable. + # Applies to retry=False calls too, where the single attempt lands straight here. + if isinstance(last_exc, requests.Timeout): + raise last_exc raise AgentXEvaluationsError(f"Request failed after retries: {last_exc}") # ------------------------------------------------------------------ @@ -227,7 +233,9 @@ def list_models(self, provider: Optional[str] = None) -> List[ModelInfo]: # ------------------------------------------------------------------ def create_dataset(self, payload: dict) -> Dataset: - data = self._request("POST", "/datasets", json=self._with_workspace(payload)) + # Server-side write: a timeout after the dataset row was created would be + # retried into a duplicate dataset, so no transport retry. + data = self._request("POST", "/datasets", json=self._with_workspace(payload), retry=False) return Dataset(**data) def delete_dataset(self, dataset_id: str) -> None: @@ -255,8 +263,9 @@ def get_dataset(self, dataset_id: str) -> Dataset: # ------------------------------------------------------------------ def create_evaluation_settings(self, payload: dict) -> EvaluationSettings: + # Server-side write - no transport retry (see init_run's comment). data = self._request( - "POST", "/evaluation-settings", json=self._with_workspace(payload) + "POST", "/evaluation-settings", json=self._with_workspace(payload), retry=False ) return EvaluationSettings(**data) @@ -281,12 +290,14 @@ def get_evaluation_settings(self, evaluation_settings_id: str) -> EvaluationSett # ------------------------------------------------------------------ # Prompt registry endpoints - see agentx.evaluations.prompts.PromptClient for the concept - # (the external-agent analog to native autotune). Deliberately read-mostly: no publish here, - # a new version only ever comes from the dashboard's human-approved propose/publish flow. + # (the external-agent analog to native autotune). propose_prompt never publishes; + # publish_prompt_version below IS the explicit approval step - call it only after a human + # reviewed the proposal. # ------------------------------------------------------------------ def create_prompt(self, payload: dict) -> Prompt: - data = self._request("POST", "/prompts", json=self._with_workspace(payload)) + # Server-side write - no transport retry (see init_run's comment). + data = self._request("POST", "/prompts", json=self._with_workspace(payload), retry=False) return Prompt(**data) def list_prompts(self) -> List[Prompt]: @@ -322,7 +333,9 @@ def init_run( alias and keeps working. ``split`` records the named case subset this run covers. ``additional_scorer_ids`` (self-host): extra judge scorers that each pass their own verdict on every result from the same single agent execution - verdicts land in each - result row's ``judgeScorerResults`` and the run's ``scorerBreakdown``.""" + result row's ``judgeScorerResults`` and the run's ``scorerBreakdown``. When + ``scorer_group_id`` is set, the engine nulls ``additionalScorerIds`` on the run too - + the group is the whole grading story, not a layer on top of extra scorers.""" from agentx.version import VERSION grader_id = _resolve_scorer_id(scorer_id, evaluation_settings_id) @@ -343,7 +356,8 @@ def init_run( if additional_scorer_ids: payload["additionalScorerIds"] = additional_scorer_ids # Scorer group grading (self-host): the group's weighted 0-10 aggregate fills the rating - # column and member verdicts land per row. Mutually exclusive with scorer_id (group wins). + # column and member verdicts land per row. Mutually exclusive with scorer_id (group + # wins), and the engine also nulls additionalScorerIds when a group grades the run. if scorer_group_id: payload["scorerGroupId"] = scorer_group_id if split: @@ -408,7 +422,10 @@ def gate_run( # scorer's own per-result verdicts. Unknown names are a hard 400 from the engine. if scorer: params["scorer"] = scorer - return self._request("GET", f"/runs/{run_id}/gate", params=params) + # record=True is a server-side write despite the GET verb (it persists a gate-history + # row): a timeout after the row was stored would be retried into a duplicate verdict, + # so no transport retry - same precedent as init_run. + return self._request("GET", f"/runs/{run_id}/gate", params=params, retry=not record) def analyze_run( self, diff --git a/agentx/evaluations/datasets.py b/agentx/evaluations/datasets.py index 20c0fcd..f96c12b 100644 --- a/agentx/evaluations/datasets.py +++ b/agentx/evaluations/datasets.py @@ -81,7 +81,8 @@ def __init__( self._payload["codeScorers"] = [ { "id": scorer.get("id") or _uuid.uuid4().hex[:12], - "name": scorer["name"], + # Name may be omitted - the engine defaults it, so don't KeyError here. + "name": scorer.get("name"), "code": scorer["code"], "enabled": scorer.get("enabled", True), } diff --git a/agentx/evaluations/evaluation_settings.py b/agentx/evaluations/evaluation_settings.py index 651cfe2..67f324a 100644 --- a/agentx/evaluations/evaluation_settings.py +++ b/agentx/evaluations/evaluation_settings.py @@ -64,7 +64,8 @@ def __init__( if rouge_score: self._payload["rougeScore"] = {"enabled": True} # Sovereignty & Portability - the models to compare when this config runs - # (use client.evaluations.list_models() to discover valid ids). + # (use client.evaluations.list_models() to discover valid ids). Self-host: accepted + # on the wire but not acted on by the engine (same caveat as DatasetBuilder's). if sovereignty_models: self._payload["sovereigntyIndex"] = { "enabled": True, diff --git a/agentx/evaluations/models.py b/agentx/evaluations/models.py index d2f69e2..b2a2cad 100644 --- a/agentx/evaluations/models.py +++ b/agentx/evaluations/models.py @@ -248,6 +248,10 @@ class LiveStatistics(BaseModel): min_rating: Optional[float] = Field(default=None, alias="minRating") max_rating: Optional[float] = Field(default=None, alias="maxRating") rated_count: int = Field(default=0, alias="ratedCount") + # Rows the judge could not score / rows submitted with an error - the difference between + # "everything rated 8" and "half the run never got a verdict". + skipped_count: int = Field(default=0, alias="skippedCount") + failed_count: int = Field(default=0, alias="failedCount") class Config: populate_by_name = True diff --git a/agentx/evaluations/prompts.py b/agentx/evaluations/prompts.py index 38e3295..bad8ebf 100644 --- a/agentx/evaluations/prompts.py +++ b/agentx/evaluations/prompts.py @@ -17,9 +17,9 @@ class PromptClient: existing version-comparison view (``client.evaluations`` run comparisons on a dataset) can tell you which published version actually scored higher. - Deliberately read-mostly from here: there is no ``publish`` on this client. A prompt only - gets a new version through the dashboard's human-approved propose/publish flow, so a - rewritten prompt never reaches your running agent without someone explicitly approving it. + ``propose()`` never publishes anything; ``publish_version()`` IS the explicit approval + step - call it only after a human reviewed the proposal, since a published version is + what your running agent pulls as its live prompt. Example:: diff --git a/agentx/evaluations/runner.py b/agentx/evaluations/runner.py index ce3d6c8..31293e4 100644 --- a/agentx/evaluations/runner.py +++ b/agentx/evaluations/runner.py @@ -342,6 +342,10 @@ def _fetch_submitted_keys(self) -> Set[str]: # ------------------------------------------------------------------ def finalize(self) -> "EvaluationRunContext": + """Mark the run completed server-side. A failed finalize is raised, not swallowed + (same fail-loud posture as _flush_batch): it leaves the run in_progress - a state a + CI pipeline MUST treat as a failure, since gates and baselines only consider + completed runs.""" _say() with Spinner("Finalizing - submitting results"): try: @@ -353,6 +357,7 @@ def finalize(self) -> "EvaluationRunContext": except Exception as exc: _say(f" {red('✗')} Finalize failed: {dim(str(exc))}") logger.error("Finalize failed: %s", exc) + raise return self def gate( diff --git a/agentx/integrations/crewai.py b/agentx/integrations/crewai.py index 1f3b7b5..54e6c18 100644 --- a/agentx/integrations/crewai.py +++ b/agentx/integrations/crewai.py @@ -212,6 +212,9 @@ def _build_steps_from_timings( "end_time": end, "input": description, "output": output_text, + # A CrewAI task is an agent turn, not a model call - without this, + # _merge_child_run's default stamped every task span "llm". + "kind": "agent", }) if description is not None or task_output_text is not None: tool_calls.append({"name": name, "input": description, "output": task_output_text}) diff --git a/agentx/integrations/google_adk.py b/agentx/integrations/google_adk.py index 16f295c..4be7c74 100644 --- a/agentx/integrations/google_adk.py +++ b/agentx/integrations/google_adk.py @@ -231,6 +231,7 @@ async def after_model_callback( model=call_start.get("model") if call_start else None, input_tokens=call_input_tokens, output_tokens=call_output_tokens, + span_kind="llm", ) async def on_model_error_callback( @@ -261,6 +262,7 @@ async def on_model_error_callback( output=f"ERROR: {error}", model=call_start.get("model") if call_start else None, error=str(error), + span_kind="llm", ) # ------------------------------------------------------------------ @@ -290,7 +292,8 @@ async def after_tool_callback( tool_input = _safe_serialize(tool_args) tool_output = str(result) if result is not None else None state["root_span"].child_span( - tool_name, start_time=start_t, end_time=end_t, input=tool_input, output=tool_output + tool_name, start_time=start_t, end_time=end_t, input=tool_input, output=tool_output, + span_kind="tool", ) async def on_tool_error_callback( @@ -311,5 +314,6 @@ async def on_tool_error_callback( tool_input = _safe_serialize(tool_args) tool_output = f"ERROR: {error}" state["root_span"].child_span( - tool_name, start_time=start_t, end_time=end_t, input=tool_input, output=tool_output, error=str(error) + tool_name, start_time=start_t, end_time=end_t, input=tool_input, output=tool_output, error=str(error), + span_kind="tool", ) diff --git a/agentx/integrations/langchain.py b/agentx/integrations/langchain.py index 7d6c353..dbd0425 100644 --- a/agentx/integrations/langchain.py +++ b/agentx/integrations/langchain.py @@ -277,6 +277,10 @@ class AgentXCallbackHandler(BaseCallbackHandler): become child spans, and each LLM call / tool call / retrieval becomes a span parented under the node that ran it - so the engine's Execution Timeline shows the actual graph trajectory (which nodes ran, in what order, and what each did), not a flat step list. + + Retriever runs are stamped ``retrieval`` - LangChain cannot distinguish memory-backed + retrievers, so a Mem0/Zep-style store exposed as a retriever classifies as retrieval too. + Use ``tracer.trace_memory`` for lookups that should classify as memory. """ def __init__( diff --git a/agentx/integrations/openai_agents.py b/agentx/integrations/openai_agents.py index ae9f22f..9fc4f3a 100644 --- a/agentx/integrations/openai_agents.py +++ b/agentx/integrations/openai_agents.py @@ -251,6 +251,7 @@ def on_span_end(self, span: Any) -> None: model=call_model, input_tokens=call_input_tokens, output_tokens=call_output_tokens, + span_kind="llm", ) elif span_type == "response": @@ -300,6 +301,7 @@ def on_span_end(self, span: Any) -> None: model=call_model, input_tokens=call_input_tokens, output_tokens=call_output_tokens, + span_kind="llm", ) elif span_type == "function": @@ -313,6 +315,7 @@ 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, + span_kind="tool", ) def force_flush(self) -> None: diff --git a/agentx/monitor/client.py b/agentx/monitor/client.py index 5cda32c..42f19d4 100644 --- a/agentx/monitor/client.py +++ b/agentx/monitor/client.py @@ -235,7 +235,9 @@ def _request( # ------------------------------------------------------------------ def create_pattern(self, payload: dict) -> MonitorPattern: - data = self._request("POST", "/patterns", json=self._with_workspace(payload)) + # Server-side write: a timeout after the pattern row was created would be + # retried into a duplicate pattern, so no transport retry. + data = self._request("POST", "/patterns", json=self._with_workspace(payload), retry=False) return MonitorPattern(**data["pattern"]) def list_patterns(self) -> List[MonitorPattern]: @@ -253,7 +255,10 @@ def get_pattern(self, pattern_id: str) -> MonitorPattern: # ------------------------------------------------------------------ def create_online_evaluator(self, payload: dict) -> MonitorOnlineEvaluator: - data = self._request("POST", "/online-evaluators", json=self._with_workspace(payload)) + # Server-side write - no transport retry (see create_pattern's comment). + data = self._request( + "POST", "/online-evaluators", json=self._with_workspace(payload), retry=False + ) return MonitorOnlineEvaluator(**data["evaluator"]) @property diff --git a/agentx/monitor/models.py b/agentx/monitor/models.py index a3b5c9b..708ac46 100644 --- a/agentx/monitor/models.py +++ b/agentx/monitor/models.py @@ -12,6 +12,11 @@ class MonitorPattern(BaseModel): A "failure" pattern (the default) raises a signal to triage; a "proper" pattern logs a healthy tally instead. Only one of ``include_terms``/``regex``/``semantic_prompt`` is meaningful at a time, selected by ``detector_kind``. + + On self-host the engine stores a pattern as a list of ``conditions`` (each with its own + detector kind and match settings) - the flat ``include_terms``/``regex``/ + ``semantic_prompt`` fields are display-only projections derived from the first condition; + ``conditions`` is the truth. """ id: str = Field(alias="_id") @@ -26,6 +31,8 @@ class MonitorPattern(BaseModel): exclude_terms: List[str] = Field(default_factory=list, alias="excludeTerms") regex: Optional[str] = None semantic_prompt: Optional[str] = Field(default=None, alias="semanticPrompt") + # The engine's stored detection rules (self-host) - see the class docstring. + conditions: List[Dict[str, Any]] = Field(default_factory=list) severity: str = "medium" polarity: str = "failure" enabled: bool = True diff --git a/agentx/monitor/profile.py b/agentx/monitor/profile.py index 7b56be4..fdc25e3 100644 --- a/agentx/monitor/profile.py +++ b/agentx/monitor/profile.py @@ -46,7 +46,7 @@ def update( """Update (and enable, if not already) this agent's Monitor profile. Only fields passed here are changed; everything else on the existing profile is left as is. - Self-host only: ``coverage_mode``/``sample_rate``/``retention_days``, + Self-host only: ``coverage_mode``/``sample_rate``/``retention_days``, ``dataset_id``, and ``threshold_overrides["latencyMs"]`` are project-level defaults now (see ``MonitorProfile``'s docstring) - set them via the dashboard's Platform Settings screen instead, passing them here is accepted but has no effect. ``enabled``/ diff --git a/agentx/tracing/tracer.py b/agentx/tracing/tracer.py index 321c62a..4071fbd 100644 --- a/agentx/tracing/tracer.py +++ b/agentx/tracing/tracer.py @@ -5,6 +5,7 @@ import functools import inspect import contextvars +import logging import threading import time from contextlib import contextmanager @@ -17,6 +18,8 @@ from agentx.tracing.eval_scope import EVAL_RUN_SOURCE, current_eval_run_id from agentx.tracing.framework_detect import detect_framework +logger = logging.getLogger(__name__) + F = TypeVar("F", bound=Callable[..., Any]) @@ -313,7 +316,7 @@ def child_span( their own real per-step identity and timing (LangChain's run_id/parent_run_id, LlamaIndex's parent_id, the OpenAI Agents SDK's own span objects) and want to parent a new child under a specific span they're holding a reference to - not just whatever's on top of - the tracer's thread-local active-span stack. + the tracer's context-local (ContextVar) active-span stack. Returns the child span (its ``.span_id`` can parent a further-nested grandchild via another ``child_span()`` call on it). The returned span is not pushed onto the @@ -441,8 +444,10 @@ def _merge_child_run( cache_read_tokens=step.get("cacheReadTokenSize"), cache_write_tokens=step.get("cacheWriteTokenSize"), # Stated, so a step named anything other than "LLM Call N" still classifies - - # the backend's name regex was the only thing holding this together. - span_kind="llm", + # the backend's name regex was the only thing holding this together. Steps + # may state their own kind (crewai.py's task steps carry "agent"); the + # default stays "llm" for callers whose steps are model calls. + span_kind=step.get("kind") or "llm", ) for tc in tool_calls or []: # Some callers' tool_calls dicts (e.g. langchain.py's, which sets these on the @@ -477,6 +482,7 @@ def _merge_child_run( }) for step in [] if not emit_steps else (retrieval_steps or []): self._child_span_count += 1 + doc_count = step.get("doc_count") self.child_span( step.get("name") or f"Retrieval {self._child_span_count}", start_time=step.get("start_time"), @@ -484,7 +490,7 @@ def _merge_child_run( duration_ms=step.get("duration_ms"), input=step.get("query"), output=step.get("output"), - metadata={"kind": "retrieval"}, + metadata={"kind": "retrieval", **({"doc_count": doc_count} if doc_count is not None else {})}, span_kind="retrieval", ) @@ -680,9 +686,11 @@ def use_span(self, span: "_TraceSpan") -> Iterator["_TraceSpan"]: """ Make ``span`` (created on another thread) the active span for the duration of this block, on *this* thread. The active-span stack is - thread-local, so work submitted to a ``ThreadPoolExecutor`` or run on - any other thread doesn't automatically see a span opened on the - calling thread - wrap the worker function body in this to attach it:: + context-local (a ContextVar): bare threads start with an empty stack, + while asyncio tasks inherit a copy of their creator's. Work submitted + to a ``ThreadPoolExecutor`` or run on any other thread therefore + doesn't automatically see a span opened on the calling thread - wrap + the worker function body in this to attach it:: with tracer.trace("orchestrator") as span: def worker(): @@ -719,7 +727,10 @@ def record_tool_call( loop where the tool executes in plain Python between two ``messages.create()`` calls. Sent as a real child-span row of the active span (see ``current_span``) immediately; queued onto the next trace's plain ``tool_calls`` list if - there's no active span to attach a child to. + there's no active span to attach a child to. That queue is tracer-wide, not + per-context: the pending record attaches to the next trace THIS TRACER sends from ANY + thread or context, so concurrent no-span use can attach it to an unrelated trace - + wrap the call in ``tracer.trace()`` when that matters. ``success``/``error`` mark a failed call. ``success=False`` is what the engine's built-in "Tool failure" Monitor check and the dashboard's Tool quality column both read; leaving @@ -824,6 +835,11 @@ def record_retrieval( (same behavior as ``record_tool_call``, covering the patched-client flow where the retrieval runs just before a standalone ``messages.create()`` / ``chat.completions.create()`` call). + + The pending queue is tracer-wide, not per-context: a record queued with no active + span attaches to the next trace THIS TRACER sends from ANY thread or context, so + concurrent no-span use can attach it to an unrelated trace. Wrap the call in + ``tracer.trace()`` when that matters. """ active_span = self.current_span if active_span is None: @@ -839,6 +855,7 @@ def record_retrieval( "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, + **({"doc_count": doc_count} if doc_count is not None else {}), }) return # The kind marker is what tells the engine (retrieval-context extraction for RAG @@ -852,7 +869,7 @@ def record_retrieval( duration_ms=duration_ms, input=query, output=output, - metadata={"kind": "retrieval"}, + metadata={"kind": "retrieval", **({"doc_count": doc_count} if doc_count is not None else {})}, span_kind="retrieval", ) @@ -874,27 +891,23 @@ def record_memory( itself stays one value so dashboards and scorers can select all memory activity at once. Deliberately NOT a retrieval: retrieval spans feed the RAG judges' ``{context}`` (knowledge grounding), while memory is recalled state - see the engine's spanKind.ts. + + With no active span the record is DROPPED (with a debug log), not queued: the only + pending queue rides the next trace's ``retrieval_steps``, and memory content must + never reach the engine's retrieval-context extraction for RAG judges. Wrap the call + in ``tracer.trace()`` to keep it. (``record_tool_call``/``record_retrieval`` queue + instead - see their docstrings.) """ 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 + # NOT the record_retrieval queue posture: _pending_retrievals rides the next + # trace's performance_summary.retrieval_steps, which the engine's + # retrieval-context extraction feeds to RAG judges - recalled memory must never + # classify as knowledge grounding. Drop, and say so. + logger.debug( + "record_memory(%r) called with no active span - wrap the call in tracer.trace(); dropped", + name, ) - 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, @@ -923,9 +936,10 @@ def trace_memory( error: Optional[str] = None try: yield recorder - except Exception as exc: + except BaseException 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. + # precedent, which also catches BaseException) - fold the error into the output + # and re-raise. error = str(exc) raise finally: diff --git a/tests/test_runner_features.py b/tests/test_runner_features.py index 8c8edbd..efc89e1 100644 --- a/tests/test_runner_features.py +++ b/tests/test_runner_features.py @@ -144,3 +144,19 @@ def test_flush_batch_failure_raises_after_one_retry(monkeypatch): with pytest.raises(EvaluationSubmissionError): ctx.execute(lambda case: "x") + + +def test_finalize_failure_raises(monkeypatch): + """A failed finalize leaves the run in_progress - swallowing it let CI pipelines pass on + a run that gates and baselines would never see (same fail-loud posture as _flush_batch).""" + monkeypatch.setenv("AGENTX_EVAL_QUIET", "1") + client = FakeClient() + + def failing_finalize(run_id): + raise RuntimeError("engine down") + + client.finalize_run = failing_finalize + ctx = make_context(client) + + with pytest.raises(RuntimeError, match="engine down"): + ctx.finalize() diff --git a/tests/test_selfhost_analysis_fallback.py b/tests/test_selfhost_analysis_fallback.py index 85d8f3e..5147a05 100644 --- a/tests/test_selfhost_analysis_fallback.py +++ b/tests/test_selfhost_analysis_fallback.py @@ -198,13 +198,37 @@ def always_times_out(method, url, **kwargs): session.request = always_times_out - with pytest.raises(AgentXEvaluationsError): + # The timeout keeps its type (requests.Timeout, not a generic wrapper) so callers can + # branch on "the engine may still be doing the billable work" - see _request. + with pytest.raises(requests.Timeout): client.analyze_run(RUN) dashboard_posts = [u for u in session.urls("POST") if "/evaluate/analyze/" in u] assert len(dashboard_posts) == 1, f"retried a billable request: {dashboard_posts}" +def test_append_results_timeout_propagates_as_timeout_with_one_post(): + """A read timeout on the batch-scoring POST must surface as requests.Timeout (the type + runner._flush_batch's double-billing guard catches - wrapping it in + AgentXEvaluationsError made that guard dead code) and must not be re-POSTed by the + transport while the engine may still be scoring the first submission.""" + import requests + + client, session = make_client({}) + + def always_times_out(method, url, **kwargs): + session.calls.append((method, url, kwargs)) + raise requests.exceptions.ReadTimeout("still scoring") + + session.request = always_times_out + + with pytest.raises(requests.Timeout): + client.append_results(RUN, "batch-1", []) + + result_posts = [u for u in session.urls("POST") if u.endswith(f"/runs/{RUN}/results")] + assert len(result_posts) == 1, f"re-POSTed a batch mid-scoring: {result_posts}" + + def test_the_fallback_request_gets_the_long_analysis_timeout(): client, session = make_client( {("POST", f"{API_ROOT}/evaluate/analyze/{RUN}"): FakeResponse(200, {"status": "completed"})} diff --git a/tests/test_span_tree.py b/tests/test_span_tree.py index d127dc2..2fa2afc 100644 --- a/tests/test_span_tree.py +++ b/tests/test_span_tree.py @@ -666,17 +666,64 @@ def test_trace_retrieval_emits_real_child_span(): assert child["output"] == "3 matching docs" -def test_record_retrieval_with_no_active_span_is_a_no_op(): - """No enclosing `with tracer.trace()` and nothing else to attach a child span to — retrieval - data has no standalone wire representation anymore (that was performance_summary-only), - so this is a documented no-op rather than a silent fabrication.""" +def test_record_retrieval_with_no_active_span_queues_onto_next_trace(): + """No enclosing `with tracer.trace()` - record_retrieval queues onto the tracer-level + pending list and rides the next trace's performance_summary.retrieval_steps (the shape + the engine's retrieval-context extraction reads), carrying doc_count along.""" tracer = make_tracer() - tracer.record_retrieval("orphan_search", query="x", output="y") + tracer.record_retrieval("orphan_search", query="x", output="y", doc_count=4) with tracer.trace("agent"): pass wires = enqueued_wires(tracer) assert len(wires) == 1 assert wires[0]["name"] == "agent" + steps = wires[0]["performance_summary"]["retrieval_steps"] + assert [s["name"] for s in steps] == ["orphan_search"] + assert steps[0]["doc_count"] == 4 + + +def test_record_retrieval_carries_doc_count_into_child_span_metadata(): + tracer = make_tracer() + with tracer.trace("agent"): + tracer.record_retrieval("kb_search", query="q", output="docs", doc_count=3) + wires = enqueued_wires(tracer) + child = wires[0] + assert child["name"] == "kb_search" + assert child["span_kind"] == "retrieval" + assert child["metadata"] == {"kind": "retrieval", "doc_count": 3} + + +def test_record_memory_with_no_active_span_drops_instead_of_queueing(): + """Memory must never ride the pending retrieval queue: it lands in + performance_summary.retrieval_steps, which feeds the engine's retrieval-context + extraction for RAG judges - recalled state is not knowledge grounding. With no active + span the record is dropped (debug-logged), not attached to the next trace.""" + tracer = make_tracer() + tracer.record_memory("orphan prefs", operation="read", query="u-1", output="secret") + with tracer.trace("agent"): + pass + wires = enqueued_wires(tracer) + assert len(wires) == 1 + assert wires[0]["name"] == "agent" + assert_no_performance_summary(wires) + + +def test_merge_child_run_execution_steps_honor_their_stated_kind(): + """crewai.py stamps its task steps "kind": "agent" - the merge loop must forward that + instead of unconditionally stamping every step "llm" (autogen/llamaindex steps carry no + kind and keep classifying as llm).""" + tracer = make_tracer() + with tracer.trace("crew") as span: + span._merge_child_run( + execution_steps=[ + {"name": "Research task", "duration_ms": 5, "kind": "agent"}, + {"name": "LLM Call 1", "duration_ms": 5}, + ], + framework="crewai", + ) + wires = enqueued_wires(tracer) + kinds = {w["name"]: w.get("span_kind") for w in wires if w["name"] != "crew"} + assert kinds == {"Research task": "agent", "LLM Call 1": "llm"} def test_record_tool_call_with_no_active_span_still_queues():