diff --git a/langgraph/checkpoint/base.py b/langgraph/checkpoint/base.py
index 35525d395..e40f3fcfc 100644
--- a/langgraph/checkpoint/base.py
+++ b/langgraph/checkpoint/base.py
@@ -5,7 +5,6 @@ from datetime import datetime, timezone
 from typing import Any, Optional, TypedDict
 
 from langchain_core.load.serializable import Serializable
-from langchain_core.pydantic_v1 import Field
 from langchain_core.runnables import RunnableConfig
 from langchain_core.runnables.utils import ConfigurableFieldSpec
 
@@ -57,9 +56,3 @@ class BaseCheckpointSaver(Serializable, ABC):
         return await asyncio.get_running_loop().run_in_executor(
             None, self.put, config, checkpoint
         )
-
-
-class CheckpointView(Serializable):
-    values: dict[str, Any] = Field(frozen=True)
-
-    step: int
diff --git a/langgraph/graph/__init__.py b/langgraph/graph/__init__.py
index a0447cf7f..1f6fdd5ce 100644
--- a/langgraph/graph/__init__.py
+++ b/langgraph/graph/__init__.py
@@ -11,6 +11,8 @@ from langchain_core.runnables.base import (
 
 from langgraph.pregel import Channel, Pregel
 
+END = "__end__"
+
 
 class Branch(NamedTuple):
     condition: Callable[..., str]
@@ -18,10 +20,8 @@ class Branch(NamedTuple):
 
     def runnable(self, input: Any) -> Runnable:
         result = self.condition(input)
-        return Channel.write_to(self.ends[result])
-
-
-END = "__end__"
+        destination = self.ends[result]
+        return Channel.write_to(f"{destination}:inbox" if destination != END else END)
 
 
 class Graph:
@@ -106,24 +106,31 @@ class Graph:
 
         outgoing_edges = defaultdict(list)
         for start, end in self.edges:
-            outgoing_edges[start].append(end)
+            outgoing_edges[start].append(f"{end}:inbox")
         if hasattr(self, "finish_point"):
             outgoing_edges[self.finish_point].append(END)
 
         nodes = {
-            key: Channel.subscribe_to(key) | node for key, node in self.nodes.items()
+            key: (Channel.subscribe_to(f"{key}:inbox") | node | Channel.write_to(key))
+            for key, node in self.nodes.items()
         }
 
-        for key, edges in outgoing_edges.items():
-            if edges:
-                nodes[key] |= Channel.write_to(*edges)
-
-        for key, branches in self.branches.items():
-            for branch in branches:
-                nodes[key] |= RunnableLambda(branch.runnable, name=f"{key}_condition")
+        for key in self.nodes:
+            outgoing = outgoing_edges[key]
+            edges_key = f"{key}:edges"
+            if outgoing or key in self.branches:
+                nodes[edges_key] = Channel.subscribe_to(key)
+            if outgoing:
+                nodes[edges_key] |= Channel.write_to(*[dest for dest in outgoing])
+            if key in self.branches:
+                for branch in self.branches[key]:
+                    nodes[edges_key] |= RunnableLambda(
+                        branch.runnable, name=f"{key}_condition"
+                    )
 
         return Pregel(
             nodes=nodes,
-            input=self.entry_point,
+            input=f"{self.entry_point}:inbox",
             output=END,
+            hidden=[f"{node}:inbox" for node in self.nodes],
         )
diff --git a/langgraph/pregel/__init__.py b/langgraph/pregel/__init__.py
index cfc74a3d3..2521708ac 100644
--- a/langgraph/pregel/__init__.py
+++ b/langgraph/pregel/__init__.py
@@ -53,7 +53,6 @@ from langgraph.checkpoint.base import (
     BaseCheckpointSaver,
     Checkpoint,
     CheckpointAt,
-    CheckpointView,
     empty_checkpoint,
 )
 from langgraph.constants import CONFIG_KEY_READ, CONFIG_KEY_SEND
@@ -153,6 +152,8 @@ class Pregel(
 
     output: Union[str, Sequence[str]] = "output"
 
+    hidden: Sequence[str] = Field(default_factory=list)
+
     input: Union[str, Sequence[str]] = "input"
 
     step_timeout: Optional[float] = None
@@ -220,11 +221,12 @@ class Pregel(
         config: RunnableConfig,
         *,
         output: Optional[Union[str, Sequence[str]]] = None,
-    ) -> Iterator[tuple[Union[dict[str, Any], Any], CheckpointView]]:
+    ) -> Iterator[Union[dict[str, Any], Any]]:
         if config["recursion_limit"] < 1:
             raise ValueError("recursion_limit must be at least 1")
         # assign defaults
-        output = output if output is not None else [chan for chan in self.channels]
+        if output is None:
+            output = [chan for chan in self.channels if chan not in self.hidden]
         # copy nodes to ignore mutations during execution
         processes = {**self.nodes}
         # get checkpoint from saver, or create an empty one
@@ -303,13 +305,12 @@ class Pregel(
                     print_checkpoint(step, channels)
 
                 # yield current value and checkpoint view
-                view = CheckpointView(
-                    values=_updateable_channel_values(channels),
-                    step=step + 1,
-                )
-                yield map_output(output, pending_writes, channels), view
-                # if view was updated, apply writes to channels
-                _apply_writes_from_view(checkpoint, channels, view)
+                if step_output := map_output(output, pending_writes, channels):
+                    yield step_output
+                    # we can detect updates when output is multiple channels (ie. dict)
+                    if not isinstance(output, str):
+                        # if view was updated, apply writes to channels
+                        _apply_writes_from_view(checkpoint, channels, step_output)
 
                 # save end of step checkpoint
                 if self.saver is not None and self.saver.at == CheckpointAt.END_OF_STEP:
@@ -328,7 +329,7 @@ class Pregel(
         config: RunnableConfig,
         *,
         output: Optional[Union[str, Sequence[str]]] = None,
-    ) -> AsyncIterator[tuple[Union[dict[str, Any], Any], CheckpointView]]:
+    ) -> AsyncIterator[Union[dict[str, Any], Any]]:
         if config["recursion_limit"] < 1:
             raise ValueError("recursion_limit must be at least 1")
         # if running from astream_log() run each proc with streaming
@@ -341,7 +342,8 @@ class Pregel(
             None,
         )
         # assign defaults
-        output = output if output is not None else [chan for chan in self.channels]
+        if output is None:
+            output = [chan for chan in self.channels if chan not in self.hidden]
         # copy nodes to ignore mutations during execution
         processes = {**self.nodes}
         # get checkpoint from saver, or create an empty one
@@ -423,13 +425,12 @@ class Pregel(
                     print_checkpoint(step, channels)
 
                 # yield current value and checkpoint view
-                view = CheckpointView(
-                    values=_updateable_channel_values(channels),
-                    step=step + 1,
-                )
-                yield map_output(output, pending_writes, channels), view
-                # if view was updated, apply writes to channels
-                _apply_writes_from_view(checkpoint, channels, view)
+                if step_output := map_output(output, pending_writes, channels):
+                    yield step_output
+                    # we can detect updates when output is multiple channels (ie. dict)
+                    if not isinstance(output, str):
+                        # if view was updated, apply writes to channels
+                        _apply_writes_from_view(checkpoint, channels, step_output)
 
                 # save end of step checkpoint
                 if self.saver is not None and self.saver.at == CheckpointAt.END_OF_STEP:
@@ -477,24 +478,10 @@ class Pregel(
         output: Optional[Union[str, Sequence[str]]] = None,
         **kwargs: Any,
     ) -> Iterator[Union[dict[str, Any], Any]]:
-        for out, _ in self._transform_stream_with_config(
+        for chunk in self._transform_stream_with_config(
             input, self._transform, config, output=output, **kwargs
         ):
-            if out is not None:
-                yield cast(Union[dict[str, Any], Any], out)
-
-    def step(
-        self,
-        input: Union[dict[str, Any], Any],
-        config: Optional[RunnableConfig] = None,
-        *,
-        output: Optional[Union[str, Sequence[str]]] = None,
-        **kwargs: Any,
-    ) -> Iterator[tuple[Union[dict[str, Any], Any], CheckpointView]]:
-        for tup in self._transform_stream_with_config(
-            iter([input]), self._transform, config, output=output, **kwargs
-        ):
-            yield cast(tuple[Union[dict[str, Any], Any], CheckpointView], tup)
+            yield chunk
 
     async def ainvoke(
         self,
@@ -538,27 +525,10 @@ class Pregel(
         output: Optional[Union[str, Sequence[str]]] = None,
         **kwargs: Any,
     ) -> AsyncIterator[Union[dict[str, Any], Any]]:
-        async for out, _ in self._atransform_stream_with_config(
+        async for chunk in self._atransform_stream_with_config(
             input, self._atransform, config, output=output, **kwargs
         ):
-            if out is not None:
-                yield out
-
-    async def astep(
-        self,
-        input: Union[dict[str, Any], Any],
-        config: Optional[RunnableConfig] = None,
-        *,
-        output: Optional[Union[str, Sequence[str]]] = None,
-        **kwargs: Any,
-    ) -> AsyncIterator[tuple[Union[dict[str, Any], Any], CheckpointView]]:
-        async def input_stream() -> AsyncIterator[Union[dict[str, Any], Any]]:
-            yield input
-
-        async for tup in self._atransform_stream_with_config(
-            input_stream(), self._atransform, config, output=output, **kwargs
-        ):
-            yield cast(tuple[Union[dict[str, Any], Any], CheckpointView], tup)
+            yield chunk
 
 
 def _interrupt_or_proceed(
@@ -629,11 +599,9 @@ def _apply_writes(
 
 
 def _apply_writes_from_view(
-    checkpoint: Checkpoint,
-    channels: Mapping[str, BaseChannel],
-    view: CheckpointView,
+    checkpoint: Checkpoint, channels: Mapping[str, BaseChannel], values: dict[str, Any]
 ) -> None:
-    for chan, value in view.values.items():
+    for chan, value in values.items():
         if value == channels[chan].get():
             continue
 
@@ -642,7 +610,7 @@ def _apply_writes_from_view(
             f"{channels[chan].__class__.__name__}"
         )
         checkpoint["channel_versions"][chan] += 1
-        channels[chan].update([view.values[chan]])
+        channels[chan].update([values[chan]])
 
 
 def _prepare_next_tasks(
@@ -703,18 +671,6 @@ def _prepare_next_tasks(
     return tasks
 
 
-def _updateable_channel_values(channels: Mapping[str, BaseChannel]) -> dict[str, Any]:
-    """Return a dictionary of updateable channel values."""
-    values: dict[str, Any] = {}
-    for k, v in channels.items():
-        if isinstance(v, LastValue) and k not in [c.value for c in ReservedChannels]:
-            try:
-                values[k] = v.get()
-            except EmptyChannelError:
-                pass
-    return values
-
-
 async def _aconsume(iterator: AsyncIterator[Any]) -> None:
     """Consume an async iterator."""
     async for _ in iterator:
diff --git a/langgraph/pregel/write.py b/langgraph/pregel/write.py
index f292303b5..2063f5018 100644
--- a/langgraph/pregel/write.py
+++ b/langgraph/pregel/write.py
@@ -32,6 +32,9 @@ class ChannelWrite(RunnablePassthrough):
         super().__init__(func=self._write, afunc=self._awrite, channels=channels)
         self.name = f"ChannelWrite<{','.join(chan for chan, _ in self.channels)}>"
 
+    def __repr_args__(self) -> Any:
+        return [("channels", self.channels)]
+
     @property
     def config_specs(self) -> list[ConfigurableFieldSpec]:
         return [
diff --git a/poetry.lock b/poetry.lock
index 1cda2d69f..aea6a8163 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 1.6.1 and should not be changed by hand.
+# This file is automatically @generated by Poetry 1.7.1 and should not be changed by hand.
 
 [[package]]
 name = "aiohttp"
@@ -121,9 +121,6 @@ files = [
     {file = "annotated_types-0.6.0.tar.gz", hash = "sha256:563339e807e53ffd9c267e99fc6d9ea23eb8443c08f112651963e24e22f84a5d"},
 ]
 
-[package.dependencies]
-typing-extensions = {version = ">=4.0.0", markers = "python_version < \"3.9\""}
-
 [[package]]
 name = "anyio"
 version = "4.2.0"
@@ -306,9 +303,6 @@ files = [
     {file = "Babel-2.14.0.tar.gz", hash = "sha256:6919867db036398ba21eb5c7a0f6b28ab8cbc3ae7a73a44ebe34ae74a4e7d363"},
 ]
 
-[package.dependencies]
-pytz = {version = ">=2015.7", markers = "python_version < \"3.9\""}
-
 [package.extras]
 dev = ["freezegun (>=1.0,<2.0)", "pytest (>=6.0)", "pytest-cov"]
 
@@ -988,24 +982,6 @@ docs = ["furo", "jaraco.packaging (>=9.3)", "jaraco.tidelift (>=1.4)", "rst.link
 perf = ["ipython"]
 testing = ["flufl.flake8", "importlib-resources (>=1.3)", "packaging", "pyfakefs", "pytest (>=6)", "pytest-black (>=0.3.7)", "pytest-checkdocs (>=2.4)", "pytest-cov", "pytest-enabler (>=2.2)", "pytest-mypy (>=0.9.1)", "pytest-perf (>=0.9.2)", "pytest-ruff"]
 
-[[package]]
-name = "importlib-resources"
-version = "6.1.1"
-description = "Read resources from Python packages"
-optional = false
-python-versions = ">=3.8"
-files = [
-    {file = "importlib_resources-6.1.1-py3-none-any.whl", hash = "sha256:e8bf90d8213b486f428c9c39714b920041cb02c184686a3dee24905aaa8105d6"},
-    {file = "importlib_resources-6.1.1.tar.gz", hash = "sha256:3893a00122eafde6894c59914446a512f728a0c1a45f9bb9b63721b6bacf0b4a"},
-]
-
-[package.dependencies]
-zipp = {version = ">=3.1.0", markers = "python_version < \"3.10\""}
-
-[package.extras]
-docs = ["furo", "jaraco.packaging (>=9.3)", "jaraco.tidelift (>=1.4)", "rst.linker (>=1.9)", "sphinx (<7.2.5)", "sphinx (>=3.5)", "sphinx-lint"]
-testing = ["pytest (>=6)", "pytest-black (>=0.3.7)", "pytest-checkdocs (>=2.4)", "pytest-cov", "pytest-enabler (>=2.2)", "pytest-mypy (>=0.9.1)", "pytest-ruff", "zipp (>=3.17)"]
-
 [[package]]
 name = "iniconfig"
 version = "2.0.0"
@@ -1214,11 +1190,9 @@ files = [
 attrs = ">=22.2.0"
 fqdn = {version = "*", optional = true, markers = "extra == \"format-nongpl\""}
 idna = {version = "*", optional = true, markers = "extra == \"format-nongpl\""}
-importlib-resources = {version = ">=1.4.0", markers = "python_version < \"3.9\""}
 isoduration = {version = "*", optional = true, markers = "extra == \"format-nongpl\""}
 jsonpointer = {version = ">1.13", optional = true, markers = "extra == \"format-nongpl\""}
 jsonschema-specifications = ">=2023.03.6"
-pkgutil-resolve-name = {version = ">=1.3.10", markers = "python_version < \"3.9\""}
 referencing = ">=0.28.4"
 rfc3339-validator = {version = "*", optional = true, markers = "extra == \"format-nongpl\""}
 rfc3986-validator = {version = ">0.1.0", optional = true, markers = "extra == \"format-nongpl\""}
@@ -1242,7 +1216,6 @@ files = [
 ]
 
 [package.dependencies]
-importlib-resources = {version = ">=1.4.0", markers = "python_version < \"3.9\""}
 referencing = ">=0.31.0"
 
 [[package]]
@@ -1441,7 +1414,6 @@ files = [
 [package.dependencies]
 async-lru = ">=1.0.0"
 importlib-metadata = {version = ">=4.8.3", markers = "python_version < \"3.10\""}
-importlib-resources = {version = ">=1.4", markers = "python_version < \"3.9\""}
 ipykernel = "*"
 jinja2 = ">=3.0.3"
 jupyter-core = "*"
@@ -2147,17 +2119,6 @@ files = [
     {file = "pickleshare-0.7.5.tar.gz", hash = "sha256:87683d47965c1da65cdacaf31c8441d12b8044cdec9aca500cd78fc2c683afca"},
 ]
 
-[[package]]
-name = "pkgutil-resolve-name"
-version = "1.3.10"
-description = "Resolve a name to an object."
-optional = false
-python-versions = ">=3.6"
-files = [
-    {file = "pkgutil_resolve_name-1.3.10-py3-none-any.whl", hash = "sha256:ca27cc078d25c5ad71a9de0a7a330146c4e014c2462d9af19c6b828280649c5e"},
-    {file = "pkgutil_resolve_name-1.3.10.tar.gz", hash = "sha256:357d6c9e6a755653cfd78893817c0853af365dd51ec97f3d358a819373bbd174"},
-]
-
 [[package]]
 name = "platformdirs"
 version = "4.1.0"
@@ -2575,17 +2536,6 @@ files = [
     {file = "python_json_logger-2.0.7-py3-none-any.whl", hash = "sha256:f380b826a991ebbe3de4d897aeec42760035ac760345e57b812938dc8b35e2bd"},
 ]
 
-[[package]]
-name = "pytz"
-version = "2023.3.post1"
-description = "World timezone definitions, modern and historical"
-optional = false
-python-versions = "*"
-files = [
-    {file = "pytz-2023.3.post1-py2.py3-none-any.whl", hash = "sha256:ce42d816b81b68506614c11e8937d3aa9e41007ceb50bfdcb0749b921bf646c7"},
-    {file = "pytz-2023.3.post1.tar.gz", hash = "sha256:7b4fddbeb94a1eba4b557da24f19fdf9db575192544270a9101d8509f9f43d7b"},
-]
-
 [[package]]
 name = "pywin32"
 version = "306"
@@ -3617,5 +3567,5 @@ testing = ["big-O", "jaraco.functools", "jaraco.itertools", "more-itertools", "p
 
 [metadata]
 lock-version = "2.0"
-python-versions = ">=3.8.1,<4.0"
-content-hash = "9aa1c4dcb484322ed2c5fc403902227865a825c9f1fd8874563edff0432133b6"
+python-versions = ">=3.9.0,<4.0"
+content-hash = "859cb37bda0ed0d3e8f94df41f0f6c5785aaf144e4e1dbb3835b945aa958ef1e"
diff --git a/tests/test_pregel.py b/tests/test_pregel.py
index 47c4ba321..cccb64fce 100644
--- a/tests/test_pregel.py
+++ b/tests/test_pregel.py
@@ -156,38 +156,28 @@ def test_invoke_two_processes_in_out(mocker: MockerFixture) -> None:
 
     assert app.invoke(2) == 4
 
-    for output, view in app.step(2):
-        if view.step == 1:
-            assert view.values == {
+    for step, values in enumerate(app.stream(2), start=1):
+        if step == 1:
+            assert values == {
                 "inbox": 3,
-                "input": 2,
             }
-            assert output == {"inbox": 3}
-        elif view.step == 2:
-            assert view.values == {
+        elif step == 2:
+            assert values == {
                 "output": 4,
-                "inbox": 3,
-                "input": 2,
             }
-            assert output == {"output": 4}
 
-    for output, view in app.step(2):
-        if view.step == 1:
-            assert view.values == {
+    for step, values in enumerate(app.stream(2), start=1):
+        if step == 1:
+            assert values == {
                 "inbox": 3,
-                "input": 2,
             }
-            assert output == {"inbox": 3}
             # modify inbox value
-            view.values["inbox"] = 5
-        elif view.step == 2:
-            assert view.values == {
-                "output": 6,
-                "inbox": 5,
-                "input": 2,
-            }
+            values["inbox"] = 5
+        elif step == 2:
             # output is different now
-            assert output == {"output": 6}
+            assert values == {
+                "output": 6,
+            }
 
     graph = Graph()
     graph.add_node("add_one", add_one)
@@ -199,38 +189,41 @@ def test_invoke_two_processes_in_out(mocker: MockerFixture) -> None:
 
     assert gapp.invoke(2) == 4
 
-    for output, view in gapp.step(2):
-        if view.step == 1:
-            assert view.values == {
-                "add_one": 2,
-                "add_one_more": 3,
+    for step, values in enumerate(gapp.stream(2), start=1):
+        if step == 1:
+            assert values == {
+                "add_one": 3,
             }
-            assert output == {"add_one_more": 3}
-        elif view.step == 2:
-            assert view.values == {
-                "add_one": 2,
-                "add_one_more": 3,
+        elif step == 2:
+            assert values == {
+                "add_one_more": 4,
+            }
+        elif step == 3:
+            assert values == {
                 "__end__": 4,
             }
-            assert output == {"__end__": 4}
+        else:
+            assert 0, f"{step}:{values}"
+    assert step == 3
 
-    for output, view in gapp.step(2):
-        if view.step == 1:
-            assert view.values == {
-                "add_one": 2,
-                "add_one_more": 3,
+    for step, values in enumerate(gapp.stream(2), start=1):
+        if step == 1:
+            assert values == {
+                "add_one": 3,
             }
-            assert output == {"add_one_more": 3}
-            # modify inbox value
-            view.values["add_one_more"] = 5
-        elif view.step == 2:
-            assert view.values == {
-                "add_one": 2,
-                "add_one_more": 5,
+            # modify value before next step
+            values["add_one"] = 5
+        elif step == 2:
+            assert values == {
+                "add_one_more": 6,
+            }
+        elif step == 3:
+            assert values == {
                 "__end__": 6,
             }
-            # output is different now
-            assert output == {"__end__": 6}
+        else:
+            assert 0, "Should not get here"
+    assert step == 3
 
 
 def test_invoke_two_processes_in_dict_out(mocker: MockerFixture) -> None:
@@ -658,14 +651,9 @@ def test_conditional_graph() -> None:
         ),
     }
 
-    assert [
-        deepcopy(c)
-        for c in app.stream(
-            {"input": "what is weather in sf"}, output=["agent", "tools"]
-        )
-    ] == [
+    assert [deepcopy(c) for c in app.stream({"input": "what is weather in sf"})] == [
         {
-            "tools": {
+            "agent": {
                 "input": "what is weather in sf",
                 "agent_outcome": AgentAction(
                     tool="search_api", tool_input="query", log="tool:search_api:query"
@@ -673,7 +661,7 @@ def test_conditional_graph() -> None:
             }
         },
         {
-            "agent": {
+            "tools": {
                 "input": "what is weather in sf",
                 "intermediate_steps": [
                     (
@@ -688,7 +676,7 @@ def test_conditional_graph() -> None:
             }
         },
         {
-            "tools": {
+            "agent": {
                 "input": "what is weather in sf",
                 "intermediate_steps": [
                     (
@@ -707,6 +695,29 @@ def test_conditional_graph() -> None:
                 ),
             }
         },
+        {
+            "tools": {
+                "input": "what is weather in sf",
+                "intermediate_steps": [
+                    (
+                        AgentAction(
+                            tool="search_api",
+                            tool_input="query",
+                            log="tool:search_api:query",
+                        ),
+                        "result for query",
+                    ),
+                    (
+                        AgentAction(
+                            tool="search_api",
+                            tool_input="another",
+                            log="tool:search_api:another",
+                        ),
+                        "result for another",
+                    ),
+                ],
+            }
+        },
         {
             "agent": {
                 "input": "what is weather in sf",
@@ -728,6 +739,35 @@ def test_conditional_graph() -> None:
                         "result for another",
                     ),
                 ],
+                "agent_outcome": AgentFinish(
+                    return_values={"answer": "answer"}, log="finish:answer"
+                ),
+            }
+        },
+        {
+            "__end__": {
+                "input": "what is weather in sf",
+                "intermediate_steps": [
+                    (
+                        AgentAction(
+                            tool="search_api",
+                            tool_input="query",
+                            log="tool:search_api:query",
+                        ),
+                        "result for query",
+                    ),
+                    (
+                        AgentAction(
+                            tool="search_api",
+                            tool_input="another",
+                            log="tool:search_api:another",
+                        ),
+                        "result for another",
+                    ),
+                ],
+                "agent_outcome": AgentFinish(
+                    return_values={"answer": "answer"}, log="finish:answer"
+                ),
             }
         },
     ]
diff --git a/tests/test_pregel_async.py b/tests/test_pregel_async.py
index 57b538398..3ee58d257 100644
--- a/tests/test_pregel_async.py
+++ b/tests/test_pregel_async.py
@@ -157,38 +157,34 @@ async def test_invoke_two_processes_in_out(mocker: MockerFixture) -> None:
 
     assert await app.ainvoke(2) == 4
 
-    async for output, view in app.astep(2):
-        if view.step == 1:
-            assert view.values == {
+    step = 0
+    async for values in app.astream(2):
+        step += 1
+        if step == 1:
+            assert values == {
                 "inbox": 3,
-                "input": 2,
             }
-            assert output == {"inbox": 3}
-        elif view.step == 2:
-            assert view.values == {
+        elif step == 2:
+            assert values == {
                 "output": 4,
-                "inbox": 3,
-                "input": 2,
             }
-            assert output == {"output": 4}
+    assert step == 2
 
-    async for output, view in app.astep(2):
-        if view.step == 1:
-            assert view.values == {
+    step = 0
+    async for values in app.astream(2):
+        step += 1
+        if step == 1:
+            assert values == {
                 "inbox": 3,
-                "input": 2,
             }
-            assert output == {"inbox": 3}
             # modify inbox value
-            view.values["inbox"] = 5
-        elif view.step == 2:
-            assert view.values == {
-                "output": 6,
-                "inbox": 5,
-                "input": 2,
-            }
+            values["inbox"] = 5
+        elif step == 2:
             # output is different now
-            assert output == {"output": 6}
+            assert values == {
+                "output": 6,
+            }
+    assert step == 2
 
     graph = Graph()
     graph.add_node("add_one", add_one)
@@ -200,38 +196,42 @@ async def test_invoke_two_processes_in_out(mocker: MockerFixture) -> None:
 
     assert await gapp.ainvoke(2) == 4
 
-    async for output, view in gapp.astep(2):
-        if view.step == 1:
-            assert view.values == {
-                "add_one": 2,
-                "add_one_more": 3,
+    step = 0
+    async for values in gapp.astream(2):
+        step += 1
+        if step == 1:
+            assert values == {
+                "add_one": 3,
             }
-            assert output == {"add_one_more": 3}
-        elif view.step == 2:
-            assert view.values == {
-                "add_one": 2,
-                "add_one_more": 3,
+        elif step == 2:
+            assert values == {
+                "add_one_more": 4,
+            }
+        elif step == 3:
+            assert values == {
                 "__end__": 4,
             }
-            assert output == {"__end__": 4}
+    assert step == 3
 
-    async for output, view in gapp.astep(2):
-        if view.step == 1:
-            assert view.values == {
-                "add_one": 2,
-                "add_one_more": 3,
+    step = 0
+    async for values in gapp.astream(2):
+        step += 1
+        if step == 1:
+            assert values == {
+                "add_one": 3,
             }
-            assert output == {"add_one_more": 3}
-            # modify inbox value
-            view.values["add_one_more"] = 5
-        elif view.step == 2:
-            assert view.values == {
-                "add_one": 2,
-                "add_one_more": 5,
+            # modify value before running next step
+            values["add_one"] = 5
+        elif step == 2:
+            # output is different now
+            assert values == {
+                "add_one_more": 6,
+            }
+        elif step == 3:
+            assert values == {
                 "__end__": 6,
             }
-            # output is different now
-            assert output == {"__end__": 6}
+    assert step == 3
 
 
 async def test_invoke_two_processes_in_dict_out(mocker: MockerFixture) -> None:
@@ -690,13 +690,10 @@ async def test_conditional_graph() -> None:
     }
 
     assert [
-        deepcopy(c)
-        async for c in app.astream(
-            {"input": "what is weather in sf"}, output=["agent", "tools"]
-        )
+        deepcopy(c) async for c in app.astream({"input": "what is weather in sf"})
     ] == [
         {
-            "tools": {
+            "agent": {
                 "input": "what is weather in sf",
                 "agent_outcome": AgentAction(
                     tool="search_api", tool_input="query", log="tool:search_api:query"
@@ -704,7 +701,7 @@ async def test_conditional_graph() -> None:
             }
         },
         {
-            "agent": {
+            "tools": {
                 "input": "what is weather in sf",
                 "intermediate_steps": [
                     (
@@ -719,7 +716,7 @@ async def test_conditional_graph() -> None:
             }
         },
         {
-            "tools": {
+            "agent": {
                 "input": "what is weather in sf",
                 "intermediate_steps": [
                     (
@@ -738,6 +735,29 @@ async def test_conditional_graph() -> None:
                 ),
             }
         },
+        {
+            "tools": {
+                "input": "what is weather in sf",
+                "intermediate_steps": [
+                    (
+                        AgentAction(
+                            tool="search_api",
+                            tool_input="query",
+                            log="tool:search_api:query",
+                        ),
+                        "result for query",
+                    ),
+                    (
+                        AgentAction(
+                            tool="search_api",
+                            tool_input="another",
+                            log="tool:search_api:another",
+                        ),
+                        "result for another",
+                    ),
+                ],
+            }
+        },
         {
             "agent": {
                 "input": "what is weather in sf",
@@ -759,6 +779,35 @@ async def test_conditional_graph() -> None:
                         "result for another",
                     ),
                 ],
+                "agent_outcome": AgentFinish(
+                    return_values={"answer": "answer"}, log="finish:answer"
+                ),
+            }
+        },
+        {
+            "__end__": {
+                "input": "what is weather in sf",
+                "intermediate_steps": [
+                    (
+                        AgentAction(
+                            tool="search_api",
+                            tool_input="query",
+                            log="tool:search_api:query",
+                        ),
+                        "result for query",
+                    ),
+                    (
+                        AgentAction(
+                            tool="search_api",
+                            tool_input="another",
+                            log="tool:search_api:another",
+                        ),
+                        "result for another",
+                    ),
+                ],
+                "agent_outcome": AgentFinish(
+                    return_values={"answer": "answer"}, log="finish:answer"
+                ),
             }
         },
     ]