From 7dc3c2308d62d967543356fe69d58005ff20d998 Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Mon, 26 Oct 2026 18:03:16 -0400 Subject: [PATCH 01/17] feat(tracing): add failure-check span attribute helper Offline failure checks read sgp.agent.* attributes the agent declares on its spans. FailureCheckSpec validates the root span declarations on construction and renders them as flat dotted keys holding native JSON values, which the SGP processor ships verbatim as span metadata. output_name, verification_result and escalates write the per-span attributes. Co-Authored-By: Claude Sonnet 5.5 --- .../10_async/00_base/030_tracing/README.md | 32 ++ src/agentex/lib/adk/__init__.py | 8 + .../lib/core/tracing/failure_attributes.py | 184 +++++++++ .../core/tracing/test_failure_attributes.py | 390 ++++++++++++++++++ 4 files changed, 614 insertions(+) create mode 100644 src/agentex/lib/core/tracing/failure_attributes.py create mode 100644 tests/lib/core/tracing/test_failure_attributes.py diff --git a/examples/tutorials/10_async/00_base/030_tracing/README.md b/examples/tutorials/10_async/00_base/030_tracing/README.md index 3c66248e8..e8b715148 100644 --- a/examples/tutorials/10_async/00_base/030_tracing/README.md +++ b/examples/tutorials/10_async/00_base/030_tracing/README.md @@ -74,6 +74,38 @@ spans stay safe to emit. Recognized token keys: `input_tokens`/`prompt_tokens`, `output_tokens`/`completion_tokens`, `cached_input_tokens`/`cached_tokens`, `reasoning_tokens`; cost is `cost_usd`. +## Failure-Check Attributes + +Offline failure checks read attributes the agent declares on its spans. Build them from one +spec instead of hand-writing keys. The spec validates its values on construction, and each +attribute lands in span metadata as a flat `sgp.agent.*` key holding a native JSON value (a +list stays a list and a step budget stays an integer). + +```python +import dataclasses + +spec = adk.FailureCheckSpec( + required_outputs=["answer"], + step_budget=20, + verification=[adk.VerificationRule(step="check_answer", verifies="answer")], + escalation_channels=["send_email.to"], + prompt_version="2026-10-26", +) +async with adk.tracing.span(trace_id=task.id, name="agent", data=spec.attributes()) as root: + ... + # The stop reason is only known at the end. + done = dataclasses.replace(spec, stop_reason="completed") + root.data = {**(root.data or {}), **done.attributes()} +``` + +A span that produces a required output, runs a verification or escalates one carries a single +attribute, passed as its `data`: `adk.failure_attributes.output_name("answer")`, +`verification_result("pass")` (or `"fail"`) and `escalates(verify_span.id)`. + +The SGP export adds `__agent_version__` when `AGENT_VERSION` is set, and `__commit_sha__` when the +process starts with `AGENT_COMMIT_SHA` set to a valid git SHA, so only `prompt_version` is yours to +declare. + ## When to Use - Debugging complex agent behaviors - Performance optimization and bottleneck identification diff --git a/src/agentex/lib/adk/__init__.py b/src/agentex/lib/adk/__init__.py index c05f8f3ea..9a3c21a3d 100644 --- a/src/agentex/lib/adk/__init__.py +++ b/src/agentex/lib/adk/__init__.py @@ -36,6 +36,10 @@ from agentex.lib.core.tracing import code_revision from agentex.lib.core.tracing.lineage import DataSourceRef, data_sources +# Span attributes the offline failure checks read, implemented in core.tracing +from agentex.lib.core.tracing import failure_attributes +from agentex.lib.core.tracing.failure_attributes import FailureCheckSpec, VerificationRule + # Unified harness surface (AGX1-375) from agentex.lib.core.harness import ( UnifiedEmitter, @@ -79,6 +83,10 @@ "code_revision", "DataSourceRef", "data_sources", + # Failure-check span attributes + "failure_attributes", + "FailureCheckSpec", + "VerificationRule", # Checkpointing / LangGraph "create_checkpointer", "stream_langgraph_events", diff --git a/src/agentex/lib/core/tracing/failure_attributes.py b/src/agentex/lib/core/tracing/failure_attributes.py new file mode 100644 index 000000000..8d6d8c229 --- /dev/null +++ b/src/agentex/lib/core/tracing/failure_attributes.py @@ -0,0 +1,184 @@ +"""Span attributes the offline failure checks read, written and validated from one spec. + +The SGP processor ships ``span.data`` verbatim as span metadata, so each attribute is a flat +dotted key holding a native JSON value, never a JSON string or a nested object. +""" + +from __future__ import annotations + +import re +import copy +import json +from typing import Any, Literal, Callable +from dataclasses import field, dataclass +from collections.abc import Sequence + +__all__ = ( + "REQUIRED_OUTPUTS_KEY", + "STEP_BUDGET_KEY", + "STOP_REASON_KEY", + "OUTPUT_NAME_KEY", + "VERIFICATION_KEY", + "ESCALATION_CHANNELS_KEY", + "VERIFICATION_RESULT_KEY", + "ESCALATES_KEY", + "OUTPUT_SCHEMA_KEY", + "QUOTE_ARGUMENTS_KEY", + "PROMPT_VERSION_KEY", + "VerificationRule", + "FailureCheckSpec", + "output_name", + "verification_result", + "escalates", +) + +REQUIRED_OUTPUTS_KEY = "sgp.agent.required_outputs" +STEP_BUDGET_KEY = "sgp.agent.step_budget" +STOP_REASON_KEY = "sgp.agent.stop_reason" +OUTPUT_NAME_KEY = "sgp.agent.output_name" +VERIFICATION_KEY = "sgp.agent.verification" +ESCALATION_CHANNELS_KEY = "sgp.agent.escalation_channels" +VERIFICATION_RESULT_KEY = "sgp.agent.verification_result" +ESCALATES_KEY = "sgp.agent.escalates" +OUTPUT_SCHEMA_KEY = "sgp.agent.output_schema" +QUOTE_ARGUMENTS_KEY = "sgp.agent.quote_arguments" +PROMPT_VERSION_KEY = "sgp.agent.prompt_version" + +_TOOL_ARGUMENT_RE = re.compile(r"[^.\s]+\.[^.\s]+") +_VERIFICATION_RESULTS: tuple[str, ...] = ("pass", "fail") + + +def _text(value: object, what: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{what} must be a non-empty string, got {value!r}") + if value != value.strip(): + raise ValueError(f"{what} must not have leading or trailing whitespace, got {value!r}") + return value + + +def _unique_texts(values: object, what: str) -> tuple[str, ...]: + if not isinstance(values, Sequence) or isinstance(values, str | bytes): + raise ValueError(f"{what} must be a list of strings, got {values!r}") + items = tuple(_text(item, f"{what} entry") for item in values) + seen: set[str] = set() + for item in items: + if item in seen: + raise ValueError(f"{what} has a duplicate entry {item!r}") + seen.add(item) + return items + + +def _tool_arguments(values: object, what: str) -> tuple[str, ...]: + items = _unique_texts(values, what) + for item in items: + if _TOOL_ARGUMENT_RE.fullmatch(item) is None: + raise ValueError(f"{what} entry must look like tool.argument, got {item!r}") + return items + + +def _step_budget(value: object) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + raise ValueError(f"step_budget must be a positive integer, got {value!r}") + return value + + +def _json_object(value: object, what: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise ValueError(f"{what} must be a JSON object (a dict), got {type(value).__name__}") + try: + return json.loads(json.dumps(value, allow_nan=False)) + except (TypeError, ValueError, RecursionError) as exc: + raise ValueError(f"{what} must be JSON serializable: {exc}") from exc + + +@dataclass(frozen=True) +class VerificationRule: + """A verification step and the name of what it verifies.""" + + step: str + verifies: str + + def __post_init__(self) -> None: + _text(self.step, "verification step") + _text(self.verifies, "verification verifies") + + +def _rules(values: object) -> tuple[VerificationRule, ...]: + if not isinstance(values, Sequence) or isinstance(values, str | bytes): + raise ValueError(f"verification must be a list of VerificationRule, got {values!r}") + for rule in values: + if not isinstance(rule, VerificationRule): + raise ValueError(f"verification entries must be VerificationRule, got {rule!r}") + rules = tuple(values) + if len(set(rules)) != len(rules): + raise ValueError("verification has a duplicate rule") + return rules + + +_NORMALIZERS: dict[str, Callable[[Any], Any]] = { + "required_outputs": lambda v: _unique_texts(v, "required_outputs"), + "step_budget": _step_budget, + "stop_reason": lambda v: _text(v, "stop_reason"), + "verification": _rules, + "escalation_channels": lambda v: _tool_arguments(v, "escalation_channels"), + "output_schema": lambda v: _json_object(v, "output_schema"), + "quote_arguments": lambda v: _tool_arguments(v, "quote_arguments"), + "prompt_version": lambda v: _text(v, "prompt_version"), +} + + +def _listed(values: Sequence[str] | None) -> list[str] | None: + return None if values is None else list(values) + + +@dataclass(frozen=True, kw_only=True) +class FailureCheckSpec: + """Declarations for the root span. None leaves one undeclared and an empty list declares none.""" + + required_outputs: Sequence[str] | None = None + step_budget: int | None = None + stop_reason: str | None = None + verification: Sequence[VerificationRule] | None = None + escalation_channels: Sequence[str] | None = None + output_schema: dict[str, Any] | None = field(default=None, hash=False) + quote_arguments: Sequence[str] | None = None + prompt_version: str | None = None + + def __post_init__(self) -> None: + for name, normalize in _NORMALIZERS.items(): + value = getattr(self, name) + if value is not None: + object.__setattr__(self, name, normalize(value)) + + def attributes(self) -> dict[str, Any]: + """The root span data, holding only the declared attributes.""" + declared: dict[str, Any] = { + REQUIRED_OUTPUTS_KEY: _listed(self.required_outputs), + STEP_BUDGET_KEY: self.step_budget, + STOP_REASON_KEY: self.stop_reason, + VERIFICATION_KEY: None + if self.verification is None + else [{"step": rule.step, "verifies": rule.verifies} for rule in self.verification], + ESCALATION_CHANNELS_KEY: _listed(self.escalation_channels), + OUTPUT_SCHEMA_KEY: None if self.output_schema is None else copy.deepcopy(self.output_schema), + QUOTE_ARGUMENTS_KEY: _listed(self.quote_arguments), + PROMPT_VERSION_KEY: self.prompt_version, + } + return {key: value for key, value in declared.items() if value is not None} + + +def output_name(name: str) -> dict[str, str]: + """Data for the span that produces a required output.""" + return {OUTPUT_NAME_KEY: _text(name, "output name")} + + +def verification_result(result: Literal["pass", "fail"]) -> dict[str, str]: + """Data for a verification span.""" + if result not in _VERIFICATION_RESULTS: + raise ValueError(f"verification result must be 'pass' or 'fail', got {result!r}") + return {VERIFICATION_RESULT_KEY: result} + + +def escalates(verification_span_id: str) -> dict[str, str]: + """Data for an escalation span, naming the verification span it escalates.""" + return {ESCALATES_KEY: _text(verification_span_id, "verification span id")} diff --git a/tests/lib/core/tracing/test_failure_attributes.py b/tests/lib/core/tracing/test_failure_attributes.py new file mode 100644 index 000000000..357a23b4f --- /dev/null +++ b/tests/lib/core/tracing/test_failure_attributes.py @@ -0,0 +1,390 @@ +"""Failure-check span attributes. + +Keys and formats are literals from the check contract, never read back from the module under test. +""" + +from __future__ import annotations + +import json +import dataclasses +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from scale_gp_beta.lib.tracing import create_span as sgp_create_span + +from agentex.lib import adk +from agentex.lib.types.tracing import SGPTracingProcessorConfig +from agentex.lib.core.tracing.trace import Trace, AsyncTrace +from agentex.lib.core.tracing.span_queue import AsyncSpanQueue +from agentex.lib.core.tracing.failure_attributes import ( + FailureCheckSpec, + VerificationRule, + escalates, + output_name, + verification_result, +) +from agentex.lib.core.tracing.processors.sgp_tracing_processor import ( + SGPSyncTracingProcessor, + SGPAsyncTracingProcessor, +) + +MODULE = "agentex.lib.core.tracing.processors.sgp_tracing_processor" + +SCHEMA = { + "type": "object", + "properties": {"answer": {"type": "string"}, "confidence": {"type": "number", "minimum": 0.0}}, + "required": ["answer"], +} + +CONTRACT = [ + ("required_outputs", ["answer", "citations"], "sgp.agent.required_outputs", ["answer", "citations"]), + ("step_budget", 25, "sgp.agent.step_budget", 25), + ("stop_reason", "completed", "sgp.agent.stop_reason", "completed"), + ( + "verification", + [VerificationRule("run_tests", "apply_patch")], + "sgp.agent.verification", + [{"step": "run_tests", "verifies": "apply_patch"}], + ), + ( + "escalation_channels", + ["send_email.to", "page_oncall.team"], + "sgp.agent.escalation_channels", + ["send_email.to", "page_oncall.team"], + ), + ("output_schema", SCHEMA, "sgp.agent.output_schema", SCHEMA), + ("quote_arguments", ["cite.quote"], "sgp.agent.quote_arguments", ["cite.quote"]), + ("prompt_version", "v3", "sgp.agent.prompt_version", "v3"), +] + +FULL_SPEC = {field: value for field, value, _, _ in CONTRACT} +FULL_ATTRIBUTES = {key: expected for _, _, key, expected in CONTRACT} + + +class TestRootSpanAttributes: + @pytest.mark.parametrize(("field", "value", "key", "expected"), CONTRACT, ids=[row[0] for row in CONTRACT]) + def test_each_attribute_renders_its_documented_key_and_format(self, field, value, key, expected): + attributes = FailureCheckSpec(**{field: value}).attributes() + assert attributes == {key: expected} + assert type(attributes[key]) is type(expected) + + def test_all_attributes_render_together_without_colliding(self): + assert FailureCheckSpec(**FULL_SPEC).attributes() == FULL_ATTRIBUTES + + def test_every_field_is_optional_and_undeclared_ones_are_absent(self): + assert FailureCheckSpec().attributes() == {} + assert FailureCheckSpec(step_budget=3).attributes() == {"sgp.agent.step_budget": 3} + + def test_an_empty_list_declares_none_and_is_kept(self): + attributes = FailureCheckSpec(required_outputs=[], escalation_channels=()).attributes() + assert attributes == {"sgp.agent.required_outputs": [], "sgp.agent.escalation_channels": []} + + def test_rendered_data_survives_a_json_round_trip(self): + attributes = FailureCheckSpec(**FULL_SPEC).attributes() + assert json.loads(json.dumps(attributes)) == attributes + + def test_the_spec_is_frozen(self): + spec = FailureCheckSpec(step_budget=3) + with pytest.raises(dataclasses.FrozenInstanceError): + spec.step_budget = 4 # type: ignore[misc] + + def test_specs_with_a_schema_stay_hashable_and_compare_by_value(self): + first = FailureCheckSpec(required_outputs=["a"], output_schema=SCHEMA) + second = FailureCheckSpec(required_outputs=("a",), output_schema=dict(SCHEMA)) + assert first == second + assert hash(first) == hash(second) + assert first != FailureCheckSpec(required_outputs=["a"], output_schema={"type": "string"}) + + def test_fields_are_keyword_only(self): + with pytest.raises(TypeError): + FailureCheckSpec(["answer"]) # type: ignore[misc] + + def test_later_changes_to_inputs_or_rendered_data_do_not_reach_the_spec(self): + outputs = ["answer"] + schema = {"type": "object", "properties": {"a": {"type": "string"}}} + spec = FailureCheckSpec(required_outputs=outputs, output_schema=schema) + + outputs.append("sneaky") + schema["properties"]["a"]["type"] = "number" + rendered = spec.attributes() + rendered["sgp.agent.required_outputs"].append("sneaky") + rendered["sgp.agent.output_schema"]["properties"]["a"]["type"] = "number" + + assert spec.attributes() == { + "sgp.agent.required_outputs": ["answer"], + "sgp.agent.output_schema": {"type": "object", "properties": {"a": {"type": "string"}}}, + } + + def test_replace_adds_the_stop_reason_known_only_at_the_end(self): + spec = FailureCheckSpec(required_outputs=["answer"]) + done = dataclasses.replace(spec, stop_reason="completed") + assert done.attributes() == { + "sgp.agent.required_outputs": ["answer"], + "sgp.agent.stop_reason": "completed", + } + + def test_replace_validates_again(self): + with pytest.raises(ValueError, match="positive integer"): + dataclasses.replace(FailureCheckSpec(step_budget=3), step_budget=0) + + +class TestRootSpanValidation: + @pytest.mark.parametrize( + ("kwargs", "message"), + [ + ({"required_outputs": ["answer", "answer"]}, "duplicate entry 'answer'"), + ({"required_outputs": [""]}, "non-empty string"), + ({"required_outputs": [" "]}, "non-empty string"), + ({"required_outputs": [3]}, "non-empty string"), + ({"required_outputs": [None]}, "non-empty string"), + ({"required_outputs": [" answer"]}, "leading or trailing whitespace"), + ({"required_outputs": ["answer\n"]}, "leading or trailing whitespace"), + ({"required_outputs": "answer"}, "list of strings"), + ({"required_outputs": {"answer"}}, "list of strings"), + ({"required_outputs": 3}, "list of strings"), + ({"required_outputs": b"answer"}, "list of strings"), + ], + ) + def test_required_outputs(self, kwargs, message): + with pytest.raises(ValueError, match=message): + FailureCheckSpec(**kwargs) + + @pytest.mark.parametrize("value", [0, -3, True, False, 2.5, "5", [5]]) + def test_step_budget_must_be_a_positive_integer(self, value): + with pytest.raises(ValueError, match="step_budget must be a positive integer"): + FailureCheckSpec(step_budget=value) + + @pytest.mark.parametrize("field", ["stop_reason", "prompt_version"]) + @pytest.mark.parametrize( + ("value", "message"), + [ + ("", "non-empty string"), + (" ", "non-empty string"), + (7, "non-empty string"), + (" v3", "leading or trailing whitespace"), + ("v3\n", "leading or trailing whitespace"), + ], + ) + def test_free_text_fields(self, field, value, message): + with pytest.raises(ValueError, match=message) as raised: + FailureCheckSpec(**{field: value}) + assert str(raised.value).startswith(field) + + @pytest.mark.parametrize( + ("value", "message"), + [ + (["run_tests"], "entries must be VerificationRule"), + ([{"step": "run_tests", "verifies": "apply_patch"}], "entries must be VerificationRule"), + ("run_tests", "list of VerificationRule"), + (VerificationRule("a", "b"), "list of VerificationRule"), + ([VerificationRule("a", "b"), VerificationRule("a", "b")], "duplicate rule"), + ], + ) + def test_verification(self, value, message): + with pytest.raises(ValueError, match=message): + FailureCheckSpec(verification=value) + + def test_one_step_may_verify_several_things_and_one_thing_have_several_verifiers(self): + rules = [VerificationRule("check", "a"), VerificationRule("check", "b"), VerificationRule("recheck", "a")] + assert len(FailureCheckSpec(verification=rules).attributes()["sgp.agent.verification"]) == 3 + + @pytest.mark.parametrize("field", ["escalation_channels", "quote_arguments"]) + @pytest.mark.parametrize( + ("value", "message"), + [ + (["send_email"], "must look like tool.argument"), + ([".to"], "must look like tool.argument"), + (["send_email."], "must look like tool.argument"), + (["send_email.to.cc"], "must look like tool.argument"), + (["send_email..to"], "must look like tool.argument"), + (["send email.to"], "must look like tool.argument"), + (["send_email.to", "send_email.to"], "duplicate entry 'send_email.to'"), + (["send_email.to", ""], "non-empty string"), + ("send_email.to", "list of strings"), + ], + ) + def test_tool_argument_paths(self, field, value, message): + with pytest.raises(ValueError, match=message): + FailureCheckSpec(**{field: value}) + + @pytest.mark.parametrize( + ("value", "message"), + [ + ([], "must be a JSON object"), + ("{}", "must be a JSON object"), + (True, "must be a JSON object"), + ({"type": {"string"}}, "must be JSON serializable"), + ({"minimum": float("nan")}, "must be JSON serializable"), + ({"maximum": float("inf")}, "must be JSON serializable"), + ({"x": object()}, "must be JSON serializable"), + ], + ) + def test_output_schema_must_be_a_json_object(self, value, message): + with pytest.raises(ValueError, match=message): + FailureCheckSpec(output_schema=value) + + def test_a_circular_schema_is_rejected(self): + schema: dict[str, Any] = {"type": "object"} + schema["self"] = schema + with pytest.raises(ValueError, match="must be JSON serializable"): + FailureCheckSpec(output_schema=schema) + + def test_a_pathologically_deep_schema_is_rejected(self): + deep: list[Any] = [] + for _ in range(200_000): + deep = [deep] + with pytest.raises(ValueError, match="must be JSON serializable"): + FailureCheckSpec(output_schema={"default": deep}) + + def test_an_empty_schema_is_a_json_object(self): + assert FailureCheckSpec(output_schema={}).attributes() == {"sgp.agent.output_schema": {}} + + +class TestVerificationRule: + @pytest.mark.parametrize( + ("step", "verifies"), [("", "a"), ("a", ""), (" ", "a"), ("a", None), (None, "a"), (" a", "b")] + ) + def test_both_names_must_be_non_empty_strings(self, step, verifies): + with pytest.raises(ValueError): + VerificationRule(step, verifies) + + def test_rules_are_frozen(self): + rule = VerificationRule("check", "answer") + with pytest.raises(dataclasses.FrozenInstanceError): + rule.step = "other" # type: ignore[misc] + + +class TestPerSpanAttributes: + def test_output_name(self): + assert output_name("answer") == {"sgp.agent.output_name": "answer"} + + @pytest.mark.parametrize("value", ["", " ", None, 3, " answer", "answer "]) + def test_output_name_rejects_bad_names(self, value): + with pytest.raises(ValueError, match="output name"): + output_name(value) + + @pytest.mark.parametrize("result", ["pass", "fail"]) + def test_verification_result(self, result): + assert verification_result(result) == {"sgp.agent.verification_result": result} + + @pytest.mark.parametrize("value", ["PASS", "passed", "ok", "", None, True, False, 1]) + def test_verification_result_accepts_only_pass_or_fail(self, value): + with pytest.raises(ValueError, match="'pass' or 'fail'"): + verification_result(value) + + def test_escalates(self): + assert escalates("3f2c1a9e-0000-4000-8000-000000000001") == { + "sgp.agent.escalates": "3f2c1a9e-0000-4000-8000-000000000001" + } + + @pytest.mark.parametrize("value", ["", " ", None, 3, " span-1"]) + def test_escalates_rejects_bad_span_ids(self, value): + with pytest.raises(ValueError, match="verification span id"): + escalates(value) + + def test_results_merge_into_a_span_data_dict(self): + data = {"note": "kept", **output_name("answer")} + assert data == {"note": "kept", "sgp.agent.output_name": "answer"} + + +class TestExports: + def test_adk_exposes_the_helper(self): + assert adk.FailureCheckSpec is FailureCheckSpec + assert adk.VerificationRule is VerificationRule + assert adk.failure_attributes.output_name is output_name + assert adk.failure_attributes.verification_result is verification_result + assert adk.failure_attributes.escalates is escalates + assert {"failure_attributes", "FailureCheckSpec", "VerificationRule"} <= set(adk.__all__) + + +AGENT_ENV = MagicMock(ACP_TYPE="async", AGENT_NAME="pilot-agent", AGENT_ID="agent-1", AGENT_VERSION="sha-abc1234") +CONFIG = SGPTracingProcessorConfig(sgp_api_key="test-key", sgp_account_id="test-account") + +SPEC_KWARGS = {field: value for field, value in FULL_SPEC.items() if field != "stop_reason"} +EXPECTED_ROOT = {**FULL_ATTRIBUTES, "sgp.agent.stop_reason": "budget_exhausted"} + + +def _offline_create_span(created: list[Any]): + def create(**kwargs: Any): + span = sgp_create_span(**kwargs, queue_manager=MagicMock()) + created.append(span) + return span + + return create + + +def _assert_attributes_landed_in_sgp_metadata(items: list[dict[str, Any]]) -> None: + by_name = {item["name"]: item for item in items} + assert sorted(by_name) == ["agent", "answer", "escalate", "verify"] + + root = by_name["agent"]["metadata"] + assert {key: value for key, value in root.items() if key.startswith("sgp.agent.")} == EXPECTED_ROOT + assert "sgp" not in root + assert root["__agent_version__"] == "sha-abc1234" + + assert by_name["answer"]["metadata"]["sgp.agent.output_name"] == "answer" + assert by_name["verify"]["metadata"]["sgp.agent.verification_result"] == "fail" + assert by_name["escalate"]["metadata"]["sgp.agent.escalates"] == by_name["verify"]["id"] + + for item in items: + assert json.loads(json.dumps(item["metadata"])) == item["metadata"] + + +class TestRealSpanPath: + async def test_async_tracer_delivers_attributes_as_flat_keys_with_native_values(self, monkeypatch): + monkeypatch.setenv("AGENTEX_TRACING_SKIP_SPAN_START", "1") + client = MagicMock() + client.spans.upsert_batch = AsyncMock() + mock_env = MagicMock() + mock_env.refresh.return_value = AGENT_ENV + with patch(f"{MODULE}.EnvironmentVariables", mock_env), patch(f"{MODULE}.AsyncSGPClient", return_value=client): + processor = SGPAsyncTracingProcessor(CONFIG) + processor._get_client = lambda: client # type: ignore[method-assign] + + queue = AsyncSpanQueue() + trace = AsyncTrace(processors=[processor], client=MagicMock(), trace_id="trace-1", span_queue=queue) + spec = FailureCheckSpec(**SPEC_KWARGS) + + with patch(f"{MODULE}.create_span", side_effect=_offline_create_span([])): + async with trace.span(name="agent", data=spec.attributes()) as root: + async with trace.span(name="answer", parent_id=root.id, data=output_name("answer")): + pass + async with trace.span(name="verify", parent_id=root.id, data=verification_result("fail")) as verify: + pass + async with trace.span(name="escalate", parent_id=root.id, data=escalates(verify.id)): + pass + stopped = dataclasses.replace(spec, stop_reason="budget_exhausted") + assert isinstance(root.data, dict) + root.data = {**root.data, **stopped.attributes()} + await queue.shutdown() + + items = [item for call in client.spans.upsert_batch.call_args_list for item in call.kwargs["items"]] + _assert_attributes_landed_in_sgp_metadata(items) + + def test_sync_tracer_delivers_attributes_as_flat_keys_with_native_values(self, monkeypatch): + monkeypatch.setenv("AGENTEX_TRACING_SKIP_SPAN_START", "1") + mock_env = MagicMock() + mock_env.refresh.return_value = AGENT_ENV + with patch(f"{MODULE}.EnvironmentVariables", mock_env), patch(f"{MODULE}.SGPClient"), patch( + f"{MODULE}.tracing" + ): + processor = SGPSyncTracingProcessor(CONFIG) + + created: list[Any] = [] + trace = Trace(processors=[processor], client=MagicMock(), trace_id="trace-1") + spec = FailureCheckSpec(**SPEC_KWARGS) + + with patch(f"{MODULE}.create_span", side_effect=_offline_create_span(created)): + with trace.span(name="agent", data=spec.attributes()) as root: + with trace.span(name="answer", parent_id=root.id, data=output_name("answer")): + pass + with trace.span(name="verify", parent_id=root.id, data=verification_result("fail")) as verify: + pass + with trace.span(name="escalate", parent_id=root.id, data=escalates(verify.id)): + pass + stopped = dataclasses.replace(spec, stop_reason="budget_exhausted") + assert isinstance(root.data, dict) + root.data = {**root.data, **stopped.attributes()} + + _assert_attributes_landed_in_sgp_metadata([span.to_request_params() for span in created]) From 1651347407fadd8ab5450bc4be7acd745a31a32f Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Mon, 26 Oct 2026 18:18:24 -0400 Subject: [PATCH 02/17] fix(tracing): write the output schema attribute as JSON text The span store keeps metadata in a ClickHouse JSON column, where a schema object spreads into many dynamic paths whose keys can collide after dot-splitting. FailureCheckSpec still takes a JSON object and validates it, then renders sgp.agent.output_schema as a JSON string. Every other attribute stays a native value. Co-Authored-By: Claude Sonnet 5.5 --- .../10_async/00_base/030_tracing/README.md | 3 +- .../lib/core/tracing/failure_attributes.py | 8 ++-- .../core/tracing/test_failure_attributes.py | 37 ++++++++++++------- 3 files changed, 30 insertions(+), 18 deletions(-) diff --git a/examples/tutorials/10_async/00_base/030_tracing/README.md b/examples/tutorials/10_async/00_base/030_tracing/README.md index e8b715148..ca3dfa261 100644 --- a/examples/tutorials/10_async/00_base/030_tracing/README.md +++ b/examples/tutorials/10_async/00_base/030_tracing/README.md @@ -79,7 +79,8 @@ spans stay safe to emit. Recognized token keys: `input_tokens`/`prompt_tokens`, Offline failure checks read attributes the agent declares on its spans. Build them from one spec instead of hand-writing keys. The spec validates its values on construction, and each attribute lands in span metadata as a flat `sgp.agent.*` key holding a native JSON value (a -list stays a list and a step budget stays an integer). +list stays a list and a step budget stays an integer). The output schema is the exception and +is written as a JSON string, because the span store would spread a schema object into many paths. ```python import dataclasses diff --git a/src/agentex/lib/core/tracing/failure_attributes.py b/src/agentex/lib/core/tracing/failure_attributes.py index 8d6d8c229..818144216 100644 --- a/src/agentex/lib/core/tracing/failure_attributes.py +++ b/src/agentex/lib/core/tracing/failure_attributes.py @@ -1,13 +1,13 @@ """Span attributes the offline failure checks read, written and validated from one spec. The SGP processor ships ``span.data`` verbatim as span metadata, so each attribute is a flat -dotted key holding a native JSON value, never a JSON string or a nested object. +dotted key holding a native JSON value. The output schema is written as JSON text instead, since +the span store would spread a schema object into many paths whose keys can collide. """ from __future__ import annotations import re -import copy import json from typing import Any, Literal, Callable from dataclasses import field, dataclass @@ -160,7 +160,9 @@ def attributes(self) -> dict[str, Any]: if self.verification is None else [{"step": rule.step, "verifies": rule.verifies} for rule in self.verification], ESCALATION_CHANNELS_KEY: _listed(self.escalation_channels), - OUTPUT_SCHEMA_KEY: None if self.output_schema is None else copy.deepcopy(self.output_schema), + OUTPUT_SCHEMA_KEY: None + if self.output_schema is None + else json.dumps(self.output_schema, separators=(",", ":")), QUOTE_ARGUMENTS_KEY: _listed(self.quote_arguments), PROMPT_VERSION_KEY: self.prompt_version, } diff --git a/tests/lib/core/tracing/test_failure_attributes.py b/tests/lib/core/tracing/test_failure_attributes.py index 357a23b4f..370eed80d 100644 --- a/tests/lib/core/tracing/test_failure_attributes.py +++ b/tests/lib/core/tracing/test_failure_attributes.py @@ -53,13 +53,13 @@ "sgp.agent.escalation_channels", ["send_email.to", "page_oncall.team"], ), - ("output_schema", SCHEMA, "sgp.agent.output_schema", SCHEMA), ("quote_arguments", ["cite.quote"], "sgp.agent.quote_arguments", ["cite.quote"]), ("prompt_version", "v3", "sgp.agent.prompt_version", "v3"), ] -FULL_SPEC = {field: value for field, value, _, _ in CONTRACT} -FULL_ATTRIBUTES = {key: expected for _, _, key, expected in CONTRACT} +SCHEMA_KEY = "sgp.agent.output_schema" +FULL_SPEC = {**{field: value for field, value, _, _ in CONTRACT}, "output_schema": SCHEMA} +NATIVE_ATTRIBUTES = {key: expected for _, _, key, expected in CONTRACT} class TestRootSpanAttributes: @@ -69,8 +69,16 @@ def test_each_attribute_renders_its_documented_key_and_format(self, field, value assert attributes == {key: expected} assert type(attributes[key]) is type(expected) + def test_output_schema_is_written_as_json_text_that_decodes_to_the_schema(self): + attributes = FailureCheckSpec(output_schema=SCHEMA).attributes() + assert list(attributes) == [SCHEMA_KEY] + assert type(attributes[SCHEMA_KEY]) is str + assert json.loads(attributes[SCHEMA_KEY]) == SCHEMA + def test_all_attributes_render_together_without_colliding(self): - assert FailureCheckSpec(**FULL_SPEC).attributes() == FULL_ATTRIBUTES + attributes = FailureCheckSpec(**FULL_SPEC).attributes() + assert json.loads(attributes.pop(SCHEMA_KEY)) == SCHEMA + assert attributes == NATIVE_ATTRIBUTES def test_every_field_is_optional_and_undeclared_ones_are_absent(self): assert FailureCheckSpec().attributes() == {} @@ -107,14 +115,11 @@ def test_later_changes_to_inputs_or_rendered_data_do_not_reach_the_spec(self): outputs.append("sneaky") schema["properties"]["a"]["type"] = "number" - rendered = spec.attributes() - rendered["sgp.agent.required_outputs"].append("sneaky") - rendered["sgp.agent.output_schema"]["properties"]["a"]["type"] = "number" + spec.attributes()["sgp.agent.required_outputs"].append("sneaky") - assert spec.attributes() == { - "sgp.agent.required_outputs": ["answer"], - "sgp.agent.output_schema": {"type": "object", "properties": {"a": {"type": "string"}}}, - } + again = spec.attributes() + assert again["sgp.agent.required_outputs"] == ["answer"] + assert json.loads(again[SCHEMA_KEY]) == {"type": "object", "properties": {"a": {"type": "string"}}} def test_replace_adds_the_stop_reason_known_only_at_the_end(self): spec = FailureCheckSpec(required_outputs=["answer"]) @@ -238,7 +243,8 @@ def test_a_pathologically_deep_schema_is_rejected(self): FailureCheckSpec(output_schema={"default": deep}) def test_an_empty_schema_is_a_json_object(self): - assert FailureCheckSpec(output_schema={}).attributes() == {"sgp.agent.output_schema": {}} + attributes = FailureCheckSpec(output_schema={}).attributes() + assert json.loads(attributes[SCHEMA_KEY]) == {} class TestVerificationRule: @@ -302,7 +308,7 @@ def test_adk_exposes_the_helper(self): CONFIG = SGPTracingProcessorConfig(sgp_api_key="test-key", sgp_account_id="test-account") SPEC_KWARGS = {field: value for field, value in FULL_SPEC.items() if field != "stop_reason"} -EXPECTED_ROOT = {**FULL_ATTRIBUTES, "sgp.agent.stop_reason": "budget_exhausted"} +EXPECTED_ROOT = {**NATIVE_ATTRIBUTES, "sgp.agent.stop_reason": "budget_exhausted"} def _offline_create_span(created: list[Any]): @@ -319,7 +325,10 @@ def _assert_attributes_landed_in_sgp_metadata(items: list[dict[str, Any]]) -> No assert sorted(by_name) == ["agent", "answer", "escalate", "verify"] root = by_name["agent"]["metadata"] - assert {key: value for key, value in root.items() if key.startswith("sgp.agent.")} == EXPECTED_ROOT + declared = {key: value for key, value in root.items() if key.startswith("sgp.agent.")} + assert type(declared[SCHEMA_KEY]) is str + assert json.loads(declared.pop(SCHEMA_KEY)) == SCHEMA + assert declared == EXPECTED_ROOT assert "sgp" not in root assert root["__agent_version__"] == "sha-abc1234" From dbb8ff63d694fd1190ae8e3b935ba3a6bf28fd03 Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Tue, 27 Oct 2026 10:31:59 -0400 Subject: [PATCH 03/17] fix(tracing): validate the output schema again when attributes render output_schema is a mutable dict on a frozen spec, so an edit after construction skipped the checks construction applies and could write NaN or invalid JSON text. attributes() now runs the same check before writing the schema, so an edited schema fails with the error construction raises. Co-Authored-By: Claude Sonnet 5.5 --- src/agentex/lib/core/tracing/failure_attributes.py | 3 ++- tests/lib/core/tracing/test_failure_attributes.py | 14 ++++++++++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/src/agentex/lib/core/tracing/failure_attributes.py b/src/agentex/lib/core/tracing/failure_attributes.py index 818144216..5db792305 100644 --- a/src/agentex/lib/core/tracing/failure_attributes.py +++ b/src/agentex/lib/core/tracing/failure_attributes.py @@ -160,9 +160,10 @@ def attributes(self) -> dict[str, Any]: if self.verification is None else [{"step": rule.step, "verifies": rule.verifies} for rule in self.verification], ESCALATION_CHANNELS_KEY: _listed(self.escalation_channels), + # The stored dict can be edited after construction, so check it again before writing it. OUTPUT_SCHEMA_KEY: None if self.output_schema is None - else json.dumps(self.output_schema, separators=(",", ":")), + else json.dumps(_json_object(self.output_schema, "output_schema"), separators=(",", ":")), QUOTE_ARGUMENTS_KEY: _listed(self.quote_arguments), PROMPT_VERSION_KEY: self.prompt_version, } diff --git a/tests/lib/core/tracing/test_failure_attributes.py b/tests/lib/core/tracing/test_failure_attributes.py index 370eed80d..9f3c2d925 100644 --- a/tests/lib/core/tracing/test_failure_attributes.py +++ b/tests/lib/core/tracing/test_failure_attributes.py @@ -242,6 +242,20 @@ def test_a_pathologically_deep_schema_is_rejected(self): with pytest.raises(ValueError, match="must be JSON serializable"): FailureCheckSpec(output_schema={"default": deep}) + @pytest.mark.parametrize( + "edit", [float("nan"), float("inf"), {"a", "b"}, object()], ids=["nan", "inf", "set", "object"] + ) + def test_a_schema_edited_after_construction_fails_the_way_construction_does(self, edit): + with pytest.raises(ValueError) as at_construction: + FailureCheckSpec(output_schema={"edited": edit}) + + spec = FailureCheckSpec(output_schema={"edited": 0}) + assert spec.output_schema is not None + spec.output_schema["edited"] = edit + with pytest.raises(ValueError) as at_render: + spec.attributes() + assert str(at_render.value) == str(at_construction.value) + def test_an_empty_schema_is_a_json_object(self): attributes = FailureCheckSpec(output_schema={}).attributes() assert json.loads(attributes[SCHEMA_KEY]) == {} From c537996238b3134fb7b5c9ceee503bdeacc5ea8e Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Tue, 27 Oct 2026 10:54:47 -0400 Subject: [PATCH 04/17] docs(tracing): guard the yielded span in the failure-check example adk.tracing.span yields None when trace_id is empty or the span fails to start, so the example reading root.data raised AttributeError on that path. It now guards the span the way the tutorial's acp.py does. Co-Authored-By: Claude Opus 5.5 --- examples/tutorials/10_async/00_base/030_tracing/README.md | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/examples/tutorials/10_async/00_base/030_tracing/README.md b/examples/tutorials/10_async/00_base/030_tracing/README.md index ca3dfa261..cc9139b5f 100644 --- a/examples/tutorials/10_async/00_base/030_tracing/README.md +++ b/examples/tutorials/10_async/00_base/030_tracing/README.md @@ -96,12 +96,15 @@ async with adk.tracing.span(trace_id=task.id, name="agent", data=spec.attributes ... # The stop reason is only known at the end. done = dataclasses.replace(spec, stop_reason="completed") - root.data = {**(root.data or {}), **done.attributes()} + if root: + root.data = {**(root.data or {}), **done.attributes()} ``` A span that produces a required output, runs a verification or escalates one carries a single attribute, passed as its `data`: `adk.failure_attributes.output_name("answer")`, -`verification_result("pass")` (or `"fail"`) and `escalates(verify_span.id)`. +`verification_result("pass")` (or `"fail"`) and `escalates(verify_span.id)`. The span +`adk.tracing.span` yields is `None` when `trace_id` is empty or the span fails to start, so +guard it before reading `.data` or `.id`. The SGP export adds `__agent_version__` when `AGENT_VERSION` is set, and `__commit_sha__` when the process starts with `AGENT_COMMIT_SHA` set to a valid git SHA, so only `prompt_version` is yours to From 815dad0522eeaa2f7d6cd6649a4178204f174e2e Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Tue, 27 Oct 2026 10:54:48 -0400 Subject: [PATCH 05/17] test(tracing): pin the attribute each validation error names Only step_budget, stop_reason and prompt_version errors were checked for the attribute they name, so relabelling escalation_channels as quote_arguments still passed. Every root field and VerificationRule now asserts the name its error starts with. Co-Authored-By: Claude Opus 5.5 --- .../core/tracing/test_failure_attributes.py | 27 ++++++++++++++----- 1 file changed, 20 insertions(+), 7 deletions(-) diff --git a/tests/lib/core/tracing/test_failure_attributes.py b/tests/lib/core/tracing/test_failure_attributes.py index 9f3c2d925..65152fd4e 100644 --- a/tests/lib/core/tracing/test_failure_attributes.py +++ b/tests/lib/core/tracing/test_failure_attributes.py @@ -152,8 +152,9 @@ class TestRootSpanValidation: ], ) def test_required_outputs(self, kwargs, message): - with pytest.raises(ValueError, match=message): + with pytest.raises(ValueError, match=message) as raised: FailureCheckSpec(**kwargs) + assert str(raised.value).startswith("required_outputs ") @pytest.mark.parametrize("value", [0, -3, True, False, 2.5, "5", [5]]) def test_step_budget_must_be_a_positive_integer(self, value): @@ -187,8 +188,9 @@ def test_free_text_fields(self, field, value, message): ], ) def test_verification(self, value, message): - with pytest.raises(ValueError, match=message): + with pytest.raises(ValueError, match=message) as raised: FailureCheckSpec(verification=value) + assert str(raised.value).startswith("verification ") def test_one_step_may_verify_several_things_and_one_thing_have_several_verifiers(self): rules = [VerificationRule("check", "a"), VerificationRule("check", "b"), VerificationRule("recheck", "a")] @@ -210,8 +212,9 @@ def test_one_step_may_verify_several_things_and_one_thing_have_several_verifiers ], ) def test_tool_argument_paths(self, field, value, message): - with pytest.raises(ValueError, match=message): + with pytest.raises(ValueError, match=message) as raised: FailureCheckSpec(**{field: value}) + assert str(raised.value).startswith(f"{field} ") @pytest.mark.parametrize( ("value", "message"), @@ -226,8 +229,9 @@ def test_tool_argument_paths(self, field, value, message): ], ) def test_output_schema_must_be_a_json_object(self, value, message): - with pytest.raises(ValueError, match=message): + with pytest.raises(ValueError, match=message) as raised: FailureCheckSpec(output_schema=value) + assert str(raised.value).startswith("output_schema ") def test_a_circular_schema_is_rejected(self): schema: dict[str, Any] = {"type": "object"} @@ -263,11 +267,20 @@ def test_an_empty_schema_is_a_json_object(self): class TestVerificationRule: @pytest.mark.parametrize( - ("step", "verifies"), [("", "a"), ("a", ""), (" ", "a"), ("a", None), (None, "a"), (" a", "b")] + ("step", "verifies", "named"), + [ + ("", "a", "verification step"), + ("a", "", "verification verifies"), + (" ", "a", "verification step"), + ("a", None, "verification verifies"), + (None, "a", "verification step"), + (" a", "b", "verification step"), + ], ) - def test_both_names_must_be_non_empty_strings(self, step, verifies): - with pytest.raises(ValueError): + def test_both_names_must_be_non_empty_strings(self, step, verifies, named): + with pytest.raises(ValueError) as raised: VerificationRule(step, verifies) + assert str(raised.value).startswith(f"{named} ") def test_rules_are_frozen(self): rule = VerificationRule("check", "answer") From c2954a88525765e0f8b3c86d021c1aaf5c4a2314 Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Tue, 27 Oct 2026 11:07:10 -0400 Subject: [PATCH 06/17] test(tracing): name the nesting depth in the deep schema test The depth was an inline number, so a reader could not tell what it controls. It is now a named class constant. Co-Authored-By: Claude Sonnet 5.5 --- tests/lib/core/tracing/test_failure_attributes.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/lib/core/tracing/test_failure_attributes.py b/tests/lib/core/tracing/test_failure_attributes.py index 65152fd4e..d4907e34c 100644 --- a/tests/lib/core/tracing/test_failure_attributes.py +++ b/tests/lib/core/tracing/test_failure_attributes.py @@ -135,6 +135,8 @@ def test_replace_validates_again(self): class TestRootSpanValidation: + NESTING_DEPTH_PAST_RECURSION_LIMIT = 200_000 + @pytest.mark.parametrize( ("kwargs", "message"), [ @@ -241,7 +243,7 @@ def test_a_circular_schema_is_rejected(self): def test_a_pathologically_deep_schema_is_rejected(self): deep: list[Any] = [] - for _ in range(200_000): + for _ in range(self.NESTING_DEPTH_PAST_RECURSION_LIMIT): deep = [deep] with pytest.raises(ValueError, match="must be JSON serializable"): FailureCheckSpec(output_schema={"default": deep}) From 7b18bb82b74509c592c90243cdb597199ebd68d6 Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Tue, 27 Oct 2026 11:15:26 -0400 Subject: [PATCH 07/17] fix(tracing): say that nested tool argument paths are not supported A path like send_email.to.cc was rejected with a message that did not say why. Tool names cannot contain dots, so an entry is a tool and one top-level argument, and the error now says nested argument paths are not supported. Co-Authored-By: Claude Sonnet 5.5 --- src/agentex/lib/core/tracing/failure_attributes.py | 4 +++- tests/lib/core/tracing/test_failure_attributes.py | 14 ++++++++++++++ 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/src/agentex/lib/core/tracing/failure_attributes.py b/src/agentex/lib/core/tracing/failure_attributes.py index 5db792305..1799fdcf6 100644 --- a/src/agentex/lib/core/tracing/failure_attributes.py +++ b/src/agentex/lib/core/tracing/failure_attributes.py @@ -44,6 +44,7 @@ QUOTE_ARGUMENTS_KEY = "sgp.agent.quote_arguments" PROMPT_VERSION_KEY = "sgp.agent.prompt_version" +# Exactly one dot: OpenAI tool names cannot contain dots, so this is a tool and one top-level argument. _TOOL_ARGUMENT_RE = re.compile(r"[^.\s]+\.[^.\s]+") _VERIFICATION_RESULTS: tuple[str, ...] = ("pass", "fail") @@ -72,7 +73,8 @@ def _tool_arguments(values: object, what: str) -> tuple[str, ...]: items = _unique_texts(values, what) for item in items: if _TOOL_ARGUMENT_RE.fullmatch(item) is None: - raise ValueError(f"{what} entry must look like tool.argument, got {item!r}") + hint = ", nested argument paths are not supported" if item.count(".") > 1 else "" + raise ValueError(f"{what} entry must look like tool.argument{hint}, got {item!r}") return items diff --git a/tests/lib/core/tracing/test_failure_attributes.py b/tests/lib/core/tracing/test_failure_attributes.py index d4907e34c..bb6247709 100644 --- a/tests/lib/core/tracing/test_failure_attributes.py +++ b/tests/lib/core/tracing/test_failure_attributes.py @@ -218,6 +218,20 @@ def test_tool_argument_paths(self, field, value, message): FailureCheckSpec(**{field: value}) assert str(raised.value).startswith(f"{field} ") + @pytest.mark.parametrize("field", ["escalation_channels", "quote_arguments"]) + @pytest.mark.parametrize("path", ["send_email.to.cc", "a.b.c.d"]) + def test_nested_argument_paths_are_rejected_and_the_error_says_so(self, field, path): + with pytest.raises(ValueError, match="nested argument paths are not supported") as raised: + FailureCheckSpec(**{field: [path]}) + assert str(raised.value).startswith(f"{field} entry must look like tool.argument") + assert repr(path) in str(raised.value) + + @pytest.mark.parametrize("path", ["send_email", ".to", "send_email.", "send email.to"]) + def test_a_path_that_is_not_nested_is_not_blamed_on_nesting(self, path): + with pytest.raises(ValueError, match="must look like tool.argument") as raised: + FailureCheckSpec(escalation_channels=[path]) + assert "nested" not in str(raised.value) + @pytest.mark.parametrize( ("value", "message"), [ From 2118a97c7f9c7f929bda1d4cdd7d1e8b5faa68d1 Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Tue, 27 Oct 2026 11:16:07 -0400 Subject: [PATCH 08/17] fix(tracing): name escalates in its span id error escalates("") failed with a message about a verification span id, which did not name the attribute being written. The message now starts with escalates, as the other validation errors start with their attribute. Co-Authored-By: Claude Sonnet 5.5 --- src/agentex/lib/core/tracing/failure_attributes.py | 2 +- tests/lib/core/tracing/test_failure_attributes.py | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/src/agentex/lib/core/tracing/failure_attributes.py b/src/agentex/lib/core/tracing/failure_attributes.py index 1799fdcf6..e2c75a717 100644 --- a/src/agentex/lib/core/tracing/failure_attributes.py +++ b/src/agentex/lib/core/tracing/failure_attributes.py @@ -186,4 +186,4 @@ def verification_result(result: Literal["pass", "fail"]) -> dict[str, str]: def escalates(verification_span_id: str) -> dict[str, str]: """Data for an escalation span, naming the verification span it escalates.""" - return {ESCALATES_KEY: _text(verification_span_id, "verification span id")} + return {ESCALATES_KEY: _text(verification_span_id, "escalates verification span id")} diff --git a/tests/lib/core/tracing/test_failure_attributes.py b/tests/lib/core/tracing/test_failure_attributes.py index bb6247709..e73784bf4 100644 --- a/tests/lib/core/tracing/test_failure_attributes.py +++ b/tests/lib/core/tracing/test_failure_attributes.py @@ -329,8 +329,9 @@ def test_escalates(self): @pytest.mark.parametrize("value", ["", " ", None, 3, " span-1"]) def test_escalates_rejects_bad_span_ids(self, value): - with pytest.raises(ValueError, match="verification span id"): + with pytest.raises(ValueError, match="verification span id") as raised: escalates(value) + assert str(raised.value).startswith("escalates ") def test_results_merge_into_a_span_data_dict(self): data = {"note": "kept", **output_name("answer")} From 57b161d461e4a965127de3401bf7762a2db1f993 Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Tue, 27 Oct 2026 11:17:03 -0400 Subject: [PATCH 09/17] fix(tracing): reject non-string keys in the output schema JSON writes every object key as a string, so the encoder quietly turned an int, bool, float or None key into text and let {1: "a", "1": "b"} collapse into one entry. Any non-string mapping key in the schema is now rejected, at construction and again when attributes render. Co-Authored-By: Claude Sonnet 5.5 --- .../lib/core/tracing/failure_attributes.py | 24 ++++++++++++++- .../core/tracing/test_failure_attributes.py | 29 +++++++++++++++++++ 2 files changed, 52 insertions(+), 1 deletion(-) diff --git a/src/agentex/lib/core/tracing/failure_attributes.py b/src/agentex/lib/core/tracing/failure_attributes.py index e2c75a717..03587c8e3 100644 --- a/src/agentex/lib/core/tracing/failure_attributes.py +++ b/src/agentex/lib/core/tracing/failure_attributes.py @@ -9,7 +9,7 @@ import re import json -from typing import Any, Literal, Callable +from typing import Any, Literal, Callable, cast from dataclasses import field, dataclass from collections.abc import Sequence @@ -84,9 +84,31 @@ def _step_budget(value: object) -> int: return value +def _require_string_keys(value: object, what: str) -> None: + # Iterative, so a very deep schema is rejected by the encoder in _json_object instead of overflowing the stack here. + pending: list[object] = [value] + seen: set[int] = set() + while pending: + node = pending.pop() + if isinstance(node, dict): + if id(node) in seen: + continue + seen.add(id(node)) + for key, child in cast("dict[object, object]", node).items(): + if not isinstance(key, str): + raise ValueError(f"{what} keys must be strings, got {key!r}") + pending.append(child) + elif isinstance(node, list | tuple): + if id(node) in seen: + continue + seen.add(id(node)) + pending.extend(cast("Sequence[object]", node)) + + def _json_object(value: object, what: str) -> dict[str, Any]: if not isinstance(value, dict): raise ValueError(f"{what} must be a JSON object (a dict), got {type(value).__name__}") + _require_string_keys(value, what) try: return json.loads(json.dumps(value, allow_nan=False)) except (TypeError, ValueError, RecursionError) as exc: diff --git a/tests/lib/core/tracing/test_failure_attributes.py b/tests/lib/core/tracing/test_failure_attributes.py index e73784bf4..825fc6147 100644 --- a/tests/lib/core/tracing/test_failure_attributes.py +++ b/tests/lib/core/tracing/test_failure_attributes.py @@ -249,6 +249,35 @@ def test_output_schema_must_be_a_json_object(self, value, message): FailureCheckSpec(output_schema=value) assert str(raised.value).startswith("output_schema ") + @pytest.mark.parametrize( + "schema", + [ + {1: "a"}, + {1: "a", "1": "b"}, + {True: "a"}, + {None: "a"}, + {1.5: "a"}, + {"properties": {2: {"type": "string"}}}, + {"anyOf": [{"properties": {3: {}}}]}, + ], + ids=["int", "int-collides-with-str", "bool", "none", "float", "nested", "nested-in-list"], + ) + def test_output_schema_keys_must_be_strings(self, schema): + with pytest.raises(ValueError, match="keys must be strings") as raised: + FailureCheckSpec(output_schema=schema) + assert str(raised.value).startswith("output_schema ") + + def test_a_non_string_key_added_after_construction_fails_the_way_construction_does(self): + with pytest.raises(ValueError) as at_construction: + FailureCheckSpec(output_schema={1: "a"}) + + spec = FailureCheckSpec(output_schema={"type": "object"}) + assert spec.output_schema is not None + spec.output_schema[1] = "a" + with pytest.raises(ValueError) as at_render: + spec.attributes() + assert str(at_render.value) == str(at_construction.value) + def test_a_circular_schema_is_rejected(self): schema: dict[str, Any] = {"type": "object"} schema["self"] = schema From 56d747dc4b18f58a229fce24403e79f0b9d4f00b Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Wed, 28 Oct 2026 07:51:48 -0400 Subject: [PATCH 10/17] fix(tracing): mark harness tool spans as tool calls A successful Claude Code or Codex tool call reports no is_error, so the failure checks could not tell its span was a tool call. Every harness tool span now carries gen_ai.operation.name=execute_tool and gen_ai.tool.name. Co-Authored-By: Claude Sonnet 5.5 --- src/agentex/lib/core/harness/tracer.py | 10 +++- tests/lib/core/harness/_fakes.py | 5 ++ .../harness/test_harness_claude_code_sync.py | 21 ++++++++- .../core/harness/test_harness_codex_sync.py | 21 ++++++++- tests/lib/core/harness/test_tracer.py | 46 +++++++++++++++++-- tests/lib/core/harness/test_tracer_lineage.py | 16 ++++++- 6 files changed, 110 insertions(+), 9 deletions(-) diff --git a/src/agentex/lib/core/harness/tracer.py b/src/agentex/lib/core/harness/tracer.py index 34cd95616..4b2cd6576 100644 --- a/src/agentex/lib/core/harness/tracer.py +++ b/src/agentex/lib/core/harness/tracer.py @@ -26,6 +26,10 @@ def merge_refs_into_data(data: dict[str, Any] | None, refs: list[dict[str, Any]] logger = logging.getLogger(__name__) +_OPERATION_NAME_KEY = "gen_ai.operation.name" +_TOOL_NAME_KEY = "gen_ai.tool.name" +_EXECUTE_TOOL = "execute_tool" + def _as_span_payload(value: Any, *, key: str) -> Any: """Coerce a span input/output payload into a dict. @@ -92,10 +96,12 @@ async def handle(self, signal: SpanSignal) -> None: ) if span is not None: if signal.kind == "tool": + data = span.data if isinstance(span.data, dict) else {} refs = resolve_refs(signal.name, signal.input if isinstance(signal.input, dict) else {}) if refs: - data = span.data if isinstance(span.data, dict) else {} - span.data = merge_refs_into_data(data, refs) + data = merge_refs_into_data(data, refs) + # A successful call reports no is_error, so the failure checks recognise a tool by these keys. + span.data = {**data, _OPERATION_NAME_KEY: _EXECUTE_TOOL, _TOOL_NAME_KEY: signal.name} self._open[signal.key] = span elif isinstance(signal, CloseSpan): span = self._open.pop(signal.key, None) diff --git a/tests/lib/core/harness/_fakes.py b/tests/lib/core/harness/_fakes.py index f9fd34a45..1d2487bd9 100644 --- a/tests/lib/core/harness/_fakes.py +++ b/tests/lib/core/harness/_fakes.py @@ -20,6 +20,11 @@ from typing import Any +def tool_marker(tool: str) -> dict[str, str]: + """The span data the offline failure checks read to recognise a tool call.""" + return {"gen_ai.operation.name": "execute_tool", "gen_ai.tool.name": tool} + + class FakeSpan: def __init__(self, name: str) -> None: self.name = name diff --git a/tests/lib/core/harness/test_harness_claude_code_sync.py b/tests/lib/core/harness/test_harness_claude_code_sync.py index b53485499..1fe8c469a 100644 --- a/tests/lib/core/harness/test_harness_claude_code_sync.py +++ b/tests/lib/core/harness/test_harness_claude_code_sync.py @@ -51,7 +51,7 @@ from agentex.types.tool_response_content import ToolResponseContent from agentex.lib.adk._modules._claude_code_turn import ClaudeCodeTurn -from ._fakes import FakeTracing +from ._fakes import FakeTracing, tool_marker # --------------------------------------------------------------------------- # Native claude-code envelope fixtures (copied from the turn + conformance tests) @@ -243,6 +243,25 @@ async def test_tool_span_output_is_tool_result(self) -> None: assert name == "Read" assert "72F" in str(output) + async def test_successful_tool_span_is_marked_as_a_tool_call(self) -> None: + fake_tracing = FakeTracing() + await _run_yield_turn( + _tool_then_text_envelopes(), + trace_id="trace1", + parent_span_id="parent-span", + fake_tracing=fake_tracing, + ) + (tool_span,) = fake_tracing.ended_spans + assert tool_span.data == tool_marker("Read") + + async def test_failed_tool_span_is_marked_as_a_tool_call(self) -> None: + envelopes = _tool_then_text_envelopes() + envelopes[1]["message"]["content"][0]["is_error"] = True + fake_tracing = FakeTracing() + await _run_yield_turn(envelopes, trace_id="trace1", parent_span_id="parent-span", fake_tracing=fake_tracing) + (tool_span,) = fake_tracing.ended_spans + assert tool_span.data == tool_marker("Read") + async def test_reasoning_span_for_thinking_block(self) -> None: """A thinking block opens and closes a reasoning span.""" fake_tracing = FakeTracing() diff --git a/tests/lib/core/harness/test_harness_codex_sync.py b/tests/lib/core/harness/test_harness_codex_sync.py index 6129716ee..56eed0588 100644 --- a/tests/lib/core/harness/test_harness_codex_sync.py +++ b/tests/lib/core/harness/test_harness_codex_sync.py @@ -55,7 +55,7 @@ from agentex.types.tool_response_content import ToolResponseContent from agentex.lib.adk._modules._codex_turn import CodexTurn -from ._fakes import FakeTracing +from ._fakes import FakeTracing, tool_marker # --------------------------------------------------------------------------- # Native codex event fixtures (copied from the turn + conformance tests) @@ -225,6 +225,25 @@ async def test_tool_span_output_is_command_output(self) -> None: _name, output = fake_tracing.ended[0] assert "72F" in str(output) + async def test_successful_tool_span_is_marked_as_a_tool_call(self) -> None: + fake_tracing = FakeTracing() + await _run_yield_turn( + _tool_then_text_events(), + trace_id="trace1", + parent_span_id="parent-span", + fake_tracing=fake_tracing, + ) + (tool_span,) = fake_tracing.ended_spans + assert tool_span.data == tool_marker("bash") + + async def test_failed_tool_span_is_marked_as_a_tool_call(self) -> None: + events = _tool_then_text_events() + events[3]["item"]["exit_code"] = 1 + fake_tracing = FakeTracing() + await _run_yield_turn(events, trace_id="trace1", parent_span_id="parent-span", fake_tracing=fake_tracing) + (tool_span,) = fake_tracing.ended_spans + assert tool_span.data == tool_marker("bash") + async def test_reasoning_span_opened_then_done_closed(self) -> None: """A codex reasoning item emits Start+Delta+Done: the reasoning span opens and is closed normally when the Done event is observed.""" diff --git a/tests/lib/core/harness/test_tracer.py b/tests/lib/core/harness/test_tracer.py index 9bd17b90c..44ef5d608 100644 --- a/tests/lib/core/harness/test_tracer.py +++ b/tests/lib/core/harness/test_tracer.py @@ -5,7 +5,7 @@ from agentex.lib.core.harness.types import OpenSpan, CloseSpan from agentex.lib.core.harness.tracer import SpanTracer -from ._fakes import FakeTracing +from ._fakes import FakeTracing, tool_marker @pytest.mark.asyncio @@ -50,16 +50,54 @@ async def test_close_records_is_error_on_span_data(): tracer = SpanTracer(trace_id="t1", parent_span_id="p1", tracing=fake) await tracer.handle(OpenSpan(key="call_err", kind="tool", name="Bash", input={})) await tracer.handle(CloseSpan(key="call_err", output="boom", is_complete=True, is_error=True)) - assert fake.ended_spans[0].data == {"is_error": True} + assert fake.ended_spans[0].data == {**tool_marker("Bash"), "is_error": True} @pytest.mark.asyncio -async def test_close_without_status_leaves_span_data_untouched(): - """is_error=None (no status reported) must not write to span.data.""" +async def test_close_without_status_writes_no_is_error(): + """is_error=None (no status reported) must not write is_error to span.data.""" fake = FakeTracing() tracer = SpanTracer(trace_id="t1", parent_span_id="p1", tracing=fake) await tracer.handle(OpenSpan(key="call_1", kind="tool", name="Bash", input={})) await tracer.handle(CloseSpan(key="call_1", output="files", is_complete=True)) + assert "is_error" not in fake.ended_spans[0].data + + +@pytest.mark.asyncio +async def test_successful_tool_span_is_marked_as_a_tool_call(): + fake = FakeTracing() + tracer = SpanTracer(trace_id="t1", parent_span_id="p1", tracing=fake) + await tracer.handle(OpenSpan(key="call_1", kind="tool", name="Bash", input={"cmd": "ls"})) + await tracer.handle(CloseSpan(key="call_1", output="files", is_complete=True)) + assert fake.ended_spans[0].data == {"gen_ai.operation.name": "execute_tool", "gen_ai.tool.name": "Bash"} + + +@pytest.mark.asyncio +async def test_incomplete_tool_span_keeps_the_tool_marker(): + fake = FakeTracing() + tracer = SpanTracer(trace_id="t1", parent_span_id="p1", tracing=fake) + await tracer.handle(OpenSpan(key="call_1", kind="tool", name="Bash", input={})) + await tracer.handle(CloseSpan(key="call_1", output=None, is_complete=False)) + assert fake.ended_spans[0].data == tool_marker("Bash") + + +@pytest.mark.asyncio +async def test_each_tool_span_names_its_own_tool(): + fake = FakeTracing() + tracer = SpanTracer(trace_id="t1", parent_span_id="p1", tracing=fake) + await tracer.handle(OpenSpan(key="a", kind="tool", name="Read", input={})) + await tracer.handle(OpenSpan(key="b", kind="tool", name="Bash", input={})) + await tracer.handle(CloseSpan(key="b", output="x")) + await tracer.handle(CloseSpan(key="a", output="y")) + assert [span.data for span in fake.ended_spans] == [tool_marker("Bash"), tool_marker("Read")] + + +@pytest.mark.asyncio +async def test_reasoning_span_is_not_marked_as_a_tool_call(): + fake = FakeTracing() + tracer = SpanTracer(trace_id="t1", parent_span_id="p1", tracing=fake) + await tracer.handle(OpenSpan(key="reasoning:0", kind="reasoning", name="reasoning", input={})) + await tracer.handle(CloseSpan(key="reasoning:0", output="thought", is_complete=True)) assert fake.ended_spans[0].data is None diff --git a/tests/lib/core/harness/test_tracer_lineage.py b/tests/lib/core/harness/test_tracer_lineage.py index 75799caee..2a42ea51b 100644 --- a/tests/lib/core/harness/test_tracer_lineage.py +++ b/tests/lib/core/harness/test_tracer_lineage.py @@ -11,7 +11,7 @@ register_tool_sources, ) -from ._fakes import FakeTracing +from ._fakes import FakeTracing, tool_marker @pytest.fixture(autouse=True) @@ -51,3 +51,17 @@ async def test_unregistered_tool_and_reasoning_spans_carry_no_refs(): for span in fake.ended_spans: assert not (isinstance(span.data, dict) and LINEAGE_REFS_KEY in span.data) + + +@pytest.mark.asyncio +async def test_tool_span_keeps_the_tool_marker_beside_the_refs(): + register_tool_sources("query_guidance", refs=[DataSourceRef("databricks://ey-tax", "guidance.rulings")]) + fake = FakeTracing() + tracer = SpanTracer(trace_id="t1", parent_span_id="p1", tracing=fake) + + await tracer.handle(OpenSpan(key="c1", kind="tool", name="query_guidance", input={})) + await tracer.handle(CloseSpan(key="c1", output={"ok": True}, is_complete=True)) + + (span,) = fake.ended_spans + assert {key: span.data[key] for key in tool_marker("query_guidance")} == tool_marker("query_guidance") + assert [ref["name"] for ref in span.data[LINEAGE_REFS_KEY]] == ["guidance.rulings"] From 2417e3466cee762e78450d7559fd9f741e9d63d0 Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Wed, 28 Oct 2026 07:51:49 -0400 Subject: [PATCH 11/17] fix(tracing): record non-string OpenAI tool results as JSON The hooks wrote str(result), a Python repr the failure checks cannot decode. Objects are now written as JSON text, strings stay as they are, and str is the fallback only when the result cannot be serialized. Co-Authored-By: Claude Sonnet 5.5 --- .../plugins/openai_agents/hooks/hooks.py | 12 +- .../openai_agents/test_hooks_tool_span.py | 104 ++++++++++++++++++ 2 files changed, 115 insertions(+), 1 deletion(-) create mode 100644 tests/lib/core/temporal/plugins/openai_agents/test_hooks_tool_span.py diff --git a/src/agentex/lib/core/temporal/plugins/openai_agents/hooks/hooks.py b/src/agentex/lib/core/temporal/plugins/openai_agents/hooks/hooks.py index 30d358cc9..4e85a0fae 100644 --- a/src/agentex/lib/core/temporal/plugins/openai_agents/hooks/hooks.py +++ b/src/agentex/lib/core/temporal/plugins/openai_agents/hooks/hooks.py @@ -46,6 +46,16 @@ _MAX_SPAN_OUTPUT_CHARS = 2000 +def _result_text(result: Any) -> str: + """The failure checks decode a result as JSON, so ``str`` is only the fallback when it cannot be serialized.""" + if isinstance(result, str): + return result + try: + return json.dumps(result) + except (TypeError, ValueError): + return str(result) + + def _get_adk() -> Any: """Lazily import the adk facade for workflow-safe tracing. @@ -355,7 +365,7 @@ async def _maybe_end_tool_span(self, tool_call_id: str, result: Any) -> None: if span is None or not self.trace_id: return try: - span.output = {"result": str(result)[:_MAX_SPAN_OUTPUT_CHARS]} + span.output = {"result": _result_text(result)[:_MAX_SPAN_OUTPUT_CHARS]} await _get_adk().tracing.end_span( trace_id=self.trace_id, span=span, diff --git a/tests/lib/core/temporal/plugins/openai_agents/test_hooks_tool_span.py b/tests/lib/core/temporal/plugins/openai_agents/test_hooks_tool_span.py new file mode 100644 index 000000000..2495ff3d1 --- /dev/null +++ b/tests/lib/core/temporal/plugins/openai_agents/test_hooks_tool_span.py @@ -0,0 +1,104 @@ +"""The tool result the OpenAI hooks record on a tool span. + +The failure checks decode the recorded text as JSON, so an object result has to be written as JSON. +""" + +from __future__ import annotations + +import json +from types import SimpleNamespace +from typing import Any, override +from datetime import UTC, datetime + +import pytest + +import agentex.lib.core.temporal.plugins.openai_agents.hooks.hooks as hooks_module +from agentex.types.span import Span +from agentex.lib.core.temporal.plugins.openai_agents.hooks.hooks import TemporalStreamingHooks + +pytestmark = pytest.mark.asyncio + + +class _FakeTracing: + def __init__(self) -> None: + self.ended: list[Span] = [] + + async def start_span( + self, *, trace_id: str, name: str, input: Any = None, parent_id: str | None = None, **_: Any + ) -> Span: + return Span( + id="span-1", name=name, start_time=datetime.now(UTC), trace_id=trace_id, parent_id=parent_id, input=input + ) + + async def end_span(self, *, trace_id: str, span: Span, **_: Any) -> Span: + self.ended.append(span) + return span + + +@pytest.fixture +def tracing(monkeypatch: pytest.MonkeyPatch) -> _FakeTracing: + fake = _FakeTracing() + monkeypatch.setattr(hooks_module, "_get_adk", lambda: SimpleNamespace(tracing=fake)) + return fake + + +async def _recorded_output(tracing: _FakeTracing, result: Any) -> Any: + hooks = TemporalStreamingHooks(task_id="task-1", trace_id="trace-1") + await hooks._maybe_start_tool_span("call-1", "lookup", {}) + await hooks._maybe_end_tool_span("call-1", result) + (span,) = tracing.ended + return span.output + + +async def test_object_result_is_recorded_as_json(tracing: _FakeTracing) -> None: + result = {"city": "Paris", "temps": [12, 15], "raining": False, "note": None} + output = await _recorded_output(tracing, result) + assert output == {"result": '{"city": "Paris", "temps": [12, 15], "raining": false, "note": null}'} + assert json.loads(output["result"]) == result + + +async def test_list_result_is_recorded_as_json(tracing: _FakeTracing) -> None: + output = await _recorded_output(tracing, [1, "two", {"three": 3}]) + assert output == {"result": '[1, "two", {"three": 3}]'} + + +@pytest.mark.parametrize(("result", "text"), [(None, "null"), (True, "true"), (3, "3"), (1.5, "1.5")]) +async def test_scalar_result_is_recorded_as_json(tracing: _FakeTracing, result: Any, text: str) -> None: + assert await _recorded_output(tracing, result) == {"result": text} + + +@pytest.mark.parametrize("text", ["plain text", '{"already": "json"}', ""]) +async def test_string_result_is_recorded_as_it_is(tracing: _FakeTracing, text: str) -> None: + assert await _recorded_output(tracing, text) == {"result": text} + + +class _Opaque: + @override + def __str__(self) -> str: + return "opaque result" + + +def _circular() -> dict[str, Any]: + result: dict[str, Any] = {} + result["self"] = result + return result + + +@pytest.mark.parametrize( + ("result", "text"), [(_Opaque(), "opaque result"), (_circular(), "{'self': {...}}")], ids=["object", "circular"] +) +async def test_result_that_cannot_be_serialized_falls_back_to_str( + tracing: _FakeTracing, result: Any, text: str +) -> None: + assert await _recorded_output(tracing, result) == {"result": text} + + +async def test_json_result_is_capped(tracing: _FakeTracing) -> None: + result = {"rows": ["x" * 100] * 50} + output = await _recorded_output(tracing, result) + assert len(output["result"]) == 2000 + assert json.dumps(result).startswith(output["result"]) + + +async def test_string_result_is_capped(tracing: _FakeTracing) -> None: + assert await _recorded_output(tracing, "y" * 5000) == {"result": "y" * 2000} From 8af1a380281a1fd467eb70072b599d2ec9bc23ff Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Wed, 28 Oct 2026 20:40:57 -0400 Subject: [PATCH 12/17] fix(tracing): let one escalation name every failure it escalates The escalation check reads a list of span ids, so escalates() now takes several and writes them as a list. Co-Authored-By: Claude Opus 5.5 --- .../10_async/00_base/030_tracing/README.md | 3 +- .../lib/core/tracing/failure_attributes.py | 8 +++-- .../core/tracing/test_failure_attributes.py | 30 ++++++++++++++----- 3 files changed, 30 insertions(+), 11 deletions(-) diff --git a/examples/tutorials/10_async/00_base/030_tracing/README.md b/examples/tutorials/10_async/00_base/030_tracing/README.md index cc9139b5f..705002cfd 100644 --- a/examples/tutorials/10_async/00_base/030_tracing/README.md +++ b/examples/tutorials/10_async/00_base/030_tracing/README.md @@ -102,7 +102,8 @@ async with adk.tracing.span(trace_id=task.id, name="agent", data=spec.attributes A span that produces a required output, runs a verification or escalates one carries a single attribute, passed as its `data`: `adk.failure_attributes.output_name("answer")`, -`verification_result("pass")` (or `"fail"`) and `escalates(verify_span.id)`. The span +`verification_result("pass")` (or `"fail"`) and `escalates(verify_span.id)`, which takes several +span ids when one escalation covers several failures. The span `adk.tracing.span` yields is `None` when `trace_id` is empty or the span fails to start, so guard it before reading `.data` or `.id`. diff --git a/src/agentex/lib/core/tracing/failure_attributes.py b/src/agentex/lib/core/tracing/failure_attributes.py index 03587c8e3..ef8395b9b 100644 --- a/src/agentex/lib/core/tracing/failure_attributes.py +++ b/src/agentex/lib/core/tracing/failure_attributes.py @@ -206,6 +206,8 @@ def verification_result(result: Literal["pass", "fail"]) -> dict[str, str]: return {VERIFICATION_RESULT_KEY: result} -def escalates(verification_span_id: str) -> dict[str, str]: - """Data for an escalation span, naming the verification span it escalates.""" - return {ESCALATES_KEY: _text(verification_span_id, "escalates verification span id")} +def escalates(*verification_span_ids: str) -> dict[str, list[str]]: + """Data for an escalation span, naming every verification span it escalates.""" + if not verification_span_ids: + raise ValueError("escalates needs at least one verification span id") + return {ESCALATES_KEY: list(_unique_texts(verification_span_ids, "escalates verification span id"))} diff --git a/tests/lib/core/tracing/test_failure_attributes.py b/tests/lib/core/tracing/test_failure_attributes.py index 825fc6147..7a82be084 100644 --- a/tests/lib/core/tracing/test_failure_attributes.py +++ b/tests/lib/core/tracing/test_failure_attributes.py @@ -353,13 +353,22 @@ def test_verification_result_accepts_only_pass_or_fail(self, value): def test_escalates(self): assert escalates("3f2c1a9e-0000-4000-8000-000000000001") == { - "sgp.agent.escalates": "3f2c1a9e-0000-4000-8000-000000000001" + "sgp.agent.escalates": ["3f2c1a9e-0000-4000-8000-000000000001"] } + def test_one_escalation_names_every_failure_it_escalates(self): + assert escalates("verify-1", "verify-2") == {"sgp.agent.escalates": ["verify-1", "verify-2"]} + @pytest.mark.parametrize("value", ["", " ", None, 3, " span-1"]) def test_escalates_rejects_bad_span_ids(self, value): with pytest.raises(ValueError, match="verification span id") as raised: - escalates(value) + escalates("verify-1", value) + assert str(raised.value).startswith("escalates ") + + @pytest.mark.parametrize("ids", [(), ("verify-1", "verify-1")]) + def test_escalates_rejects_no_or_repeated_span_ids(self, ids): + with pytest.raises(ValueError, match="verification span id") as raised: + escalates(*ids) assert str(raised.value).startswith("escalates ") def test_results_merge_into_a_span_data_dict(self): @@ -395,7 +404,7 @@ def create(**kwargs: Any): def _assert_attributes_landed_in_sgp_metadata(items: list[dict[str, Any]]) -> None: by_name = {item["name"]: item for item in items} - assert sorted(by_name) == ["agent", "answer", "escalate", "verify"] + assert sorted(by_name) == ["agent", "answer", "escalate", "recheck", "verify"] root = by_name["agent"]["metadata"] declared = {key: value for key, value in root.items() if key.startswith("sgp.agent.")} @@ -406,8 +415,11 @@ def _assert_attributes_landed_in_sgp_metadata(items: list[dict[str, Any]]) -> No assert root["__agent_version__"] == "sha-abc1234" assert by_name["answer"]["metadata"]["sgp.agent.output_name"] == "answer" - assert by_name["verify"]["metadata"]["sgp.agent.verification_result"] == "fail" - assert by_name["escalate"]["metadata"]["sgp.agent.escalates"] == by_name["verify"]["id"] + for verify in ("verify", "recheck"): + assert by_name[verify]["metadata"]["sgp.agent.verification_result"] == "fail" + # The check reads a list of span ids, so one escalation covers both failures. + escalated = [by_name["verify"]["id"], by_name["recheck"]["id"]] + assert by_name["escalate"]["metadata"]["sgp.agent.escalates"] == escalated for item in items: assert json.loads(json.dumps(item["metadata"])) == item["metadata"] @@ -434,7 +446,9 @@ async def test_async_tracer_delivers_attributes_as_flat_keys_with_native_values( pass async with trace.span(name="verify", parent_id=root.id, data=verification_result("fail")) as verify: pass - async with trace.span(name="escalate", parent_id=root.id, data=escalates(verify.id)): + async with trace.span(name="recheck", parent_id=root.id, data=verification_result("fail")) as recheck: + pass + async with trace.span(name="escalate", parent_id=root.id, data=escalates(verify.id, recheck.id)): pass stopped = dataclasses.replace(spec, stop_reason="budget_exhausted") assert isinstance(root.data, dict) @@ -463,7 +477,9 @@ def test_sync_tracer_delivers_attributes_as_flat_keys_with_native_values(self, m pass with trace.span(name="verify", parent_id=root.id, data=verification_result("fail")) as verify: pass - with trace.span(name="escalate", parent_id=root.id, data=escalates(verify.id)): + with trace.span(name="recheck", parent_id=root.id, data=verification_result("fail")) as recheck: + pass + with trace.span(name="escalate", parent_id=root.id, data=escalates(verify.id, recheck.id)): pass stopped = dataclasses.replace(spec, stop_reason="budget_exhausted") assert isinstance(root.data, dict) From 410c7f530a8aedb432e17691b74516aec7b5456b Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Wed, 28 Oct 2026 20:41:26 -0400 Subject: [PATCH 13/17] fix(tracing): mark a span cancelled mid-call as failed The span context managers caught only Exception, so a cancelled or interrupted call ended its span as SUCCESS. They now record any BaseException and re-raise it. Co-Authored-By: Claude Opus 5.5 --- src/agentex/lib/adk/_modules/tracing.py | 2 +- src/agentex/lib/core/tracing/trace.py | 6 ++- tests/lib/core/tracing/test_span_error.py | 45 +++++++++++++++++++++++ tests/test_adk_tracing_span_error.py | 38 ++++++++++++++++++- 4 files changed, 87 insertions(+), 4 deletions(-) diff --git a/src/agentex/lib/adk/_modules/tracing.py b/src/agentex/lib/adk/_modules/tracing.py index 9b89d076e..45ad9359c 100644 --- a/src/agentex/lib/adk/_modules/tracing.py +++ b/src/agentex/lib/adk/_modules/tracing.py @@ -237,7 +237,7 @@ async def span( ) try: yield span - except Exception as exc: + except BaseException as exc: # Record the failure on the span so the obs span reflects the error # instead of a false green. Agents use THIS context manager (not # AsyncTrace.span, which is the only other place set_span_error is diff --git a/src/agentex/lib/core/tracing/trace.py b/src/agentex/lib/core/tracing/trace.py index 8f5260913..48a88aa58 100644 --- a/src/agentex/lib/core/tracing/trace.py +++ b/src/agentex/lib/core/tracing/trace.py @@ -371,7 +371,8 @@ def span( span = self.start_span(name, parent_id, input, data, task_id=task_id) try: yield span - except Exception as exc: + except BaseException as exc: + # A cancelled call did not succeed, and cancellation is not an Exception. set_span_error(span, exc) raise finally: @@ -560,7 +561,8 @@ async def span( span = await self.start_span(name, parent_id, input, data, task_id=task_id) try: yield span - except Exception as exc: + except BaseException as exc: + # A cancelled call did not succeed, and cancellation is not an Exception. set_span_error(span, exc) raise finally: diff --git a/tests/lib/core/tracing/test_span_error.py b/tests/lib/core/tracing/test_span_error.py index 02e9645a4..1012c5e57 100644 --- a/tests/lib/core/tracing/test_span_error.py +++ b/tests/lib/core/tracing/test_span_error.py @@ -1,12 +1,14 @@ from __future__ import annotations import uuid +import asyncio from typing import Any from datetime import UTC, datetime from unittest.mock import MagicMock, patch import pytest from scale_gp_beta.lib.tracing import ( + create_span as sgp_create_span, PlatformError as SGPPlatformError, ApplicationError as SGPApplicationError, CategorizedError as SGPCategorizedError, @@ -25,6 +27,21 @@ PROCESSOR_MODULE = "agentex.lib.core.tracing.processors.sgp_tracing_processor" +# The failure checks read a span with either status as failed. +FAILED_STATUSES = {"ERROR", "CANCELED"} + + +def _sgp_status(span: Span) -> str: + """The status the span is written to SGP with.""" + from agentex.lib.core.tracing.processors.sgp_tracing_processor import _build_sgp_span + + def create(**kwargs: Any) -> Any: + return sgp_create_span(**kwargs, queue_manager=MagicMock()) + + env = MagicMock(ACP_TYPE=None, AGENT_NAME=None, AGENT_ID=None, AGENT_VERSION=None) + with patch(f"{PROCESSOR_MODULE}.create_span", side_effect=create): + return _build_sgp_span(span, env).status + def _make_span(data=None) -> Span: return Span( @@ -151,6 +168,34 @@ async def test_async_span_records_error_and_reraises(self): "category": "unknown", } + @pytest.mark.asyncio + async def test_async_span_cancelled_mid_call_reads_failed(self): + trace = AsyncTrace(processors=[], client=MagicMock(), trace_id="t1") + captured = {} + entered = asyncio.Event() + + async def call() -> None: + async with trace.span("op") as span: + captured["span"] = span + entered.set() + await asyncio.Event().wait() + + task = asyncio.create_task(call()) + await entered.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert _sgp_status(captured["span"]) in FAILED_STATUSES + + def test_sync_span_interrupted_mid_call_reads_failed(self): + trace = Trace(processors=[], client=MagicMock(), trace_id="t1") + captured = {} + with pytest.raises(KeyboardInterrupt): + with trace.span("op") as span: + captured["span"] = span + raise KeyboardInterrupt + assert _sgp_status(captured["span"]) in FAILED_STATUSES + # --------------------------------------------------------------------------- # Map: _build_sgp_span translates the recorded error into SGP status=ERROR diff --git a/tests/test_adk_tracing_span_error.py b/tests/test_adk_tracing_span_error.py index c81015142..81117ed04 100644 --- a/tests/test_adk_tracing_span_error.py +++ b/tests/test_adk_tracing_span_error.py @@ -17,13 +17,29 @@ from __future__ import annotations -from unittest.mock import AsyncMock +import asyncio +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch import pytest +from scale_gp_beta.lib.tracing import create_span as sgp_create_span from agentex.types.span import Span from agentex.lib.adk._modules.tracing import TracingModule from agentex.lib.core.tracing.span_error import get_span_error +from agentex.lib.core.tracing.processors.sgp_tracing_processor import _build_sgp_span + +# The failure checks read a span with either status as failed. +FAILED_STATUSES = {"ERROR", "CANCELED"} + + +def _sgp_status(span: Span) -> str: + def create(**kwargs: Any) -> Any: + return sgp_create_span(**kwargs, queue_manager=MagicMock()) + + env = MagicMock(ACP_TYPE=None, AGENT_NAME=None, AGENT_ID=None, AGENT_VERSION=None) + with patch("agentex.lib.core.tracing.processors.sgp_tracing_processor.create_span", side_effect=create): + return _build_sgp_span(span, env).status def _make_module() -> tuple[TracingModule, Span, AsyncMock]: @@ -93,6 +109,26 @@ def _boom(*_args: object, **_kwargs: object) -> None: end_span.assert_awaited_once() +async def test_span_cancelled_mid_call_is_persisted_as_failed() -> None: + module, span, end_span = _make_module() + entered = asyncio.Event() + + async def step() -> None: + async with module.span(trace_id="trace-1", name="step"): + entered.set() + await asyncio.Event().wait() + + task = asyncio.create_task(step()) + await entered.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + end_span.assert_awaited_once() + persisted_span = end_span.await_args.kwargs["span"] + assert _sgp_status(persisted_span) in FAILED_STATUSES + + async def test_span_noop_when_trace_id_falsy() -> None: module, _span, end_span = _make_module() From 769485c026d30b095e39cf5599c73213a2e397dd Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Wed, 28 Oct 2026 20:41:54 -0400 Subject: [PATCH 14/17] feat(tracing): stamp the model id on model call spans The LiteLLM completion spans and the OpenAI Agents streaming model span now carry gen_ai.request.model, the attribute the failure checks read as the model a call was made to. Co-Authored-By: Claude Opus 5.5 --- .../core/services/adk/providers/litellm.py | 6 ++++ .../models/temporal_streaming_model.py | 2 ++ tests/lib/adk/providers/test_litellm_usage.py | 32 +++++++++++++++++++ .../plugins/openai_agents/test_model_usage.py | 28 ++++++++++++++++ 4 files changed, 68 insertions(+) diff --git a/src/agentex/lib/core/services/adk/providers/litellm.py b/src/agentex/lib/core/services/adk/providers/litellm.py index 416077e8f..947de89b8 100644 --- a/src/agentex/lib/core/services/adk/providers/litellm.py +++ b/src/agentex/lib/core/services/adk/providers/litellm.py @@ -24,6 +24,8 @@ logger = logging.make_logger(__name__) +_REQUEST_MODEL_KEY = "gen_ai.request.model" + def _stream_kwargs_with_usage(llm_config: LLMConfig) -> dict: """Completion kwargs with usage reporting enabled on the final stream chunk. @@ -63,6 +65,7 @@ async def chat_completion( parent_id=parent_span_id, name="chat_completion", input=llm_config.model_dump(), + data={_REQUEST_MODEL_KEY: llm_config.model}, ) as span: heartbeat_if_in_workflow("chat completion") if self.llm_gateway is None: @@ -104,6 +107,7 @@ async def chat_completion_auto_send( parent_id=parent_span_id, name="chat_completion_auto_send", input=llm_config.model_dump(), + data={_REQUEST_MODEL_KEY: llm_config.model}, ) as span: heartbeat_if_in_workflow("chat completion auto send") @@ -174,6 +178,7 @@ async def chat_completion_stream( parent_id=parent_span_id, name="chat_completion_stream", input=completion_kwargs, + data={_REQUEST_MODEL_KEY: llm_config.model}, ) as span: # Direct streaming outside temporal - yield each chunk as it comes chunks: list[Completion] = [] @@ -217,6 +222,7 @@ async def chat_completion_stream_auto_send( parent_id=parent_span_id, name="chat_completion_stream_auto_send", input=completion_kwargs, + data={_REQUEST_MODEL_KEY: llm_config.model}, ) as span: # Use streaming context manager async with self.streaming_service.streaming_task_message_context( diff --git a/src/agentex/lib/core/temporal/plugins/openai_agents/models/temporal_streaming_model.py b/src/agentex/lib/core/temporal/plugins/openai_agents/models/temporal_streaming_model.py index c985d5e65..0a8874379 100644 --- a/src/agentex/lib/core/temporal/plugins/openai_agents/models/temporal_streaming_model.py +++ b/src/agentex/lib/core/temporal/plugins/openai_agents/models/temporal_streaming_model.py @@ -143,6 +143,7 @@ def _serialize_item(item: Any) -> dict[str, Any]: # Cap on the rendered hosted-tool result string (UI / trace readability). _HOSTED_TOOL_RESULT_CAP = 2000 +_REQUEST_MODEL_KEY = "gen_ai.request.model" def _coerce_args(raw: Any) -> dict[str, Any]: @@ -652,6 +653,7 @@ async def get_response( "tools_count": len(tools) if tools else 0, "handoffs_count": len(handoffs) if handoffs else 0, }, + data={_REQUEST_MODEL_KEY: self.model_name}, ) as span: # Always use Responses API for streaming if not task_id: diff --git a/tests/lib/adk/providers/test_litellm_usage.py b/tests/lib/adk/providers/test_litellm_usage.py index 5f5d480d9..7405a484e 100644 --- a/tests/lib/adk/providers/test_litellm_usage.py +++ b/tests/lib/adk/providers/test_litellm_usage.py @@ -42,6 +42,7 @@ async def span(self, name, parent_id=None, input=None, data=None, task_id=None): trace_id="trace-1", parent_id=parent_id, input=input, + data=data, ) self.spans.append(span) yield span @@ -217,3 +218,34 @@ async def test_stream_without_usage_chunk_omits_usage(self): ) assert "usage" not in _output_dict(tracer.trace_obj.spans[0]) + + +class TestModelIdOnSpans: + """The failure checks read ``gen_ai.request.model`` as the model a call was made to.""" + + async def test_every_completion_span_names_its_model(self): + completion = Completion( + choices=[Choice(index=0, message=AssistantMessage(content="Hi"), finish_reason="stop")], + ) + gateway = _stream_gateway([_delta_chunk("Hi", role="assistant")], {}) + gateway.acompletion = AsyncMock(return_value=completion) + service, tracer = _make_service(gateway) + model = "openai/gpt-4o-mini" + + await service.chat_completion(llm_config=LLMConfig(model=model, messages=[]), trace_id="trace-1") + await service.chat_completion_auto_send( + task_id="task-1", llm_config=LLMConfig(model=model, messages=[]), trace_id="trace-1" + ) + async for _ in service.chat_completion_stream( + llm_config=LLMConfig(model=model, messages=[], stream=True), trace_id="trace-1" + ): + pass + await service.chat_completion_stream_auto_send( + task_id="task-1", llm_config=LLMConfig(model=model, messages=[], stream=True), trace_id="trace-1" + ) + + spans = tracer.trace_obj.spans + assert len(spans) == 4 + for span in spans: + assert isinstance(span.data, dict) + assert span.data["gen_ai.request.model"] == model diff --git a/tests/lib/core/temporal/plugins/openai_agents/test_model_usage.py b/tests/lib/core/temporal/plugins/openai_agents/test_model_usage.py index bf3dc8006..770d49810 100644 --- a/tests/lib/core/temporal/plugins/openai_agents/test_model_usage.py +++ b/tests/lib/core/temporal/plugins/openai_agents/test_model_usage.py @@ -180,3 +180,31 @@ async def test_streaming_model_writes_zero_usage_when_api_reports_none(self, tra "cached_input_tokens": 0, "reasoning_tokens": 0, } + + async def test_streaming_model_span_names_its_model(self, tracing_contextvars): + completed = ResponseCompletedEvent.model_construct( + type="response.completed", + response=Response.model_construct(output=[], usage=None), + ) + fake_tracer = FakeTracer() + openai_client = MagicMock() + openai_client.responses.create = AsyncMock(return_value=FakeStream([completed])) + + with patch.object(tsm, "create_async_agentex_client", return_value=MagicMock()): + with patch.object(tsm, "AsyncTracer", return_value=fake_tracer): + model = tsm.TemporalStreamingModel(model_name="gpt-4o-mini", openai_client=openai_client) + + await model.get_response( + system_instructions=None, + input="hello", + model_settings=ModelSettings(), + tools=[], + output_schema=None, + handoffs=[], + tracing=None, + ) + + # The failure checks read gen_ai.request.model as the model a call was made to. + span = fake_tracer.trace_obj.spans[0] + assert isinstance(span.data, dict) + assert span.data["gen_ai.request.model"] == "gpt-4o-mini" From 8d8c9b343c9a0f2ded6812c0254234fdbd978406 Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Wed, 28 Oct 2026 20:42:48 -0400 Subject: [PATCH 15/17] test(tracing): sort the span error test imports Co-Authored-By: Claude Opus 5.5 --- tests/lib/core/tracing/test_span_error.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/lib/core/tracing/test_span_error.py b/tests/lib/core/tracing/test_span_error.py index 1012c5e57..50dee5b74 100644 --- a/tests/lib/core/tracing/test_span_error.py +++ b/tests/lib/core/tracing/test_span_error.py @@ -8,10 +8,10 @@ import pytest from scale_gp_beta.lib.tracing import ( - create_span as sgp_create_span, PlatformError as SGPPlatformError, ApplicationError as SGPApplicationError, CategorizedError as SGPCategorizedError, + create_span as sgp_create_span, ) from agentex.types.span import Span From 447298999f278cc5a11ec4b1778e29569793bb6d Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Wed, 28 Oct 2026 20:47:38 -0400 Subject: [PATCH 16/17] test(tracing): drop a docstring that restates its helper Co-Authored-By: Claude Opus 5.5 --- tests/lib/core/tracing/test_span_error.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/lib/core/tracing/test_span_error.py b/tests/lib/core/tracing/test_span_error.py index 50dee5b74..93779b460 100644 --- a/tests/lib/core/tracing/test_span_error.py +++ b/tests/lib/core/tracing/test_span_error.py @@ -32,7 +32,6 @@ def _sgp_status(span: Span) -> str: - """The status the span is written to SGP with.""" from agentex.lib.core.tracing.processors.sgp_tracing_processor import _build_sgp_span def create(**kwargs: Any) -> Any: From f12a3569d34a0716479bd576d3d996ddca8a8584 Mon Sep 17 00:00:00 2001 From: Mohammad Atallah Date: Thu, 29 Oct 2026 07:39:34 -0400 Subject: [PATCH 17/17] feat(tracing): write failure-check attributes onto OpenTelemetry spans otel_attributes turns the helpers' data into OpenTelemetry attribute values, writing the verification rules and any other object as JSON text, and otel_span_id gives the id escalates takes for an OpenTelemetry span. Co-Authored-By: Claude Opus 5.5 --- .../10_async/00_base/030_tracing/README.md | 4 ++ .../lib/core/tracing/failure_attributes.py | 47 ++++++++++++- .../core/tracing/test_failure_attributes.py | 66 +++++++++++++++++++ 3 files changed, 115 insertions(+), 2 deletions(-) diff --git a/examples/tutorials/10_async/00_base/030_tracing/README.md b/examples/tutorials/10_async/00_base/030_tracing/README.md index 705002cfd..43d899391 100644 --- a/examples/tutorials/10_async/00_base/030_tracing/README.md +++ b/examples/tutorials/10_async/00_base/030_tracing/README.md @@ -107,6 +107,10 @@ span ids when one escalation covers several failures. The span `adk.tracing.span` yields is `None` when `trace_id` is empty or the span fails to start, so guard it before reading `.data` or `.id`. +An OpenTelemetry producer passes any of these dicts through `failure_attributes.otel_attributes` +before setting them on a span, and names an OpenTelemetry span to `escalates` as +`failure_attributes.otel_span_id(span)`. + The SGP export adds `__agent_version__` when `AGENT_VERSION` is set, and `__commit_sha__` when the process starts with `AGENT_COMMIT_SHA` set to a valid git SHA, so only `prompt_version` is yours to declare. diff --git a/src/agentex/lib/core/tracing/failure_attributes.py b/src/agentex/lib/core/tracing/failure_attributes.py index ef8395b9b..54f0a25bf 100644 --- a/src/agentex/lib/core/tracing/failure_attributes.py +++ b/src/agentex/lib/core/tracing/failure_attributes.py @@ -3,15 +3,18 @@ The SGP processor ships ``span.data`` verbatim as span metadata, so each attribute is a flat dotted key holding a native JSON value. The output schema is written as JSON text instead, since the span store would spread a schema object into many paths whose keys can collide. + +An OpenTelemetry span carries only scalars and lists of one scalar type, so ``otel_attributes`` writes any +other value as JSON text, which the checks also read. This module does not import OpenTelemetry. """ from __future__ import annotations import re import json -from typing import Any, Literal, Callable, cast +from typing import Any, Literal, Callable, Protocol, cast from dataclasses import field, dataclass -from collections.abc import Sequence +from collections.abc import Mapping, Sequence __all__ = ( "REQUIRED_OUTPUTS_KEY", @@ -30,6 +33,8 @@ "output_name", "verification_result", "escalates", + "otel_attributes", + "otel_span_id", ) REQUIRED_OUTPUTS_KEY = "sgp.agent.required_outputs" @@ -211,3 +216,41 @@ def escalates(*verification_span_ids: str) -> dict[str, list[str]]: if not verification_span_ids: raise ValueError("escalates needs at least one verification span id") return {ESCALATES_KEY: list(_unique_texts(verification_span_ids, "escalates verification span id"))} + + +_OTEL_SCALARS: frozenset[type] = frozenset({str, bool, int, float}) + + +def _otel_value(key: str, value: object) -> Any: + if type(value) in _OTEL_SCALARS: + return value + if isinstance(value, list | tuple): + items = cast("Sequence[object]", value) + kinds = {type(item) for item in items} + if len(kinds) <= 1 and kinds <= _OTEL_SCALARS: + return list(items) + elif not isinstance(value, dict): + raise ValueError(f"{key} has no OpenTelemetry attribute form, got {type(value).__name__}") + try: + return json.dumps(value, separators=(",", ":"), allow_nan=False) + except (TypeError, ValueError, RecursionError) as exc: + raise ValueError(f"{key} must be JSON serializable: {exc}") from exc + + +def otel_attributes(attributes: Mapping[str, Any]) -> dict[str, Any]: + """These helpers' data as OpenTelemetry span attributes, with objects and mixed lists as JSON text.""" + return {key: _otel_value(key, value) for key, value in attributes.items()} + + +class _OTelSpanContext(Protocol): + @property + def span_id(self) -> int: ... + + +class _OTelSpan(Protocol): + def get_span_context(self) -> _OTelSpanContext: ... + + +def otel_span_id(span: _OTelSpan) -> str: + """The id the span store gives an OpenTelemetry span, which ``escalates`` takes.""" + return format(span.get_span_context().span_id, "016x") diff --git a/tests/lib/core/tracing/test_failure_attributes.py b/tests/lib/core/tracing/test_failure_attributes.py index 7a82be084..905c9b5dd 100644 --- a/tests/lib/core/tracing/test_failure_attributes.py +++ b/tests/lib/core/tracing/test_failure_attributes.py @@ -11,7 +11,11 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest +from opentelemetry.trace import SpanContext, NonRecordingSpan +from opentelemetry.sdk.trace import TracerProvider from scale_gp_beta.lib.tracing import create_span as sgp_create_span +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter from agentex.lib import adk from agentex.lib.types.tracing import SGPTracingProcessorConfig @@ -22,6 +26,8 @@ VerificationRule, escalates, output_name, + otel_span_id, + otel_attributes, verification_result, ) from agentex.lib.core.tracing.processors.sgp_tracing_processor import ( @@ -376,6 +382,65 @@ def test_results_merge_into_a_span_data_dict(self): assert data == {"note": "kept", "sgp.agent.output_name": "answer"} +class TestOpenTelemetryAdapter: + def test_attributes_reach_an_exported_otel_span_without_loss(self): + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + tracer = provider.get_tracer("test") + + with tracer.start_as_current_span( + "agent", attributes=otel_attributes(FailureCheckSpec(**FULL_SPEC).attributes()) + ): + with tracer.start_as_current_span( + "verify", attributes=otel_attributes(verification_result("fail")) + ) as verify: + pass + with tracer.start_as_current_span( + "recheck", attributes=otel_attributes(verification_result("fail")) + ) as recheck: + pass + escalated = escalates(otel_span_id(verify), otel_span_id(recheck)) + with tracer.start_as_current_span("escalate", attributes=otel_attributes(escalated)): + pass + + by_name = {span.name: dict(span.attributes or {}) for span in exporter.get_finished_spans()} + root = {key: list(value) if isinstance(value, tuple) else value for key, value in by_name["agent"].items()} + assert json.loads(root.pop(SCHEMA_KEY)) == SCHEMA + assert json.loads(root.pop("sgp.agent.verification")) == [{"step": "run_tests", "verifies": "apply_patch"}] + assert root == {key: value for key, value in NATIVE_ATTRIBUTES.items() if key != "sgp.agent.verification"} + assert by_name["verify"] == {"sgp.agent.verification_result": "fail"} + ids = [format(span.context.span_id, "016x") for span in exporter.get_finished_spans() if span.name != "agent"] + assert list(by_name["escalate"]["sgp.agent.escalates"]) == ids[:2] + + def test_the_span_id_is_the_otlp_span_id_as_16_lowercase_hex_digits(self): + span = NonRecordingSpan(SpanContext(trace_id=1, span_id=0xAB, is_remote=False)) + assert otel_span_id(span) == "00000000000000ab" + + @pytest.mark.parametrize( + ("value", "expected"), + [ + ("text", "text"), + (3, 3), + (True, True), + (["a", "b"], ["a", "b"]), + ((), []), + ([1, "a"], '[1,"a"]'), + ([True, 1], "[true,1]"), + ({"a": [1]}, '{"a":[1]}'), + ], + ) + def test_values_otel_cannot_carry_become_json_text(self, value, expected): + converted = otel_attributes({"k": value})["k"] + assert converted == expected + assert type(converted) is type(expected) + + @pytest.mark.parametrize("value", [None, object(), {"a": float("nan")}, [{"a": {1, 2}}]]) + def test_a_value_with_no_otel_form_is_refused(self, value): + with pytest.raises(ValueError, match="^k "): + otel_attributes({"k": value}) + + class TestExports: def test_adk_exposes_the_helper(self): assert adk.FailureCheckSpec is FailureCheckSpec @@ -383,6 +448,7 @@ def test_adk_exposes_the_helper(self): assert adk.failure_attributes.output_name is output_name assert adk.failure_attributes.verification_result is verification_result assert adk.failure_attributes.escalates is escalates + assert adk.failure_attributes.otel_attributes is otel_attributes assert {"failure_attributes", "FailureCheckSpec", "VerificationRule"} <= set(adk.__all__)