diff --git a/CHANGELOG.md b/CHANGELOG.md
index fec4614c..0237a062 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Nested `@log(span_type="tool")` calls now retain the calling tool as their parent and expose their own active
+  tracing context during execution. Tool, workflow, and agent generators restore the consumer context between
+  yields and keep their own outputs when interleaved.
+
 ## [0.6.0]
 
 ### Added
diff --git a/src/splunk_ao/decorator.py b/src/splunk_ao/decorator.py
index 6188e839..d5821786 100644
--- a/src/splunk_ao/decorator.py
+++ b/src/splunk_ao/decorator.py
@@ -48,19 +48,21 @@ def call_llm(prompt, temperature=0.7):
 import json
 import logging
 from collections.abc import AsyncGenerator, Callable, Generator
-from contextlib import contextmanager
-from contextvars import ContextVar
-from dataclasses import dataclass
+from contextlib import ExitStack, contextmanager, nullcontext
+from contextvars import Context, ContextVar, copy_context
+from dataclasses import dataclass, replace
 from functools import wraps
 from types import TracebackType
 from typing import Any, TypeVar, cast, overload
 
+from opentelemetry import context as otel_context
 from typing_extensions import ParamSpec
 
-from galileo_core.schemas.logging.span import WorkflowSpan
+from galileo_core.schemas.logging.span import StepWithChildSpans, ToolSpan, WorkflowSpan
 from galileo_core.schemas.logging.trace import Trace
 from splunk_ao.constants import LoggerModeType
 from splunk_ao.logger import SplunkAOLogger
+from splunk_ao.logger.logger import _otel_context_state
 from splunk_ao.schema.content_blocks import is_content_block_list
 from splunk_ao.schema.datasets import DatasetRecord
 from splunk_ao.schema.metrics import LocalMetricConfig
@@ -123,6 +125,8 @@ class _CallState:
     owns_trace: bool
     set_trace_context: bool
     span_stack_depth: int
+    span: StepWithChildSpans | None = None
+    scope: ExitStack | None = None
 
 
 def _get_or_init_list(context_var: ContextVar, default_factory: Callable = list) -> list:
@@ -732,15 +736,50 @@ def _safe_prepare_call(
         _CallState | None
             Per-call ownership state when preparation succeeds.
         """
+        scope = ExitStack() if span_type in (None, "workflow", "agent", "tool") else None
         try:
-            return self._prepare_call(span_type, span_params, dataset_record)
+            if scope is not None:
+                scope.enter_context(self._operation_execution_context())
+            state = self._prepare_call(span_type, span_params, dataset_record)
+            return replace(state, scope=scope) if scope is not None else state
         except Exception as e:
+            if scope is not None:
+                scope.close()
             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 None
 
+    @contextmanager
+    def _operation_execution_context(
+        self, state: _CallState | None = None, execution: Context | None = None
+    ) -> Generator[None, None, None]:
+        """Activate an operation only while its function or generator step executes."""
+        logger = state.logger if state is not None else self.get_logger_instance()
+        caller = logger.current_parent()
+        with ExitStack() as restore:
+            token = otel_context.attach(otel_context.get_current())
+            restore.callback(otel_context.detach, token)
+            # Rebuilding operation context must not consume the caller's activation tokens.
+            tracking_token = _otel_context_state.set(None)
+            restore.callback(_otel_context_state.reset, tracking_token)
+            trace_token = _trace_context.set(state.trace if state is not None else _trace_context.get())
+            restore.callback(_trace_context.reset, trace_token)
+            stack_token = _span_stack_context.set(list(_span_stack_context.get() or []))
+            restore.callback(_span_stack_context.reset, stack_token)
+            restore.callback(logger._set_current_parent, caller)
+            if state is not None:
+                parent = execution.run(logger.current_parent) if execution is not None else state.span
+                logger._set_current_parent(parent)
+                if execution is not None:
+                    _span_stack_context.set(list(execution.get(_span_stack_context) or []))
+                    token = otel_context.attach(execution.run(otel_context.get_current))
+                    restore.callback(otel_context.detach, token)
+                elif state.span is not None and not isinstance(state.span, ToolSpan):
+                    _get_or_init_list(_span_stack_context).append(state.span)
+            yield
+
     def _prepare_call(
         self, span_type: SPAN_TYPE | None, span_params: dict[str, Any], dataset_record: DatasetRecord | None
     ) -> _CallState:
@@ -801,6 +840,7 @@ def _prepare_call(
             set_trace_context=set_trace_context,
             span_stack_depth=span_stack_depth,
         )
+        span = None
         try:
             if not span_type or span_type in ["workflow", "agent"]:
                 created_at = span_params.get("created_at", _get_timestamp())
@@ -812,6 +852,20 @@ def _prepare_call(
                 else:
                     span = client_instance.add_workflow_span(input=input_, name=name, created_at=created_at)
                 _get_or_init_list(_span_stack_context).append(span)
+            elif span_type == "tool":
+                # A tool must be active while its function runs so nested calls inherit it.
+                valid_params = inspect.signature(client_instance.add_tool_span).parameters
+                kwargs = {**span_params, "input": input_}
+                span = client_instance.add_tool_span(
+                    **{
+                        key: value
+                        for key, value in kwargs.items()
+                        if key in valid_params and key not in {"output", "redacted_output"}
+                    }
+                )
+                if span is None:
+                    raise RuntimeError("Unable to start a Splunk AO tool span")
+                client_instance._set_current_parent(span)
         except Exception:
             if owns_trace:
                 self._conclude_owned_trace(call_state, "", 500)
@@ -819,7 +873,7 @@ def _prepare_call(
                 _trace_context.set(None)
             raise
 
-        return call_state
+        return replace(call_state, span=span) if span is not None else call_state
 
     def _get_input_from_func_args(
         self, *, is_method: bool = False, func_args: tuple = (), func_kwargs: dict | None = None
@@ -877,6 +931,9 @@ def _finalize_call(
         -------
         The original result, possibly wrapped if it's a generator
         """
+        if (inspect.isgenerator(result) or inspect.isasyncgen(result)) and call_state.scope is not None:
+            # A factory has already executed; suspend its operation before returning to the consumer.
+            call_state.scope.close()
         if inspect.isgenerator(result):
             return self._wrap_sync_generator_result(span_type, span_params, result, call_state=call_state)
         if inspect.isasyncgen(result):
@@ -918,6 +975,8 @@ def _complete_call(
                     _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)
+                if call_state.scope is not None:
+                    call_state.scope.close()
 
         return result
 
@@ -1035,20 +1094,21 @@ def _handle_call_result(
                 span_params["created_at"] = created_at
                 span_params["duration_ns"] = 0
 
-            # Workflow and agent spans are "concludable" - they need to be concluded
-            # Default (no span_type) is treated as workflow
-            if not span_type or is_concludable_span_type(span_type):
-                # Pop from stack and conclude the workflow/agent span
-                if stack:
+            # Decorated tools, workflows, and agents start before execution.
+            if not span_type or is_concludable_span_type(span_type) or span_type == "tool":
+                if span_type != "tool" and stack:
                     stack.pop()
                     _span_stack_context.set(stack)
 
                 status_code = span_params.get("status_code")
-                logger.conclude(output=output, duration_ns=span_params["duration_ns"], status_code=status_code)
+                conclude_kwargs = {"redacted_output": redacted_output} if span_type == "tool" else {}
+                logger.conclude(
+                    output=output, duration_ns=span_params["duration_ns"], status_code=status_code, **conclude_kwargs
+                )
 
             else:
-                # Non-concludable spans (llm, tool, retriever) are  added to the parent
-                span_methods = {"llm": "add_llm_span", "tool": "add_tool_span", "retriever": "add_retriever_span"}
+                # LLM and retriever spans are added to the parent after execution.
+                span_methods = {"llm": "add_llm_span", "retriever": "add_retriever_span"}
 
                 if span_type in span_methods:
                     method_name = span_methods[span_type]
@@ -1107,15 +1167,27 @@ def _wrap_sync_generator_result(
         """
         items = []
         state = call_state or self._safe_prepare_call(span_type, span_params, dataset_record)
+        has_operation = state is not None and state.span is not None
+        execution: Context | None = None
+        if state is not None and state.scope is not None:
+            state.scope.close()
         error: BaseException | None = None
 
         try:
-            for item in generator:
+            while True:
+                with self._operation_execution_context(state, execution) if has_operation else nullcontext():
+                    try:
+                        item = next(generator)
+                    except StopIteration:
+                        break
+                    if has_operation:
+                        execution = copy_context()
                 items.append(item)
 
                 yield item
         except GeneratorExit:
-            generator.close()
+            with self._operation_execution_context(state, execution) if has_operation else nullcontext():
+                generator.close()
             raise
         except BaseException as exc:
             error = exc
@@ -1127,7 +1199,8 @@ def _wrap_sync_generator_result(
                 output = "".join(items)
 
             if state is not None:
-                self._complete_call(span_type, span_params, output, state, error=error)
+                with self._operation_execution_context(state) if has_operation else nullcontext():
+                    self._complete_call(span_type, span_params, output, state, error=error)
 
     async def _wrap_async_generator_result(
         self,
@@ -1162,15 +1235,27 @@ async def _wrap_async_generator_result(
 
         items = []
         state = call_state or self._safe_prepare_call(span_type, span_params, dataset_record)
+        has_operation = state is not None and state.span is not None
+        execution: Context | None = None
+        if state is not None and state.scope is not None:
+            state.scope.close()
         error: BaseException | None = None
 
         try:
-            async for item in generator:
+            while True:
+                with self._operation_execution_context(state, execution) if has_operation else nullcontext():
+                    try:
+                        item = await anext(generator)
+                    except StopAsyncIteration:
+                        break
+                    if has_operation:
+                        execution = copy_context()
                 items.append(item)
 
                 yield item
         except GeneratorExit:
-            await generator.aclose()
+            with self._operation_execution_context(state, execution) if has_operation else nullcontext():
+                await generator.aclose()
             raise
         except BaseException as exc:
             error = exc
@@ -1182,7 +1267,8 @@ async def _wrap_async_generator_result(
                 output = "".join(items)
 
             if state is not None:
-                self._complete_call(span_type, span_params, output, state, error=error)
+                with self._operation_execution_context(state) if has_operation else nullcontext():
+                    self._complete_call(span_type, span_params, output, state, error=error)
 
     def get_logger_instance(
         self,
diff --git a/tests/test_decorator_distributed.py b/tests/test_decorator_distributed.py
index 3b6a2e77..d4ad53f4 100644
--- a/tests/test_decorator_distributed.py
+++ b/tests/test_decorator_distributed.py
@@ -9,6 +9,7 @@
 from galileo_core.schemas.shared.multimodal import ContentModality
 from splunk_ao import Message, MessageRole, extract_tracing_context, log, splunk_ao_context
 from splunk_ao.schema.content_blocks import DataContentBlock, TextContentBlock
+from splunk_ao.schema.trace import SPAN_TYPE
 from splunk_ao.tracing import get_tracing_headers
 from tests.testutils.setup import setup_mock_logstreams_client, setup_mock_projects_client, setup_mock_traces_client
 
@@ -43,10 +44,11 @@ def init_logger():
     return splunk_ao_context.get_logger_instance()
 
 
-def test_decorator_get_tracing_headers(reset_context: None, distributed_clients: Mock) -> None:
+@pytest.mark.parametrize("span_type", ["workflow", "tool"])
+def test_decorator_get_tracing_headers(reset_context: None, distributed_clients: Mock, span_type: SPAN_TYPE) -> None:
     logger = init_logger()
 
-    @log(span_type="workflow")
+    @log(span_type=span_type)
     def orchestrator(query: str) -> dict:
         return {"result": query, "headers": get_tracing_headers()}
 
@@ -62,12 +64,15 @@ def orchestrator(query: str) -> dict:
     assert logger.current_parent() is None
 
 
-def test_decorator_respects_incoming_distributed_context(reset_context: None, distributed_clients: Mock) -> None:
+@pytest.mark.parametrize("span_type, operation", [("workflow", "invoke_workflow"), ("tool", "execute_tool")])
+def test_decorator_respects_incoming_distributed_context(
+    reset_context: None, distributed_clients: Mock, span_type: SPAN_TYPE, operation: str
+) -> None:
     trace_id = "4bf92f3577b34da6a3ce929d0e0e4736"
     parent_id = "00f067aa0ba902b7"
     logger = init_logger()
 
-    @log(span_type="workflow")
+    @log(span_type=span_type)
     def downstream_service(query: str) -> str:
         return f"processed: {query}"
 
@@ -81,7 +86,7 @@ def downstream_service(query: str) -> str:
     exported = logger._sink.spans[-1]
     assert exported.context.trace_id == int(trace_id, 16)
     assert exported.parent.span_id == int(parent_id, 16)
-    assert (exported.attributes or {})["gen_ai.operation.name"] == "invoke_workflow"
+    assert (exported.attributes or {})["gen_ai.operation.name"] == operation
 
 
 def test_completed_workflow_is_enqueued_and_flush_does_not_change_ownership(
diff --git a/tests/test_decorator_operation_ownership.py b/tests/test_decorator_operation_ownership.py
index 04305ad0..2b9a939d 100644
--- a/tests/test_decorator_operation_ownership.py
+++ b/tests/test_decorator_operation_ownership.py
@@ -4,8 +4,10 @@
 from unittest.mock import patch
 
 import pytest
+from opentelemetry import trace as otel_trace
+from opentelemetry.sdk.trace import TracerProvider
 
-from splunk_ao import log, splunk_ao_context
+from splunk_ao import get_tracing_headers, log, splunk_ao_context
 from splunk_ao.schema.trace import SPAN_TYPE
 from tests.testutils.setup import setup_mock_logstreams_client, setup_mock_projects_client, setup_mock_traces_client
 
@@ -62,6 +64,111 @@ def operation(value: str) -> str:
     assert logger.current_parent() is None
 
 
+def test_nested_tools_activate_their_own_context_and_preserve_parenting(initialized_context: None) -> None:
+    # Given: two decorated tools nested beneath an agent, with no framework callbacks.
+    headers = {}
+
+    @log(span_type="tool", name="inner")
+    def inner() -> str:
+        headers["inner"] = get_tracing_headers()["traceparent"]
+        return "done"
+
+    @log(span_type="tool", name="outer")
+    def outer() -> str:
+        headers["outer"] = get_tracing_headers()["traceparent"]
+        return inner()
+
+    @log(span_type="agent", name="agent")
+    def agent() -> str:
+        return outer()
+
+    # When: the nested calls finish normally.
+    assert agent() == "done"
+
+    # Then: every tool is active during execution and keeps the actual calling tool as parent.
+    logger = splunk_ao_context.get_logger_instance()
+    spans = {span.name: span for span in logger._sink.spans}
+    assert len(spans) == 3
+    agent_span = spans["invoke_agent agent"]
+    outer_span = spans["execute_tool outer"]
+    inner_span = spans["execute_tool inner"]
+    assert outer_span.parent == agent_span.context
+    assert inner_span.parent == outer_span.context
+    assert agent_span.context.trace_id == outer_span.context.trace_id == inner_span.context.trace_id
+    for name, span in (("inner", inner_span), ("outer", outer_span)):
+        assert int(headers[name].split("-")[2], 16) == span.context.span_id
+        assert span.attributes["gen_ai.operation.name"] == "execute_tool"
+    assert logger.current_parent() is None
+    assert not otel_trace.get_current_span().get_span_context().is_valid
+
+
+@pytest.mark.asyncio
+async def test_async_nested_tools_preserve_parenting_and_caller_context(initialized_context: None) -> None:
+    # Given: nested async tools within a caller-owned agent operation.
+    logger = splunk_ao_context.get_logger_instance()
+    logger.start_trace(input="request")
+    caller = logger.add_agent_span(input="request", name="caller")
+    caller_context = otel_trace.get_current_span().get_span_context()
+    contexts = {}
+
+    @log(span_type="tool", name="inner")
+    async def inner() -> str:
+        contexts["inner"] = otel_trace.get_current_span().get_span_context()
+        await asyncio.sleep(0)
+        return "done"
+
+    @log(span_type="tool", name="outer")
+    async def outer() -> str:
+        contexts["outer"] = otel_trace.get_current_span().get_span_context()
+        return await inner()
+
+    # When: async decorated work completes beneath that caller.
+    assert await outer() == "done"
+
+    # Then: only decorator-owned operations complete, with their original active identities.
+    assert logger.current_parent() is caller
+    assert otel_trace.get_current_span().get_span_context() == caller_context
+    logger.conclude(output="done", conclude_all=True)
+    spans = {span.name: span for span in logger._sink.spans}
+    assert len(spans) == 3
+    assert spans["execute_tool outer"].parent == spans["invoke_agent caller"].context
+    assert spans["execute_tool inner"].parent == spans["execute_tool outer"].context
+    assert contexts["outer"] == spans["execute_tool outer"].context
+    assert contexts["inner"] == spans["execute_tool inner"].context
+
+
+@pytest.mark.parametrize("sibling_type", ["tool", "llm", "retriever", "workflow", "agent"])
+def test_calls_after_nested_tool_remain_siblings(initialized_context: None, sibling_type: SPAN_TYPE) -> None:
+    @log(span_type="tool", name="inner")
+    def inner() -> str:
+        return "inner result"
+
+    @log(span_type="tool", name="outer")
+    def outer() -> str:
+        return inner()
+
+    @log(span_type=sibling_type, name="sibling")
+    def sibling() -> str:
+        return "sibling result"
+
+    @log(span_type="agent", name="caller")
+    def caller() -> str:
+        outer()
+        return sibling()
+
+    assert caller() == "sibling result"
+    logger = splunk_ao_context.get_logger_instance()
+    spans = logger._sink.spans
+    assert len(spans) == 4
+    agent = next(span for span in spans if span.name == "invoke_agent caller")
+    outer_span = next(span for span in spans if span.name == "execute_tool outer")
+    inner_span = next(span for span in spans if span.name == "execute_tool inner")
+    sibling_span = next(span for span in spans if span not in (agent, outer_span, inner_span))
+    assert inner_span.parent == outer_span.context
+    assert outer_span.parent == sibling_span.parent == agent.context
+    assert logger.current_parent() is None
+
+
 @pytest.mark.parametrize("span_type", ["agent", "workflow"])
 def test_operation_without_return_inherits_last_child_output(initialized_context: None, span_type: SPAN_TYPE) -> None:
     # Given: an operation that prints its child's answer without returning it
@@ -164,8 +271,11 @@ def nested_operation() -> str:
     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")
+@pytest.mark.parametrize("span_type", ["workflow", "tool"])
+def test_user_exception_is_preserved_and_owned_trace_is_concluded(
+    initialized_context: None, span_type: SPAN_TYPE
+) -> None:
+    @log(span_type=span_type)
     def failing_operation() -> None:
         raise RuntimeError("application failure")
 
@@ -177,10 +287,17 @@ def failing_operation() -> 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")
+@pytest.mark.parametrize("span_type", ["workflow", "agent", "tool", None])
+def test_sync_generator_concludes_on_close_and_preserves_errors(
+    initialized_context: None, span_type: SPAN_TYPE | None
+) -> None:
+    @log(span_type="tool", name="child")
+    def child() -> str:
+        return "first"
+
+    @log(span_type=span_type, name="stream")
     def stream(fail: bool = False):
-        yield "first"
+        yield child()
         if fail:
             raise ValueError("stream failure")
         yield "second"
@@ -188,7 +305,7 @@ def stream(fail: bool = False):
     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
+    assert splunk_ao_context.get_current_trace() is None
     closed_stream.close()
     assert splunk_ao_context.get_current_trace() is None
 
@@ -197,14 +314,23 @@ def stream(fail: bool = False):
     with pytest.raises(ValueError, match="stream failure"):
         next(failing_stream)
     assert splunk_ao_context.get_current_trace() is None
+    spans = splunk_ao_context.get_logger_instance()._sink.spans
+    parents = [span.context for span in spans if span.name.endswith(" stream")]
+    children = [span for span in spans if span.name == "execute_tool child"]
+    assert len(spans) == 4
+    assert len(children) == 2
+    assert all(child.parent in parents for child in children)
 
 
 @pytest.mark.asyncio
-async def test_async_operations_have_invocation_local_ownership(initialized_context: None) -> None:
+@pytest.mark.parametrize("span_type", ["workflow", "agent", "tool", None])
+async def test_async_operations_have_invocation_local_ownership(
+    initialized_context: None, span_type: SPAN_TYPE | None
+) -> None:
     entered = 0
     both_entered = asyncio.Event()
 
-    @log(span_type="workflow")
+    @log(span_type=span_type)
     async def operation(value: str) -> str:
         nonlocal entered
         entered += 1
@@ -223,20 +349,446 @@ async def operation(value: str) -> str:
 
 
 @pytest.mark.asyncio
-async def test_async_generator_concludes_on_close(initialized_context: None) -> None:
-    @log(span_type="workflow")
+@pytest.mark.parametrize("span_type", ["workflow", "agent", "tool", None])
+async def test_async_generator_concludes_on_close(initialized_context: None, span_type: SPAN_TYPE | None) -> None:
+    @log(span_type="tool", name="child")
+    async def child() -> str:
+        return "first"
+
+    @log(span_type=span_type, name="stream")
     async def stream():
-        yield "first"
+        yield await child()
         yield "second"
 
     result = stream()
     assert await anext(result) == "first"
-    assert splunk_ao_context.get_current_trace() is not None
+    assert splunk_ao_context.get_current_trace() is 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
+    spans = logger._sink.spans
+    assert len(spans) == 2
+    parent = next(span for span in spans if span.name.endswith(" stream"))
+    child_span = next(span for span in spans if span.name == "execute_tool child")
+    assert child_span.parent == parent.context
+
+
+@pytest.mark.asyncio
+async def test_cancelled_tool_restores_caller_context(initialized_context: None) -> None:
+    logger = splunk_ao_context.get_logger_instance()
+    caller = logger.start_trace(input="request")
+    caller_context = otel_trace.get_current_span().get_span_context()
+
+    @log(span_type="tool")
+    async def cancelled() -> None:
+        raise asyncio.CancelledError()
+
+    with pytest.raises(asyncio.CancelledError):
+        await cancelled()
+
+    assert logger.current_parent() is caller
+    assert otel_trace.get_current_span().get_span_context() == caller_context
+    assert len(logger._sink.spans) == 1
+    assert logger._sink.spans[0].status.status_code == otel_trace.StatusCode.ERROR
+    logger.conclude(output="done")
+
+
+def test_manual_tool_keeps_caller_active(initialized_context: None) -> None:
+    # Decorating tools must not change the leaf semantics of the public logger method.
+    logger = splunk_ao_context.get_logger_instance()
+    logger.start_trace(input="request")
+    caller = logger.add_agent_span(input="request", name="caller")
+    caller_context = otel_trace.get_current_span().get_span_context()
+    logger.add_tool_span(input="input", output="output", name="manual")
+
+    assert logger.current_parent() is caller
+    assert otel_trace.get_current_span().get_span_context() == caller_context
+    logger.conclude(output="done", conclude_all=True)
+
+
+@pytest.mark.parametrize("factory", [False, True])
+@pytest.mark.parametrize(
+    "span_type, operation",
+    [("tool", "execute_tool"), ("workflow", "invoke_workflow"), ("agent", "invoke_agent"), (None, "invoke_workflow")],
+)
+def test_generator_restores_consumer_between_yields(
+    initialized_context: None, factory: bool, span_type: SPAN_TYPE | None, operation: str
+) -> None:
+    # Given: an operation yields nested work, then its consumer makes a sibling call.
+    logger = splunk_ao_context.get_logger_instance()
+    contexts = []
+
+    @log(span_type="tool", name="nested")
+    def nested() -> str:
+        return "chunk"
+
+    def generate():
+        contexts.append(otel_trace.get_current_span().get_span_context())
+        yield nested()
+        contexts.append(otel_trace.get_current_span().get_span_context())
+        yield nested()
+
+    def create_stream():
+        return generate()
+
+    stream = log(span_type=span_type, name="stream")(create_stream if factory else generate)
+
+    @log(span_type="tool", name="sibling")
+    def sibling() -> str:
+        return "sibling"
+
+    @log(span_type="agent", name="caller")
+    def caller() -> str:
+        parent = logger.current_parent()
+        caller_context = otel_trace.get_current_span().get_span_context()
+        iterator = stream()
+        assert logger.current_parent() is parent
+        for chunk in iterator:
+            assert chunk == "chunk"
+            assert logger.current_parent() is parent
+            assert otel_trace.get_current_span().get_span_context() == caller_context
+            sibling()
+        return "done"
+
+    assert caller() == "done"
+    spans = logger._sink.spans
+    agent = next(span for span in spans if span.name == "invoke_agent caller")
+    tool = next(span for span in spans if span.name == f"{operation} stream")
+    assert len(spans) == 6
+    assert tool.parent == agent.context
+    assert contexts == [tool.context, tool.context]
+    assert all(span.parent == tool.context for span in spans if span.name == "execute_tool nested")
+    assert all(span.parent == agent.context for span in spans if span.name == "execute_tool sibling")
+    assert logger.current_parent() is None
+    assert not otel_trace.get_current_span().get_span_context().is_valid
+
+
+@pytest.mark.parametrize("factory", [False, True])
+@pytest.mark.parametrize(
+    "span_type, operation",
+    [("tool", "execute_tool"), ("workflow", "invoke_workflow"), ("agent", "invoke_agent"), (None, "invoke_workflow")],
+)
+def test_interleaved_generators_keep_their_own_output(
+    initialized_context: None, factory: bool, span_type: SPAN_TYPE | None, operation: str
+) -> None:
+    # Given: two generators that finish in a different order from their last activation.
+    def generate(prefix):
+        yield prefix + "1"
+        yield prefix + "2"
+
+    if factory:
+        first = log(span_type=span_type, name="first")(lambda: generate("A"))
+        second = log(span_type=span_type, name="second")(lambda: generate("B"))
+    else:
+
+        @log(span_type=span_type, name="first")
+        def first():
+            yield from generate("A")
+
+        @log(span_type=span_type, name="second")
+        def second():
+            yield from generate("B")
+
+    @log(span_type="agent", name="caller")
+    def caller() -> None:
+        a, b = first(), second()
+        try:
+            assert list(zip(a, b, strict=False)) == [("A1", "B1"), ("A2", "B2")]
+        finally:
+            a.close()
+            b.close()
+
+    assert caller() is None
+    spans = {span.name: span for span in splunk_ao_context.get_logger_instance()._sink.spans}
+    assert len(spans) == 3
+    agent = spans["invoke_agent caller"]
+    assert json.loads(agent.attributes["gen_ai.output.messages"])[0]["parts"] == [{"type": "text", "content": "B1B2"}]
+    for name, output in (("first", "A1A2"), ("second", "B1B2")):
+        span = spans[f"{operation} {name}"]
+        assert span.parent == agent.context
+        if span_type == "tool":
+            assert json.loads(span.attributes["gen_ai.tool.call.result"]) == {"value": output}
+        else:
+            assert json.loads(span.attributes["gen_ai.output.messages"])[0]["parts"] == [
+                {"type": "text", "content": output}
+            ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("factory", ["generator", "sync", "async"])
+@pytest.mark.parametrize(
+    "span_type, operation",
+    [("tool", "execute_tool"), ("workflow", "invoke_workflow"), ("agent", "invoke_agent"), (None, "invoke_workflow")],
+)
+async def test_async_generator_restores_consumer_between_yields(
+    initialized_context: None, factory: str, span_type: SPAN_TYPE | None, operation: str
+) -> None:
+    logger = splunk_ao_context.get_logger_instance()
+    contexts = []
+
+    @log(span_type="tool", name="nested")
+    async def nested() -> str:
+        await asyncio.sleep(0)
+        return "chunk"
+
+    async def generate():
+        contexts.append(otel_trace.get_current_span().get_span_context())
+        yield await nested()
+        contexts.append(otel_trace.get_current_span().get_span_context())
+        yield await nested()
+
+    if factory == "sync":
+
+        @log(span_type=span_type, name="stream")
+        def stream():
+            return generate()
+    elif factory == "async":
+
+        @log(span_type=span_type, name="stream")
+        async def stream():
+            return generate()
+    else:
+        stream = log(span_type=span_type, name="stream")(generate)
+
+    @log(span_type="tool", name="sibling")
+    async def sibling() -> str:
+        return "sibling"
+
+    @log(span_type="agent", name="caller")
+    async def caller() -> str:
+        parent = logger.current_parent()
+        caller_context = otel_trace.get_current_span().get_span_context()
+        iterator = await stream() if factory == "async" else stream()
+        assert logger.current_parent() is parent
+        async for chunk in iterator:
+            assert chunk == "chunk"
+            assert logger.current_parent() is parent
+            assert otel_trace.get_current_span().get_span_context() == caller_context
+            await sibling()
+        return "done"
+
+    assert await caller() == "done"
+    spans = logger._sink.spans
+    agent = next(span for span in spans if span.name == "invoke_agent caller")
+    tool = next(span for span in spans if span.name == f"{operation} stream")
+    assert len(spans) == 6
+    assert tool.parent == agent.context
+    assert contexts == [tool.context, tool.context]
+    assert all(span.parent == tool.context for span in spans if span.name == "execute_tool nested")
+    assert all(span.parent == agent.context for span in spans if span.name == "execute_tool sibling")
+    assert logger.current_parent() is None
+    assert not otel_trace.get_current_span().get_span_context().is_valid
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "span_type, operation",
+    [("tool", "execute_tool"), ("workflow", "invoke_workflow"), ("agent", "invoke_agent"), (None, "invoke_workflow")],
+)
+async def test_interleaved_async_generators_keep_their_own_output(
+    initialized_context: None, span_type: SPAN_TYPE | None, operation: str
+) -> None:
+    @log(span_type=span_type, name="stream")
+    async def stream(prefix):
+        yield prefix + "1"
+        await asyncio.sleep(0)
+        yield prefix + "2"
+
+    @log(span_type="agent", name="caller")
+    async def caller() -> None:
+        first, second = stream("A"), stream("B")
+        try:
+            assert await anext(first) == "A1"
+            assert await anext(second) == "B1"
+            assert await anext(first) == "A2"
+            assert await anext(second) == "B2"
+            with pytest.raises(StopAsyncIteration):
+                await anext(first)
+            with pytest.raises(StopAsyncIteration):
+                await anext(second)
+        finally:
+            await first.aclose()
+            await second.aclose()
+
+    assert await caller() is None
+    spans = splunk_ao_context.get_logger_instance()._sink.spans
+    agent = next(span for span in spans if span.name == "invoke_agent caller")
+    assert json.loads(agent.attributes["gen_ai.output.messages"])[0]["parts"] == [{"type": "text", "content": "B1B2"}]
+    tools = [span for span in spans if span.name == f"{operation} stream"]
+    assert len(spans) == 3
+    assert all(span.parent == agent.context for span in tools)
+    for span, output in zip(tools, ["A1A2", "B1B2"], strict=True):
+        if span_type == "tool":
+            assert json.loads(span.attributes["gen_ai.tool.call.result"]) == {"value": output}
+        else:
+            assert json.loads(span.attributes["gen_ai.output.messages"])[0]["parts"] == [
+                {"type": "text", "content": output}
+            ]
+
+
+@pytest.mark.parametrize(
+    "span_type, operation",
+    [("tool", "execute_tool"), ("workflow", "invoke_workflow"), ("agent", "invoke_agent"), (None, "invoke_workflow")],
+)
+def test_partially_consumed_generator_restores_native_caller(
+    initialized_context: None, span_type: SPAN_TYPE | None, operation: str
+) -> None:
+    # Given: the consumer retains a real native recording span while a tool is suspended.
+    logger = splunk_ao_context.get_logger_instance()
+    provider = TracerProvider()
+    observed = []
+
+    @log(span_type="tool", name="cleanup")
+    def cleanup() -> str:
+        return "closed"
+
+    @log(span_type=span_type, name="stream")
+    def stream():
+        try:
+            with provider.get_tracer(__name__).start_as_current_span("generator native span") as inner:
+                yield "first"
+                observed.append(otel_trace.get_current_span() is inner)
+                yield "second"
+                yield "third"
+        finally:
+            cleanup()
+
+    try:
+        with provider.get_tracer(__name__).start_as_current_span("native caller") as native:
+            iterator = stream()
+            assert next(iterator) == "first"
+            assert logger.current_parent() is None
+            assert splunk_ao_context.get_current_trace() is None
+            assert otel_trace.get_current_span() is native
+            otel_trace.get_current_span().set_attribute("after_yield", True)
+            assert next(iterator) == "second"
+            assert observed == [True]
+            assert otel_trace.get_current_span() is native
+            iterator.close()
+            assert otel_trace.get_current_span() is native
+            assert native.attributes["after_yield"] is True
+        spans = {span.name: span for span in logger._sink.spans}
+        assert len(spans) == 2
+        assert spans["execute_tool cleanup"].parent == spans[f"{operation} stream"].context
+        assert spans[f"{operation} stream"].parent == native.get_span_context()
+        assert logger.current_parent() is None
+        assert not logger._otel_ids
+        assert not logger._pending_otel_steps
+    finally:
+        provider.shutdown()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("finish", ["close", "error", "cancel"])
+@pytest.mark.parametrize(
+    "span_type, operation",
+    [("tool", "execute_tool"), ("workflow", "invoke_workflow"), ("agent", "invoke_agent"), (None, "invoke_workflow")],
+)
+async def test_suspended_async_generator_cleanup_restores_caller(
+    initialized_context: None, finish: str, span_type: SPAN_TYPE | None, operation: str
+) -> None:
+    logger = splunk_ao_context.get_logger_instance()
+    entered = asyncio.Event()
+
+    @log(span_type="tool", name="cleanup")
+    async def cleanup() -> str:
+        return "closed"
+
+    @log(span_type=span_type, name="stream")
+    async def stream():
+        try:
+            yield "first"
+            if finish == "error":
+                raise ValueError("stream failure")
+            entered.set()
+            await asyncio.Event().wait()
+        finally:
+            await cleanup()
+
+    @log(span_type="agent", name="caller")
+    async def caller() -> str:
+        parent = logger.current_parent()
+        caller_context = otel_trace.get_current_span().get_span_context()
+        iterator = stream()
+        assert await anext(iterator) == "first"
+        assert logger.current_parent() is parent
+        assert otel_trace.get_current_span().get_span_context() == caller_context
+        if finish == "close":
+            await iterator.aclose()
+        elif finish == "error":
+            with pytest.raises(ValueError, match="stream failure"):
+                await anext(iterator)
+        else:
+            task = asyncio.create_task(anext(iterator))
+            await entered.wait()
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+        assert logger.current_parent() is parent
+        assert otel_trace.get_current_span().get_span_context() == caller_context
+        return "done"
+
+    assert await caller() == "done"
+    spans = {span.name: span for span in logger._sink.spans}
+    assert len(spans) == 3
+    tool = spans[f"{operation} stream"]
+    assert tool.parent == spans["invoke_agent caller"].context
+    assert spans["execute_tool cleanup"].parent == tool.context
+    assert (tool.status.status_code == otel_trace.StatusCode.ERROR) == (finish != "close")
+    assert logger.current_parent() is None
+    assert not logger._otel_ids
+    assert not logger._pending_otel_steps
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("span_type", ["tool", "workflow", "agent", None])
+@pytest.mark.parametrize("asynchronous", [False, True])
+async def test_interleaved_top_level_generators_own_separate_traces(
+    initialized_context: None, span_type: SPAN_TYPE | None, asynchronous: bool
+) -> None:
+    logger = splunk_ao_context.get_logger_instance()
+    if asynchronous:
+
+        @log(span_type=span_type, name="stream")
+        async def stream(prefix):
+            yield prefix + "1"
+            await asyncio.sleep(0)
+            yield prefix + "2"
+    else:
+
+        @log(span_type=span_type, name="stream")
+        def stream(prefix):
+            yield prefix + "1"
+            yield prefix + "2"
+
+    first, second = stream("A"), stream("B")
+    for expected, iterator in [("A1", first), ("B1", second), ("A2", first), ("B2", second)]:
+        assert (await anext(iterator) if asynchronous else next(iterator)) == expected
+        assert logger.current_parent() is None
+        assert splunk_ao_context.get_current_trace() is None
+        assert not otel_trace.get_current_span().get_span_context().is_valid
+    for iterator in (first, second):
+        if asynchronous:
+            with pytest.raises(StopAsyncIteration):
+                await anext(iterator)
+        else:
+            with pytest.raises(StopIteration):
+                next(iterator)
+    spans = logger._sink.spans
+    assert len(spans) == 2
+    assert all(span.parent is None for span in spans)
+    assert len({span.context.trace_id for span in spans}) == 2
+    for span, output in zip(spans, ["A1A2", "B1B2"], strict=True):
+        if span_type == "tool":
+            assert json.loads(span.attributes["gen_ai.tool.call.result"]) == {"value": output}
+        else:
+            assert json.loads(span.attributes["gen_ai.output.messages"])[0]["parts"] == [
+                {"type": "text", "content": output}
+            ]
+    assert not logger._otel_ids
+    assert not logger._pending_otel_steps
 
 
 def test_flush_and_flush_all_do_not_conclude_active_trace(initialized_context: None) -> None: