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

examples/tutorials/10_async/00_base/030_tracing/README.md

@@ -74,6 +74,47 @@ 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)`, 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`.
+
+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.
+
 ## 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/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

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

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/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: