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..cc9139b5f 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,42 @@ 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). 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 + +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") + 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)`. 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 +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/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/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/src/agentex/lib/core/tracing/failure_attributes.py b/src/agentex/lib/core/tracing/failure_attributes.py new file mode 100644 index 000000000..03587c8e3 --- /dev/null +++ b/src/agentex/lib/core/tracing/failure_attributes.py @@ -0,0 +1,211 @@ +"""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. 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 json +from typing import Any, Literal, Callable, cast +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" + +# 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") + + +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: + 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 + + +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 _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: + 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), + # 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(_json_object(self.output_schema, "output_schema"), separators=(",", ":")), + 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, "escalates verification span id")} 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"] 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} 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..825fc6147 --- /dev/null +++ b/tests/lib/core/tracing/test_failure_attributes.py @@ -0,0 +1,472 @@ +"""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"], + ), + ("quote_arguments", ["cite.quote"], "sgp.agent.quote_arguments", ["cite.quote"]), + ("prompt_version", "v3", "sgp.agent.prompt_version", "v3"), +] + +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: + @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_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): + 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() == {} + 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" + spec.attributes()["sgp.agent.required_outputs"].append("sneaky") + + 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"]) + 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: + NESTING_DEPTH_PAST_RECURSION_LIMIT = 200_000 + + @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) 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): + 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) 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")] + 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) as raised: + 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"), + [ + ([], "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) as raised: + 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 + 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(self.NESTING_DEPTH_PAST_RECURSION_LIMIT): + deep = [deep] + 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]) == {} + + +class TestVerificationRule: + @pytest.mark.parametrize( + ("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, 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") + 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") 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")} + 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 = {**NATIVE_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"] + 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" + + 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])