From 3bbcf1925b8eb19b80ee14cb2d938aca8df8325c Mon Sep 17 00:00:00 2001
From: pradystar <pradeepn@splunk.com>
Date: Wed, 29 Jul 2026 19:18:18 -0700
Subject: [PATCH 1/4] fix(otel): enforce operation trace ownership and
 drain-only flush

---
 src/splunk_ao/decorator.py                    | 345 ++++++++++++++----
 src/splunk_ao/exporter/config.py              |  19 +-
 src/splunk_ao/handlers/base_async_handler.py  |  53 +--
 src/splunk_ao/handlers/base_handler.py        |  24 +-
 src/splunk_ao/handlers/crewai/handler.py      |   2 +-
 .../handlers/langchain/async_handler.py       |   2 +-
 src/splunk_ao/handlers/langchain/handler.py   |   2 +-
 .../handlers/langchain/middleware.py          |   2 +-
 .../handlers/openai_agents/handler.py         |  43 ++-
 tests/test_async_base_handler.py              |  16 +-
 tests/test_base_handler.py                    |  29 +-
 tests/test_decorator.py                       |  58 ++-
 tests/test_decorator_distributed.py           |  61 ++--
 tests/test_experiments.py                     |  12 +-
 tests/test_exporter_config.py                 |  31 ++
 tests/test_langchain.py                       |   2 +-
 tests/test_langchain_async.py                 |   2 +-
 tests/test_langchain_middleware.py            |   2 +-
 18 files changed, 513 insertions(+), 192 deletions(-)

diff --git a/src/splunk_ao/decorator.py b/src/splunk_ao/decorator.py
index 80398c65..d2bf9051 100644
--- a/src/splunk_ao/decorator.py
+++ b/src/splunk_ao/decorator.py
@@ -50,6 +50,7 @@ def call_llm(prompt, temperature=0.7):
 from collections.abc import AsyncGenerator, Callable, Generator
 from contextlib import contextmanager
 from contextvars import ContextVar
+from dataclasses import dataclass
 from functools import wraps
 from types import TracebackType
 from typing import Any, TypeVar, cast, overload
@@ -114,6 +115,15 @@ def call_llm(prompt, temperature=0.7):
 _span_stack_stack: ContextVar[list[list[WorkflowSpan]] | None] = ContextVar("span_stack_stack", default=None)
 
 
+@dataclass(frozen=True)
+class _CallState:
+    logger: SplunkAOLogger
+    trace: Trace
+    owns_trace: bool
+    set_trace_context: bool
+    span_stack_depth: int
+
+
 def _get_or_init_list(context_var: ContextVar, default_factory: Callable = list) -> list:
     """Helper function to get or initialize a context variable with a list."""
     value = context_var.get()
@@ -303,17 +313,32 @@ def log(
         """
 
         def decorator(func: Callable[P, R]) -> Callable[P, R]:
-            return (
+            if inspect.isasyncgenfunction(func):
+                return cast(
+                    Callable[P, R],
+                    self._async_generator_log(
+                        func, name=name, span_type=span_type, params=params, dataset_record=dataset_record
+                    ),
+                )
+            if inspect.isgeneratorfunction(func):
+                return cast(
+                    Callable[P, R],
+                    self._sync_generator_log(
+                        func, name=name, span_type=span_type, params=params, dataset_record=dataset_record
+                    ),
+                )
+            wrapped = (
                 self._async_log(func, name=name, span_type=span_type, params=params, dataset_record=dataset_record)
                 if asyncio.iscoroutinefunction(func)
                 else self._sync_log(func, name=name, span_type=span_type, params=params, dataset_record=dataset_record)
             )
+            return cast(Callable[P, R], wrapped)
 
         # If the decorator is called without arguments, return the decorator function itself.
         # This allows the decorator to be used with or without arguments.
         if func is None:
             return decorator
-        return decorator(func)
+        return cast(Callable[[Callable[P, R]], Callable[P, R]], decorator(func))
 
     def _async_log(
         self,
@@ -344,7 +369,7 @@ def _async_log(
         """
 
         @wraps(func)
-        async def async_wrapper(*args, **kwargs) -> Any:
+        async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
             # Copy the span stack to isolate parallel async tasks
             # This prevents concurrent tasks from interfering with each other's span stacks
             current_stack = _get_or_init_list(_span_stack_context)
@@ -363,20 +388,22 @@ async def async_wrapper(*args, **kwargs) -> Any:
                 func_args=args,
                 func_kwargs=kwargs,
             )
+            if span_params is None:
+                return await func(*args, **kwargs)
 
-            logging_enabled = self._safe_prepare_call(span_type, span_params, dataset_record)
+            call_state = self._safe_prepare_call(span_type, span_params, dataset_record)
 
-            result = None
             try:
                 result = await func(*args, **kwargs)
-            except Exception:
+            except BaseException as exc:
                 _logger.error("Error while executing function in async_wrapper", exc_info=True)
+                if call_state is not None:
+                    self._finalize_call(span_type, span_params, None, call_state, error=exc)
                 raise
-            finally:
-                if logging_enabled:
-                    result = self._finalize_call(span_type, span_params, result)
 
-            return result
+            if call_state is None:
+                return result
+            return self._finalize_call(span_type, span_params, result, call_state)
 
         return cast(F, async_wrapper)
 
@@ -409,7 +436,7 @@ def _sync_log(
         """
 
         @wraps(func)
-        def sync_wrapper(*args, **kwargs) -> Any:
+        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
             span_params = self._prepare_input(
                 func=func,
                 name=name or func.__name__,
@@ -419,22 +446,79 @@ def sync_wrapper(*args, **kwargs) -> Any:
                 func_args=args,
                 func_kwargs=kwargs,
             )
+            if span_params is None:
+                return func(*args, **kwargs)
 
-            logging_enabled = self._safe_prepare_call(span_type, span_params, dataset_record)
+            call_state = self._safe_prepare_call(span_type, span_params, dataset_record)
 
-            result = None
             try:
                 result = func(*args, **kwargs)
-            except Exception:
+            except BaseException as exc:
                 _logger.error("Error while executing function in sync_wrapper", exc_info=True)
+                if call_state is not None:
+                    self._finalize_call(span_type, span_params, None, call_state, error=exc)
                 raise
-            finally:
-                if logging_enabled:
-                    self._finalize_call(span_type, span_params, result)
-            return result
+
+            if call_state is None:
+                return result
+            return self._finalize_call(span_type, span_params, result, call_state)
 
         return cast(F, sync_wrapper)
 
+    def _sync_generator_log(
+        self,
+        func: F,
+        *,
+        name: str | None,
+        span_type: SPAN_TYPE | None,
+        params: dict[str, str | Callable] | None = None,
+        dataset_record: DatasetRecord | None = None,
+    ) -> F:
+        @wraps(func)
+        def generator_wrapper(*args: Any, **kwargs: Any) -> Generator:
+            span_params = self._prepare_input(
+                func=func,
+                name=name or func.__name__,
+                span_type=span_type,
+                params=params,
+                is_method=self._is_method(func),
+                func_args=args,
+                func_kwargs=kwargs,
+            )
+            generator = func(*args, **kwargs)
+            if span_params is None:
+                return generator
+            return self._wrap_sync_generator_result(span_type, span_params, generator, dataset_record=dataset_record)
+
+        return cast(F, generator_wrapper)
+
+    def _async_generator_log(
+        self,
+        func: F,
+        *,
+        name: str | None,
+        span_type: SPAN_TYPE | None,
+        params: dict[str, str | Callable] | None = None,
+        dataset_record: DatasetRecord | None = None,
+    ) -> F:
+        @wraps(func)
+        def async_generator_wrapper(*args: Any, **kwargs: Any) -> AsyncGenerator:
+            span_params = self._prepare_input(
+                func=func,
+                name=name or func.__name__,
+                span_type=span_type,
+                params=params,
+                is_method=self._is_method(func),
+                func_args=args,
+                func_kwargs=kwargs,
+            )
+            generator = func(*args, **kwargs)
+            if span_params is None:
+                return generator
+            return self._wrap_async_generator_result(span_type, span_params, generator, dataset_record=dataset_record)
+
+        return cast(F, async_generator_wrapper)
+
     @staticmethod
     def _is_method(func: Callable) -> bool:
         """
@@ -610,9 +694,9 @@ def _get_span_param_names(self, span_type: SPAN_TYPE) -> list[str]:
 
     def _safe_prepare_call(
         self, span_type: SPAN_TYPE | None, span_params: dict[str, Any], dataset_record: DatasetRecord | None
-    ) -> bool:
+    ) -> _CallState | None:
         """
-        Safely prepare telemetry, returning False if initialization fails.
+        Safely prepare telemetry, returning None if initialization fails.
 
         This method wraps _prepare_call with exception handling to ensure that
         telemetry initialization errors do not crash user code. Any exception
@@ -630,22 +714,21 @@ def _safe_prepare_call(
 
         Returns
         -------
-        bool
-            True if preparation succeeded, False if it failed
+        _CallState | None
+            Per-call ownership state when preparation succeeds.
         """
         try:
-            self._prepare_call(span_type, span_params, dataset_record)
-            return True
+            return self._prepare_call(span_type, span_params, dataset_record)
         except Exception as e:
             if isinstance(e, ConfigurationError):
                 _logger.error("Splunk AO logging initialization failed: %s", e, exc_info=True)
             else:
                 _logger.warning("Splunk AO logging initialization failed, continuing without logging: %s", e)
-            return False
+            return None
 
     def _prepare_call(
         self, span_type: SPAN_TYPE | None, span_params: dict[str, Any], dataset_record: DatasetRecord | None
-    ) -> None:
+    ) -> _CallState:
         """
         Prepare the call for logging by setting up trace and span contexts.
 
@@ -663,18 +746,23 @@ def _prepare_call(
         name = span_params.get("name", "")
 
         existing_trace = _trace_context.get()
+        current_parent = client_instance.current_parent()
+        span_stack_depth = len(_get_or_init_list(_span_stack_context))
 
-        # Check if existing trace is still valid (not concluded/flushed)
-        if existing_trace and client_instance.current_parent() is None:
+        if existing_trace and current_parent is None:
             existing_trace = None
             _trace_context.set(None)
 
+        owns_trace = False
+        set_trace_context = False
         if not existing_trace:
-            # If the singleton logger has an active trace, use it
-            if client_instance.has_active_trace():
-                trace = client_instance.traces[-1]
+            if current_parent is not None:
+                trace = current_parent
+                while trace._parent is not None:
+                    trace = trace._parent
+                if not isinstance(trace, Trace):
+                    raise RuntimeError("Active Splunk AO operation does not have a trace root")
             else:
-                # If no trace is available, start a new one
                 trace = client_instance.start_trace(
                     input=input_,
                     name=name,
@@ -683,20 +771,40 @@ def _prepare_call(
                     dataset_output=dataset_record.output if dataset_record else None,
                     dataset_metadata=dataset_record.metadata if dataset_record else None,
                 )
+                owns_trace = True
+            if not isinstance(trace, Trace):
+                raise RuntimeError("Unable to start a Splunk AO operation trace")
             _trace_context.set(trace)
-
-        # Start a workflow or agent span here
-        # If the user hasn't specified a span type, create and add a workflow span
-        if not span_type or span_type in ["workflow", "agent"]:
-            created_at = span_params.get("created_at", _get_timestamp())
-            if span_type == "agent":
-                agent_type = span_params.get("agent_type")
-                span = client_instance.add_agent_span(
-                    input=input_, name=name, agent_type=agent_type, created_at=created_at
-                )
-            else:
-                span = client_instance.add_workflow_span(input=input_, name=name, created_at=created_at)
-            _get_or_init_list(_span_stack_context).append(span)
+            set_trace_context = True
+        else:
+            trace = existing_trace
+
+        call_state = _CallState(
+            logger=client_instance,
+            trace=trace,
+            owns_trace=owns_trace,
+            set_trace_context=set_trace_context,
+            span_stack_depth=span_stack_depth,
+        )
+        try:
+            if not span_type or span_type in ["workflow", "agent"]:
+                created_at = span_params.get("created_at", _get_timestamp())
+                if span_type == "agent":
+                    agent_type = span_params.get("agent_type")
+                    span = client_instance.add_agent_span(
+                        input=input_, name=name, agent_type=agent_type, created_at=created_at
+                    )
+                else:
+                    span = client_instance.add_workflow_span(input=input_, name=name, created_at=created_at)
+                _get_or_init_list(_span_stack_context).append(span)
+        except Exception:
+            if owns_trace:
+                self._conclude_owned_trace(call_state, "", 500)
+            if set_trace_context and _trace_context.get() is trace:
+                _trace_context.set(None)
+            raise
+
+        return call_state
 
     def _get_input_from_func_args(
         self, *, is_method: bool = False, func_args: tuple = (), func_kwargs: dict | None = None
@@ -727,7 +835,13 @@ def _get_input_from_func_args(
         return json.loads(json.dumps(raw_input, cls=EventSerializer))
 
     def _finalize_call(
-        self, span_type: SPAN_TYPE | None, span_params: dict[str, Any], result: Any
+        self,
+        span_type: SPAN_TYPE | None,
+        span_params: dict[str, Any],
+        result: Any,
+        call_state: _CallState,
+        *,
+        error: BaseException | None = None,
     ) -> Generator | AsyncGenerator | Any:
         """
         Finalize the call logging by handling the result appropriately.
@@ -749,10 +863,66 @@ def _finalize_call(
         The original result, possibly wrapped if it's a generator
         """
         if inspect.isgenerator(result):
-            return self._wrap_sync_generator_result(span_type, span_params, result)
+            return self._wrap_sync_generator_result(span_type, span_params, result, call_state=call_state)
         if inspect.isasyncgen(result):
-            return self._wrap_async_generator_result(span_type, span_params, result)
-        return self._handle_call_result(span_type, span_params, result)
+            return self._wrap_async_generator_result(span_type, span_params, result, call_state=call_state)
+        return self._complete_call(span_type, span_params, result, call_state, error=error)
+
+    def _complete_call(
+        self,
+        span_type: SPAN_TYPE | None,
+        span_params: dict[str, Any],
+        result: Any,
+        call_state: _CallState,
+        *,
+        error: BaseException | None = None,
+    ) -> Any:
+        final_params = span_params
+        if error is not None and span_params.get("status_code") is None:
+            final_params = {**span_params, "status_code": 500}
+
+        output = final_params.get("output")
+        if output is None:
+            output = result if result is not None else ""
+
+        try:
+            self._handle_call_result(span_type, final_params, result, logger=call_state.logger)
+        finally:
+            try:
+                if call_state.owns_trace:
+                    self._conclude_owned_trace(call_state, output, final_params.get("status_code"))
+            except Exception:
+                _logger.warning("Failed to conclude Splunk AO operation trace", exc_info=True)
+            finally:
+                stack = _get_or_init_list(_span_stack_context)
+                if len(stack) > call_state.span_stack_depth:
+                    _span_stack_context.set(stack[: call_state.span_stack_depth])
+                if call_state.set_trace_context and _trace_context.get() is call_state.trace:
+                    _trace_context.set(None)
+
+        return result
+
+    def _conclude_owned_trace(self, call_state: _CallState, output: Any, status_code: int | None) -> None:
+        current_parent = call_state.logger.current_parent()
+        root = current_parent
+        while root is not None and root._parent is not None:
+            root = root._parent
+
+        if root is not call_state.trace:
+            return
+
+        try:
+            trace_output = self._serialize_output(output, None)
+        except Exception:
+            trace_output = ""
+
+        duration_ns = None
+        if call_state.trace.created_at is not None:
+            duration_ns = convert_time_delta_to_ns(_get_timestamp() - call_state.trace.created_at)
+
+        call_state.logger.conclude(
+            output=trace_output, duration_ns=duration_ns, status_code=status_code, conclude_all=True
+        )
 
     def _serialize_output(self, output: Any, span_type: SPAN_TYPE | None) -> Any:
         """
@@ -788,7 +958,14 @@ def _serialize_output(self, output: Any, span_type: SPAN_TYPE | None) -> Any:
         # Serialize and deserialize to ensure proper JSON serialization
         return json.loads(json.dumps(output, cls=EventSerializer))
 
-    def _handle_call_result(self, span_type: SPAN_TYPE | None, span_params: dict[str, Any], result: Any) -> Any:
+    def _handle_call_result(
+        self,
+        span_type: SPAN_TYPE | None,
+        span_params: dict[str, Any],
+        result: Any,
+        *,
+        logger: SplunkAOLogger | None = None,
+    ) -> Any:
         """
         Handle the result of a function call for logging.
 
@@ -809,7 +986,7 @@ def _handle_call_result(self, span_type: SPAN_TYPE | None, span_params: dict[str
         The original result
         """
         # Initialize logger before try block
-        logger = self.get_logger_instance()
+        logger = logger or self.get_logger_instance()
 
         # Serialize output and redacted_output - set to None if serialization fails
         output = span_params.get("output")
@@ -907,7 +1084,13 @@ def _handle_call_result(self, span_type: SPAN_TYPE | None, span_params: dict[str
         return result
 
     def _wrap_sync_generator_result(
-        self, span_type: SPAN_TYPE | None, span_params: dict[str, Any], generator: Generator
+        self,
+        span_type: SPAN_TYPE | None,
+        span_params: dict[str, Any],
+        generator: Generator,
+        *,
+        dataset_record: DatasetRecord | None = None,
+        call_state: _CallState | None = None,
     ) -> Generator:
         """
         Wrap a synchronous generator to log its results.
@@ -929,24 +1112,37 @@ def _wrap_sync_generator_result(
         A wrapped generator that yields the same items as the original
         """
         items = []
+        state = call_state or self._safe_prepare_call(span_type, span_params, dataset_record)
+        error: BaseException | None = None
 
         try:
             for item in generator:
                 items.append(item)
 
                 yield item
-        except Exception as e:
-            _logger.error(f"Failed to wrap generator result: {e}", exc_info=True)
+        except GeneratorExit:
+            generator.close()
+            raise
+        except BaseException as exc:
+            error = exc
+            raise
         finally:
-            output = items
+            output: Any = items
 
             if all(isinstance(item, str) for item in items):
                 output = "".join(items)
 
-            self._handle_call_result(span_type, span_params, output)
+            if state is not None:
+                self._complete_call(span_type, span_params, output, state, error=error)
 
     async def _wrap_async_generator_result(
-        self, span_type: SPAN_TYPE | None, span_params: dict[str, Any], generator: AsyncGenerator
+        self,
+        span_type: SPAN_TYPE | None,
+        span_params: dict[str, Any],
+        generator: AsyncGenerator,
+        *,
+        dataset_record: DatasetRecord | None = None,
+        call_state: _CallState | None = None,
     ) -> AsyncGenerator:
         """
         Wrap an asynchronous generator to log its results.
@@ -967,22 +1163,32 @@ async def _wrap_async_generator_result(
         -------
         A wrapped async generator that yields the same items as the original
         """
+        current_stack = _get_or_init_list(_span_stack_context)
+        _span_stack_context.set(current_stack.copy())
+
         items = []
+        state = call_state or self._safe_prepare_call(span_type, span_params, dataset_record)
+        error: BaseException | None = None
 
         try:
             async for item in generator:
                 items.append(item)
 
                 yield item
-        except Exception as e:
-            _logger.error(f"Failed to wrap generator result: {e}", exc_info=True)
+        except GeneratorExit:
+            await generator.aclose()
+            raise
+        except BaseException as exc:
+            error = exc
+            raise
         finally:
-            output = items
+            output: Any = items
 
             if all(isinstance(item, str) for item in items):
                 output = "".join(items)
 
-            self._handle_call_result(span_type, span_params, output)
+            if state is not None:
+                self._complete_call(span_type, span_params, output, state, error=error)
 
     def get_logger_instance(
         self,
@@ -1141,21 +1347,6 @@ def _on_flush_error(exc: Exception) -> None:
         except Exception as e:
             _on_flush_error(e)
 
-        # Reset trace state if we're flushing the current context
-        current_mode = _get_mode_or_default(mode) if mode is not None else _mode_context.get()
-        resolved_project = project if project is not None else _project_context.get()
-        resolved_log_stream = log_stream if log_stream is not None else _log_stream_context.get()
-        resolved_experiment_id = experiment_id if experiment_id is not None else _experiment_id_context.get()
-
-        if (
-            current_mode == _mode_context.get()
-            and resolved_project == _project_context.get()
-            and resolved_log_stream == _log_stream_context.get()
-            and resolved_experiment_id == _experiment_id_context.get()
-        ):
-            _span_stack_context.set([])
-            _trace_context.set(None)
-
     def flush_all(self) -> None:
         """
         Upload all captured traces under all contexts to Splunk AO.
@@ -1163,8 +1354,6 @@ def flush_all(self) -> None:
         This method flushes all traces regardless of project or log stream.
         """
         SplunkAOLoggerSingleton().flush_all()
-        _span_stack_context.set([])
-        _trace_context.set(None)
 
     def reset(self) -> None:
         """
diff --git a/src/splunk_ao/exporter/config.py b/src/splunk_ao/exporter/config.py
index 7a26b7c4..077a97cc 100644
--- a/src/splunk_ao/exporter/config.py
+++ b/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(
diff --git a/src/splunk_ao/handlers/base_async_handler.py b/src/splunk_ao/handlers/base_async_handler.py
index 6a00fad0..19d0f91f 100644
--- a/src/splunk_ao/handlers/base_async_handler.py
+++ b/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:
         """
diff --git a/src/splunk_ao/handlers/base_handler.py b/src/splunk_ao/handlers/base_handler.py
index 56b1af06..c095e717 100644
--- a/src/splunk_ao/handlers/base_handler.py
+++ b/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.
diff --git a/src/splunk_ao/handlers/crewai/handler.py b/src/splunk_ao/handlers/crewai/handler.py
index 0f477530..6adc870c 100644
--- a/src/splunk_ao/handlers/crewai/handler.py
+++ b/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()
 
diff --git a/src/splunk_ao/handlers/langchain/async_handler.py b/src/splunk_ao/handlers/langchain/async_handler.py
index 2c681438..6c21cd16 100644
--- a/src/splunk_ao/handlers/langchain/async_handler.py
+++ b/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(
diff --git a/src/splunk_ao/handlers/langchain/handler.py b/src/splunk_ao/handlers/langchain/handler.py
index 30376336..5a957065 100644
--- a/src/splunk_ao/handlers/langchain/handler.py
+++ b/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(
diff --git a/src/splunk_ao/handlers/langchain/middleware.py b/src/splunk_ao/handlers/langchain/middleware.py
index df0fb382..70497d60 100644
--- a/src/splunk_ao/handlers/langchain/middleware.py
+++ b/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:
diff --git a/src/splunk_ao/handlers/openai_agents/handler.py b/src/splunk_ao/handlers/openai_agents/handler.py
index 9ce91da7..cf54ddd3 100644
--- a/src/splunk_ao/handlers/openai_agents/handler.py
+++ b/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"),
diff --git a/tests/test_async_base_handler.py b/tests/test_async_base_handler.py
index 87f7fd75..b0fac381 100644
--- a/tests/test_async_base_handler.py
+++ b/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
diff --git a/tests/test_base_handler.py b/tests/test_base_handler.py
index e6fe20fc..9147a699 100644
--- a/tests/test_base_handler.py
+++ b/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")
diff --git a/tests/test_decorator.py b/tests/test_decorator.py
index a55a105e..4c25fa77 100644
--- a/tests/test_decorator.py
+++ b/tests/test_decorator.py
@@ -42,7 +42,7 @@ def llm_call(query: str) -> str:
     llm_call(query="input")
 
     assert len(splunk_ao_context.get_logger_instance().traces) == 1
-    assert splunk_ao_context.get_current_trace() is not None
+    assert splunk_ao_context.get_current_trace() is None
     assert splunk_ao_context.get_current_project() == "project-X"
     assert splunk_ao_context.get_current_log_stream() == "log-stream-X"
 
@@ -93,7 +93,7 @@ def llm_call(query: str) -> str:
 
     llm_call(query="input")
 
-    assert splunk_ao_context.get_current_trace() is not None
+    assert splunk_ao_context.get_current_trace() is None
 
     splunk_ao_context.flush()
 
@@ -129,7 +129,7 @@ def llm_call(query: str) -> str:
 
     llm_call(query="input")
 
-    assert splunk_ao_context.get_current_trace() is not None
+    assert splunk_ao_context.get_current_trace() is None
 
     splunk_ao_context.init(project="project-Y", log_stream="log-stream-Y")
 
@@ -137,7 +137,7 @@ def llm_call(query: str) -> str:
 
     llm_call(query="input")
 
-    assert splunk_ao_context.get_current_trace() is not None
+    assert splunk_ao_context.get_current_trace() is None
 
     splunk_ao_context.flush(project="project-X", log_stream="log-stream-X")
 
@@ -146,7 +146,7 @@ def llm_call(query: str) -> str:
     assert len(payload.traces) == 1
     assert len(payload.traces[0].spans) == 1
 
-    assert splunk_ao_context.get_current_trace() is not None
+    assert splunk_ao_context.get_current_trace() is None
 
     splunk_ao_context.flush(project="project-Y", log_stream="log-stream-Y")
 
@@ -176,20 +176,20 @@ def llm_call(query: str) -> str:
 
     llm_call(query="input_X")
 
-    trace_X = splunk_ao_context.get_current_trace()
+    logger_X = splunk_ao_context.get_logger_instance(project="project-X", log_stream="log-stream-X")
+    trace_X = logger_X.traces[-1]
     assert trace_X.input == '{"query": "input_X"}'
 
-    logger_X = splunk_ao_context.get_logger_instance(project="project-X", log_stream="log-stream-X")
     assert len(logger_X.traces) == 1
 
     splunk_ao_context.init(project="project-Y", log_stream="log-stream-Y")
 
     llm_call(query="input_Y")
 
-    trace_Y = splunk_ao_context.get_current_trace()
+    logger_Y = splunk_ao_context.get_logger_instance(project="project-Y", log_stream="log-stream-Y")
+    trace_Y = logger_Y.traces[-1]
     assert trace_Y.input == '{"query": "input_Y"}'
 
-    logger_Y = splunk_ao_context.get_logger_instance(project="project-Y", log_stream="log-stream-Y")
     assert len(logger_Y.traces) == 1
 
     # Flush both loggers
@@ -1293,7 +1293,7 @@ def llm_call(query: str) -> str:
         return "response"
 
     llm_call(query="input")
-    assert splunk_ao_context.get_current_trace() is not None
+    assert splunk_ao_context.get_current_trace() is None
 
     # Flush with explicit mode
     splunk_ao_context.flush(project="project-X", log_stream="log-stream-X", mode="batch")
@@ -1308,10 +1308,10 @@ def llm_call(query: str) -> str:
 @patch("splunk_ao.logger.logger.AgentStreams")
 @patch("splunk_ao.logger.logger.Projects")
 @patch("splunk_ao.logger.logger.Traces")
-def test_mode_flush_different_mode_no_reset(
+def test_mode_flush_different_mode_preserves_completed_operation_state(
     mock_traces_client: Mock, mock_projects_client: Mock, mock_logstreams_client: Mock, reset_context
 ) -> None:
-    """Test that flush with different mode doesn't reset current trace context."""
+    """Test that flushing another logger does not create or restore operation state."""
     setup_mock_traces_client(mock_traces_client)
     setup_mock_projects_client(mock_projects_client)
     setup_mock_logstreams_client(mock_logstreams_client)
@@ -1325,14 +1325,13 @@ def llm_call(query: str) -> str:
 
     llm_call(query="input")
     current_trace = splunk_ao_context.get_current_trace()
-    assert current_trace is not None
+    assert current_trace is None
 
     # Flush with a different mode shouldn't reset the current trace context
     # Since the logger instances are different
     splunk_ao_context.flush(project="project-X", log_stream="log-stream-X", mode="distributed")
 
-    # Current trace should still exist since we didn't flush the batch mode instance
-    assert splunk_ao_context.get_current_trace() is not None
+    # The completed operation remains concluded.
     assert splunk_ao_context.get_current_trace() == current_trace
 
 
@@ -1401,11 +1400,11 @@ def test_get_logger_instance_with_explicit_mode(
 @patch("splunk_ao.logger.logger.AgentStreams")
 @patch("splunk_ao.logger.logger.Projects")
 @patch("splunk_ao.logger.logger.Traces")
-def test_multiple_workflow_calls_create_one_trace_with_multiple_spans(
+def test_multiple_workflow_calls_create_independent_traces(
     mock_traces_client: Mock, mock_projects_client: Mock, mock_logstreams_client: Mock, reset_context
 ) -> None:
     """
-    Test that multiple workflow-level function calls are added as spans to a single trace in batch mode.
+    Test that each top-level decorated operation owns an independent trace.
     """
     mock_traces_client_instance = setup_mock_traces_client(mock_traces_client)
     setup_mock_projects_client(mock_projects_client)
@@ -1427,26 +1426,23 @@ def process_query(query: str) -> str:
     assert result2 == "Processed: query 2"
     assert result3 == "Processed: query 3"
 
-    # Before flush, verify only 1 trace was created with 3 workflow spans
+    # Before flush, verify each call produced one concluded trace.
     logger = splunk_ao_context.get_logger_instance()
-    assert len(logger.traces) == 1, f"Expected 1 trace, got {len(logger.traces)}"
-
-    # Verify the single trace has 3 workflow spans
-    trace = logger.traces[0]
-    assert len(trace.spans) == 3, f"Expected 3 spans, got {len(trace.spans)}"
-
-    # Verify each span has the correct input
-    assert trace.spans[0].input == '{"query": "query 1"}'
-    assert trace.spans[1].input == '{"query": "query 2"}'
-    assert trace.spans[2].input == '{"query": "query 3"}'
+    assert len(logger.traces) == 3
+    assert all(len(trace.spans) == 1 for trace in logger.traces)
+    assert [trace.spans[0].input for trace in logger.traces] == [
+        '{"query": "query 1"}',
+        '{"query": "query 2"}',
+        '{"query": "query 3"}',
+    ]
 
     # Flush the trace
     splunk_ao_context.flush()
 
-    # Verify ingest_traces was called with 1 trace containing 3 spans
+    # Verify all three operation traces are uploaded.
     payload = mock_traces_client_instance.ingest_traces.call_args[0][0]
-    assert len(payload.traces) == 1
-    assert len(payload.traces[0].spans) == 3
+    assert len(payload.traces) == 3
+    assert all(len(trace.spans) == 1 for trace in payload.traces)
 
     # After flush, trace context should be cleared
     assert splunk_ao_context.get_current_trace() is None
diff --git a/tests/test_decorator_distributed.py b/tests/test_decorator_distributed.py
index 137533e6..9cb685cc 100644
--- a/tests/test_decorator_distributed.py
+++ b/tests/test_decorator_distributed.py
@@ -49,13 +49,15 @@ def test_decorator_get_tracing_headers(reset_context: None, distributed_clients:
 
     @log(span_type="workflow")
     def orchestrator(query: str) -> dict:
-        return {"result": query, "headers": get_tracing_headers()}
+        trace = splunk_ao_context.get_current_trace()
+        return {"result": query, "headers": get_tracing_headers(), "trace_id": str(trace.id)}
 
     result = orchestrator("test input")
 
     headers = result["headers"]
-    assert headers[TRACE_ID_HEADER] == str(logger.traces[0].id)
+    assert headers[TRACE_ID_HEADER] == result["trace_id"]
     assert PARENT_ID_HEADER in headers
+    assert logger.current_parent() is None
 
 
 def test_decorator_respects_incoming_distributed_context(reset_context: None, distributed_clients: Mock) -> None:
@@ -76,7 +78,7 @@ def downstream_service(query: str) -> str:
     assert (logger._sink.spans[-1].attributes or {})["gen_ai.operation.name"] == "invoke_workflow"
 
 
-def test_completed_workflow_is_enqueued_and_flush_does_not_complete_root(
+def test_completed_workflow_is_enqueued_and_flush_does_not_change_ownership(
     reset_context: None, distributed_clients: Mock
 ) -> None:
     logger = init_logger()
@@ -95,14 +97,12 @@ def workflow(input_value: str) -> str:
         {"finish_reason": "unknown", "parts": [{"content": "output: test input", "type": "text"}], "role": "assistant"}
     ]
     assert emitted[0].end_time >= emitted[0].start_time
-    root = logger.current_parent()
-    assert root is not None
-    assert root.output == "output: test input"
+    assert logger.current_parent() is None
 
     logger.flush()
 
     assert logger._sink.spans == emitted
-    assert logger.current_parent() is root
+    assert logger.current_parent() is None
     distributed_clients.ingest_traces.assert_not_called()
     distributed_clients.ingest_spans.assert_not_called()
     distributed_clients.update_trace.assert_not_called()
@@ -120,10 +120,12 @@ def workflow() -> str:
     assert json.loads((logger._sink.spans[-1].attributes or {})["gen_ai.output.messages"]) == [
         {"finish_reason": "unknown", "parts": [{"content": "", "type": "text"}], "role": "assistant"}
     ]
-    assert logger.current_parent().output == ""
+    assert logger.current_parent() is None
 
 
-def test_trace_duration_accumulates_without_proprietary_updates(reset_context: None, distributed_clients: Mock) -> None:
+def test_top_level_workflows_have_independent_durations_and_traces(
+    reset_context: None, distributed_clients: Mock
+) -> None:
     logger = init_logger()
 
     @log(span_type="workflow")
@@ -135,17 +137,17 @@ def second() -> str:
         return "second"
 
     first()
-    first_duration = logger.current_parent().metrics.duration_ns
     second()
-    second_duration = logger.current_parent().metrics.duration_ns
+    first_span, second_span = logger._sink.spans
 
-    assert first_duration is not None and first_duration >= 0
-    assert second_duration is not None and second_duration >= first_duration
-    assert len(logger._sink.spans) == 2
+    assert first_span.end_time >= first_span.start_time
+    assert second_span.end_time >= second_span.start_time
+    assert first_span.context.trace_id != second_span.context.trace_id
+    assert logger.current_parent() is None
     distributed_clients.update_trace.assert_not_called()
 
 
-def test_content_blocks_are_preserved_on_open_trace(reset_context: None, distributed_clients: Mock) -> None:
+def test_content_blocks_are_preserved_on_completed_operation(reset_context: None, distributed_clients: Mock) -> None:
     logger = init_logger()
     blocks = [
         TextContentBlock(text="Here is the result"),
@@ -158,24 +160,33 @@ def workflow() -> list:
 
     workflow()
 
-    assert logger.current_parent().output == blocks
+    output_messages = (logger._sink.spans[-1].attributes or {})["gen_ai.output.messages"]
+    assert "Here is the result" in output_messages
+    assert "https://example.com/img.png" in output_messages
+    assert logger.current_parent() is None
 
 
 @pytest.mark.parametrize(
-    ("output", "expected_values"),
+    ("output", "expected_values", "expect_messages"),
     [
         (
             [Message(content="Hello", role=MessageRole.user), Message(content="Hi!", role=MessageRole.assistant)],
             ("Hello", "Hi!"),
+            True,
         ),
         (
             [Document(content="Tokyo is the capital of Japan."), Document(content="Mount Fuji is 3776m tall.")],
             ("Tokyo", "Mount Fuji"),
+            False,
         ),
     ],
 )
-def test_structured_outputs_are_serialized_on_open_trace(
-    reset_context: None, distributed_clients: Mock, output: list, expected_values: tuple[str, str]
+def test_only_message_outputs_use_otel_message_schema(
+    reset_context: None,
+    distributed_clients: Mock,
+    output: list,
+    expected_values: tuple[str, str],
+    expect_messages: bool,
 ) -> None:
     logger = init_logger()
 
@@ -184,7 +195,11 @@ def workflow() -> list:
         return output
 
     workflow()
-    trace_output = logger.current_parent().output
-
-    assert isinstance(trace_output, str)
-    assert all(value in trace_output for value in expected_values)
+    trace_output = (logger._sink.spans[-1].attributes or {}).get("gen_ai.output.messages")
+
+    if expect_messages:
+        assert isinstance(trace_output, str)
+        assert all(value in trace_output for value in expected_values)
+    else:
+        assert trace_output is None
+    assert logger.current_parent() is None
diff --git a/tests/test_experiments.py b/tests/test_experiments.py
index 77e1fb31..50e9b7d2 100644
--- a/tests/test_experiments.py
+++ b/tests/test_experiments.py
@@ -1591,18 +1591,16 @@ def simple_function(input: dict) -> str:
             payload = call[0][0]
             all_traces.extend(payload.traces)
 
-        # Each record creates a trace, and they're all sent in one batch
-        assert len(all_traces) >= 1  # At least one trace object
+        # Each record creates an independent operation trace.
+        assert len(all_traces) == len(local_dataset)
 
         # Count the actual workflow spans (each represents a processed record)
         total_spans = sum(len(trace.spans) for trace in all_traces)
         assert total_spans == len(local_dataset)
 
-        # Verify both records were processed (checking the input data in the spans)
-        first_trace = all_traces[0]
-        assert len(first_trace.spans) == 2
-        # Check that we have inputs for both Spain and Japan questions
-        span_inputs = [span.input for span in first_trace.spans]
+        # Verify both records were processed without sharing a trace.
+        assert all(len(trace.spans) == 1 for trace in all_traces)
+        span_inputs = [trace.spans[0].input for trace in all_traces]
         assert '{"input": "Which continent is Spain in?"}' in span_inputs[0]
         assert '{"input": "Which continent is Japan in?"}' in span_inputs[1]
 
diff --git a/tests/test_exporter_config.py b/tests/test_exporter_config.py
index 94e48910..dce16c1b 100644
--- a/tests/test_exporter_config.py
+++ b/tests/test_exporter_config.py
@@ -151,6 +151,37 @@ def test_explicit_routing_overrides_environment_resource_routing(monkeypatch: An
     assert resource.attributes["deployment.environment.name"] == "test"
 
 
+def test_resource_drops_reserved_environment_routing_when_sdk_routing_absent(monkeypatch: Any) -> None:
+    monkeypatch.setenv(
+        "OTEL_RESOURCE_ATTRIBUTES",
+        (
+            "splunk_ao.project.name=environment-project,"
+            "splunk_ao.logstream.id=environment-stream,"
+            "splunk_ao.experiment.id=environment-experiment,"
+            "deployment.environment.name=test"
+        ),
+    )
+
+    resource = create_otel_resource(make_routing())
+
+    for key in ("splunk_ao.project.name", "splunk_ao.logstream.id", "splunk_ao.experiment.id"):
+        assert key not in resource.attributes
+    assert resource.attributes["deployment.environment.name"] == "test"
+
+
+def test_resource_removes_conflicting_routing_forms(monkeypatch: Any) -> None:
+    monkeypatch.setenv(
+        "OTEL_RESOURCE_ATTRIBUTES", "splunk_ao.project.name=stale-project,splunk_ao.logstream.name=stale-stream"
+    )
+
+    resource = create_otel_resource(make_routing(project_id="project-id", log_stream_id="stream-id"))
+
+    assert "splunk_ao.project.name" not in resource.attributes
+    assert "splunk_ao.logstream.name" not in resource.attributes
+    assert resource.attributes["splunk_ao.project.id"] == "project-id"
+    assert resource.attributes["splunk_ao.logstream.id"] == "stream-id"
+
+
 def test_build_standalone_exporter_passes_resolved_public_config_to_factory() -> None:
     captured: dict[str, Any] = {}
     expected_exporter = object()
diff --git a/tests/test_langchain.py b/tests/test_langchain.py
index 677dbe54..6201e5dc 100644
--- a/tests/test_langchain.py
+++ b/tests/test_langchain.py
@@ -47,7 +47,7 @@ def test_initialization(self, splunk_ao_logger: SplunkAOLogger) -> None:
         callback = SplunkAOCallback(splunk_ao_logger=splunk_ao_logger)
         assert callback._handler._splunk_ao_logger == splunk_ao_logger
         assert callback._handler._start_new_trace is True
-        assert callback._handler._flush_on_chain_end is True
+        assert callback._handler._flush_on_chain_end is False
         assert callback._handler._nodes == {}
 
         # Custom initialization
diff --git a/tests/test_langchain_async.py b/tests/test_langchain_async.py
index ebe3a705..531c0213 100644
--- a/tests/test_langchain_async.py
+++ b/tests/test_langchain_async.py
@@ -43,7 +43,7 @@ async def test_initialization(self, splunk_ao_logger: SplunkAOLogger) -> None:
         callback = SplunkAOAsyncCallback(splunk_ao_logger=splunk_ao_logger)
         assert callback._handler._splunk_ao_logger == splunk_ao_logger
         assert callback._handler._start_new_trace is True
-        assert callback._handler._flush_on_chain_end is True
+        assert callback._handler._flush_on_chain_end is False
         assert callback._handler._nodes == {}
 
         # Custom initialization
diff --git a/tests/test_langchain_middleware.py b/tests/test_langchain_middleware.py
index 4ad180e3..4af3e7a3 100644
--- a/tests/test_langchain_middleware.py
+++ b/tests/test_langchain_middleware.py
@@ -131,7 +131,7 @@ def test_default_initialization(self, splunk_ao_logger: SplunkAOLogger) -> None:
         assert middleware._handler._splunk_ao_logger == splunk_ao_logger
         assert middleware._async_handler._splunk_ao_logger == splunk_ao_logger
         assert middleware._handler._start_new_trace is True
-        assert middleware._handler._flush_on_chain_end is True
+        assert middleware._handler._flush_on_chain_end is False
         assert middleware._root_run_id is None
 
     def test_custom_initialization(self, splunk_ao_logger: SplunkAOLogger) -> None:

From 0eaebdbed9d1790f567f8fce530832b158bfb8cd Mon Sep 17 00:00:00 2001
From: pradystar <pradeepn@splunk.com>
Date: Wed, 29 Jul 2026 19:18:44 -0700
Subject: [PATCH 2/4] add decorator ownership test

---
 tests/test_decorator_operation_ownership.py | 163 ++++++++++++++++++++
 1 file changed, 163 insertions(+)
 create mode 100644 tests/test_decorator_operation_ownership.py

diff --git a/tests/test_decorator_operation_ownership.py b/tests/test_decorator_operation_ownership.py
new file mode 100644
index 00000000..841bd08e
--- /dev/null
+++ b/tests/test_decorator_operation_ownership.py
@@ -0,0 +1,163 @@
+import asyncio
+from collections.abc import Generator
+from unittest.mock import patch
+
+import pytest
+
+from splunk_ao import log, splunk_ao_context
+from tests.testutils.setup import setup_mock_logstreams_client, setup_mock_projects_client, setup_mock_traces_client
+
+
+@pytest.fixture
+def initialized_context() -> Generator[None, None, None]:
+    with (
+        patch("splunk_ao.logger.logger.Traces") as traces,
+        patch("splunk_ao.logger.logger.Projects") as projects,
+        patch("splunk_ao.logger.logger.AgentStreams") as agent_streams,
+    ):
+        setup_mock_traces_client(traces)
+        setup_mock_projects_client(projects)
+        setup_mock_logstreams_client(agent_streams)
+        splunk_ao_context.init(project="project", log_stream="stream")
+        yield
+        splunk_ao_context.reset()
+
+
+def test_top_level_calls_create_independent_otel_traces(initialized_context: None) -> None:
+    @log(span_type="workflow")
+    def operation(value: str) -> str:
+        return value
+
+    assert operation("first") == "first"
+    assert operation("second") == "second"
+
+    logger = splunk_ao_context.get_logger_instance()
+    spans = logger._sink.spans
+
+    assert logger.current_parent() is None
+    assert splunk_ao_context.get_current_trace() is None
+    assert len(spans) == 2
+    assert spans[0].context.trace_id != spans[1].context.trace_id
+
+
+def test_nested_decorators_share_trace_and_preserve_parenting(initialized_context: None) -> None:
+    @log(span_type="llm")
+    def model_call(value: str) -> str:
+        return value.upper()
+
+    @log(span_type="workflow")
+    def operation(value: str) -> str:
+        return model_call(value)
+
+    assert operation("hello") == "HELLO"
+
+    logger = splunk_ao_context.get_logger_instance()
+    workflow = next(span for span in logger._sink.spans if span.name.endswith("operation"))
+    llm = next(span for span in logger._sink.spans if span.name == "chat")
+
+    assert workflow.context.trace_id == llm.context.trace_id
+    assert llm.parent.span_id == workflow.context.span_id
+    assert logger.current_parent() is None
+
+
+def test_decorator_does_not_conclude_caller_owned_trace(initialized_context: None) -> None:
+    logger = splunk_ao_context.get_logger_instance()
+    caller_trace = logger.start_trace(input="request", name="caller")
+
+    @log(span_type="workflow")
+    def nested_operation() -> str:
+        return "done"
+
+    assert nested_operation() == "done"
+    assert logger.current_parent() is caller_trace
+
+    logger.conclude(output="done")
+    assert logger.current_parent() is None
+
+
+def test_user_exception_is_preserved_and_owned_trace_is_concluded(initialized_context: None) -> None:
+    @log(span_type="workflow")
+    def failing_operation() -> None:
+        raise RuntimeError("application failure")
+
+    with pytest.raises(RuntimeError, match="application failure"):
+        failing_operation()
+
+    logger = splunk_ao_context.get_logger_instance()
+    assert logger.current_parent() is None
+    assert splunk_ao_context.get_current_trace() is None
+
+
+def test_sync_generator_concludes_on_close_and_preserves_errors(initialized_context: None) -> None:
+    @log(span_type="workflow")
+    def stream(fail: bool = False):
+        yield "first"
+        if fail:
+            raise ValueError("stream failure")
+        yield "second"
+
+    closed_stream = stream()
+    assert splunk_ao_context.get_current_trace() is None
+    assert next(closed_stream) == "first"
+    assert splunk_ao_context.get_current_trace() is not None
+    closed_stream.close()
+    assert splunk_ao_context.get_current_trace() is None
+
+    failing_stream = stream(fail=True)
+    assert next(failing_stream) == "first"
+    with pytest.raises(ValueError, match="stream failure"):
+        next(failing_stream)
+    assert splunk_ao_context.get_current_trace() is None
+
+
+@pytest.mark.asyncio
+async def test_async_operations_have_invocation_local_ownership(initialized_context: None) -> None:
+    entered = 0
+    both_entered = asyncio.Event()
+
+    @log(span_type="workflow")
+    async def operation(value: str) -> str:
+        nonlocal entered
+        entered += 1
+        if entered == 2:
+            both_entered.set()
+        await both_entered.wait()
+        return value
+
+    assert await asyncio.gather(operation("first"), operation("second")) == ["first", "second"]
+
+    logger = splunk_ao_context.get_logger_instance()
+    spans = logger._sink.spans
+    assert len(spans) == 2
+    assert spans[0].context.trace_id != spans[1].context.trace_id
+    assert logger.current_parent() is None
+
+
+@pytest.mark.asyncio
+async def test_async_generator_concludes_on_close(initialized_context: None) -> None:
+    @log(span_type="workflow")
+    async def stream():
+        yield "first"
+        yield "second"
+
+    result = stream()
+    assert await anext(result) == "first"
+    assert splunk_ao_context.get_current_trace() is not None
+    await result.aclose()
+
+    logger = splunk_ao_context.get_logger_instance()
+    assert logger.current_parent() is None
+    assert splunk_ao_context.get_current_trace() is None
+
+
+def test_flush_and_flush_all_do_not_conclude_active_trace(initialized_context: None) -> None:
+    logger = splunk_ao_context.get_logger_instance()
+    caller_trace = logger.start_trace(input="request", name="caller")
+
+    splunk_ao_context.flush()
+    assert logger.current_parent() is caller_trace
+
+    splunk_ao_context.flush_all()
+    assert logger.current_parent() is caller_trace
+
+    logger.conclude(output="done")

From c3d77888f08d3853daef41a73cb4d56f1ec0b6a0 Mon Sep 17 00:00:00 2001
From: pradystar <pradeepn@splunk.com>
Date: Thu, 30 Jul 2026 12:34:25 -0700
Subject: [PATCH 3/4] add test for openai agents

---
 tests/test_openai_agents.py | 53 +++++++++++++++++++++++++++++++++++++
 1 file changed, 53 insertions(+)

diff --git a/tests/test_openai_agents.py b/tests/test_openai_agents.py
index e8b2ee40..16ed813e 100644
--- a/tests/test_openai_agents.py
+++ b/tests/test_openai_agents.py
@@ -169,6 +169,59 @@ def test_processor_marks_direct_trace_child_agent(
     logger.conclude(output="output")
 
 
+@patch("splunk_ao.logger.logger.AgentStreams")
+@patch("splunk_ao.logger.logger.Projects")
+@patch("splunk_ao.logger.logger.Traces")
+def test_commit_failure_concludes_handler_owned_trace(
+    mock_traces_client: Mock, mock_projects_client: Mock, mock_logstreams_client: Mock
+) -> None:
+    setup_mock_traces_client(mock_traces_client)
+    setup_mock_projects_client(mock_projects_client)
+    setup_mock_logstreams_client(mock_logstreams_client)
+    logger = SplunkAOLogger(project="test", agent_stream="test")
+    processor = SplunkAOTracingProcessor(splunk_ao_logger=logger)
+    trace = MagicMock(trace_id="trace-id", name="Agent trace", metadata={})
+    processor.on_trace_start(trace)
+    owned_traces = []
+
+    def fail_after_starting_trace(*args, **kwargs):
+        processor._owned_trace = logger.add_trace(input="input", name="Trace")
+        owned_traces.append(processor._owned_trace)
+        raise RuntimeError("conversion failed")
+
+    with patch.object(processor, "_log_node_tree", side_effect=fail_after_starting_trace):
+        processor.on_trace_end(trace)
+
+    assert logger.current_parent() is None
+    assert owned_traces[0].status_code == 500
+    assert processor._nodes == {}
+    assert processor._owned_trace is None
+
+
+@patch("splunk_ao.logger.logger.AgentStreams")
+@patch("splunk_ao.logger.logger.Projects")
+@patch("splunk_ao.logger.logger.Traces")
+def test_commit_failure_preserves_caller_owned_trace(
+    mock_traces_client: Mock, mock_projects_client: Mock, mock_logstreams_client: Mock
+) -> None:
+    setup_mock_traces_client(mock_traces_client)
+    setup_mock_projects_client(mock_projects_client)
+    setup_mock_logstreams_client(mock_logstreams_client)
+    logger = SplunkAOLogger(project="test", agent_stream="test")
+    processor = SplunkAOTracingProcessor(splunk_ao_logger=logger)
+    caller_trace = logger.start_trace(input="request", name="caller")
+    trace = MagicMock(trace_id="trace-id", name="Agent trace", metadata={})
+    processor.on_trace_start(trace)
+
+    with patch.object(processor, "_commit_trace", side_effect=RuntimeError("conversion failed")):
+        processor.on_trace_end(trace)
+
+    assert logger.current_parent() is caller_trace
+    assert processor._nodes == {}
+    assert processor._owned_trace is None
+    logger.conclude(output="done")
+
+
 def _create_mock_response_with_tools(tool_calls: list[dict]) -> dict:
     """Create a mock OpenAI API response with embedded tool calls."""
     return {

From 72437a726df45007a0ffb4ded66338a7da3f7b02 Mon Sep 17 00:00:00 2001
From: pradystar <pradeepn@splunk.com>
Date: Thu, 30 Jul 2026 13:00:37 -0700
Subject: [PATCH 4/4] fix tests

---
 tests/test_exporter_config.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/test_exporter_config.py b/tests/test_exporter_config.py
index e873b633..335e3501 100644
--- a/tests/test_exporter_config.py
+++ b/tests/test_exporter_config.py
@@ -174,7 +174,7 @@ def test_resource_removes_conflicting_routing_forms(monkeypatch: Any) -> None:
         "OTEL_RESOURCE_ATTRIBUTES", "splunk_ao.project.name=stale-project,splunk_ao.logstream.name=stale-stream"
     )
 
-    resource = create_otel_resource(make_routing(project_id="project-id", log_stream_id="stream-id"))
+    resource = create_otel_resource(make_routing(project_id="project-id", agent_stream_id="stream-id"))
 
     assert "splunk_ao.project.name" not in resource.attributes
     assert "splunk_ao.logstream.name" not in resource.attributes