Fix/SAO-259 operation trace ownership

src/splunk_ao/exporter/config.py

@@ -40,6 +40,16 @@ class RoutingAttrs:
 
 ExporterFactory = Callable[..., SpanExporter]
 
+RESERVED_ROUTING_RESOURCE_ATTRIBUTES = frozenset(
+    {
+        "splunk_ao.project.name",
+        "splunk_ao.project.id",
+        "splunk_ao.logstream.name",
+        "splunk_ao.logstream.id",
+        "splunk_ao.experiment.id",
+    }
+)
+
 
 def _resolve_name_or_id(
     label: str,
@@ -149,7 +159,14 @@ def routing_resource_attributes(routing: RoutingAttrs) -> dict[str, str]:
 
 def create_otel_resource(routing: RoutingAttrs) -> Resource:
     """Create a Resource with standard OTel detection and Splunk AO routing."""
-    return Resource.create(routing_resource_attributes(routing))
+    detected_resource = Resource.create()
+    attributes = {
+        key: value
+        for key, value in detected_resource.attributes.items()
+        if key not in RESERVED_ROUTING_RESOURCE_ATTRIBUTES
+    }
+    attributes.update(routing_resource_attributes(routing))
+    return Resource(attributes, schema_url=detected_resource.schema_url)
 
 
 def build_exporter(

src/splunk_ao/handlers/base_async_handler.py

@@ -42,31 +42,34 @@ async def async_commit(self) -> None:
             _logger.warning("Unable to add nodes to trace: Root node does not exist")
             return
 
-        if self._start_new_trace:
-            self._splunk_ao_logger.start_trace(
-                input=serialize_to_str(root_node.span_params.get("input", "")),
-                name=root_node.span_params.get("name"),
-                metadata=root_node.span_params.get("metadata"),
-            )
-
-        self.log_node_tree(root_node)
-
-        # Conclude the trace with the root node's output
-        root_output = root_node.span_params.get("output", "")
-
-        if self._start_new_trace:
-            # If we started a new trace, we need to conclude it
-            self._splunk_ao_logger.conclude(
-                output=serialize_to_str(root_output), status_code=root_node.span_params.get("status_code")
-            )
-
-        if self._flush_on_chain_end:
-            # Upload the trace to Splunk AO
-            await self._splunk_ao_logger.async_flush()
-
-        # Clear nodes after successful commit
-        self._nodes.clear()
-        self._root_node = None
+        owned_trace = None
+        try:
+            if self._start_new_trace:
+                owned_trace = self._splunk_ao_logger.start_trace(
+                    input=serialize_to_str(root_node.span_params.get("input", "")),
+                    name=root_node.span_params.get("name"),
+                    metadata=root_node.span_params.get("metadata"),
+                )
+
+            self.log_node_tree(root_node)
+            root_output = root_node.span_params.get("output", "")
+
+            if self._start_new_trace:
+                self._conclude_owned_trace(
+                    owned_trace,
+                    output=serialize_to_str(root_output),
+                    status_code=root_node.span_params.get("status_code"),
+                )
+
+            if self._flush_on_chain_end:
+                await self._splunk_ao_logger.async_flush()
+        except Exception:
+            if owned_trace is not None:
+                self._conclude_owned_trace(owned_trace, output="", status_code=500)
+            _logger.warning("Failed to commit async handler telemetry", exc_info=True)
+        finally:
+            self._nodes.clear()
+            self._root_node = None
 
     async def async_end_node(self, run_id: UUID, **kwargs: Any) -> None:
         """

src/splunk_ao/handlers/base_handler.py

@@ -41,7 +41,7 @@ def __init__(
         integration: INTEGRATION = "langchain",
         splunk_ao_logger: SplunkAOLogger | None = None,
         start_new_trace: bool = True,
-        flush_on_chain_end: bool = True,
+        flush_on_chain_end: bool | None = None,
         ingestion_hook: Callable[[TracesIngestRequest], None] | None = None,
     ):
         self._splunk_ao_logger: SplunkAOLogger = splunk_ao_logger or splunk_ao_context.get_logger_instance(
@@ -52,7 +52,7 @@ def __init__(
                 raise ValueError("ingestion_hook can only be used in batch mode")
             self._splunk_ao_logger._ingestion_hook = ingestion_hook
         self._start_new_trace: bool = start_new_trace
-        self._flush_on_chain_end: bool = flush_on_chain_end
+        self._flush_on_chain_end = ingestion_hook is not None if flush_on_chain_end is None else flush_on_chain_end
         self._nodes: dict[str, Node] = {}
         self._root_node: Node | None = None
         self._integration: INTEGRATION = integration
@@ -73,9 +73,10 @@ def commit(self) -> None:
             _logger.warning("Unable to add nodes to trace: Root node does not exist")
             return
 
+        owned_trace = None
         try:
             if self._start_new_trace:
-                self._splunk_ao_logger.start_trace(
+                owned_trace = self._splunk_ao_logger.start_trace(
                     input=SplunkAOLogger._coerce_output(root_node.span_params.get("input", "")),
                     name=root_node.span_params.get("name"),
                     metadata=root_node.span_params.get("metadata"),
@@ -87,18 +88,31 @@ def commit(self) -> None:
             root_output = root_node.span_params.get("output", "")
 
             if self._start_new_trace:
-                self._splunk_ao_logger.conclude(
+                self._conclude_owned_trace(
+                    owned_trace,
                     output=SplunkAOLogger._coerce_output(root_output),
                     status_code=root_node.span_params.get("status_code"),
                 )
 
             if self._flush_on_chain_end:
                 self._splunk_ao_logger.flush()
+        except Exception:
+            if owned_trace is not None:
+                self._conclude_owned_trace(owned_trace, output="", status_code=500)
+            _logger.warning("Failed to commit handler telemetry", exc_info=True)
         finally:
-            # Always clean up, even if trace building or flush fails
             self._nodes.clear()
             self._root_node = None
 
+    def _conclude_owned_trace(self, trace: Any, output: Any, status_code: int | None) -> None:
+        current_parent = self._splunk_ao_logger.current_parent()
+        root = current_parent
+        while root is not None and root._parent is not None:
+            root = root._parent
+
+        if root is trace:
+            self._splunk_ao_logger.conclude(output=output, status_code=status_code, conclude_all=True)
+
     def log_node_tree(self, node: Node) -> None:
         """
         Log a node and its children recursively.

src/splunk_ao/handlers/crewai/handler.py

@@ -67,7 +67,7 @@ def __init__(
         self,
         splunk_ao_logger: SplunkAOLogger | None = None,
         start_new_trace: bool = True,
-        flush_on_crew_completed: bool = True,
+        flush_on_crew_completed: bool = False,
     ):
         _resolve_crewai_imports()
 

src/splunk_ao/handlers/langchain/async_handler.py

@@ -46,7 +46,7 @@ def __init__(
         self,
         splunk_ao_logger: SplunkAOLogger | None = None,
         start_new_trace: bool = True,
-        flush_on_chain_end: bool = True,
+        flush_on_chain_end: bool | None = None,
         ingestion_hook: Callable[[TracesIngestRequest], None] | None = None,
     ):
         self._handler = SplunkAOAsyncBaseHandler(

src/splunk_ao/handlers/langchain/handler.py

@@ -45,7 +45,7 @@ def __init__(
         self,
         splunk_ao_logger: SplunkAOLogger | None = None,
         start_new_trace: bool = True,
-        flush_on_chain_end: bool = True,
+        flush_on_chain_end: bool | None = None,
         ingestion_hook: Callable[[TracesIngestRequest], None] | None = None,
     ):
         self._handler = SplunkAOBaseHandler(

src/splunk_ao/handlers/langchain/middleware.py

@@ -41,7 +41,7 @@ def __init__(
         self,
         splunk_ao_logger: SplunkAOLogger | None = None,
         start_new_trace: bool = True,
-        flush_on_chain_end: bool = True,
+        flush_on_chain_end: bool | None = None,
         ingestion_hook: Callable[[TracesIngestRequest], None] | None = None,
     ) -> None:
         if not HAS_LANGCHAIN:

src/splunk_ao/handlers/openai_agents/handler.py

@@ -41,7 +41,7 @@ class SplunkAOTracingProcessor(TracingProcessor):
         Stores Node objects keyed by their OpenAI span_id or trace_id (for root).
     """
 
-    def __init__(self, splunk_ao_logger: SplunkAOLogger | None = None, flush_on_trace_end: bool = True):
+    def __init__(self, splunk_ao_logger: SplunkAOLogger | None = None, flush_on_trace_end: bool = False):
         """
         OpenAI Agents TracingProcessor for logging traces to Splunk AO.
 
@@ -58,6 +58,7 @@ def __init__(self, splunk_ao_logger: SplunkAOLogger | None = None, flush_on_trac
         self._last_output: Any = None
         self._last_status_code: int | None = None
         self._first_input: Any = None
+        self._owned_trace: Any = None
 
     def on_trace_start(self, trace: Trace) -> None:
         """Called when an OpenAI Agent trace starts."""
@@ -82,17 +83,19 @@ def on_trace_end(self, trace: Trace) -> None:
 
         node.span_params["duration_ns"] = convert_time_delta_to_ns(_get_timestamp() - node.span_params["start_time"])
 
-        # Log the trace to Splunk AO (this includes concluding the trace)
-        self._commit_trace(trace)
-        self._nodes = {}
-
-        self._last_output = None
-        self._last_status_code = None
-        self._first_input = None
-
-        # Optionally flush the log batch
-        if self._flush_on_trace_end:
-            self._splunk_ao_logger.flush()
+        try:
+            self._commit_trace(trace)
+            if self._flush_on_trace_end:
+                self._splunk_ao_logger.flush()
+        except Exception:
+            self._conclude_current_trace_on_failure()
+            _logger.warning("Failed to commit OpenAI Agents telemetry", exc_info=True)
+        finally:
+            self._nodes = {}
+            self._last_output = None
+            self._last_status_code = None
+            self._first_input = None
+            self._owned_trace = None
 
     def _commit_trace(self, trace: Trace) -> None:
         if not self._nodes:
@@ -106,6 +109,20 @@ def _commit_trace(self, trace: Trace) -> None:
             _logger.warning(f"Root node {trace.trace_id} not found")
         self._splunk_ao_logger.conclude(output=self._last_output, status_code=self._last_status_code)
 
+    def _conclude_current_trace_on_failure(self) -> None:
+        if self._owned_trace is None:
+            return
+
+        current_parent = self._splunk_ao_logger.current_parent()
+        if current_parent is None:
+            return
+
+        root = current_parent
+        while root._parent is not None:
+            root = root._parent
+        if root is self._owned_trace:
+            self._splunk_ao_logger.conclude(output="", status_code=500, conclude_all=True)
+
     def _log_node_tree(self, node: Node, first_node: bool = False) -> None:
         """
         Log a node and its children recursively.
@@ -129,7 +146,7 @@ def _log_node_tree(self, node: Node, first_node: bool = False) -> None:
         if metadata is not None:
             metadata = convert_to_string_dict(metadata)
         if first_node:
-            self._splunk_ao_logger.add_trace(
+            self._owned_trace = self._splunk_ao_logger.add_trace(
                 input=self._first_input or "Agent Workflow",
                 output=self._last_output,
                 duration_ns=node.span_params.get("duration_ns"),

tests/test_async_base_handler.py

@@ -38,7 +38,7 @@ async def test_initialization(self, splunk_ao_logger: SplunkAOLogger) -> None:
         callback = SplunkAOAsyncBaseHandler(splunk_ao_logger=splunk_ao_logger)
         assert callback._splunk_ao_logger == splunk_ao_logger
         assert callback._start_new_trace is True
-        assert callback._flush_on_chain_end is True
+        assert callback._flush_on_chain_end is False
         assert callback._nodes == {}
 
         # Custom initialization
@@ -101,3 +101,17 @@ async def test_end_node(self, handler: SplunkAOAsyncBaseHandler, splunk_ao_logge
         assert traces[0].spans[0].type == "workflow"
         assert traces[0].spans[0].input == '{"query": "test"}'
         assert traces[0].spans[0].output == '{"result": "test result"}'
+
+    @pytest.mark.asyncio
+    async def test_commit_failure_concludes_handler_owned_trace(
+        self, handler: SplunkAOAsyncBaseHandler, splunk_ao_logger: SplunkAOLogger
+    ) -> None:
+        run_id = uuid.uuid4()
+        await handler.async_start_node(node_type="chain", parent_run_id=None, run_id=run_id, name="Test", input="test")
+
+        with patch.object(handler, "log_node_tree", side_effect=RuntimeError("conversion failed")):
+            await handler.async_end_node(run_id, output="result")
+
+        assert splunk_ao_logger.current_parent() is None
+        assert handler._nodes == {}
+        assert handler._root_node is None

tests/test_base_handler.py

@@ -34,7 +34,7 @@ def test_initialization(self, splunk_ao_logger: SplunkAOLogger) -> None:
         handler = SplunkAOBaseHandler(splunk_ao_logger=splunk_ao_logger)
         assert handler._splunk_ao_logger == splunk_ao_logger
         assert handler._start_new_trace is True
-        assert handler._flush_on_chain_end is True
+        assert handler._flush_on_chain_end is False
         assert handler._nodes == {}
 
         # Custom initialization
@@ -139,3 +139,30 @@ def test_commit_no_flush_when_disabled(self) -> None:
         # Then: neither flush nor terminate is called
         mock_logger.flush.assert_not_called()
         mock_logger.terminate.assert_not_called()
+
+    def test_commit_failure_concludes_only_handler_owned_trace(
+        self, handler: SplunkAOBaseHandler, splunk_ao_logger: SplunkAOLogger
+    ) -> None:
+        run_id = uuid.uuid4()
+        handler.start_node(node_type="chain", parent_run_id=None, run_id=run_id, name="Test", input="test")
+
+        with patch.object(handler, "log_node_tree", side_effect=RuntimeError("conversion failed")):
+            handler.end_node(run_id, output="result")
+
+        assert splunk_ao_logger.current_parent() is None
+        assert handler._nodes == {}
+        assert handler._root_node is None
+
+    def test_commit_failure_preserves_caller_owned_trace(self, splunk_ao_logger: SplunkAOLogger) -> None:
+        caller_trace = splunk_ao_logger.start_trace(input="request", name="caller")
+        handler = SplunkAOBaseHandler(
+            splunk_ao_logger=splunk_ao_logger, start_new_trace=False, flush_on_chain_end=False
+        )
+        run_id = uuid.uuid4()
+        handler.start_node(node_type="chain", parent_run_id=None, run_id=run_id, name="Test", input="test")
+
+        with patch.object(handler, "log_node_tree", side_effect=RuntimeError("conversion failed")):
+            handler.end_node(run_id, output="result")
+
+        assert splunk_ao_logger.current_parent() is caller_trace
+        splunk_ao_logger.conclude(output="done")