feat(tracing): write the span attributes the failure checks read

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

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",

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)

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,

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")}

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

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()