diff --git a/.github/instructions/instrumentation.instructions.md b/.github/instructions/instrumentation.instructions.md
index 513f637ba..6e56f01c0 100644
--- a/.github/instructions/instrumentation.instructions.md
+++ b/.github/instructions/instrumentation.instructions.md
@@ -54,6 +54,9 @@ prefer opt-in or additive. Breaking changes need explicit justification in the P
 - Message content, prompts, and tool call arguments must only be set through the util's content
   capture path — never as unconditional span/log attributes.
 - Adding attributes to invocations produced by the util is fine.
+- Flag captured request iterators that bypass `IteratorCapture`, eager consumption or replay,
+  wrapping unrelated arguments, capture when disabled, and snapshots taken before SDK consumption.
+  See [request property capture](../../AGENTS.md#capturing-properties-that-may-contain-iterators).
 - Streaming responses must be instrumented by subclassing the util's `SyncStreamWrapper` /
   `AsyncStreamWrapper` (`opentelemetry.util.genai.stream`). Flag hand-rolled stream wrappers, and
   invocation-backed wrappers that do not pass the invocation to `super().__init__(stream, invocation)`
@@ -109,6 +112,7 @@ Flag, with a link to the rule:
 
 - For every public API instrumented, cover sync/async variants when both exist.
 - Cover streaming and non-streaming variants when both exist.
+- Flag missing iterator tests for nested values, partial reads, errors, or file-backed content.
 - Cover happy path and error scenarios. For error scenarios, at minimum include: provider error /
   endpoint unavailable, stream interrupted by network, stream closed early by the caller, and a
   caller-side exception raised inside the `with …stream(…) as stream:` block before the stream is drained.
diff --git a/AGENTS.md b/AGENTS.md
index 75662ed0f..9373ef629 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -233,6 +233,32 @@ as the reference:
   `instrument(completion_hook=…)` override) and ship a `custom_hook.py` example mirroring the
   OpenAI package.
 
+#### Capturing properties that may contain iterators
+
+- Use `opentelemetry.util.genai.iterator_capture.IteratorCapture` for captured request properties
+  that may contain iterators, including nested values. Wrap those properties, not all `kwargs`.
+  Don't drain or replay them for telemetry.
+- Use one instance per named property: `IteratorCapture.from_property(kwargs, "messages", enabled=capture_content)`.
+  It replaces that property in `kwargs`. Use `capture.extract(extractor)`
+  after SDK consumption, including failures and lazy stream entry.
+  The extractor receives a snapshot with `()` for unread iterators and `None`
+  for disabled or missing properties; `capture.snapshot()` exposes the raw snapshot.
+- Preserve file objects. Extend the util for missing support; don't write local wrappers.
+- Test nested generators, partial reads, errors, and files across supported sync/async and
+  streaming variants.
+
+```python
+from opentelemetry.util.genai.iterator_capture import IteratorCapture
+
+capture = IteratorCapture.from_property(
+    kwargs, "messages", enabled=capture_content
+)
+try:
+    response = sdk_call(**kwargs)
+finally:
+    captured_messages = capture.extract(get_input_messages)
+```
+
 #### Streaming responses
 
 A streamed response only finishes once the caller has drained the stream, so the invocation must
diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/873.fixed b/instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/873.fixed
new file mode 100644
index 000000000..e696059ce
--- /dev/null
+++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/873.fixed
@@ -0,0 +1 @@
+Capture one-shot message, content, system, and tool iterators lazily as the Anthropic SDK consumes them.
diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/README.rst b/instrumentation/opentelemetry-instrumentation-genai-anthropic/README.rst
index 978a5f7fe..d45cc57e6 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/README.rst
+++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/README.rst
@@ -85,4 +85,3 @@ References
 * `OpenTelemetry GenAI semantic conventions <https://opentelemetry.io/docs/specs/semconv/gen-ai/>`_
 * `Anthropic SDK (Python) <https://github.com/anthropics/anthropic-sdk-python>`_
 * `Anthropic Documentation <https://docs.anthropic.com/>`_
-
diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/pyproject.toml b/instrumentation/opentelemetry-instrumentation-genai-anthropic/pyproject.toml
index aef6be05c..3045a9b3f 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/pyproject.toml
+++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/pyproject.toml
@@ -28,7 +28,7 @@ dependencies = [
   "opentelemetry-api ~= 1.43",
   "opentelemetry-instrumentation >= 0.64b0, <1",
   "opentelemetry-semantic-conventions >= 0.64b0, <1",
-  "opentelemetry-util-genai >= 1.2b0, <2",
+  "opentelemetry-util-genai >= 1.3b0.dev, <2",
 ]
 
 [project.optional-dependencies]
diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/messages_extractors.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/messages_extractors.py
index 0b65ae464..342483dc5 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/messages_extractors.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/messages_extractors.py
@@ -120,6 +120,8 @@ def get_input_messages(
         return []
     result: list[InputMessage] = []
     for message in messages:
+        if "role" not in message or "content" not in message:
+            continue
         role = message["role"]
         parts = convert_content_to_parts(message["content"])
         result.append(InputMessage(role=role, parts=parts))
diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/patch.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/patch.py
index 6169b7639..056b4fa56 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/patch.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/patch.py
@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Awaitable, Callable, Iterator
+from collections.abc import Awaitable, Callable
 from typing import TYPE_CHECKING, Any, cast
 
 from anthropic.types import Message as AnthropicMessage
@@ -16,6 +16,7 @@
 )
 from opentelemetry.util.genai.handler import TelemetryHandler
 from opentelemetry.util.genai.invocation import InferenceInvocation
+from opentelemetry.util.genai.iterator_capture import IteratorCapture
 
 from ._raw_response import wrap_raw_response
 from .messages_extractors import (
@@ -96,19 +97,6 @@ def _fail_context_manager_response(
         )
 
 
-def _materialize_tools(kwargs: dict[str, Any]) -> None:
-    """Replace a one-shot ``tools`` iterator with a list.
-
-    ``tools`` is typed as an iterable, so it may be a generator, and both the
-    SDK and the tool definitions read it. Call this before either of them does:
-    ``create`` reads the request after this wrapper, but ``stream`` serializes
-    it while building the manager, before the invocation is created.
-    """
-    tools = kwargs.get("tools")
-    if isinstance(tools, Iterator):
-        kwargs["tools"] = list(cast("Iterator[Any]", tools))
-
-
 def _is_raw_response(result: object) -> bool:
     """Whether ``result`` is a raw-response object to route through the proxy.
 
@@ -144,15 +132,31 @@ def traced_method(
         | AnthropicStream[RawMessageStreamEvent]
         | MessagesStreamWrapper[None]
     ):
-        _materialize_tools(kwargs)
-        invocation = _create_invocation(
-            handler, instance, args, kwargs, capture_content
+        messages_capture = IteratorCapture.from_property(
+            kwargs, "messages", capture_content
+        )
+        system_capture = IteratorCapture.from_property(
+            kwargs, "system", capture_content
         )
+        tools_capture = IteratorCapture.from_property(
+            kwargs, "tools", capture_content
+        )
+        invocation = _create_invocation(handler, instance, args, kwargs)
         try:
-            # ``Any``: through ``with_raw_response`` / ``with_streaming_response``
-            # this returns a response object that ``create``'s annotations do
-            # not describe, so the checks below cannot be narrowed away.
-            result: Any = wrapped(*args, **kwargs)
+            try:
+                # Raw response variants return types outside create's
+                # annotations, so Any keeps the checks below reachable.
+                result: Any = wrapped(*args, **kwargs)
+            finally:
+                invocation.input_messages = (
+                    messages_capture.extract(get_input_messages) or []
+                )
+                invocation.system_instruction = (
+                    system_capture.extract(get_system_instruction) or []
+                )
+                invocation.tool_definitions = tools_capture.extract(
+                    get_tool_definitions
+                )
         except BaseException as exc:
             invocation.fail(exc)
             raise
@@ -203,13 +207,30 @@ async def traced_method(
         | AnthropicAsyncStream[RawMessageStreamEvent]
         | AsyncMessagesStreamWrapper[None]
     ):
-        _materialize_tools(kwargs)
-        invocation = _create_invocation(
-            handler, instance, args, kwargs, capture_content
+        messages_capture = IteratorCapture.from_property(
+            kwargs, "messages", capture_content
+        )
+        system_capture = IteratorCapture.from_property(
+            kwargs, "system", capture_content
         )
+        tools_capture = IteratorCapture.from_property(
+            kwargs, "tools", capture_content
+        )
+        invocation = _create_invocation(handler, instance, args, kwargs)
         try:
-            # See ``messages_create`` on why this is Any.
-            result: Any = await wrapped(*args, **kwargs)
+            try:
+                # See ``messages_create`` on why this is Any.
+                result: Any = await wrapped(*args, **kwargs)
+            finally:
+                invocation.input_messages = (
+                    messages_capture.extract(get_input_messages) or []
+                )
+                invocation.system_instruction = (
+                    system_capture.extract(get_system_instruction) or []
+                )
+                invocation.tool_definitions = tools_capture.extract(
+                    get_tool_definitions
+                )
         except BaseException as exc:
             invocation.fail(exc)
             raise
@@ -239,7 +260,6 @@ def _create_invocation(
     instance: Messages | AsyncMessages,
     args: tuple[Any, ...],
     kwargs: dict[str, Any],
-    capture_content: bool,
 ) -> InferenceInvocation:
     params = extract_params(*args, **kwargs)
     attributes = get_llm_request_attributes(params, instance)
@@ -259,15 +279,6 @@ def _create_invocation(
         server_address=server_address,
         server_port=server_port,
     )
-    invocation.input_messages = (
-        get_input_messages(params.messages) if capture_content else []
-    )
-    invocation.system_instruction = (
-        get_system_instruction(params.system) if capture_content else []
-    )
-    invocation.tool_definitions = (
-        get_tool_definitions(params.tools) if capture_content else None
-    )
     invocation.attributes = attributes
     return invocation
 
@@ -284,13 +295,22 @@ def traced_method(
         args: tuple[Any, ...],
         kwargs: dict[str, Any],
     ) -> MessagesStreamManagerWrapper[Any]:
-        _materialize_tools(kwargs)
+        messages_capture = IteratorCapture.from_property(
+            kwargs, "messages", capture_content
+        )
+        system_capture = IteratorCapture.from_property(
+            kwargs, "system", capture_content
+        )
+        tools_capture = IteratorCapture.from_property(
+            kwargs, "tools", capture_content
+        )
         return MessagesStreamManagerWrapper(
             wrapped(*args, **kwargs),
-            lambda: _create_invocation(
-                handler, instance, args, kwargs, capture_content
-            ),
+            lambda: _create_invocation(handler, instance, args, kwargs),
             capture_content,
+            messages_capture=messages_capture,
+            system_capture=system_capture,
+            tools_capture=tools_capture,
         )
 
     return cast(
@@ -310,13 +330,22 @@ def traced_method(
         args: tuple[Any, ...],
         kwargs: dict[str, Any],
     ) -> AsyncMessagesStreamManagerWrapper[Any]:
-        _materialize_tools(kwargs)
+        messages_capture = IteratorCapture.from_property(
+            kwargs, "messages", capture_content
+        )
+        system_capture = IteratorCapture.from_property(
+            kwargs, "system", capture_content
+        )
+        tools_capture = IteratorCapture.from_property(
+            kwargs, "tools", capture_content
+        )
         return AsyncMessagesStreamManagerWrapper(
             wrapped(*args, **kwargs),
-            lambda: _create_invocation(
-                handler, instance, args, kwargs, capture_content
-            ),
+            lambda: _create_invocation(handler, instance, args, kwargs),
             capture_content,
+            messages_capture=messages_capture,
+            system_capture=system_capture,
+            tools_capture=tools_capture,
         )
 
     return cast(
diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py
index 80af9cc9e..c3c5de8be 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py
@@ -29,7 +29,12 @@
     finalize_on_close,
 )
 
-from .messages_extractors import set_invocation_response_attributes
+from .messages_extractors import (
+    get_input_messages,
+    get_system_instruction,
+    get_tool_definitions,
+    set_invocation_response_attributes,
+)
 
 _logger = logging.getLogger(__name__)
 
@@ -58,6 +63,7 @@
     from anthropic.types.parsed_message import ParsedMessage
 
     from opentelemetry.util.genai.invocation import InferenceInvocation
+    from opentelemetry.util.genai.iterator_capture import IteratorCapture
 
 
 ResponseFormatT = TypeVar("ResponseFormatT")
@@ -307,9 +313,32 @@ def __init__(
         manager: MessageStreamManager[ResponseFormatT],
         invocation_factory: Callable[[], InferenceInvocation],
         capture_content: bool,
+        *,
+        messages_capture: IteratorCapture,
+        system_capture: IteratorCapture,
+        tools_capture: IteratorCapture,
     ):
         super().__init__(manager, invocation_factory)
         self._self_capture_content = capture_content
+        self._self_messages_capture = messages_capture
+        self._self_system_capture = system_capture
+        self._self_tools_capture = tools_capture
+
+    def _enter_manager(
+        self, invocation: InferenceInvocation
+    ) -> MessageStream[ResponseFormatT]:
+        try:
+            return super()._enter_manager(invocation)
+        finally:
+            invocation.input_messages = (
+                self._self_messages_capture.extract(get_input_messages) or []
+            )
+            invocation.system_instruction = (
+                self._self_system_capture.extract(get_system_instruction) or []
+            )
+            invocation.tool_definitions = self._self_tools_capture.extract(
+                get_tool_definitions
+            )
 
     def _wrap_stream(
         self,
@@ -340,9 +369,32 @@ def __init__(
         manager: AsyncMessageStreamManager[ResponseFormatT],
         invocation_factory: Callable[[], InferenceInvocation],
         capture_content: bool,
+        *,
+        messages_capture: IteratorCapture,
+        system_capture: IteratorCapture,
+        tools_capture: IteratorCapture,
     ):
         super().__init__(manager, invocation_factory)
         self._self_capture_content = capture_content
+        self._self_messages_capture = messages_capture
+        self._self_system_capture = system_capture
+        self._self_tools_capture = tools_capture
+
+    async def _enter_manager(
+        self, invocation: InferenceInvocation
+    ) -> AsyncMessageStream[ResponseFormatT]:
+        try:
+            return await super()._enter_manager(invocation)
+        finally:
+            invocation.input_messages = (
+                self._self_messages_capture.extract(get_input_messages) or []
+            )
+            invocation.system_instruction = (
+                self._self_system_capture.extract(get_system_instruction) or []
+            )
+            invocation.tool_definitions = self._self_tools_capture.extract(
+                get_tool_definitions
+            )
 
     def _wrap_stream(
         self,
diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/requirements.oldest.txt b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/requirements.oldest.txt
index 504210574..688d04bfe 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/requirements.oldest.txt
+++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/requirements.oldest.txt
@@ -20,3 +20,5 @@
 # factor, so they are NOT pinned here — pyproject.toml is the single source of truth. The
 # OpenTelemetry SDK and test utilities come transitively from opentelemetry-test-util-genai, which
 # every oldest env installs. Pin here only test-only deps that nothing else already provides.
+
+-e util/opentelemetry-util-genai
diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_wrappers.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_wrappers.py
index 548041c0d..666538e3b 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_wrappers.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_wrappers.py
@@ -15,6 +15,7 @@
     gen_ai_attributes as GenAIAttributes,
 )
 from opentelemetry.util.genai.handler import TelemetryHandler
+from opentelemetry.util.genai.iterator_capture import IteratorCapture
 
 
 def _make_invocation():
@@ -360,6 +361,15 @@ def test_sync_manager_enter_constructs_stream_wrapper():
         manager=_FakeSyncManager(stream=stream),
         invocation_factory=_make_invocation,
         capture_content=False,
+        messages_capture=IteratorCapture.from_property(
+            {}, "messages", enabled=False
+        ),
+        system_capture=IteratorCapture.from_property(
+            {}, "system", enabled=False
+        ),
+        tools_capture=IteratorCapture.from_property(
+            {}, "tools", enabled=False
+        ),
     )
 
     with wrapper as result:
@@ -376,6 +386,15 @@ def test_sync_manager_does_not_create_invocation_until_enter():
             factory_calls.append(True) or _make_invocation()
         ),
         capture_content=False,
+        messages_capture=IteratorCapture.from_property(
+            {}, "messages", enabled=False
+        ),
+        system_capture=IteratorCapture.from_property(
+            {}, "system", enabled=False
+        ),
+        tools_capture=IteratorCapture.from_property(
+            {}, "tools", enabled=False
+        ),
     )
 
     assert not factory_calls
@@ -538,6 +557,15 @@ async def test_async_manager_enter_constructs_async_stream_wrapper():
         manager=_FakeAsyncManager(stream=stream),
         invocation_factory=_make_invocation,
         capture_content=False,
+        messages_capture=IteratorCapture.from_property(
+            {}, "messages", enabled=False
+        ),
+        system_capture=IteratorCapture.from_property(
+            {}, "system", enabled=False
+        ),
+        tools_capture=IteratorCapture.from_property(
+            {}, "tools", enabled=False
+        ),
     )
 
     async with wrapper as result:
diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_iterator_transparency.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_iterator_transparency.py
new file mode 100644
index 000000000..34158052d
--- /dev/null
+++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_iterator_transparency.py
@@ -0,0 +1,283 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+import json
+from collections.abc import Iterator
+from io import BytesIO
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+from anthropic import Anthropic, AsyncAnthropic
+
+try:
+    import httpx2 as httpx
+except ImportError:
+    import httpx
+
+from opentelemetry.instrumentation.genai.anthropic import patch, wrappers
+from opentelemetry.semconv._incubating.attributes import (
+    gen_ai_attributes as GenAI,
+)
+
+from .test_async_wrappers import (
+    _FakeAsyncManager,
+    _FakeAsyncStream,
+    _FakeSyncManager,
+    _FakeSyncStream,
+)
+from .test_sync_messages import (
+    _STREAM_SSE_BODY,
+    _build_message,
+    _load_span_messages,
+)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "wrapper_name",
+    [
+        "messages_create",
+        "async_messages_create",
+        "messages_stream",
+        "async_messages_stream",
+    ],
+)
+@pytest.mark.parametrize("sdk_failure", [False, True])
+async def test_extraction_failure_preserves_sdk_outcome(
+    monkeypatch: pytest.MonkeyPatch, wrapper_name: str, sdk_failure: bool
+) -> None:
+    extractor = MagicMock(side_effect=ValueError("extraction failed"))
+    monkeypatch.setattr(patch, "get_input_messages", extractor)
+    monkeypatch.setattr(wrappers, "get_input_messages", extractor)
+    handler = MagicMock()
+    handler.should_capture_content.return_value = True
+    error = RuntimeError("SDK failed")
+    asynchronous = wrapper_name.startswith("async")
+    streaming = wrapper_name.endswith("stream")
+    stream = _FakeAsyncStream() if asynchronous else _FakeSyncStream()
+    manager_type = _FakeAsyncManager if asynchronous else _FakeSyncManager
+    response = object()
+    message = {"role": "user", "content": "Hi"}
+    kwargs = {
+        "messages": iter([message]),
+        "system": [{"type": "text", "text": "Be brief"}],
+    }
+
+    def sdk_call(**kwargs: Any) -> Any:
+        assert list(kwargs["messages"]) == [message]
+        if streaming:
+            return manager_type(
+                stream, enter_error=error if sdk_failure else None
+            )
+        if sdk_failure:
+            raise error
+        return response
+
+    async def async_sdk_call(**kwargs: Any) -> Any:
+        return sdk_call(**kwargs)
+
+    wrapper = getattr(patch, wrapper_name)(handler)
+
+    async def invoke() -> Any:
+        if wrapper_name == "async_messages_create":
+            return await wrapper(async_sdk_call, MagicMock(), (), kwargs)
+        result = wrapper(sdk_call, MagicMock(), (), kwargs)
+        if streaming:
+            if asynchronous:
+                async with result as wrapped_stream:
+                    assert wrapped_stream.__wrapped__ is stream
+            else:
+                with result as wrapped_stream:
+                    assert wrapped_stream.__wrapped__ is stream
+        return result
+
+    invocation = handler.inference.return_value
+    if sdk_failure:
+        with pytest.raises(RuntimeError) as caught:
+            await invoke()
+        assert caught.value is error
+        invocation.fail.assert_called_once_with(error)
+    else:
+        result = await invoke()
+        if not streaming:
+            assert result is response
+        invocation.fail.assert_not_called()
+
+    extractor.assert_called_once_with([message])
+    assert invocation.input_messages == []
+    assert invocation.system_instruction[0].content == "Be brief"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("capture_content", [False, True])
+@pytest.mark.parametrize("iterator_failure", [False, True])
+@pytest.mark.parametrize(
+    "wrapper_name",
+    [
+        "messages_create",
+        "async_messages_create",
+        "messages_stream",
+        "async_messages_stream",
+    ],
+)
+async def test_wrappers_leave_iterators_for_sdk(
+    wrapper_name: str, capture_content: bool, iterator_failure: bool
+) -> None:
+    handler = MagicMock()
+    handler.should_capture_content.return_value = capture_content
+    consumed: list[str] = []
+    error = ValueError("SDK request failed")
+
+    def generate(name: str, value: Any) -> Iterator[Any]:
+        consumed.append(name)
+        if iterator_failure:
+            raise error
+        yield value
+
+    inputs = {
+        "messages": generate("messages", {"role": "user", "content": "Hi"}),
+        "system": generate("system", {"type": "text", "text": "Be brief"}),
+        "tools": generate("tools", {"name": "weather", "input_schema": {}}),
+    }
+    other_kwargs = {
+        "stop_sequences": iter(["END"]),
+        "extra_body": {"custom": iter(["value"])},
+    }
+
+    def sdk_call(*args: Any, **kwargs: Any) -> Any:
+        assert consumed == []
+        for key, value in other_kwargs.items():
+            assert kwargs[key] is value
+        for key, value in inputs.items():
+            if not capture_content:
+                assert kwargs[key] is value
+            assert isinstance(kwargs[key], type(value))
+            assert next(kwargs[key])
+        raise error
+
+    async def async_sdk_call(*args: Any, **kwargs: Any) -> Any:
+        return sdk_call(*args, **kwargs)
+
+    wrapper = getattr(patch, wrapper_name)(handler)
+    with pytest.raises(ValueError) as caught:
+        if wrapper_name == "async_messages_create":
+            await wrapper(
+                async_sdk_call, MagicMock(), (), {**inputs, **other_kwargs}
+            )
+        else:
+            wrapper(sdk_call, MagicMock(), (), {**inputs, **other_kwargs})
+    assert caught.value is error
+    assert consumed == (
+        ["messages"] if iterator_failure else ["messages", "system", "tools"]
+    )
+    if wrapper_name.endswith("create"):
+        handler.inference.return_value.fail.assert_called_once_with(error)
+        if capture_content and not iterator_failure:
+            invocation = handler.inference.return_value
+            assert invocation.input_messages[0].parts[0].content == "Hi"
+            assert invocation.system_instruction[0].content == "Be brief"
+            assert invocation.tool_definitions[0].name == "weather"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize("streaming", [False, True])
+@pytest.mark.parametrize("message_iterator", [False, True])
+@pytest.mark.parametrize("block_type", ["text", "image", "document"])
+async def test_sdk_consumes_and_captures_request_iterators(
+    span_exporter: Any,
+    instrument_with_content: Any,
+    asynchronous: bool,
+    streaming: bool,
+    message_iterator: bool,
+    block_type: str,
+) -> None:
+    seen: list[dict[str, Any]] = []
+    blocks: list[dict[str, Any]] = [{"type": "text", "text": "Hello"}]
+    expected_blocks = blocks
+    expected_parts = [{"type": "text", "content": "Hello"}]
+    if block_type != "text":
+        source = {
+            "type": "base64",
+            "media_type": (
+                "image/png" if block_type == "image" else "application/pdf"
+            ),
+        }
+        blocks = [
+            {
+                "type": block_type,
+                "source": {**source, "data": BytesIO(b"content")},
+            }
+        ]
+        expected_blocks = [
+            {
+                "type": block_type,
+                "source": {**source, "data": "Y29udGVudA=="},
+            }
+        ]
+        expected_parts = [{"type": "blob"}]
+    system = [{"type": "text", "text": "Be brief"}]
+    tools = [{"name": "weather", "input_schema": {"type": "object"}}]
+    messages = [{"role": "user", "content": iter(blocks)}]
+    kwargs = {
+        "model": "claude-sonnet-4-20250514",
+        "max_tokens": 32,
+        "messages": iter(messages) if message_iterator else messages,
+        "system": iter(system),
+        "tools": iter(tools),
+    }
+
+    def respond(request: Any) -> Any:
+        seen.append(json.loads(request.content))
+        if streaming:
+            return httpx.Response(
+                200,
+                headers={"content-type": "text/event-stream"},
+                content=_STREAM_SSE_BODY,
+            )
+        return httpx.Response(200, json=_build_message().model_dump())
+
+    if asynchronous:
+        async with AsyncAnthropic(
+            api_key="test",
+            http_client=httpx.AsyncClient(
+                transport=httpx.MockTransport(respond)
+            ),
+        ) as client:
+            if streaming:
+                async with client.messages.stream(**kwargs) as stream:
+                    await stream.until_done()
+            else:
+                await client.messages.create(**kwargs)
+    else:
+        with Anthropic(
+            api_key="test",
+            http_client=httpx.Client(transport=httpx.MockTransport(respond)),
+        ) as client:
+            if streaming:
+                with client.messages.stream(**kwargs) as stream:
+                    stream.until_done()
+            else:
+                client.messages.create(**kwargs)
+
+    assert len(seen) == 1
+    assert seen[0]["messages"] == [
+        {"role": "user", "content": expected_blocks}
+    ]
+    assert seen[0]["system"] == system
+    assert seen[0]["tools"] == tools
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 1
+    captured_messages = _load_span_messages(
+        spans[0], GenAI.GEN_AI_INPUT_MESSAGES
+    )
+    assert captured_messages[0]["parts"] == expected_parts
+    system_instructions = _load_span_messages(
+        spans[0], GenAI.GEN_AI_SYSTEM_INSTRUCTIONS
+    )
+    assert system_instructions == [{"type": "text", "content": "Be brief"}]
+    tool_definitions = _load_span_messages(
+        spans[0], GenAI.GEN_AI_TOOL_DEFINITIONS
+    )
+    assert tool_definitions[0]["name"] == "weather"
diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_messages_extractors.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_messages_extractors.py
index 3eb56a8ee..2d1270c03 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_messages_extractors.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_messages_extractors.py
@@ -39,6 +39,29 @@
 )
 
 
+@pytest.mark.parametrize(
+    "invalid",
+    [
+        {},
+        {"role": "user"},
+        {"content": "Hello"},
+    ],
+)
+def test_input_messages_skip_missing_fields(invalid: dict) -> None:
+    assert get_input_messages([invalid]) == []
+    messages = get_input_messages(
+        [
+            {"role": "user", "content": "First"},
+            invalid,
+            {"role": "user", "content": "Last"},
+        ]
+    )
+    assert [message.parts for message in messages] == [
+        [TextPart(content="First")],
+        [TextPart(content="Last")],
+    ]
+
+
 def test_extract_params_reads_sampling_params_from_extra_body():
     params = extract_params(
         extra_body={"temperature": 0.7, "top_p": 0.9, "top_k": 40}
diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py
index 9ec93f3e9..67f463305 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py
@@ -533,6 +533,7 @@ def test_sync_messages_create_captures_multimodal_content(
 
 def test_sync_messages_create_preserves_generator_document_content(
     instrument_with_content,
+    span_exporter,
 ):
     received_content = None
 
@@ -586,6 +587,15 @@ def document_content():
         {"type": "text", "text": "First"},
         {"type": "text", "text": "Second"},
     ]
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 1
+    messages = json.loads(
+        spans[0].attributes[GenAIAttributes.GEN_AI_INPUT_MESSAGES]
+    )
+    assert messages[0]["parts"] == [
+        {"type": "text", "content": "First"},
+        {"type": "text", "content": "Second"},
+    ]
 
 
 @pytest.mark.vcr()
@@ -1390,14 +1400,9 @@ def respond(request):
     _assert_weather_tool_definitions(spans[0])
 
 
-def test_sync_messages_stream_tools_generator_is_recorded(
+def test_sync_messages_stream_tools_generator_reaches_the_sdk(
     span_exporter, instrument_with_content
 ):
-    """``stream`` serializes the request before the invocation exists.
-
-    The generator has to be frozen in the wrapper, not while the invocation is
-    built, or the SDK drains it first and the span records no tools.
-    """
     seen = {}
 
     def respond(request):
diff --git a/util/opentelemetry-util-genai/.changelog/873.added b/util/opentelemetry-util-genai/.changelog/873.added
new file mode 100644
index 000000000..71aa926f4
--- /dev/null
+++ b/util/opentelemetry-util-genai/.changelog/873.added
@@ -0,0 +1 @@
+Add IteratorCapture to observe nested request iterators lazily and expose snapshots for telemetry extraction.
diff --git a/util/opentelemetry-util-genai/README.rst b/util/opentelemetry-util-genai/README.rst
index 6e61193f1..7d8be4e0c 100644
--- a/util/opentelemetry-util-genai/README.rst
+++ b/util/opentelemetry-util-genai/README.rst
@@ -204,6 +204,47 @@ For embedding invocations, the following attributes are set:
 - ``gen_ai.usage.input_tokens``: Int(24)
 
 
+Capturing request iterators
+---------------------------
+
+Use ``IteratorCapture.from_property()`` for a named request property. It replaces
+that property's value in ``kwargs`` and records values as the SDK reads them.
+Disabled captures and missing properties leave ``kwargs`` untouched and return
+``None`` from ``snapshot()``.
+
+.. code-block:: python
+
+    from opentelemetry.util.genai.iterator_capture import IteratorCapture
+
+    messages_capture = IteratorCapture.from_property(
+        kwargs, "messages", enabled=capture_content
+    )
+    try:
+        response = sdk_call(**kwargs)
+    finally:
+        messages = messages_capture.extract(get_input_messages)
+
+For a bare value, pass ``capture.wrapped`` to the SDK:
+
+.. code-block:: python
+
+    capture = IteratorCapture(messages, enabled=capture_content)
+    try:
+        response = sdk_call(messages=capture.wrapped)
+    finally:
+        messages = capture.extract(get_input_messages)
+
+``extract(extractor)`` calls the instrumentation's extractor with a snapshot,
+using ``()`` for unread iterators and ``None`` for disabled or missing properties.
+The extractor must accept that shape and return the telemetry content.
+``extract()`` logs snapshot or conversion errors and returns ``None``. Use
+``snapshot()`` directly when no conversion is needed.
+
+Nested built-in dictionaries, lists, and tuples are supported. File objects
+pass through unchanged. Snapshots contain only the consumed portion of each
+iterator; async iterators and custom containers are not captured.
+
+
 Installation
 ------------
 
diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/iterator_capture.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/iterator_capture.py
new file mode 100644
index 000000000..ea187a040
--- /dev/null
+++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/iterator_capture.py
@@ -0,0 +1,260 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Record request iterator values as the SDK consumes them."""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable, Iterator, MutableMapping, Sequence
+from io import IOBase
+from typing import TYPE_CHECKING, TypeVar, cast
+
+if TYPE_CHECKING:
+
+    class _ObjectProxy:
+        __wrapped__: object
+
+        def __init__(self, wrapped: object) -> None: ...
+
+else:
+    from wrapt import ObjectProxy as _ObjectProxy
+
+_logger = logging.getLogger(__name__)
+_InputT = TypeVar("_InputT")
+_ResultT = TypeVar("_ResultT")
+UNCONSUMED: object = object()
+"""Snapshot placeholder for an iterator the SDK has not read."""
+
+
+class _Node:
+    """A captured request value, with depth capped to avoid recursion errors."""
+
+    _MAX_DEPTH = 64
+
+    @staticmethod
+    def capture(value: object, ancestors: set[int], depth: int = 0) -> _Node:
+        if depth >= _Node._MAX_DEPTH:
+            return _Node(value, unavailable=True)
+        if isinstance(value, IOBase):
+            return _Node(value)
+        if isinstance(value, Iterator):
+            return _IteratorNode(cast("Iterator[object]", value), depth)
+        if type(value) not in (dict, list, tuple):
+            return _Node(value)
+        value_id = id(value)
+        if value_id in ancestors:
+            return _Node(value, unavailable=True)
+        ancestors.add(value_id)
+        try:
+            if type(value) is dict:
+                mapping = cast("dict[object, object]", value)
+                children = {
+                    key: _Node.capture(item, ancestors, depth + 1)
+                    for key, item in mapping.items()
+                }
+                wrapped = {
+                    key: child.wrapped for key, child in children.items()
+                }
+                return _Node(
+                    mapping
+                    if all(
+                        wrapped[key] is item for key, item in mapping.items()
+                    )
+                    else wrapped,
+                    children,
+                )
+            sequence = cast("Sequence[object]", value)
+            items = [
+                _Node.capture(item, ancestors, depth + 1) for item in sequence
+            ]
+            wrapped_items = [item.wrapped for item in items]
+            wrapped_sequence: object = value
+            if any(
+                new is not old for new, old in zip(wrapped_items, sequence)
+            ):
+                wrapped_sequence = (
+                    tuple(wrapped_items)
+                    if type(value) is tuple
+                    else wrapped_items
+                )
+            return _Node(wrapped_sequence, items)
+        finally:
+            ancestors.remove(value_id)
+
+    def __init__(
+        self,
+        wrapped: object,
+        children: dict[object, _Node] | list[_Node] | None = None,
+        *,
+        unavailable: bool = False,
+    ) -> None:
+        self.wrapped: object = wrapped
+        self.children: dict[object, _Node] | list[_Node] | None = children
+        self.unavailable: bool = unavailable
+
+    @property
+    def complete(self) -> bool:
+        if self.unavailable:
+            return False
+        children = self.children
+        if isinstance(children, dict):
+            return all(child.complete for child in children.values())
+        if isinstance(children, list):
+            return all(child.complete for child in children)
+        return True
+
+    def snapshot(self, unconsumed: object) -> object:
+        if self.unavailable:
+            return unconsumed
+        children = self.children
+        if isinstance(children, dict):
+            return {
+                key: child.snapshot(unconsumed)
+                for key, child in children.items()
+            }
+        if isinstance(children, list):
+            items = [child.snapshot(unconsumed) for child in children]
+            return tuple(items) if isinstance(self.wrapped, tuple) else items
+        return self.wrapped
+
+
+class _IteratorNode(_Node):
+    def __init__(self, iterator: Iterator[object], depth: int) -> None:
+        super().__init__(iterator)
+        self.depth: int = depth
+        self.items: list[_Node] = []
+        self.started: bool = False
+        self.exhausted: bool = False
+        self.wrapped = _RecordingIterator(iterator, self)
+
+    @property
+    def complete(self) -> bool:
+        return self.exhausted and all(item.complete for item in self.items)
+
+    def snapshot(self, unconsumed: object) -> object:
+        if not self.started:
+            return unconsumed
+        return [item.snapshot(unconsumed) for item in self.items]
+
+
+class _RecordingIterator(_ObjectProxy):
+    _self_node: _IteratorNode
+
+    def __init__(
+        self, iterator: Iterator[object], node: _IteratorNode
+    ) -> None:
+        super().__init__(iterator)
+        self._self_node = node
+
+    def __iter__(self) -> _RecordingIterator:
+        return self
+
+    def __next__(self) -> object:
+        iterator = cast("Iterator[object]", self.__wrapped__)
+        return self._self_advance(lambda: next(iterator))
+
+    def send(self, value: object) -> object:
+        send = cast(
+            "Callable[[object], object]", getattr(self.__wrapped__, "send")
+        )
+        return self._self_advance(lambda: send(value))
+
+    def throw(self, *args: object) -> object:
+        throw = cast(
+            "Callable[..., object]", getattr(self.__wrapped__, "throw")
+        )
+        return self._self_advance(lambda: throw(*args))
+
+    def _self_advance(self, advance: Callable[[], object]) -> object:
+        node = self._self_node
+        node.started = True
+        try:
+            item = advance()
+        except StopIteration:
+            node.exhausted = True
+            raise
+        # Snapshot before the next advance: generators can mutate and yield
+        # the same dictionary repeatedly.
+        captured = _Node.capture(item, set(), node.depth + 1)
+        node.items.append(captured)
+        return captured.wrapped
+
+
+class IteratorCapture:
+    """Capture input iterators as the SDK consumes them.
+
+    Wrap synchronous iterators in the input, including those
+    nested in built-in dicts, lists, and tuples. Copy containers that need
+    wrappers; leave the originals, files, and other objects untouched.
+    Store consumed values in memory, copying each yielded container.
+
+    Read ``snapshot()`` or ``extract()`` after the SDK call, including on
+    failure. Iterator exceptions propagate unchanged. Disabled captures pass
+    the input through unchanged and return ``None`` snapshots.
+    """
+
+    def __init__(
+        self,
+        value: object,
+        enabled: bool = True,
+    ) -> None:
+        self.property_name: str | None = None
+        self._node: _Node | None = (
+            _Node.capture(value, set()) if enabled else None
+        )
+        self.wrapped: object = self._node.wrapped if self._node else value
+
+    @classmethod
+    def from_property(
+        cls,
+        properties: MutableMapping[str, object],
+        property_name: str,
+        enabled: bool = True,
+    ) -> IteratorCapture:
+        """Replace one property's value; leave missing properties absent."""
+        present = property_name in properties
+        capture = cls(properties.get(property_name), enabled and present)
+        capture.property_name = property_name
+        if enabled and present:
+            properties[property_name] = capture.wrapped
+        return capture
+
+    @property
+    def complete(self) -> bool:
+        """True if capture succeeded and all captured iterators are exhausted."""
+        return self._node is not None and self._node.complete
+
+    def snapshot(self, *, unconsumed: object = UNCONSUMED) -> object:
+        """Return a snapshot without advancing iterators.
+
+        Each iterator becomes a list of items consumed so far. Unread
+        iterators, cycles, and values beyond the depth limit become
+        ``unconsumed``.
+        Empty iterators become ``[]`` after the SDK reads them.
+        Disabled or missing properties return ``None``.
+        """
+        return self._node.snapshot(unconsumed) if self._node else None
+
+    def extract(
+        self, extractor: Callable[[_InputT | None], _ResultT]
+    ) -> _ResultT | None:
+        """Pass a snapshot to ``extractor`` and return its result.
+
+        The extractor must accept lists for consumed iterators, ``()`` for
+        unread or unavailable values, and ``None`` for disabled or missing
+        properties. Log snapshot or conversion errors and return ``None``.
+        """
+        try:
+            snapshot = cast("_InputT | None", self.snapshot(unconsumed=()))
+            return extractor(snapshot)
+        except Exception:
+            # Extractors should handle user input defensively, but cannot guard
+            # against every unexpected value or custom object. Extraction often
+            # runs in finally blocks and its errors must not hide SDK failures.
+            _logger.debug(
+                "Failed to extract %s",
+                self.property_name or "captured input",
+                exc_info=True,
+            )
+            return None
diff --git a/util/opentelemetry-util-genai/tests/test_iterator_capture.py b/util/opentelemetry-util-genai/tests/test_iterator_capture.py
new file mode 100644
index 000000000..02ccd3751
--- /dev/null
+++ b/util/opentelemetry-util-genai/tests/test_iterator_capture.py
@@ -0,0 +1,413 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+import asyncio
+import logging
+from collections.abc import Generator, Iterable, Iterator, Sequence
+from io import BytesIO, StringIO
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from opentelemetry.util.genai.iterator_capture import (
+    UNCONSUMED,
+    IteratorCapture,
+)
+
+
+@pytest.mark.parametrize("enabled", [False, True])
+def test_bare_iterator_capture(enabled: bool) -> None:
+    source = iter(["first", "second"])
+    capture = IteratorCapture(source, enabled=enabled)
+    assert capture.property_name is None
+    if not enabled:
+        assert capture.wrapped is source
+    assert capture.snapshot() is (UNCONSUMED if enabled else None)
+    assert next(capture.wrapped) == "first"
+    assert capture.snapshot() == (["first"] if enabled else None)
+    assert list(capture.wrapped) == ["second"]
+    assert capture.complete is enabled
+
+
+def test_bare_nested_value_preserves_original_container() -> None:
+    source = {"content": iter(["Hello"])}
+    capture = IteratorCapture(source)
+    assert list(capture.wrapped["content"]) == ["Hello"]
+    assert capture.snapshot() == {"content": ["Hello"]}
+    assert source["content"] is not capture.wrapped["content"]
+
+
+@pytest.mark.parametrize(
+    "state", ["disabled", "missing", "none", "unread", "empty"]
+)
+def test_extract_converts_snapshot_defaults(state: str) -> None:
+    source = iter(["Hello"] if state != "empty" else [])
+    kwargs: dict[str, Any] = {"messages": source}
+    if state == "missing":
+        kwargs.clear()
+    elif state == "none":
+        kwargs["messages"] = None
+    capture = IteratorCapture.from_property(
+        kwargs, "messages", enabled=state != "disabled"
+    )
+    if state == "empty":
+        assert list(kwargs["messages"]) == []
+
+    def extract(values: Iterable[str] | None) -> str:
+        return "absent" if values is None else ",".join(values)
+
+    expected = "absent" if state in ("disabled", "missing", "none") else ""
+    assert capture.extract(extract) == expected
+    if state != "empty":
+        assert next(source) == "Hello"
+
+
+def test_extract_converts_partial_nested_input_after_failure() -> None:
+    error = ValueError("request failed")
+    unread = iter(["unread"])
+
+    def generate() -> Iterator[dict[str, Iterator[str]]]:
+        yield {"content": iter(["first", "second"]), "other": unread}
+        raise error
+
+    kwargs = {"messages": generate()}
+    capture = IteratorCapture.from_property(kwargs, "messages")
+    message = next(kwargs["messages"])
+    assert next(message["content"]) == "first"
+    with pytest.raises(ValueError) as caught:
+        next(kwargs["messages"])
+    assert caught.value is error
+
+    def extract(
+        messages: Iterable[dict[str, Sequence[str]]] | None,
+    ) -> list[str]:
+        return [
+            text
+            for message in messages or ()
+            for content in message.values()
+            for text in content
+        ]
+
+    assert capture.extract(extract) == ["first"]
+    assert not capture.complete
+    assert next(message["content"]) == "second"
+    assert next(unread) == "unread"
+
+
+@pytest.mark.parametrize("failure", ["snapshot", "extractor"])
+def test_extract_errors_return_none(
+    failure: str,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    capture = IteratorCapture.from_property({"messages": [1]}, "messages")
+
+    def extract(value: object) -> str:
+        raise ValueError("invalid content")
+
+    if failure == "snapshot":
+
+        def fail_snapshot(*, unconsumed: object) -> object:
+            raise ValueError("invalid snapshot")
+
+        monkeypatch.setattr(capture, "snapshot", fail_snapshot)
+
+    with caplog.at_level(logging.DEBUG):
+        assert capture.extract(extract) is None
+    assert "Failed to extract messages" in caplog.text
+
+
+def test_extract_does_not_swallow_cancellation() -> None:
+    capture = IteratorCapture.from_property({}, "messages")
+    error = asyncio.CancelledError()
+
+    def extract(value: object) -> str:
+        raise error
+
+    with pytest.raises(asyncio.CancelledError) as caught:
+        capture.extract(extract)
+    assert caught.value is error
+
+
+def test_disabled_capture_leaves_request_untouched(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def fail_capture(value: object, ancestors: set[int]) -> None:
+        pytest.fail("disabled capture must not inspect request values")
+
+    monkeypatch.setattr(
+        "opentelemetry.util.genai.iterator_capture._Node.capture", fail_capture
+    )
+    messages = {"content": iter(["Hello"])}
+    kwargs = {"messages": messages}
+    capture = IteratorCapture.from_property(kwargs, "messages", enabled=False)
+    assert kwargs["messages"] is messages
+    assert capture.snapshot() is None
+    assert not capture.complete
+
+
+def test_missing_property_is_not_added() -> None:
+    kwargs: dict[str, object] = {}
+    capture = IteratorCapture.from_property(kwargs, "messages")
+    assert kwargs == {}
+    assert capture.snapshot() is None
+    assert not capture.complete
+
+
+def test_captures_are_independent_and_leave_other_properties_untouched() -> (
+    None
+):
+    messages = iter(["Hello", "Goodbye"])
+    tools = iter(["weather"])
+    unrelated = iter(["stop"])
+    kwargs = {"messages": messages, "tools": tools, "stop": unrelated}
+    messages_capture = IteratorCapture.from_property(kwargs, "messages")
+    tools_capture = IteratorCapture.from_property(kwargs, "tools")
+
+    assert messages_capture.property_name == "messages"
+    assert tools_capture.property_name == "tools"
+    assert kwargs["stop"] is unrelated
+    assert next(kwargs["messages"]) == "Hello"
+    assert messages_capture.snapshot() == ["Hello"]
+    assert tools_capture.snapshot() is UNCONSUMED
+    assert list(kwargs["tools"]) == ["weather"]
+    assert tools_capture.snapshot() == ["weather"]
+    assert tools_capture.complete
+    assert not messages_capture.complete
+
+
+@pytest.mark.parametrize("on_disk", [False, True])
+@pytest.mark.parametrize("binary", [False, True])
+def test_file_objects_are_preserved(
+    tmp_path: Path, on_disk: bool, binary: bool
+) -> None:
+    path = tmp_path / "content.txt"
+    path.write_bytes(b"content")
+    if on_disk:
+        data = path.open("rb" if binary else "r")
+    else:
+        data = BytesIO(b"content") if binary else StringIO("content")
+    with data:
+        kwargs = {"value": {"data": data}}
+        capture = IteratorCapture.from_property(kwargs, "value")
+        assert kwargs["value"]["data"] is data
+        assert data.tell() == 0
+        assert capture.snapshot(unconsumed=()) == {"data": data}
+        assert kwargs["value"]["data"].read() == (
+            b"content" if binary else "content"
+        )
+        assert capture.snapshot(unconsumed=()) == {"data": data}
+        assert capture.complete
+
+
+def test_capture_is_lazy_and_distinguishes_empty_from_unread() -> None:
+    reads: list[str] = []
+
+    def generate() -> Iterator[str]:
+        reads.append("read")
+        yield from ()
+
+    kwargs = {"value": generate()}
+    capture = IteratorCapture.from_property(kwargs, "value")
+    assert reads == []
+    assert capture.snapshot() is UNCONSUMED
+    assert capture.snapshot(unconsumed=()) == ()
+    assert not capture.complete
+    assert list(kwargs["value"]) == []
+    assert reads == ["read"]
+    assert capture.snapshot() == []
+    assert capture.complete
+
+
+def test_nested_capture_preserves_containers_and_partial_input() -> None:
+    inner = iter(["first", "second"])
+    source = {"arbitrary": [{"payload": (inner,)}]}
+    kwargs = {"value": source}
+    capture = IteratorCapture.from_property(kwargs, "value")
+    wrapped = kwargs["value"]["arbitrary"][0]["payload"][0]
+    assert isinstance(wrapped, type(inner))
+    assert wrapped.__class__ is inner.__class__
+    assert source["arbitrary"][0]["payload"][0] is inner
+    assert capture.snapshot() == {"arbitrary": [{"payload": (UNCONSUMED,)}]}
+    assert next(wrapped) == "first"
+    assert capture.snapshot() == {"arbitrary": [{"payload": (["first"],)}]}
+    assert not capture.complete
+    assert list(wrapped) == ["second"]
+    assert capture.complete
+    assert capture.snapshot() == {
+        "arbitrary": [{"payload": (["first", "second"],)}]
+    }
+
+
+def test_yielded_mutable_objects_are_snapshotted_per_item() -> None:
+    shared: dict[str, object] = {"value": [0]}
+
+    def generate() -> Iterator[dict[str, object]]:
+        yield shared
+        shared["value"] = [1]
+        yield shared
+        shared.clear()
+
+    kwargs = {"value": generate()}
+    capture = IteratorCapture.from_property(kwargs, "value")
+    iterator = kwargs["value"]
+    assert next(iterator) is shared
+    assert next(iterator) is shared
+    assert list(iterator) == []
+    assert capture.snapshot() == [{"value": [0]}, {"value": [1]}]
+
+
+def test_nested_iterators_yielded_by_iterators_stay_lazy() -> None:
+    reads: list[str] = []
+
+    def inner() -> Iterator[str]:
+        reads.append("inner")
+        yield "value"
+
+    kwargs = {"value": iter([{"data": inner()}])}
+    capture = IteratorCapture.from_property(kwargs, "value")
+    item = next(kwargs["value"])
+    assert reads == []
+    assert capture.snapshot() == [{"data": UNCONSUMED}]
+    assert list(item["data"]) == ["value"]
+    assert capture.snapshot() == [{"data": ["value"]}]
+    assert not capture.complete
+    assert list(kwargs["value"]) == []
+    assert capture.complete
+
+
+@pytest.mark.parametrize(
+    "error", [ValueError("iterator failed"), asyncio.CancelledError()]
+)
+def test_iterator_errors_propagate_unchanged(error: BaseException) -> None:
+    def generate() -> Iterator[str]:
+        yield "first"
+        raise error
+
+    kwargs = {"value": generate()}
+    capture = IteratorCapture.from_property(kwargs, "value")
+    assert next(kwargs["value"]) == "first"
+    with pytest.raises(type(error)) as caught:
+        next(kwargs["value"])
+    assert caught.value is error
+    assert capture.snapshot() == ["first"]
+    assert not capture.complete
+
+
+def test_generator_protocol_and_return_value() -> None:
+    closed: list[bool] = []
+
+    def generate() -> Generator[object, object, str]:
+        try:
+            received = yield "first"
+            try:
+                yield received
+            except ValueError:
+                yield "handled"
+            return "finished"
+        finally:
+            closed.append(True)
+
+    kwargs = {"value": generate()}
+    capture = IteratorCapture.from_property(kwargs, "value")
+    assert next(kwargs["value"]) == "first"
+    assert kwargs["value"].send("sent") == "sent"
+    assert kwargs["value"].throw(ValueError()) == "handled"
+    with pytest.raises(StopIteration) as caught:
+        next(kwargs["value"])
+    assert caught.value.value == "finished"
+    assert capture.snapshot() == ["first", "sent", "handled"]
+    assert capture.complete
+    assert closed == [True]
+
+
+def test_close_does_not_drain_input() -> None:
+    def generate() -> Iterator[int]:
+        yield 1
+        pytest.fail("close must not advance the iterator")
+
+    kwargs = {"value": generate()}
+    capture = IteratorCapture.from_property(kwargs, "value")
+    assert next(kwargs["value"]) == 1
+    kwargs["value"].close()
+    assert capture.snapshot() == [1]
+    assert not capture.complete
+
+
+def test_repeated_iterator_reference_does_not_replay_items() -> None:
+    iterator = iter([1, 2])
+    kwargs = {"value": [iterator, iterator]}
+    capture = IteratorCapture.from_property(kwargs, "value")
+    assert list(kwargs["value"][0]) == [1, 2]
+    assert list(kwargs["value"][1]) == []
+    assert capture.snapshot() == [[1, 2], []]
+
+
+def test_plain_containers_are_not_replaced() -> None:
+    source = {"values": [1, {"value": "text"}]}
+    kwargs = {"value": source}
+    capture = IteratorCapture.from_property(kwargs, "value")
+    assert kwargs["value"] is source
+    snapshot = capture.snapshot()
+    source["values"].clear()
+    assert snapshot == {"values": [1, {"value": "text"}]}
+    assert capture.snapshot() == snapshot
+
+
+def test_opaque_objects_are_not_inspected() -> None:
+    class Opaque:
+        def model_dump(self) -> None:
+            pytest.fail("capture must not invoke model methods")
+
+        def __iter__(self) -> Iterator[object]:
+            pytest.fail("capture must not iterate unknown containers")
+
+    value = Opaque()
+    kwargs = {"value": value}
+    capture = IteratorCapture.from_property(kwargs, "value")
+    assert kwargs["value"] is value
+    assert capture.snapshot() is value
+
+
+@pytest.mark.parametrize("kind", ["dict", "list", "iterator"])
+def test_deep_input_is_preserved_and_snapshot_is_bounded(kind: str) -> None:
+    value: Any = "leaf"
+    for _ in range(2000):
+        if kind == "dict":
+            value = {"child": value}
+        elif kind == "iterator":
+            value = iter([value])
+        else:
+            value = [value]
+
+    kwargs = {"value": value}
+    capture = IteratorCapture.from_property(kwargs, "value")
+    current = kwargs["value"]
+    for _ in range(2000):
+        if kind == "iterator":
+            child = next(current)
+            assert next(current, UNCONSUMED) is UNCONSUMED
+            current = child
+        else:
+            current = current["child"] if kind == "dict" else current[0]
+    assert current == "leaf"
+
+    snapshot = capture.snapshot()
+    depth = 0
+    while isinstance(snapshot, (dict, list)):
+        snapshot = snapshot["child"] if kind == "dict" else snapshot[0]
+        depth += 1
+    assert snapshot is UNCONSUMED
+    assert 0 < depth < 2000
+    assert not capture.complete
+
+
+def test_cycles_do_not_break_the_request() -> None:
+    value: list[Any] = []
+    value.append(value)
+    kwargs = {"value": value}
+    capture = IteratorCapture.from_property(kwargs, "value")
+    assert kwargs["value"] is value
+    assert capture.snapshot() == [UNCONSUMED]
+    assert not capture.complete