feat: emit runtime events from third-party OTel spans

examples/realtime_demo/README.md

@@ -32,6 +32,34 @@ No external OTel Collector or ClickHouse is required. The demo attaches the
 `SessionStateMaterializer` in-process via the runtime-event listener so it runs
 with a single command.
 
+## Third-party OTel SDK mapping example
+
+`third_party_otel.py` creates and owns its own `TracerProvider`, instruments
+agent, model, and tool spans using current `gen_ai.*` semantic-convention
+attributes, and sends traces and mapped runtime-event logs to an OTLP/HTTP
+collector:
+
+```bash
+uv run python examples/realtime_demo/third_party_otel.py
+```
+
+Start an OTLP/HTTP collector on port `4318` before running this example.
+The endpoint defaults to `http://localhost:4318` and honors
+`OTEL_EXPORTER_OTLP_ENDPOINT`, `OTEL_EXPORTER_OTLP_LOGS_ENDPOINT`, and
+`OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`.
+
+The application keeps ownership of its provider and existing span processor.
+Observe's processor owns the default GenAI mapper and private Logs pipeline:
+
+```python
+provider.add_span_processor(
+    RuntimeEventSpanProcessor(
+        endpoint="http://localhost:4318/v1/logs",
+        resource=resource,
+    )
+)
+```
+
 ## How it maps to production
 
 In production the materializer lives **downstream** and consumes the same OTel

examples/realtime_demo/third_party_otel.py

@@ -0,0 +1,127 @@
+# Copyright AGNTCY Contributors (https://github.com/agntcy)
+# SPDX-License-Identifier: Apache-2.0
+
+"""Instrument GenAI work with an application-owned OpenTelemetry SDK."""
+
+from __future__ import annotations
+
+import json
+import os
+
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+from opentelemetry.trace import SpanKind, Tracer
+
+from ioa_observe.sdk.tracing import (
+    RuntimeEventSpanProcessor,
+)
+
+
+def run_weather_agent(tracer: Tracer) -> None:
+    conversation_id = "conversation-123"
+    with tracer.start_as_current_span(
+        "invoke_agent Weather Agent",
+        kind=SpanKind.INTERNAL,
+        attributes={
+            "gen_ai.operation.name": "invoke_agent",
+            "gen_ai.conversation.id": conversation_id,
+            "gen_ai.agent.name": "Weather Agent",
+        },
+    ):
+        with tracer.start_as_current_span(
+            "chat gpt-5",
+            kind=SpanKind.CLIENT,
+            attributes={
+                "gen_ai.operation.name": "chat",
+                "gen_ai.provider.name": "openai",
+                "gen_ai.conversation.id": conversation_id,
+                "gen_ai.request.model": "gpt-5",
+                "gen_ai.input.messages": json.dumps(
+                    [
+                        {
+                            "role": "user",
+                            "parts": [{"type": "text", "content": "Weather?"}],
+                        }
+                    ]
+                ),
+            },
+        ) as model_span:
+            model_span.set_attribute("gen_ai.response.model", "gpt-5")
+            model_span.set_attribute("gen_ai.response.id", "response-123")
+            model_span.set_attribute(
+                "gen_ai.output.messages",
+                json.dumps(
+                    [
+                        {
+                            "role": "assistant",
+                            "parts": [
+                                {
+                                    "type": "tool_call",
+                                    "id": "call-123",
+                                    "name": "get_weather",
+                                }
+                            ],
+                        }
+                    ]
+                ),
+            )
+
+        with tracer.start_as_current_span(
+            "execute_tool get_weather",
+            kind=SpanKind.INTERNAL,
+            attributes={
+                "gen_ai.operation.name": "execute_tool",
+                "gen_ai.conversation.id": conversation_id,
+                "gen_ai.agent.name": "Weather Agent",
+                "gen_ai.tool.name": "get_weather",
+                "gen_ai.tool.call.id": "call-123",
+                "gen_ai.tool.call.arguments": json.dumps({"city": "London"}),
+            },
+        ) as tool_span:
+            tool_span.set_attribute(
+                "gen_ai.tool.call.result",
+                json.dumps({"temperature_celsius": 18, "conditions": "cloudy"}),
+            )
+
+
+def main() -> None:
+    collector_endpoint = os.getenv(
+        "OTEL_EXPORTER_OTLP_ENDPOINT",
+        "http://localhost:4318",
+    ).rstrip("/")
+    logs_endpoint = os.getenv(
+        "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT",
+        f"{collector_endpoint}/v1/logs",
+    )
+    traces_endpoint = os.getenv(
+        "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
+        f"{collector_endpoint}/v1/traces",
+    )
+    resource = Resource.create(
+        {
+            "service.name": "third-party-weather-agent",
+            "service.version": "1.0.0",
+        }
+    )
+    provider = TracerProvider(resource=resource)
+    provider.add_span_processor(
+        RuntimeEventSpanProcessor(
+            endpoint=logs_endpoint,
+            resource=resource,
+        )
+    )
+    provider.add_span_processor(
+        BatchSpanProcessor(
+            OTLPSpanExporter(endpoint=traces_endpoint),
+            schedule_delay_millis=5_000,
+        )
+    )
+
+    run_weather_agent(provider.get_tracer("example.third_party_genai", "1.0.0"))
+    provider.shutdown()
+
+
+if __name__ == "__main__":
+    main()

ioa_observe/materializer/session_state.py

@@ -79,6 +79,7 @@ class SessionNodeState:
     started_at: datetime | None = None
     completed_at: datetime | None = None
     input: str | None = None
+    output: str | None = None
 
 
 @dataclass
@@ -321,6 +322,11 @@ def _apply_node_event(
             node.status = "completed"
             node.started_at = node.started_at or record.event_time
             node.completed_at = record.event_time
+            agent_output = _optional_attribute(
+                record, RuntimeEventAttribute.AGENT_OUTPUT.value
+            )
+            if agent_output is not None:
+                node.output = agent_output
 
     def _apply_topology_edge_event(
         self,

ioa_observe/sdk/decorators/base.py

@@ -615,7 +615,11 @@ def _cleanup_span(span, ctx_token):
         unregister_active_span(session_id, agent_seq)
         agent_name = getattr(span, "_ioa_agent_name", None)
         if agent_name:
-            record_node_completed(session_id, agent_name)
+            record_node_completed(
+                session_id,
+                agent_name,
+                agent_output=span.attributes.get(OBSERVE_ENTITY_OUTPUT),
+            )
 
     # Mark tool as no longer in-flight for tool-level fork detection
     tool_parent_hex = getattr(span, "_ioa_tool_parent_hex", None)

ioa_observe/sdk/tracing/__init__.py

@@ -20,6 +20,14 @@
     register_runtime_event_listener,
     unregister_runtime_event_listener,
 )
+from ioa_observe.sdk.tracing.runtime_event_processor import (
+    GenAIRuntimeEventMapper,
+    InstrumentationScope,
+    RuntimeEventMapper,
+    RuntimeEventSpanProcessor,
+    SpanLifecycle,
+    SpanLifecycleObservation,
+)
 from ioa_observe.sdk.tracing.topology import (
     get_live_topology_snapshot,
     record_session_completed,
@@ -39,6 +47,12 @@
     "register_runtime_event_listener",
     "unregister_runtime_event_listener",
     "clear_runtime_event_listeners",
+    "GenAIRuntimeEventMapper",
+    "InstrumentationScope",
+    "RuntimeEventMapper",
+    "RuntimeEventSpanProcessor",
+    "SpanLifecycle",
+    "SpanLifecycleObservation",
     "get_live_topology_snapshot",
     "record_session_completed",
 ]