diff --git a/CHANGELOG.md b/CHANGELOG.md
index ef615a7..5f2cba1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,7 +4,7 @@ All notable changes to this project will be documented in this file.
 
 The format is based on Keep a Changelog and this project adheres to Semantic Versioning.
 
-## [0.1.97b1] - 2026-07-28
+## [0.1.97] -
 
 - **Label evaluation/simulation traces on the root span** - Root spans produced by evaluation test runs (`Netra.evaluation`) and simulation runs (`Netra.simulation`) now carry a `netra.trace.origin` attribute set to `evaluation`, letting the backend and frontend distinguish these traces from normal workflow invocations.
 
@@ -20,7 +20,7 @@ The format is based on Keep a Changelog and this project adheres to Semantic Ver
 
 - **Fix span attributes in OpenAI instrumentation** - Assistant completions no longer emit empty entries when the model returns `content: null` alongside tool calls, request messages now correctly handle non-dictionary objects (such as Pydantic ChatCompletionMessage instances) by converting them with model_as_dict() instead of skipping them, and assistant `tool_calls` arrays as well as `tool_call_id` values on tool messages are now captured and serialized as indexed prompt and completion span attributes.
 
-- **Fix set_root_output_stream handling** - `set_root_output_stream` now reliably commits output for streams, even when iteration ends early (for example, via `break` or `.close()`), and correctly handles plain iterables by setting their output immediately with a warning recommending `Netra.set_root_output()`. Only true single-pass iterators are wrapped as streams.
+- **Fix set_root_output_stream handling** – `set_root_output_stream` now forces a commit of the inner stream to capture output when a stream exits early (for example, via `break` or `.close()`). It also correctly handles plain iterables by setting their output immediately with a warning recommending `Netra.set_root_output()`. Only true single-pass iterators are wrapped as streams.
 
 - **Refactor stream wrapper architecture to use callback injection** - `stream_utils` is now a pure utility module with no Netra-internal imports. The commit logic (serialize and set attribute on root span) is injected as a callback from `SessionManager`, eliminating the circular dependency between `stream_utils` and `SessionManager`.
 
@@ -336,4 +336,4 @@ Users can be now overwrite the input and ouput attributes of spans created by in
 
 - Added utility to set input and output data for any active span in a trace
 
-[0.1.97b1]: https://github.com/KeyValueSoftwareSystems/netra-sdk-py/tree/main
+[0.1.97]: https://github.com/KeyValueSoftwareSystems/netra-sdk-py/tree/main
diff --git a/netra/instrumentation/stream_utils.py b/netra/instrumentation/stream_utils.py
index 6338e66..c2fe066 100644
--- a/netra/instrumentation/stream_utils.py
+++ b/netra/instrumentation/stream_utils.py
@@ -53,6 +53,109 @@ def _generic_extractor(wrapper: Union["RootOutputSyncStreamWrapper", "RootOutput
     return "".join(wrapper._chunks)
 
 
+def _finalize_via_method(stream: Any) -> None:
+    """Try to finalize a Netra wrapper by calling ``_finalize`` or ``_finalize_span``.
+
+    ``_finalize`` is tried first because wrappers that use it (e.g. Agno)
+    have an idempotency guard (``_finalized`` flag), making them safe to
+    call unconditionally.  ``_finalize_span`` is the fallback for LLM
+    wrappers (openai, groq, cerebras, litellm, google_genai) that lack
+    such a guard.
+    """
+    if stream is None or not getattr(stream, "_netra_stream_wrapper", False):
+        return
+    for method_name in ("_finalize", "_finalize_span"):
+        fn = getattr(stream, method_name, None)
+        if fn is not None and callable(fn):
+            try:
+                fn()
+            except Exception:
+                logger.warning("_force_finalize_inner_stream: %s() failed", method_name, exc_info=True)
+            return
+
+
+def _force_finalize_inner_stream(iterator: Any, stream: Any) -> None:
+    """Force-finalize a **sync** inner stream so ``_netra_output`` is populated
+    before the outer wrapper's extractor reads it.
+
+    On early ``break`` the outer wrapper's ``_commit`` fires before the
+    inner wrapper has finalized.  This helper covers every inner iterator
+    variant without requiring changes to any instrumentation wrapper:
+
+    1. **Generator-based iterators** — the inner ``__iter__`` returns a
+       generator (either a plain non-Netra generator, or a Netra wrapper
+       that delegates to an internal plain generator).  Calling
+       ``iterator.close()`` throws ``GeneratorExit`` into the generator;
+       if its ``finally`` block triggers the wrapper's finalization,
+       ``_netra_output`` is set and we are done.
+
+    2. **Netra return-self iterators** — the inner wrapper's ``__iter__``
+       returns ``self`` and has no ``close()`` method.  We fall back to
+       calling the finalization method directly on *stream* (the original
+       inner wrapper).  All Netra instrumentation wrappers use either
+       ``_finalize()`` or ``_finalize_span()``; we try both.
+
+    Path 2 also serves as a fallback when path 1 closes a generator but
+    ``_netra_output`` is still unset (e.g. the generator catches
+    ``GeneratorExit`` without calling finalization).
+    """
+    if iterator is None:
+        return
+
+    # Path 1: generator-based inner iterator (sync only).
+    # Async iterators use ``aclose()`` (a coroutine) instead of ``close()``.
+    # Calling ``close()`` on an ObjectProxy-based async wrapper would close
+    # the underlying transport without triggering wrapper finalization.
+    if hasattr(iterator, "close") and not hasattr(iterator, "__anext__"):
+        try:
+            iterator.close()
+        except Exception:
+            logger.debug("_force_finalize_inner_stream: failed to close iterator", exc_info=True)
+
+    # If the inner wrapper already set _netra_output (either because
+    # path 1 triggered finalization, or because __next__'s StopIteration
+    # handler already called _finalize/_finalize_span on full exhaustion),
+    # there is nothing left to do.  Skipping path 2 here is essential
+    # because several instrumentation wrappers (openai, groq, cerebras,
+    # litellm, google_genai, elevenlabs) have no idempotency guard in
+    # _finalize_span — calling it twice would double-end the span.
+    if getattr(stream, "_netra_output", None) is not None:
+        return
+
+    # Path 2: return-self wrappers — call the finalization method directly.
+    # Only reached when _netra_output is still unset (i.e. early break).
+    _finalize_via_method(stream)
+
+
+async def _aforce_finalize_inner_stream(iterator: Any, stream: Any) -> None:
+    """Async variant of :func:`_force_finalize_inner_stream`.
+
+    Handles async generators via ``aclose()`` (path 1a) in addition to
+    sync generators (path 1b) and direct finalization (path 2).
+    """
+    if iterator is None:
+        return
+
+    # Path 1a: async generator — ``aclose()`` is a coroutine.
+    if hasattr(iterator, "aclose"):
+        try:
+            await iterator.aclose()
+        except Exception:
+            logger.debug("_aforce_finalize_inner_stream: failed to aclose iterator", exc_info=True)
+    # Path 1b: sync generator attached to an async wrapper (unusual but possible).
+    elif hasattr(iterator, "close") and not hasattr(iterator, "__anext__"):
+        try:
+            iterator.close()
+        except Exception:
+            logger.debug("_aforce_finalize_inner_stream: failed to close iterator", exc_info=True)
+
+    if getattr(stream, "_netra_output", None) is not None:
+        return
+
+    # Path 2: return-self wrappers — call the finalization method directly.
+    _finalize_via_method(stream)
+
+
 # Sync wrapper
 class RootOutputSyncStreamWrapper:
     """Wraps a **single-pass** sync iterator; on exhaustion commits the output
@@ -124,6 +227,7 @@ def _commit(self) -> None:
             return
         self._committed = True
         try:
+            _force_finalize_inner_stream(self._iterator, self._stream)
             self._commit_fn(self._extractor(self))
         except Exception:
             logger.debug("RootOutputSyncWrapper: failed to commit output", exc_info=True)
@@ -191,7 +295,7 @@ async def _aiter_gen(self) -> Any:
                     self._chunks.append(str(chunk))
                 yield chunk
         finally:
-            self._commit()
+            await self._acommit()
 
     def __aiter__(self) -> Any:
         return self._aiter_gen()
@@ -203,7 +307,7 @@ async def __anext__(self) -> Any:
                 self._chunks.append(str(chunk))
             return chunk
         except StopAsyncIteration:
-            self._commit()
+            await self._acommit()
             raise
 
     async def __aenter__(self) -> "RootOutputAsyncStreamWrapper":
@@ -215,7 +319,7 @@ async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
         if hasattr(self._stream, "__aexit__"):
             await self._stream.__aexit__(exc_type, exc_val, exc_tb)
         if exc_type is None:
-            self._commit()
+            await self._acommit()
 
     def __getattr__(self, name: str) -> Any:
         return getattr(self._stream, name)
@@ -224,11 +328,24 @@ def __del__(self) -> None:
         if not self._committed:
             self._commit()
 
+    async def _acommit(self) -> None:
+        """Async commit — uses ``aclose()`` to finalize async generator inner streams."""
+        if self._committed:
+            return
+        self._committed = True
+        try:
+            await _aforce_finalize_inner_stream(self._aiterator, self._stream)
+            self._commit_fn(self._extractor(self))
+        except Exception:
+            logger.debug("RootOutputAsyncWrapper: failed to commit output", exc_info=True)
+
     def _commit(self) -> None:
+        """Sync fallback for ``__del__`` and other non-async contexts."""
         if self._committed:
             return
         self._committed = True
         try:
+            _force_finalize_inner_stream(self._aiterator, self._stream)
             self._commit_fn(self._extractor(self))
         except Exception:
             logger.debug("RootOutputAsyncWrapper: failed to commit output", exc_info=True)
diff --git a/netra/version.py b/netra/version.py
index 1b6f4db..1646b62 100644
--- a/netra/version.py
+++ b/netra/version.py
@@ -1 +1 @@
-__version__ = "0.1.97b1"
+__version__ = "0.1.97dev2"
diff --git a/pyproject.toml b/pyproject.toml
index af628bd..40d136a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [project]
 name = "netra-sdk"
-version = "0.1.97b1"
+version = "0.1.97dev2"
 description = "A Python SDK for AI application observability that provides OpenTelemetry-based monitoring, tracing, and PII protection for LLM and vector database applications. Enables easy instrumentation, session tracking, and privacy-focused data collection for AI systems in production environments."
 authors = [
     {name = "Sooraj Thomas",email = "sooraj@keyvalue.systems"}
diff --git a/tests/test_stream_utils.py b/tests/test_stream_utils.py
index 88646b3..fe4088d 100644
--- a/tests/test_stream_utils.py
+++ b/tests/test_stream_utils.py
@@ -16,6 +16,8 @@
 from netra.instrumentation.stream_utils import (
     RootOutputAsyncStreamWrapper,
     RootOutputSyncStreamWrapper,
+    _aforce_finalize_inner_stream,
+    _force_finalize_inner_stream,
     _generic_extractor,
     _netra_extractor,
     wrap_stream_for_root_output,
@@ -392,3 +394,425 @@ def test_generator_is_wrapped(self) -> None:
         gen = (x for x in [1, 2, 3])
         wrapped = wrap_stream_for_root_output(gen, commit_fn)
         assert isinstance(wrapped, RootOutputSyncStreamWrapper)
+
+
+class _ReturnSelfSyncWrapper:
+    """Mimics an OpenAI-style return-self sync wrapper with ``_finalize_span``
+    that sets ``_netra_output`` (no idempotency guard)."""
+
+    _netra_stream_wrapper = True
+
+    def __init__(self, items: List[Any]) -> None:
+        self._items = items
+        self._netra_output: Any = None
+        self._finalize_span_called = False
+
+    def __iter__(self) -> "_ReturnSelfSyncWrapper":
+        return self
+
+    def __next__(self) -> Any:
+        if not self._items:
+            self._finalize_span()
+            raise StopIteration
+        return self._items.pop(0)
+
+    def _finalize_span(self) -> None:
+        self._finalize_span_called = True
+        self._netra_output = "finalized_output"
+
+
+class _ReturnSelfAsyncWrapper:
+    """Mimics an OpenAI-style return-self async wrapper with ``_finalize_span``."""
+
+    _netra_stream_wrapper = True
+
+    def __init__(self, items: List[Any]) -> None:
+        self._items = list(items)
+        self._netra_output: Any = None
+        self._finalize_span_called = False
+
+    def __aiter__(self) -> "_ReturnSelfAsyncWrapper":
+        return self
+
+    async def __anext__(self) -> Any:
+        if not self._items:
+            self._finalize_span()
+            raise StopAsyncIteration
+        return self._items.pop(0)
+
+    def _finalize_span(self) -> None:
+        self._finalize_span_called = True
+        self._netra_output = "async_finalized_output"
+
+
+class _IdempotentFinalizeWrapper:
+    """Mimics an Agno-style wrapper with ``_finalize`` and idempotency guard."""
+
+    _netra_stream_wrapper = True
+
+    def __init__(self, items: List[Any]) -> None:
+        self._items = items
+        self._netra_output: Any = None
+        self._finalized = False
+        self._finalize_call_count = 0
+
+    def __iter__(self) -> "_IdempotentFinalizeWrapper":
+        return self
+
+    def __next__(self) -> Any:
+        if not self._items:
+            self._finalize()
+            raise StopIteration
+        return self._items.pop(0)
+
+    def _finalize(self) -> None:
+        self._finalize_call_count += 1
+        if self._finalized:
+            return
+        self._finalized = True
+        self._netra_output = "agno_output"
+
+
+# --- _force_finalize_inner_stream unit tests ---
+
+
+class TestForceFinalize:
+
+    def test_none_iterator_returns_immediately(self) -> None:
+        """Passing ``iterator=None`` is a no-op."""
+        stream = _ReturnSelfSyncWrapper(["a"])
+        _force_finalize_inner_stream(None, stream)
+        assert stream._finalize_span_called is False
+        assert stream._netra_output is None
+
+    def test_path2_calls_finalize_span_on_return_self_wrapper(self) -> None:
+        """When ``_netra_output`` is ``None``, path 2 calls ``_finalize_span``."""
+        stream = _ReturnSelfSyncWrapper(["a", "b"])
+        iterator = iter(stream)
+        next(iterator)
+        _force_finalize_inner_stream(iterator, stream)
+        assert stream._finalize_span_called is True
+        assert stream._netra_output == "finalized_output"
+
+    def test_path1_generator_close_triggers_finalization(self) -> None:
+        """Closing a generator-based inner iterator triggers its ``finally`` block."""
+        finalized = {"called": False, "output": None}
+
+        class _GenBasedStream:
+            _netra_stream_wrapper = True
+
+            def __init__(self) -> None:
+                self._netra_output: Any = None
+
+            def __iter__(self) -> Iterator[Any]:
+                try:
+                    yield "a"
+                    yield "b"
+                finally:
+                    self._netra_output = "gen_finalized"
+                    finalized["called"] = True
+
+        stream = _GenBasedStream()
+        iterator = iter(stream)
+        next(iterator)
+        _force_finalize_inner_stream(iterator, stream)
+        assert finalized["called"] is True
+        assert stream._netra_output == "gen_finalized"
+
+    def test_skips_path2_when_netra_output_already_set(self) -> None:
+        """If ``_netra_output`` is already populated, path 2 is skipped."""
+        stream = _ReturnSelfSyncWrapper(["a"])
+        list(stream)  # exhaust fully — _finalize_span sets _netra_output
+        assert stream._netra_output == "finalized_output"
+        stream._finalize_span_called = False  # reset for tracking
+        _force_finalize_inner_stream(iter([]), stream)
+        assert stream._finalize_span_called is False
+
+    def test_path2_prefers_finalize_over_finalize_span(self) -> None:
+        """``_finalize`` is tried before ``_finalize_span`` for idempotency-safe wrappers."""
+        stream = _IdempotentFinalizeWrapper(["a", "b"])
+        iterator = iter(stream)
+        next(iterator)
+        _force_finalize_inner_stream(iterator, stream)
+        assert stream._finalized is True
+        assert stream._netra_output == "agno_output"
+        assert stream._finalize_call_count == 1
+
+    def test_path2_skips_non_netra_wrappers(self) -> None:
+        """Streams without ``_netra_stream_wrapper`` never trigger path 2."""
+
+        class _PlainIterator:
+            def __init__(self) -> None:
+                self._finalize_span_called = False
+                self._netra_output: Any = None
+
+            def _finalize_span(self) -> None:
+                self._finalize_span_called = True
+
+            def __iter__(self) -> "_PlainIterator":
+                return self
+
+            def __next__(self) -> Any:
+                raise StopIteration
+
+        stream = _PlainIterator()
+        _force_finalize_inner_stream(iter([]), stream)
+        assert stream._finalize_span_called is False
+
+    def test_async_iterator_skips_path1_close(self) -> None:
+        """Async iterators (with ``__anext__``) should not have ``close()`` called."""
+        close_called = {"value": False}
+
+        class _AsyncWithClose:
+            _netra_stream_wrapper = True
+
+            def __init__(self) -> None:
+                self._netra_output: Any = None
+
+            def __aiter__(self) -> "_AsyncWithClose":
+                return self
+
+            async def __anext__(self) -> Any:
+                raise StopAsyncIteration
+
+            def close(self) -> None:
+                close_called["value"] = True
+
+            def _finalize_span(self) -> None:
+                self._netra_output = "async_output"
+
+        stream = _AsyncWithClose()
+        _force_finalize_inner_stream(stream, stream)
+        assert close_called["value"] is False
+        assert stream._netra_output == "async_output"
+
+    def test_finalize_span_exception_is_logged_not_raised(self) -> None:
+        """If ``_finalize_span()`` raises, it is caught and does not propagate."""
+
+        class _BrokenWrapper:
+            _netra_stream_wrapper = True
+
+            def __init__(self) -> None:
+                self._netra_output: Any = None
+
+            def __iter__(self) -> "_BrokenWrapper":
+                return self
+
+            def __next__(self) -> Any:
+                raise StopIteration
+
+            def _finalize_span(self) -> None:
+                raise RuntimeError("span already ended")
+
+        stream = _BrokenWrapper()
+        _force_finalize_inner_stream(iter([]), stream)
+
+
+# --- Integration tests: early break with _force_finalize_inner_stream ---
+
+
+class TestEarlyBreakIntegration:
+
+    def test_sync_early_break_captures_output_from_return_self_wrapper(self) -> None:
+        """Early ``break`` on a sync wrapper around a return-self Netra inner stream
+        correctly captures the inner output via ``_force_finalize_inner_stream``."""
+        commit_fn = _make_commit_fn()
+        inner = _ReturnSelfSyncWrapper(["chunk1", "chunk2", "chunk3"])
+        wrapper = RootOutputSyncStreamWrapper(inner, commit_fn, _netra_extractor)
+        for _ in wrapper:
+            break
+        assert wrapper._committed is True
+        assert inner._finalize_span_called is True
+        commit_fn.assert_called_once_with("finalized_output")
+
+    def test_sync_full_exhaustion_with_return_self_wrapper(self) -> None:
+        """Full exhaustion of a return-self wrapper commits output without
+        double-calling ``_finalize_span`` (the ``_netra_output is not None``
+        guard in ``_force_finalize_inner_stream`` prevents it)."""
+        commit_fn = _make_commit_fn()
+        inner = _ReturnSelfSyncWrapper(["a", "b"])
+        wrapper = RootOutputSyncStreamWrapper(inner, commit_fn, _netra_extractor)
+        result = list(wrapper)
+        assert result == ["a", "b"]
+        assert wrapper._committed is True
+        commit_fn.assert_called_once_with("finalized_output")
+
+    def test_async_early_break_captures_output_from_return_self_wrapper(self) -> None:
+        """Early ``break`` on an async wrapper around a return-self Netra inner
+        stream correctly captures the inner output."""
+        commit_fn = _make_commit_fn()
+        inner = _ReturnSelfAsyncWrapper(["c1", "c2", "c3"])
+        wrapper = RootOutputAsyncStreamWrapper(inner, commit_fn, _netra_extractor)
+
+        async def _break_early() -> None:
+            async for _ in wrapper:
+                break
+
+        asyncio.run(_break_early())
+        assert wrapper._committed is True
+        assert inner._finalize_span_called is True
+        commit_fn.assert_called_once_with("async_finalized_output")
+
+    def test_sync_early_break_idempotent_wrapper(self) -> None:
+        """Early ``break`` with an Agno-style idempotent wrapper calls
+        ``_finalize`` exactly once."""
+        commit_fn = _make_commit_fn()
+        inner = _IdempotentFinalizeWrapper(["x", "y", "z"])
+        wrapper = RootOutputSyncStreamWrapper(inner, commit_fn, _netra_extractor)
+        for _ in wrapper:
+            break
+        assert wrapper._committed is True
+        assert inner._finalized is True
+        assert inner._finalize_call_count == 1
+        commit_fn.assert_called_once_with("agno_output")
+
+
+# --- _aforce_finalize_inner_stream unit tests ---
+
+
+class TestAsyncForceFinalize:
+
+    def test_none_iterator_returns_immediately(self) -> None:
+        """Passing ``iterator=None`` is a no-op."""
+        stream = _ReturnSelfAsyncWrapper(["a"])
+
+        async def _run() -> None:
+            await _aforce_finalize_inner_stream(None, stream)
+
+        asyncio.run(_run())
+        assert stream._finalize_span_called is False
+        assert stream._netra_output is None
+
+    def test_aclose_called_on_async_generator(self) -> None:
+        """``aclose()`` is awaited on async generator inner iterators."""
+        finalized = {"called": False}
+
+        class _AsyncGenStream:
+            _netra_stream_wrapper = True
+
+            def __init__(self) -> None:
+                self._netra_output: Any = None
+
+            async def __aiter__(self) -> Any:
+                try:
+                    yield "a"
+                    yield "b"
+                finally:
+                    self._netra_output = "async_gen_finalized"
+                    finalized["called"] = True
+
+        stream = _AsyncGenStream()
+
+        async def _run() -> None:
+            ait = stream.__aiter__()
+            await ait.__anext__()
+            await _aforce_finalize_inner_stream(ait, stream)
+
+        asyncio.run(_run())
+        assert finalized["called"] is True
+        assert stream._netra_output == "async_gen_finalized"
+
+    def test_path2_on_return_self_async_wrapper(self) -> None:
+        """Return-self async wrappers trigger path 2 (direct finalization)."""
+        stream = _ReturnSelfAsyncWrapper(["a", "b"])
+
+        async def _run() -> None:
+            ait = aiter(stream)
+            await ait.__anext__()
+            await _aforce_finalize_inner_stream(ait, stream)
+
+        asyncio.run(_run())
+        assert stream._finalize_span_called is True
+        assert stream._netra_output == "async_finalized_output"
+
+    def test_skips_path2_when_netra_output_already_set(self) -> None:
+        """If ``_netra_output`` is already populated, path 2 is skipped."""
+        stream = _ReturnSelfAsyncWrapper(["a"])
+
+        async def _exhaust_and_check() -> None:
+            ait = aiter(stream)
+            try:
+                while True:
+                    await ait.__anext__()
+            except StopAsyncIteration:
+                pass
+            assert stream._netra_output == "async_finalized_output"
+            stream._finalize_span_called = False
+            await _aforce_finalize_inner_stream(ait, stream)
+
+        asyncio.run(_exhaust_and_check())
+        assert stream._finalize_span_called is False
+
+    def test_aclose_exception_is_caught(self) -> None:
+        """If ``aclose()`` raises, it is caught and does not propagate."""
+
+        class _BrokenAsyncGen:
+            _netra_stream_wrapper = True
+
+            def __init__(self) -> None:
+                self._netra_output: Any = None
+
+            async def aclose(self) -> None:
+                raise RuntimeError("aclose failed")
+
+            def _finalize_span(self) -> None:
+                self._netra_output = "recovered"
+
+        stream = _BrokenAsyncGen()
+
+        async def _run() -> None:
+            await _aforce_finalize_inner_stream(stream, stream)
+
+        asyncio.run(_run())
+        assert stream._netra_output == "recovered"
+
+
+class TestAsyncEarlyBreakIntegration:
+
+    def test_async_early_break_with_async_gen_inner_stream(self) -> None:
+        """Early ``break`` on an async wrapper around an async-generator-based
+        Netra inner stream correctly triggers ``aclose()`` and captures output."""
+
+        class _AsyncGenNetraStream:
+            _netra_stream_wrapper = True
+
+            def __init__(self) -> None:
+                self._netra_output: Any = None
+
+            async def __aiter__(self) -> Any:
+                try:
+                    yield "chunk1"
+                    yield "chunk2"
+                    yield "chunk3"
+                finally:
+                    self._netra_output = "async_gen_output"
+
+        commit_fn = _make_commit_fn()
+        inner = _AsyncGenNetraStream()
+        wrapper = RootOutputAsyncStreamWrapper(inner, commit_fn, _netra_extractor)
+
+        async def _break_early() -> None:
+            async for _ in wrapper:
+                break
+
+        asyncio.run(_break_early())
+        assert wrapper._committed is True
+        assert inner._netra_output == "async_gen_output"
+        commit_fn.assert_called_once_with("async_gen_output")
+
+    def test_async_full_exhaustion_with_return_self_wrapper(self) -> None:
+        """Full exhaustion of an async return-self wrapper commits correctly."""
+        commit_fn = _make_commit_fn()
+        inner = _ReturnSelfAsyncWrapper(["a", "b"])
+        wrapper = RootOutputAsyncStreamWrapper(inner, commit_fn, _netra_extractor)
+
+        async def _consume() -> List[Any]:
+            result = []
+            async for chunk in wrapper:
+                result.append(chunk)
+            return result
+
+        result = asyncio.run(_consume())
+        assert result == ["a", "b"]
+        assert wrapper._committed is True
+        commit_fn.assert_called_once_with("async_finalized_output")