diff --git a/libs/core/tests/unit_tests/language_models/chat_models/test_base.py b/libs/core/tests/unit_tests/language_models/chat_models/test_base.py
index 3f15efe4b27..d874bc2b9cd 100644
--- a/libs/core/tests/unit_tests/language_models/chat_models/test_base.py
+++ b/libs/core/tests/unit_tests/language_models/chat_models/test_base.py
@@ -1561,16 +1561,17 @@ def test_invocation_params_passed_to_tracer_metadata() -> None:
     assert run.metadata == run.extra["metadata"]
 
 
-def test_stream_v2_invocation_params_passed_to_tracer_metadata() -> None:
-    """`stream_v2()` must preserve filtered invocation params for tracing."""
+def test_stream_events_v3_invocation_params_passed_to_tracer_metadata() -> None:
+    """`stream_events(version="v3")` must preserve filtered invocation params for tracing."""
     llm = FakeStreamingChatModelWithInvocationParams()
     collector = LangChainTracerRunCollector()
 
     with collector.tracing_callback() as tracer:
-        _ = llm.stream_v2(
+        _ = llm.stream_events(
             [HumanMessage(content="Hello")],
             config={"callbacks": [tracer]},
             stop=["done"],
+            version="v3",
         ).output
 
     assert len(collector.runs) == 1
@@ -1581,16 +1582,17 @@ def test_stream_v2_invocation_params_passed_to_tracer_metadata() -> None:
     assert metadata["temperature"] == 0.7
 
 
-async def test_astream_v2_invocation_params_passed_to_tracer_metadata() -> None:
-    """`astream_v2()` must preserve filtered invocation params for tracing."""
+async def test_astream_events_v3_invocation_params_passed_to_tracer_metadata() -> None:
+    """`astream_events(version="v3")` must preserve filtered invocation params for tracing."""
     llm = FakeStreamingChatModelWithInvocationParams()
     collector = LangChainTracerRunCollector()
 
     with collector.tracing_callback() as tracer:
-        stream = await llm.astream_v2(
+        stream = await llm.astream_events(
             [HumanMessage(content="Hello")],
             config={"callbacks": [tracer]},
             stop=["done"],
+            version="v3",
         )
         _ = await stream
 
diff --git a/libs/core/tests/unit_tests/language_models/test_chat_model_streamer.py b/libs/core/tests/unit_tests/language_models/test_chat_model_streamer.py
index 4390825beef..643e3c67bfc 100644
--- a/libs/core/tests/unit_tests/language_models/test_chat_model_streamer.py
+++ b/libs/core/tests/unit_tests/language_models/test_chat_model_streamer.py
@@ -1,4 +1,4 @@
-"""Tests for BaseChatModel.stream_v2() / astream_v2()."""
+"""Tests for `BaseChatModel.stream_events(version="v3")` / `astream_events(version="v3")`."""
 
 from __future__ import annotations
 
@@ -32,11 +32,11 @@ if TYPE_CHECKING:
 
 
 class TestStreamV2Sync:
-    """Test BaseChatModel.stream_v2() with FakeListChatModel."""
+    """Test `BaseChatModel.stream_events(version="v3")` with `FakeListChatModel`."""
 
     def test_stream_text(self) -> None:
         model = FakeListChatModel(responses=["Hello world!"])
-        stream = model.stream_v2("test")
+        stream = model.stream_events("test", version="v3")
 
         assert isinstance(stream, ChatModelStream)
         deltas = list(stream.text)
@@ -45,7 +45,7 @@ class TestStreamV2Sync:
 
     def test_stream_output(self) -> None:
         model = FakeListChatModel(responses=["Hello!"])
-        stream = model.stream_v2("test")
+        stream = model.stream_events("test", version="v3")
 
         msg = stream.output
         assert isinstance(msg.content, list)
@@ -54,7 +54,7 @@ class TestStreamV2Sync:
 
     def test_stream_usage_none_for_fake(self) -> None:
         model = FakeListChatModel(responses=["Hi"])
-        stream = model.stream_v2("test")
+        stream = model.stream_events("test", version="v3")
         # Drain
         for _ in stream.text:
             pass
@@ -62,7 +62,7 @@ class TestStreamV2Sync:
 
     def test_stream_raw_events(self) -> None:
         model = FakeListChatModel(responses=["ab"])
-        stream = model.stream_v2("test")
+        stream = model.stream_events("test", version="v3")
 
         events = list(stream)
         event_types = [e.get("event") for e in events]
@@ -72,12 +72,12 @@ class TestStreamV2Sync:
 
 
 class TestAstreamV2:
-    """Test BaseChatModel.astream_v2() with FakeListChatModel."""
+    """Test `BaseChatModel.astream_events(version="v3")` with `FakeListChatModel`."""
 
     @pytest.mark.asyncio
     async def test_astream_text_await(self) -> None:
         model = FakeListChatModel(responses=["Hello!"])
-        stream = await model.astream_v2("test")
+        stream = await model.astream_events("test", version="v3")
 
         assert isinstance(stream, AsyncChatModelStream)
         full = await stream.text
@@ -86,7 +86,7 @@ class TestAstreamV2:
     @pytest.mark.asyncio
     async def test_astream_text_deltas(self) -> None:
         model = FakeListChatModel(responses=["Hi"])
-        stream = await model.astream_v2("test")
+        stream = await model.astream_events("test", version="v3")
 
         deltas = [d async for d in stream.text]
         assert "".join(deltas) == "Hi"
@@ -94,7 +94,7 @@ class TestAstreamV2:
     @pytest.mark.asyncio
     async def test_astream_await_output(self) -> None:
         model = FakeListChatModel(responses=["Hey"])
-        stream = await model.astream_v2("test")
+        stream = await model.astream_events("test", version="v3")
 
         msg = await stream
         assert msg.content == [{"type": "text", "text": "Hey", "index": 0}]
@@ -193,13 +193,13 @@ class _EmptyStreamModel(BaseChatModel):
 
 
 class TestCallbacks:
-    """Verify stream_v2 fires on_llm_end / on_llm_error callbacks."""
+    """Verify `stream_events(version="v3")` fires `on_llm_end` / `on_llm_error` callbacks."""
 
-    def test_stream_v2_defers_on_chat_model_start_until_consumed(self) -> None:
+    def test_stream_events_v3_defers_on_chat_model_start_until_consumed(self) -> None:
         handler = _RecordingHandler()
         model = FakeListChatModel(responses=["done"], callbacks=[handler])
 
-        stream = model.stream_v2("test")
+        stream = model.stream_events("test", version="v3")
 
         assert handler.events == []
 
@@ -210,7 +210,7 @@ class TestCallbacks:
     def test_on_llm_end_fires_after_drain(self) -> None:
         handler = _RecordingHandler()
         model = FakeListChatModel(responses=["done"], callbacks=[handler])
-        stream = model.stream_v2("test")
+        stream = model.stream_events("test", version="v3")
         for _ in stream.text:
             pass
         _ = stream.output
@@ -225,18 +225,18 @@ class TestCallbacks:
     async def test_on_llm_end_fires_async(self) -> None:
         handler = _AsyncRecordingHandler()
         model = FakeListChatModel(responses=["done"], callbacks=[handler])
-        stream = await model.astream_v2("test")
+        stream = await model.astream_events("test", version="v3")
         _ = await stream
 
         assert "on_chat_model_start" in handler.events
         assert "on_llm_end" in handler.events
 
     @pytest.mark.asyncio
-    async def test_astream_v2_defers_on_chat_model_start_until_consumed(self) -> None:
+    async def test_astream_events_v3_defers_on_chat_model_start_until_consumed(self) -> None:
         handler = _AsyncRecordingHandler()
         model = FakeListChatModel(responses=["done"], callbacks=[handler])
 
-        stream = await model.astream_v2("test")
+        stream = await model.astream_events("test", version="v3")
 
         assert handler.events == []
 
@@ -251,7 +251,7 @@ class TestCallbacks:
         """
         handler = _RecordingHandler()
         model = FakeListChatModel(responses=["hello"], callbacks=[handler])
-        stream = model.stream_v2("test")
+        stream = model.stream_events("test", version="v3")
         _ = stream.output
 
         response = handler.last_llm_end_response
@@ -265,7 +265,7 @@ class TestCallbacks:
     async def test_on_llm_end_receives_assembled_message_async(self) -> None:
         handler = _AsyncRecordingHandler()
         model = FakeListChatModel(responses=["hello"], callbacks=[handler])
-        stream = await model.astream_v2("test")
+        stream = await model.astream_events("test", version="v3")
         _ = await stream
 
         response = handler.last_llm_end_response
@@ -277,7 +277,7 @@ class TestCallbacks:
 
     def test_empty_stream_reports_error_without_finish_only_lifecycle(self) -> None:
         handler = _RecordingHandler()
-        stream = _EmptyStreamModel(callbacks=[handler]).stream_v2("test")
+        stream = _EmptyStreamModel(callbacks=[handler]).stream_events("test", version="v3")
 
         with pytest.raises(ValueError, match="No generation chunks were returned"):
             list(stream)
@@ -289,7 +289,7 @@ class TestCallbacks:
     @pytest.mark.asyncio
     async def test_empty_astream_reports_error(self) -> None:
         handler = _AsyncRecordingHandler()
-        stream = await _EmptyStreamModel(callbacks=[handler]).astream_v2("test")
+        stream = await _EmptyStreamModel(callbacks=[handler]).astream_events("test", version="v3")
 
         with pytest.raises(ValueError, match="No generation chunks were returned"):
             await stream
@@ -303,12 +303,12 @@ class TestCallbacks:
 
 
 class TestOnStreamEvent:
-    """`on_stream_event` must fire once per protocol event from stream_v2."""
+    """`on_stream_event` must fire once per protocol event from `stream_events(version="v3")`."""
 
     def test_on_stream_event_fires_for_every_event_sync(self) -> None:
         handler = _RecordingHandler()
         model = FakeListChatModel(responses=["Hi"], callbacks=[handler])
-        stream = model.stream_v2("test")
+        stream = model.stream_events("test", version="v3")
         _ = stream.output
 
         # Every event the stream sees should also reach the observer.
@@ -322,7 +322,7 @@ class TestOnStreamEvent:
     async def test_on_stream_event_fires_for_every_event_async(self) -> None:
         handler = _AsyncRecordingHandler()
         model = FakeListChatModel(responses=["Hi"], callbacks=[handler])
-        stream = await model.astream_v2("test")
+        stream = await model.astream_events("test", version="v3")
         _ = await stream
 
         event_types = [e["event"] for e in handler.stream_events]
@@ -334,7 +334,7 @@ class TestOnStreamEvent:
         """Stream events must all fire between on_chat_model_start and on_llm_end."""
         handler = _RecordingHandler()
         model = FakeListChatModel(responses=["Hi"], callbacks=[handler])
-        stream = model.stream_v2("test")
+        stream = model.stream_events("test", version="v3")
         _ = stream.output
 
         # on_stream_event doesn't show up in `events` (different list), but
@@ -346,10 +346,10 @@ class TestOnStreamEvent:
 
 
 class TestCancellation:
-    """Cancellation of `astream_v2` must propagate, not be swallowed."""
+    """Cancellation of `astream_events(version="v3")` must propagate, not be swallowed."""
 
     @pytest.mark.asyncio
-    async def test_astream_v2_cancellation_propagates(self) -> None:
+    async def test_astream_events_v3_cancellation_propagates(self) -> None:
         """Cancelling the producer task must raise CancelledError.
 
         Regression test: the producer's `except BaseException` previously
@@ -357,7 +357,7 @@ class TestCancellation:
         `on_llm_error` + `stream._fail` pair that never propagated.
         """
         model = FakeListChatModel(responses=["abcdefghij"], sleep=0.05)
-        stream = await model.astream_v2("test")
+        stream = await model.astream_events("test", version="v3")
         aiter_ = stream.text.__aiter__()
         await aiter_.__anext__()
         task = stream._producer_task
@@ -401,7 +401,7 @@ class _KwargRecordingModel(FakeListChatModel):
 
 
 class TestRunnableBindingForwarding:
-    """`RunnableBinding.stream_v2` must merge bound kwargs into the call.
+    """`RunnableBinding.stream_events(version="v3")` must merge bound kwargs into the call.
 
     Without the explicit override on `RunnableBinding`, `__getattr__`
     forwards the call but drops `self.kwargs` — so tools bound via
@@ -409,12 +409,12 @@ class TestRunnableBindingForwarding:
     ignored.
     """
 
-    def test_bound_kwargs_reach_stream_v2(self) -> None:
+    def test_bound_kwargs_reach_stream_events_v3(self) -> None:
         model = _KwargRecordingModel(responses=["hi"])
         model.received_kwargs = []
         bound = model.bind(my_marker="sentinel-42")
 
-        stream = bound.stream_v2("test")
+        stream = bound.stream_events("test", version="v3")
         for _ in stream.text:
             pass
 
@@ -426,19 +426,19 @@ class TestRunnableBindingForwarding:
         model.received_kwargs = []
         bound = model.bind(my_marker="from-bind")
 
-        stream = bound.stream_v2("test", my_marker="from-call")
+        stream = bound.stream_events("test", my_marker="from-call", version="v3")
         for _ in stream.text:
             pass
 
         assert model.received_kwargs[0].get("my_marker") == "from-call"
 
     @pytest.mark.asyncio
-    async def test_bound_kwargs_reach_astream_v2(self) -> None:
+    async def test_bound_kwargs_reach_astream_events_v3(self) -> None:
         model = _KwargRecordingModel(responses=["hi"])
         model.received_kwargs = []
         bound = model.bind(my_marker="sentinel-async")
 
-        stream = await bound.astream_v2("test")
+        stream = await bound.astream_events("test", version="v3")
         _ = await stream
 
         assert len(model.received_kwargs) == 1
diff --git a/libs/core/tests/unit_tests/language_models/test_stream_v2.py b/libs/core/tests/unit_tests/language_models/test_chat_model_v3_stream.py
similarity index 94%
rename from libs/core/tests/unit_tests/language_models/test_stream_v2.py
rename to libs/core/tests/unit_tests/language_models/test_chat_model_v3_stream.py
index 9e847019d09..399b9ac1c5a 100644
--- a/libs/core/tests/unit_tests/language_models/test_stream_v2.py
+++ b/libs/core/tests/unit_tests/language_models/test_chat_model_v3_stream.py
@@ -1,4 +1,4 @@
-"""Tests for stream_v2 / astream_v2 and ChatModelStream."""
+"""Tests for `stream_events(version="v3")` / `astream_events(version="v3")` and `ChatModelStream`."""
 
 from __future__ import annotations
 
@@ -301,20 +301,20 @@ class TestAsyncChatModelStream:
 
 
 class TestStreamV2:
-    """Test BaseChatModel.stream_v2() with FakeListChatModel."""
+    """Test `BaseChatModel.stream_events(version="v3")` with `FakeListChatModel`."""
 
-    def test_stream_v2_text(self) -> None:
+    def test_stream_events_v3_text(self) -> None:
         model = FakeListChatModel(responses=["Hello world!"])
-        stream = model.stream_v2("test")
+        stream = model.stream_events("test", version="v3")
 
         assert isinstance(stream, ChatModelStream)
         deltas = list(stream.text)
         assert "".join(deltas) == "Hello world!"
         assert stream.done
 
-    def test_stream_v2_usage(self) -> None:
+    def test_stream_events_v3_usage(self) -> None:
         model = FakeListChatModel(responses=["Hi"])
-        stream = model.stream_v2("test")
+        stream = model.stream_events("test", version="v3")
 
         # Drain stream
         for _ in stream.text:
@@ -323,10 +323,10 @@ class TestStreamV2:
         assert stream.output.usage_metadata is None
         assert stream.done
 
-    def test_stream_v2_malformed_tool_args_produce_invalid_tool_call(self) -> None:
+    def test_stream_events_v3_malformed_tool_args_produce_invalid_tool_call(self) -> None:
         """End-to-end: malformed tool-call JSON becomes invalid_tool_calls."""
         model = _MalformedToolCallModel()
-        stream = model.stream_v2("test")
+        stream = model.stream_events("test", version="v3")
         msg = stream.output
 
         assert msg.tool_calls == []
@@ -336,10 +336,10 @@ class TestStreamV2:
         assert itc["args"] == '{"q": '
         assert itc["id"] == "call_1"
 
-    def test_stream_v2_translates_anthropic_server_tool_use_to_protocol(self) -> None:
+    def test_stream_events_v3_translates_anthropic_server_tool_use_to_protocol(self) -> None:
         """Phase E end-to-end: server_tool_use becomes server_tool_call in output."""
         model = _AnthropicStyleServerToolModel()
-        stream = model.stream_v2("weather?")
+        stream = model.stream_events("weather?", version="v3")
         msg = stream.output
 
         assert isinstance(msg.content, list)
@@ -353,21 +353,21 @@ class TestStreamV2:
 
 
 class TestAstreamV2:
-    """Test BaseChatModel.astream_v2() with FakeListChatModel."""
+    """Test `BaseChatModel.astream_events(version="v3")` with `FakeListChatModel`."""
 
     @pytest.mark.asyncio
-    async def test_astream_v2_text(self) -> None:
+    async def test_astream_events_v3_text(self) -> None:
         model = FakeListChatModel(responses=["Hello!"])
-        stream = await model.astream_v2("test")
+        stream = await model.astream_events("test", version="v3")
 
         assert isinstance(stream, AsyncChatModelStream)
         full = await stream.text
         assert full == "Hello!"
 
     @pytest.mark.asyncio
-    async def test_astream_v2_deltas(self) -> None:
+    async def test_astream_events_v3_deltas(self) -> None:
         model = FakeListChatModel(responses=["Hi"])
-        stream = await model.astream_v2("test")
+        stream = await model.astream_events("test", version="v3")
 
         deltas = [d async for d in stream.text]
         assert "".join(deltas) == "Hi"
@@ -690,33 +690,34 @@ class TestStructuredOutputKwargStripping:
 
     `stream()` / `astream()` pop `ls_structured_output_format` and
     `structured_output_format` before forwarding kwargs to `_stream` —
-    provider clients reject unknown kwargs. `stream_v2` / `astream_v2`
-    must do the same, or `.with_structured_output().stream_v2()` breaks.
+    provider clients reject unknown kwargs. `stream_events(version="v3")` /
+    `astream_events(version="v3")` must do the same, or
+    `.with_structured_output().stream_events(version="v3")` breaks.
     """
 
-    def test_stream_v2_strips_ls_structured_output_format(self) -> None:
+    def test_stream_events_v3_strips_ls_structured_output_format(self) -> None:
         model = _RecordingStreamModel()
         bound = model.bind(ls_structured_output_format={"schema": {"type": "object"}})
-        stream = bound.stream_v2("test")
+        stream = bound.stream_events("test", version="v3")
         _ = stream.output  # drain
         recorded = _RecordingStreamModel.last_stream_kwargs
         assert "ls_structured_output_format" not in recorded
         assert "structured_output_format" not in recorded
 
-    def test_stream_v2_strips_structured_output_format(self) -> None:
+    def test_stream_events_v3_strips_structured_output_format(self) -> None:
         model = _RecordingStreamModel()
         bound = model.bind(structured_output_format={"schema": {"type": "object"}})
-        stream = bound.stream_v2("test")
+        stream = bound.stream_events("test", version="v3")
         _ = stream.output
         recorded = _RecordingStreamModel.last_stream_kwargs
         assert "ls_structured_output_format" not in recorded
         assert "structured_output_format" not in recorded
 
     @pytest.mark.asyncio
-    async def test_astream_v2_strips_ls_structured_output_format(self) -> None:
+    async def test_astream_events_v3_strips_ls_structured_output_format(self) -> None:
         model = _RecordingStreamModel()
         bound = model.bind(ls_structured_output_format={"schema": {"type": "object"}})
-        stream = await bound.astream_v2("test")
+        stream = await bound.astream_events("test", version="v3")
         _ = await stream
         assert (
             "ls_structured_output_format"
@@ -727,10 +728,10 @@ class TestStructuredOutputKwargStripping:
         )
 
     @pytest.mark.asyncio
-    async def test_astream_v2_strips_structured_output_format(self) -> None:
+    async def test_astream_events_v3_strips_structured_output_format(self) -> None:
         model = _RecordingStreamModel()
         bound = model.bind(structured_output_format={"schema": {"type": "object"}})
-        stream = await bound.astream_v2("test")
+        stream = await bound.astream_events("test", version="v3")
         _ = await stream
         assert (
             "ls_structured_output_format"
@@ -845,7 +846,7 @@ class TestAsyncStreamAclose:
     async def test_aclose_cancels_producer_task(self) -> None:
         gate = asyncio.Event()
         model = _GatedStreamModel(gate=gate)
-        stream = await model.astream_v2("test")
+        stream = await model.astream_events("test", version="v3")
 
         # Pull the first delta so the producer enters the gated section.
         aiter_ = stream.text.__aiter__()
@@ -863,7 +864,7 @@ class TestAsyncStreamAclose:
     async def test_aclose_is_idempotent(self) -> None:
         gate = asyncio.Event()
         model = _GatedStreamModel(gate=gate)
-        stream = await model.astream_v2("test")
+        stream = await model.astream_events("test", version="v3")
         aiter_ = stream.text.__aiter__()
         await aiter_.__anext__()
 
@@ -874,7 +875,7 @@ class TestAsyncStreamAclose:
     async def test_async_context_manager_closes_stream(self) -> None:
         gate = asyncio.Event()
         model = _GatedStreamModel(gate=gate)
-        stream = await model.astream_v2("test")
+        stream = await model.astream_events("test", version="v3")
 
         async with stream as s:
             assert s is stream
@@ -897,7 +898,7 @@ class TestAsyncStreamAclose:
         """
         teardown_gate = asyncio.Event()
         model = _SlowTeardownModel(teardown_gate=teardown_gate)
-        stream = await model.astream_v2("test")
+        stream = await model.astream_events("test", version="v3")
 
         # Prime the producer so it enters `_astream`'s forever-blocking
         # await.
@@ -947,7 +948,7 @@ class TestAsyncStreamAclose:
     async def test_aclose_before_producer_starts_resolves_projections(self) -> None:
         """Early-cancel path: `_produce` never runs.
 
-        If a consumer calls `astream_v2()` and immediately `aclose()`
+        If a consumer calls `astream_events(version="v3")` and immediately `aclose()`
         (or `async with` exits before the loop schedules `_produce`),
         `task.cancel()` marks the task cancelled without ever invoking
         its body — so neither `stream.fail` nor `on_llm_error` fires.
@@ -963,8 +964,8 @@ class TestAsyncStreamAclose:
         handler = RecordingHandler()
         gate = asyncio.Event()
         model = _GatedStreamModel(gate=gate)
-        stream = await model.astream_v2("test", config={"callbacks": [handler]})
-        # No yield to the event loop between `astream_v2` returning and
+        stream = await model.astream_events("test", config={"callbacks": [handler]}, version="v3")
+        # No yield to the event loop between `astream_events(version="v3")` returning and
         # `aclose()` — the producer task has been created but its body
         # has not executed.
         await stream.aclose()
@@ -1003,7 +1004,7 @@ class TestAsyncStreamAclose:
         handler = RecordingHandler()
         gate = asyncio.Event()
         model = _GatedStreamModel(gate=gate)
-        stream = await model.astream_v2("test", config={"callbacks": [handler]})
+        stream = await model.astream_events("test", config={"callbacks": [handler]}, version="v3")
 
         aiter_ = stream.text.__aiter__()
         await aiter_.__anext__()
@@ -1043,7 +1044,7 @@ class TestAsyncStreamAclose:
 
         handler = SlowEndHandler()
         model = FakeListChatModel(responses=["ok"])
-        stream = await model.astream_v2("test", config={"callbacks": [handler]})
+        stream = await model.astream_events("test", config={"callbacks": [handler]}, version="v3")
 
         # Wait until the stream has assembled the message and the
         # slow on_llm_end handler has started running.
@@ -1163,7 +1164,7 @@ class TestCacheHitV2Replay:
 class _ProviderMetadataStreamModel(BaseChatModel):
     """Fake model that advertises `output_version="responses/v1"` in metadata.
 
-    Verifies `stream_v2` pins the assembled message's `output_version` to
+    Verifies `stream_events(version="v3")` pins the assembled message's `output_version` to
     `"v1"` — the shape it actually produces — regardless of what the
     provider's chunk metadata claims.
     """
@@ -1199,11 +1200,11 @@ class _ProviderMetadataStreamModel(BaseChatModel):
 
 
 class TestOutputVersionPinning:
-    """`stream_v2().output` always serializes as v1 content blocks."""
+    """`stream_events(version="v3").output` always serializes as v1 content blocks."""
 
     def test_output_version_pinned_to_v1(self) -> None:
         model = _ProviderMetadataStreamModel()
-        stream = model.stream_v2("hi")
+        stream = model.stream_events("hi", version="v3")
         msg = stream.output
         # Assembled message must claim `"v1"` even though the provider
         # chunk metadata advertised `"responses/v1"`.
@@ -1325,7 +1326,7 @@ class TestBedrockConverseToolCallArgs:
 
     def test_bedrock_tool_call_assembles_without_error(self) -> None:
         model = _BedrockConverseToolCallModel()
-        stream = model.stream_v2("What's the weather in Boston?")
+        stream = model.stream_events("What's the weather in Boston?", version="v3")
         # Drive the stream to completion — the raise would have surfaced here.
         events = list(stream)
 
diff --git a/libs/core/tests/unit_tests/language_models/test_v1_parity.py b/libs/core/tests/unit_tests/language_models/test_v1_parity.py
index d7f3f3a602c..4c7ed261080 100644
--- a/libs/core/tests/unit_tests/language_models/test_v1_parity.py
+++ b/libs/core/tests/unit_tests/language_models/test_v1_parity.py
@@ -1,7 +1,7 @@
-"""V1 parity tests: stream_v2() output must match model.stream() output.
+"""V1 parity tests: `stream_events(version="v3")` output must match `model.stream()` output.
 
-These are the acceptance criteria for streaming v2 — if any test fails,
-v2 has a regression vs v1.
+These are the acceptance criteria for the v3 streaming API — if any test fails,
+v3 has a regression vs v1.
 """
 
 from __future__ import annotations
@@ -128,8 +128,8 @@ def _collect_v1_message(model: BaseChatModel, input_text: str) -> AIMessage:
 
 
 def _collect_v2_message(model: BaseChatModel, input_text: str) -> AIMessage:
-    """Run model.stream_v2() and get .output."""
-    stream = model.stream_v2(input_text)
+    """Run `model.stream_events(version="v3")` and get `.output`."""
+    stream = model.stream_events(input_text, version="v3")
     return stream.output
 
 
@@ -156,13 +156,13 @@ class TestV1ParityBasic:
     def test_empty_response(self) -> None:
         """A truly empty stream is an error, matching `stream()` parity.
 
-        `stream_v2` distinguishes "producer emitted events but no terminal
-        `message-finish`" (which is synthesized, for native-event providers
+        `stream_events(version="v3")` distinguishes "producer emitted events but no
+        terminal `message-finish`" (which is synthesized, for native-event providers
         that omit it) from "producer emitted nothing at all" (which fails
         with `ValueError`, same as `stream()`).
         """
         model = FakeListChatModel(responses=[""])
-        stream = model.stream_v2("test")
+        stream = model.stream_events("test", version="v3")
         with pytest.raises(ValueError, match="No generation chunks"):
             _ = stream.output
 
@@ -179,7 +179,7 @@ class TestV1ParityBasic:
 
     def test_text_deltas_reconstruct_content(self) -> None:
         model = FakeListChatModel(responses=["Hello!"])
-        stream = model.stream_v2("test")
+        stream = model.stream_events("test", version="v3")
 
         deltas = list(stream.text)
         content = stream.output.content
@@ -233,7 +233,7 @@ class TestV1ParityToolCalls:
 
     def test_tool_calls_via_projection(self) -> None:
         model = self._make_model()
-        stream = model.stream_v2("weather?")
+        stream = model.stream_events("weather?", version="v3")
         finalized = stream.tool_calls.get()
         assert len(finalized) == 1
         assert finalized[0]["name"] == "get_weather"
@@ -276,7 +276,7 @@ class TestV1ParityUsage:
         assert v1.usage_metadata["total_tokens"] == v2.usage_metadata["total_tokens"]
 
     def test_usage_projection_matches(self) -> None:
-        stream = self._make_model().stream_v2("hello")
+        stream = self._make_model().stream_events("hello", version="v3")
         # Drain so usage is available
         for _ in stream.text:
             pass
@@ -352,7 +352,7 @@ class TestV1ParityReasoning:
         assert types_in_order == ["reasoning", "text"]
 
     def test_reasoning_projection(self) -> None:
-        stream = self._make_model().stream_v2("think")
+        stream = self._make_model().stream_events("think", version="v3")
         full_reasoning = str(stream.reasoning)
         assert full_reasoning == "Let me think. Done."
 
@@ -366,7 +366,7 @@ class TestV1ParityError:
         ]
         model = _ScriptedChunkModel(scripted_chunks=chunks, raise_after=True)
 
-        stream = model.stream_v2("boom")
+        stream = model.stream_events("boom", version="v3")
         # Drain first; error may surface here or at .output access.
         try:
             list(stream.text)
@@ -382,7 +382,7 @@ class TestV1ParityError:
         ]
         model = _ScriptedChunkModel(scripted_chunks=chunks, raise_after=True)
 
-        stream = await model.astream_v2("boom")
+        stream = await model.astream_events("boom", version="v3")
         try:
             async for _ in stream.text:
                 pass
diff --git a/libs/langchain_v1/tests/unit_tests/agents/test_agent_streaming.py b/libs/langchain_v1/tests/unit_tests/agents/test_agent_streaming.py
index 9450b6a5494..a573c76f343 100644
--- a/libs/langchain_v1/tests/unit_tests/agents/test_agent_streaming.py
+++ b/libs/langchain_v1/tests/unit_tests/agents/test_agent_streaming.py
@@ -1,4 +1,4 @@
-"""Unit tests for create_agent graphs streaming via stream_v2."""
+"""Unit tests for create_agent graphs streaming via `stream_events(version="v3")`."""
 
 from __future__ import annotations
 
@@ -60,7 +60,7 @@ class TestAgentStreamV2Sync:
         model = FakeToolCallingModel(tool_calls=_single_tool_call_script("echo", text="x"))
         agent = create_agent(model, [echo])
 
-        run = agent.stream_v2({"messages": [HumanMessage("hi")]})
+        run = agent.stream_events({"messages": [HumanMessage("hi")]}, version="v3")
 
         # Drain so the run closes cleanly.
         list(run.tool_calls)
@@ -70,7 +70,7 @@ class TestAgentStreamV2Sync:
         model = FakeToolCallingModel(tool_calls=_single_tool_call_script("echo", text="x"))
         agent = create_agent(model, [echo])
 
-        run = agent.stream_v2({"messages": [HumanMessage("hi")]})
+        run = agent.stream_events({"messages": [HumanMessage("hi")]}, version="v3")
 
         collected: list[ToolCallStream] = list(run.tool_calls)
         assert len(collected) == 1
@@ -84,7 +84,7 @@ class TestAgentStreamV2Sync:
         model = FakeToolCallingModel(tool_calls=_single_tool_call_script("streamer", text="x"))
         agent = create_agent(model, [streamer])
 
-        run = agent.stream_v2({"messages": [HumanMessage("hi")]})
+        run = agent.stream_events({"messages": [HumanMessage("hi")]}, version="v3")
 
         tool_calls: list[ToolCallStream] = []
         for tc in run.tool_calls:
@@ -97,7 +97,7 @@ class TestAgentStreamV2Sync:
         model = FakeToolCallingModel()  # no tool calls scripted
         agent = create_agent(model, [])
 
-        run = agent.stream_v2({"messages": [HumanMessage("hi")]})
+        run = agent.stream_events({"messages": [HumanMessage("hi")]}, version="v3")
         assert list(run.tool_calls) == []
         assert run.output is not None
 
@@ -106,7 +106,7 @@ class TestAgentStreamV2Sync:
         model = FakeToolCallingModel(tool_calls=_single_tool_call_script("echo", text="x"))
         agent = create_agent(model, [echo])
 
-        run = agent.stream_v2({"messages": [HumanMessage("hi")]})
+        run = agent.stream_events({"messages": [HumanMessage("hi")]}, version="v3")
         # The native `messages` projection is bound as an instance attribute
         # by `BaseRunStream.__init__` whenever `MessagesTransformer` is
         # registered. Content population is covered by langgraph tests —
@@ -137,9 +137,10 @@ class TestAgentStreamV2Sync:
         model = FakeToolCallingModel(tool_calls=_single_tool_call_script("echo", text="x"))
         agent = create_agent(model, [echo])
 
-        run = agent.stream_v2(
+        run = agent.stream_events(
             {"messages": [HumanMessage("hi")]},
             transformers=[_Marker],
+            version="v3",
         )
         # Both the agent default and the user transformer are registered.
         assert "tool_calls" in run._mux.extensions  # type: ignore[attr-defined]
@@ -156,7 +157,7 @@ class TestAgentStreamV2Sync:
         model = FakeToolCallingModel(tool_calls=_single_tool_call_script("boom"))
         agent = create_agent(model, [boom])
 
-        run = agent.stream_v2({"messages": [HumanMessage("hi")]})
+        run = agent.stream_events({"messages": [HumanMessage("hi")]}, version="v3")
 
         collected: list[ToolCallStream] = []
 
@@ -179,7 +180,7 @@ class TestAgentStreamV2Async:
         model = FakeToolCallingModel(tool_calls=_single_tool_call_script("echo", text="x"))
         agent = create_agent(model, [echo])
 
-        run = await agent.astream_v2({"messages": [HumanMessage("hi")]})
+        run = await agent.astream_events({"messages": [HumanMessage("hi")]}, version="v3")
         async for tc in run.tool_calls:
             async for _ in tc.output_deltas:
                 pass
@@ -189,7 +190,7 @@ class TestAgentStreamV2Async:
         model = FakeToolCallingModel(tool_calls=_single_tool_call_script("astreamer", text="hi"))
         agent = create_agent(model, [astreamer])
 
-        run = await agent.astream_v2({"messages": [HumanMessage("hi")]})
+        run = await agent.astream_events({"messages": [HumanMessage("hi")]}, version="v3")
 
         collected: list[ToolCallStream] = []
         async for tc in run.tool_calls:
diff --git a/libs/partners/anthropic/tests/integration_tests/test_chat_models.py b/libs/partners/anthropic/tests/integration_tests/test_chat_models.py
index 113ae45c791..1ede8fba9c6 100644
--- a/libs/partners/anthropic/tests/integration_tests/test_chat_models.py
+++ b/libs/partners/anthropic/tests/integration_tests/test_chat_models.py
@@ -927,7 +927,7 @@ def test_agent_loop_streaming(
     llm_with_tools = llm.bind_tools([get_weather])
     input_message = HumanMessage("What is the weather in San Francisco, CA?")
     if use_v2_stream:
-        tool_call_message = llm_with_tools.stream_v2([input_message]).output
+        tool_call_message = llm_with_tools.stream_events([input_message], version="v3").output
     else:
         tool_call_message = llm_with_tools.invoke([input_message])
     assert isinstance(tool_call_message, AIMessage)
@@ -938,8 +938,9 @@ def test_agent_loop_streaming(
     tool_message = get_weather.invoke(tool_call)
     assert isinstance(tool_message, ToolMessage)
     if use_v2_stream:
-        response = llm_with_tools.stream_v2(
-            [input_message, tool_call_message, tool_message]
+        response = llm_with_tools.stream_events(
+            [input_message, tool_call_message, tool_message],
+            version="v3",
         ).output
     else:
         response = llm_with_tools.invoke(
@@ -954,8 +955,8 @@ def test_agent_loop_streaming(
 
 @pytest.mark.default_cassette("test_agent_loop_streaming.yaml.gz")
 @pytest.mark.vcr
-async def test_agent_loop_streaming_astream_v2_v1() -> None:
-    """Async multi-turn through `astream_v2`.
+async def test_agent_loop_streaming_astream_events_v3_v1() -> None:
+    """Async multi-turn through `astream_events(version="v3")`.
 
     Mirrors `test_agent_loop_streaming` for `output_version="v1"` but
     exercises `AsyncChatModelStream` end-to-end.
@@ -973,7 +974,7 @@ async def test_agent_loop_streaming_astream_v2_v1() -> None:
     )
     llm_with_tools = llm.bind_tools([get_weather])
     input_message = HumanMessage("What is the weather in San Francisco, CA?")
-    tool_call_message = await (await llm_with_tools.astream_v2([input_message]))
+    tool_call_message = await (await llm_with_tools.astream_events([input_message], version="v3"))
     assert isinstance(tool_call_message, AIMessage)
     tool_calls = tool_call_message.tool_calls
     assert len(tool_calls) == 1
@@ -981,8 +982,9 @@ async def test_agent_loop_streaming_astream_v2_v1() -> None:
     tool_message = get_weather.invoke(tool_call)
     assert isinstance(tool_message, ToolMessage)
     response = await (
-        await llm_with_tools.astream_v2(
-            [input_message, tool_call_message, tool_message]
+        await llm_with_tools.astream_events(
+            [input_message, tool_call_message, tool_message],
+            version="v3",
         )
     )
     assert isinstance(response, AIMessage)
@@ -1030,7 +1032,7 @@ def test_citations(output_version: Literal["v0", "v1"], *, use_v2_stream: bool)
     # Test streaming
     full: BaseMessage
     if use_v2_stream:
-        full = llm.stream_v2(messages).output
+        full = llm.stream_events(messages, version="v3").output
     else:
         aggregated: BaseMessageChunk | None = None
         for chunk in llm.stream(messages):
@@ -1124,7 +1126,7 @@ def test_thinking_v1(*, use_v2_stream: bool) -> None:
     # Test streaming
     full: BaseMessage
     if use_v2_stream:
-        full = llm.stream_v2([input_message]).output
+        full = llm.stream_events([input_message], version="v3").output
     else:
         aggregated: BaseMessageChunk | None = None
         for chunk in llm.stream([input_message]):
@@ -2618,10 +2620,10 @@ def _stable_blocks(blocks: Any) -> list[dict[str, Any]]:
 @pytest.mark.default_cassette("test_streaming_tool_call_v1_v2_parity.yaml.gz")
 @pytest.mark.vcr
 def test_streaming_tool_call_v1_v2_parity() -> None:
-    """`AIMessage` parity between `stream()` reduction and `stream_v2().output`.
+    """`AIMessage` parity between `stream()` reduction and `stream_events(version="v3").output`.
 
     Runs the same forced-tool-call prompt through both the legacy chunk
-    stream (reduced with `AIMessageChunk.__add__`) and the `stream_v2`
+    stream (reduced with `AIMessageChunk.__add__`) and the `stream_events(version="v3")`
     bridge path on a `v1`-output `ChatAnthropic`, then compares the
     resulting messages on path-independent invariants:
 
@@ -2649,7 +2651,7 @@ def test_streaming_tool_call_v1_v2_parity() -> None:
         v1_full = chunk if v1_full is None else v1_full + chunk
     assert isinstance(v1_full, AIMessageChunk)
 
-    stream = with_tool.stream_v2(prompt)
+    stream = with_tool.stream_events(prompt, version="v3")
     events = list(stream)
     assert_valid_event_stream(events)
     v2_message = stream.output
diff --git a/libs/partners/anthropic/tests/unit_tests/test_chat_models.py b/libs/partners/anthropic/tests/unit_tests/test_chat_models.py
index 9d42e9bad89..61549b645f7 100644
--- a/libs/partners/anthropic/tests/unit_tests/test_chat_models.py
+++ b/libs/partners/anthropic/tests/unit_tests/test_chat_models.py
@@ -3053,15 +3053,15 @@ def test_no_task_budget_no_beta() -> None:
         assert "task-budgets-2026-03-13" not in betas
 
 
-def test_anthropic_stream_v2_lifecycle() -> None:
+def test_anthropic_stream_events_v3_lifecycle() -> None:
     """Validate lifecycle events across a thinking + text + tool_use stream.
 
     Anthropic emits raw `content_block_start` / `content_block_delta` /
     `content_block_stop` events with integer `index` fields, interleaved
     with `message_start` and `message_delta`. This test threads a
     realistic event sequence through `_stream` via a mocked raw client
-    and asserts that `stream_v2` produces a spec-conformant event
-    stream: paired start/finish per block, no interleaving, sequential
+    and asserts that `stream_events(version="v3")` produces a spec-conformant
+    event stream: paired start/finish per block, no interleaving, sequential
     `uint` wire indices.
     """
     from unittest.mock import patch
@@ -3182,7 +3182,7 @@ def test_anthropic_stream_v2_lifecycle() -> None:
         return events
 
     with patch.object(llm, "_create", mock_create):
-        stream_events = list(llm.stream_v2("Test query"))
+        stream_events = list(llm.stream_events("Test query", version="v3"))
 
     assert_valid_event_stream(stream_events)
 
diff --git a/libs/partners/openai/tests/integration_tests/chat_models/test_base.py b/libs/partners/openai/tests/integration_tests/chat_models/test_base.py
index d939877346b..a4ddb6ef386 100644
--- a/libs/partners/openai/tests/integration_tests/chat_models/test_base.py
+++ b/libs/partners/openai/tests/integration_tests/chat_models/test_base.py
@@ -1289,7 +1289,7 @@ class _Person(BaseModel):
 
 @pytest.mark.vcr
 def test_streaming_tool_call_v1_v2_parity() -> None:
-    """`stream()` and `stream_v2()` must agree on their final `AIMessage`.
+    """`stream()` and `stream_events(version="v3")` must agree on their final `AIMessage`.
 
     Both paths are invoked against the same HTTP response (the cassette's
     single recorded interaction, replayed for both calls via
@@ -1310,7 +1310,7 @@ def test_streaming_tool_call_v1_v2_parity() -> None:
         v1 = chunk if v1 is None else v1 + chunk
     assert isinstance(v1, AIMessageChunk)
 
-    stream = with_tool.stream_v2(prompt)
+    stream = with_tool.stream_events(prompt, version="v3")
     events = list(stream)
     assert_valid_event_stream(events)
     v2 = stream.output
diff --git a/libs/partners/openai/tests/integration_tests/chat_models/test_responses_api.py b/libs/partners/openai/tests/integration_tests/chat_models/test_responses_api.py
index 33531867c8f..26c923cf2d3 100644
--- a/libs/partners/openai/tests/integration_tests/chat_models/test_responses_api.py
+++ b/libs/partners/openai/tests/integration_tests/chat_models/test_responses_api.py
@@ -113,9 +113,10 @@ def test_web_search(
     # Test streaming
     full: BaseMessage
     if use_v2_stream:
-        full = llm.stream_v2(
+        full = llm.stream_events(
             "What was a positive news story from today?",
             tools=[{"type": "web_search_preview"}],
+            version="v3",
         ).output
     else:
         aggregated: BaseMessageChunk | None = None
@@ -284,7 +285,7 @@ def test_agent_loop_streaming(
     llm_with_tools = llm.bind_tools([get_weather])
     input_message = HumanMessage("What is the weather in San Francisco, CA?")
     if use_v2_stream:
-        tool_call_message = llm_with_tools.stream_v2([input_message]).output
+        tool_call_message = llm_with_tools.stream_events([input_message], version="v3").output
     else:
         tool_call_message = llm_with_tools.invoke([input_message])
     assert isinstance(tool_call_message, AIMessage)
@@ -294,8 +295,9 @@ def test_agent_loop_streaming(
     tool_message = get_weather.invoke(tool_call)
     assert isinstance(tool_message, ToolMessage)
     if use_v2_stream:
-        response = llm_with_tools.stream_v2(
-            [input_message, tool_call_message, tool_message]
+        response = llm_with_tools.stream_events(
+            [input_message, tool_call_message, tool_message],
+            version="v3",
         ).output
     else:
         response = llm_with_tools.invoke(
@@ -310,8 +312,8 @@ def test_agent_loop_streaming(
 
 @pytest.mark.default_cassette("test_agent_loop_streaming.yaml.gz")
 @pytest.mark.vcr
-async def test_agent_loop_streaming_astream_v2_v1() -> None:
-    """Async multi-turn through `astream_v2`.
+async def test_agent_loop_streaming_astream_events_v3_v1() -> None:
+    """Async multi-turn through `astream_events(version="v3")`.
 
     Mirrors `test_agent_loop_streaming` for `output_version="v1"` but
     exercises `AsyncChatModelStream` end-to-end: aggregation in the
@@ -335,7 +337,7 @@ async def test_agent_loop_streaming_astream_v2_v1() -> None:
     )
     llm_with_tools = llm.bind_tools([get_weather])
     input_message = HumanMessage("What is the weather in San Francisco, CA?")
-    stream = await llm_with_tools.astream_v2([input_message])
+    stream = await llm_with_tools.astream_events([input_message], version="v3")
     tool_call_message = await stream
     assert isinstance(tool_call_message, AIMessage)
     tool_calls = tool_call_message.tool_calls
@@ -343,8 +345,9 @@ async def test_agent_loop_streaming_astream_v2_v1() -> None:
     tool_call = tool_calls[0]
     tool_message = get_weather.invoke(tool_call)
     assert isinstance(tool_message, ToolMessage)
-    stream = await llm_with_tools.astream_v2(
-        [input_message, tool_call_message, tool_message]
+    stream = await llm_with_tools.astream_events(
+        [input_message, tool_call_message, tool_message],
+        version="v3",
     )
     response = await stream
     assert isinstance(response, AIMessage)
@@ -647,7 +650,7 @@ def test_stream_reasoning_summary(
     }
     response_1: BaseMessage
     if use_v2_stream:
-        response_1 = llm.stream_v2([message_1]).output
+        response_1 = llm.stream_events([message_1], version="v3").output
     else:
         aggregated: BaseMessageChunk | None = None
         for chunk in llm.stream([message_1]):
@@ -769,7 +772,7 @@ def test_code_interpreter(
 
     full: BaseMessage
     if use_v2_stream:
-        full = llm_with_tools.stream_v2([input_message]).output
+        full = llm_with_tools.stream_events([input_message], version="v3").output
     else:
         aggregated: BaseMessageChunk | None = None
         for chunk in llm_with_tools.stream([input_message]):
@@ -933,7 +936,7 @@ def test_mcp_builtin_zdr_v1(use_v2_stream: bool) -> None:
     }
     full: BaseMessage
     if use_v2_stream:
-        full = llm_with_tools.stream_v2([input_message]).output
+        full = llm_with_tools.stream_events([input_message], version="v3").output
     else:
         aggregated: BaseMessageChunk | None = None
         for chunk in llm_with_tools.stream([input_message]):
@@ -1365,7 +1368,7 @@ def test_compaction_streaming(
 
     def _run(messages: list) -> AIMessage:
         if use_v2_stream:
-            return llm.stream_v2(messages).output
+            return llm.stream_events(messages, version="v3").output
         result = llm.invoke(messages)
         assert isinstance(result, AIMessage)
         return result
@@ -1746,7 +1749,7 @@ def test_client_executed_tool_search() -> None:
 @pytest.mark.default_cassette("test_reasoning_text_v1_v2_parity.yaml.gz")
 @pytest.mark.vcr
 def test_reasoning_text_v1_v2_parity() -> None:
-    """`stream()` and `stream_v2()` must agree on reasoning + text output.
+    """`stream()` and `stream_events(version="v3")` must agree on reasoning + text output.
 
     Exercises the non-tool-call branch of the parity claim: a reasoning
     model (`o4-mini` via the Responses API) produces one or more
@@ -1767,7 +1770,7 @@ def test_reasoning_text_v1_v2_parity() -> None:
         v1 = chunk if v1 is None else v1 + chunk
     assert isinstance(v1, AIMessageChunk)
 
-    stream = llm.stream_v2([prompt])
+    stream = llm.stream_events([prompt], version="v3")
     events = list(stream)
     assert_valid_event_stream(events)
     v2 = stream.output
diff --git a/libs/partners/openai/tests/unit_tests/chat_models/test_base.py b/libs/partners/openai/tests/unit_tests/chat_models/test_base.py
index 9b673e1b5d7..7f9e96a00c5 100644
--- a/libs/partners/openai/tests/unit_tests/chat_models/test_base.py
+++ b/libs/partners/openai/tests/unit_tests/chat_models/test_base.py
@@ -616,8 +616,8 @@ def test_openai_stream(mock_openai_completion: list) -> None:
         assert "stream_options" not in call_kwargs[-1]
 
 
-def test_openai_stream_v2_lifecycle(mock_openai_completion: list) -> None:
-    """`stream_v2` on chat completions emits a spec-conformant lifecycle."""
+def test_openai_stream_events_v3_lifecycle(mock_openai_completion: list) -> None:
+    """`stream_events(version="v3")` on chat completions emits a spec-conformant lifecycle."""
     from langchain_tests.utils.stream_lifecycle import assert_valid_event_stream
 
     llm = ChatOpenAI(model="gpt-4o")
@@ -628,7 +628,7 @@ def test_openai_stream_v2_lifecycle(mock_openai_completion: list) -> None:
 
     mock_client.create = mock_create
     with patch.object(llm, "client", mock_client):
-        events = list(llm.stream_v2("你的名字叫什么？只回答名字"))
+        events = list(llm.stream_events("你的名字叫什么？只回答名字", version="v3"))
 
     assert_valid_event_stream(events)
     # At minimum, a text block with the accumulated answer.
diff --git a/libs/partners/openai/tests/unit_tests/chat_models/test_responses_stream.py b/libs/partners/openai/tests/unit_tests/chat_models/test_responses_stream.py
index 98abdce630c..181797aebde 100644
--- a/libs/partners/openai/tests/unit_tests/chat_models/test_responses_stream.py
+++ b/libs/partners/openai/tests/unit_tests/chat_models/test_responses_stream.py
@@ -763,8 +763,8 @@ def test_responses_stream(output_version: str, expected_content: list[dict]) ->
         assert dumped == payload["input"][idx]
 
 
-def test_responses_stream_v2_emits_reasoning_lifecycle() -> None:
-    """`stream_v2` must emit `content-block-finish` events for reasoning blocks.
+def test_responses_stream_events_v3_emits_reasoning_lifecycle() -> None:
+    """`stream_events(version="v3")` must emit `content-block-finish` events for reasoning blocks.
 
     Regression test: the protocol bridge should surface the full lifecycle
     (`content-block-start` / `content-block-delta` / `content-block-finish`)
@@ -779,7 +779,7 @@ def test_responses_stream_v2_emits_reasoning_lifecycle() -> None:
     mock_client.responses.create = mock_create
 
     with patch.object(llm, "root_client", mock_client):
-        events = list(llm.stream_v2("test"))
+        events = list(llm.stream_events("test", version="v3"))
 
     assert_valid_event_stream(events)
 
diff --git a/libs/standard-tests/langchain_tests/integration_tests/chat_models.py b/libs/standard-tests/langchain_tests/integration_tests/chat_models.py
index 02d842ca0ad..c28d0b7afe4 100644
--- a/libs/standard-tests/langchain_tests/integration_tests/chat_models.py
+++ b/libs/standard-tests/langchain_tests/integration_tests/chat_models.py
@@ -913,24 +913,24 @@ class ChatModelIntegrationTests(ChatModelTests):
         )
 
     def test_stream_v2(self, model: BaseChatModel) -> None:
-        """Test that `model.stream_v2(simple_message)` works.
+        """Test that `model.stream_events("Hello", version="v3")` works.
 
         Exercises the content-block-centric streaming protocol. Passing this
-        test indicates the model participates in `stream_v2` either natively
-        (via `_stream_chat_model_events`) or through the compat bridge that
+        test indicates the model participates in `stream_events(version="v3")` either
+        natively (via `_stream_chat_model_events`) or through the compat bridge that
         converts `_stream` chunks into protocol events.
 
         ??? question "Troubleshooting"
 
             First, debug
             `langchain_tests.integration_tests.chat_models.ChatModelIntegrationTests.test_stream`
-            — `stream_v2` falls back to the same `_stream` path via the compat
-            bridge when the model does not implement
+            — `stream_events(version="v3")` falls back to the same `_stream` path via the
+            compat bridge when the model does not implement
             `_stream_chat_model_events`. If `test_stream` passes but this does
             not, inspect the raised lifecycle violation: it identifies the
             event index and the rule broken.
         """
-        stream = model.stream_v2("Hello")
+        stream = model.stream_events("Hello", version="v3")
         assert isinstance(stream, ChatModelStream)
 
         events = list(stream)
@@ -942,11 +942,11 @@ class ChatModelIntegrationTests(ChatModelTests):
         assert message.content
         assert len(message.content_blocks) == 1
         assert message.content_blocks[0]["type"] == "text"
-        # `stream_v2` always assembles content as v1 protocol blocks.
+        # `stream_events(version="v3")` always assembles content as v1 protocol blocks.
         assert message.response_metadata.get("output_version") == "v1"
 
     async def test_astream_v2(self, model: BaseChatModel) -> None:
-        """Test that `await model.astream_v2(simple_message)` works.
+        """Test that `await model.astream_events("Hello", version="v3")` works.
 
         Async counterpart to `test_stream_v2`. Exercises the
         `AsyncChatModelStream` path end-to-end: the background producer task,
@@ -961,7 +961,7 @@ class ChatModelIntegrationTests(ChatModelTests):
             lifecycle violation; it identifies the event index and the rule
             broken.
         """
-        stream = await model.astream_v2("Hello")
+        stream = await model.astream_events("Hello", version="v3")
         assert isinstance(stream, AsyncChatModelStream)
 
         events = [event async for event in stream]
diff --git a/libs/standard-tests/langchain_tests/utils/stream_lifecycle.py b/libs/standard-tests/langchain_tests/utils/stream_lifecycle.py
index cb241827f03..048d71ec5e4 100644
--- a/libs/standard-tests/langchain_tests/utils/stream_lifecycle.py
+++ b/libs/standard-tests/langchain_tests/utils/stream_lifecycle.py
@@ -1,6 +1,6 @@
 """Validator for LangChain content-block protocol event streams.
 
-Checks that an event stream emitted by a chat model (via `stream_v2`,
+Checks that an event stream emitted by a chat model (via `stream_events(version="v3")`,
 or by the compat bridge's `chunks_to_events` / `message_to_events`)
 conforms to the protocol lifecycle rules:
 
@@ -37,7 +37,7 @@ def assert_valid_event_stream(events: Iterable[Any]) -> None:
 
     Args:
         events: Iterable of protocol event dicts (as yielded by
-            `stream_v2` or `chunks_to_events`).
+            `stream_events(version="v3")` or `chunks_to_events`).
 
     Raises:
         AssertionError: On the first lifecycle violation found. The