diff --git a/tests/test_pregel.py b/tests/test_pregel.py
index 2a8025f9e..d12ce4898 100644
--- a/tests/test_pregel.py
+++ b/tests/test_pregel.py
@@ -740,7 +740,7 @@ def test_invoke_two_processes_two_in_join_two_out(mocker: MockerFixture) -> None
         assert [*executor.map(app.invoke, [2] * 100)] == [[13, 13]] * 100
 
 
-def test_invoke_join_then_call_other_app(mocker: MockerFixture) -> None:
+def test_invoke_join_then_call_other_pregel(mocker: MockerFixture) -> None:
     add_one = mocker.Mock(side_effect=lambda x: x + 1)
     add_10_each = mocker.Mock(side_effect=lambda x: [y + 10 for y in x])
 
diff --git a/tests/test_pregel_async.py b/tests/test_pregel_async.py
index 1196de6d6..238dfb5e9 100644
--- a/tests/test_pregel_async.py
+++ b/tests/test_pregel_async.py
@@ -80,19 +80,6 @@ async def test_invoke_single_process_in_out_falsy_values(falsy_value: Any) -> No
     assert falsy_value == await gapp.ainvoke(1)
 
 
-async def test_invoke_single_process_in_out_implicit_channels(
-    mocker: MockerFixture,
-) -> None:
-    add_one = mocker.Mock(side_effect=lambda x: x + 1)
-    chain = Channel.subscribe_to("input") | add_one | Channel.write_to("output")
-
-    app = Pregel(nodes={"one": chain})
-
-    assert app.input_schema.schema() == {"title": "LangGraphInput"}
-    assert app.output_schema.schema() == {"title": "LangGraphOutput"}
-    assert await app.ainvoke(2) == 3
-
-
 async def test_invoke_single_process_in_write_kwargs(mocker: MockerFixture) -> None:
     add_one = mocker.Mock(side_effect=lambda x: x + 1)
     chain = (
@@ -102,17 +89,25 @@ async def test_invoke_single_process_in_write_kwargs(mocker: MockerFixture) -> N
     )
 
     app = Pregel(
-        nodes={"one": chain}, output_channels=["output", "fixed", "output_plus_one"]
+        nodes={"one": chain},
+        channels={
+            "input": LastValue(int),
+            "output": LastValue(int),
+            "fixed": LastValue(int),
+            "output_plus_one": LastValue(int),
+        },
+        output_channels=["output", "fixed", "output_plus_one"],
+        input_channels="input",
     )
 
-    assert app.input_schema.schema() == {"title": "LangGraphInput"}
+    assert app.input_schema.schema() == {"title": "LangGraphInput", "type": "integer"}
     assert app.output_schema.schema() == {
         "title": "LangGraphOutput",
         "type": "object",
         "properties": {
-            "output": {"title": "Output"},
-            "fixed": {"title": "Fixed"},
-            "output_plus_one": {"title": "Output Plus One"},
+            "output": {"title": "Output", "type": "integer"},
+            "fixed": {"title": "Fixed", "type": "integer"},
+            "output_plus_one": {"title": "Output Plus One", "type": "integer"},
         },
     }
     assert await app.ainvoke(2) == {"output": 3, "fixed": 5, "output_plus_one": 4}
@@ -124,14 +119,16 @@ async def test_invoke_single_process_in_out_dict(mocker: MockerFixture) -> None:
 
     app = Pregel(
         nodes={"one": chain},
+        channels={"input": LastValue(int), "output": LastValue(int)},
+        input_channels="input",
         output_channels=["output"],
     )
 
-    assert app.input_schema.schema() == {"title": "LangGraphInput"}
+    assert app.input_schema.schema() == {"title": "LangGraphInput", "type": "integer"}
     assert app.output_schema.schema() == {
         "title": "LangGraphOutput",
         "type": "object",
-        "properties": {"output": {"title": "Output"}},
+        "properties": {"output": {"title": "Output", "type": "integer"}},
     }
     assert await app.ainvoke(2) == {"output": 3}
 
@@ -141,9 +138,8 @@ async def test_invoke_single_process_in_dict_out_dict(mocker: MockerFixture) ->
     chain = Channel.subscribe_to("input") | add_one | Channel.write_to("output")
 
     app = Pregel(
-        nodes={
-            "one": chain,
-        },
+        nodes={"one": chain},
+        channels={"input": LastValue(int), "output": LastValue(int)},
         input_channels=["input"],
         output_channels=["output"],
     )
@@ -151,12 +147,12 @@ async def test_invoke_single_process_in_dict_out_dict(mocker: MockerFixture) ->
     assert app.input_schema.schema() == {
         "title": "LangGraphInput",
         "type": "object",
-        "properties": {"input": {"title": "Input"}},
+        "properties": {"input": {"title": "Input", "type": "integer"}},
     }
     assert app.output_schema.schema() == {
         "title": "LangGraphOutput",
         "type": "object",
-        "properties": {"output": {"title": "Output"}},
+        "properties": {"output": {"title": "Output", "type": "integer"}},
     }
     assert await app.ainvoke({"input": 2}) == {"output": 3}
 
@@ -166,7 +162,17 @@ async def test_invoke_two_processes_in_out(mocker: MockerFixture) -> None:
     one = Channel.subscribe_to("input") | add_one | Channel.write_to("inbox")
     two = Channel.subscribe_to("inbox") | add_one | Channel.write_to("output")
 
-    app = Pregel(nodes={"one": one, "two": two}, stream_channels=["inbox", "output"])
+    app = Pregel(
+        nodes={"one": one, "two": two},
+        channels={
+            "inbox": LastValue(int),
+            "output": LastValue(int),
+            "input": LastValue(int),
+        },
+        input_channels="input",
+        output_channels="output",
+        stream_channels=["inbox", "output"],
+    )
 
     assert await app.ainvoke(2) == 4
 
@@ -226,6 +232,13 @@ async def test_invoke_two_processes_in_out_interrupt(
     memory = MemorySaverAssertImmutable(at=checkpoint_at)
     app = Pregel(
         nodes={"one": one, "two": two},
+        channels={
+            "inbox": LastValue(int),
+            "output": LastValue(int),
+            "input": LastValue(int),
+        },
+        input_channels="input",
+        output_channels="output",
         checkpointer=memory,
         interrupt_after_nodes=["one"],
     )
@@ -281,9 +294,14 @@ async def test_invoke_two_processes_in_dict_out(mocker: MockerFixture) -> None:
 
     app = Pregel(
         nodes={"one": one, "two": two},
-        channels={"inbox": Topic(int)},
+        channels={
+            "inbox": Topic(int),
+            "output": LastValue(int),
+            "input": LastValue(int),
+        },
         input_channels=["input", "inbox"],
-        stream_channels=["inbox", "output"],
+        stream_channels=["output", "inbox"],
+        output_channels=["output"],
     )
 
     # [12 + 1, 2 + 1 + 1]
@@ -322,7 +340,13 @@ async def test_batch_two_processes_in_out() -> None:
 
     app = Pregel(
         nodes={"one": one, "two": two},
-        channels={"one": LastValue(int)},
+        channels={
+            "one": LastValue(int),
+            "output": LastValue(int),
+            "input": LastValue(int),
+        },
+        input_channels="input",
+        output_channels="output",
     )
 
     assert await app.abatch([3, 2, 1, 3, 5]) == [5, 4, 3, 5, 7]
@@ -356,7 +380,13 @@ async def test_invoke_many_processes_in_out(mocker: MockerFixture) -> None:
         )
     nodes["last"] = Channel.subscribe_to(str(i)) | add_one | Channel.write_to("output")
 
-    app = Pregel(nodes=nodes)
+    app = Pregel(
+        nodes=nodes,
+        channels={str(i): LastValue(int) for i in range(-1, test_size - 2)}
+        | {"input": LastValue(int), "output": LastValue(int)},
+        input_channels="input",
+        output_channels="output",
+    )
 
     # No state is left over from previous invocations
     for _ in range(10):
@@ -379,7 +409,13 @@ async def test_batch_many_processes_in_out(mocker: MockerFixture) -> None:
         )
     nodes["last"] = Channel.subscribe_to(str(i)) | add_one | Channel.write_to("output")
 
-    app = Pregel(nodes=nodes)
+    app = Pregel(
+        nodes=nodes,
+        channels={str(i): LastValue(int) for i in range(-1, test_size - 2)}
+        | {"input": LastValue(int), "output": LastValue(int)},
+        input_channels="input",
+        output_channels="output",
+    )
 
     # No state is left over from previous invocations
     for _ in range(3):
@@ -409,7 +445,12 @@ async def test_invoke_two_processes_two_in_two_out_invalid(
     one = Channel.subscribe_to("input") | add_one | Channel.write_to("output")
     two = Channel.subscribe_to("input") | add_one | Channel.write_to("output")
 
-    app = Pregel(nodes={"one": one, "two": two})
+    app = Pregel(
+        nodes={"one": one, "two": two},
+        channels={"output": LastValue(int), "input": LastValue(int)},
+        input_channels="input",
+        output_channels="output",
+    )
 
     with pytest.raises(InvalidUpdateError):
         # LastValue channels can only be updated once per iteration
@@ -424,7 +465,12 @@ async def test_invoke_two_processes_two_in_two_out_valid(mocker: MockerFixture)
 
     app = Pregel(
         nodes={"one": one, "two": two},
-        channels={"output": Topic(int)},
+        channels={
+            "input": LastValue(int),
+            "output": Topic(int),
+        },
+        input_channels="input",
+        output_channels="output",
     )
 
     # An Topic channel accumulates updates into a sequence
@@ -455,7 +501,13 @@ async def test_invoke_checkpoint(
 
     app = Pregel(
         nodes={"one": one},
-        channels={"total": BinaryOperatorAggregate(int, operator.add)},
+        channels={
+            "total": BinaryOperatorAggregate(int, operator.add),
+            "input": LastValue(int),
+            "output": LastValue(int),
+        },
+        input_channels="input",
+        output_channels="output",
         checkpointer=memory,
     )
 
@@ -510,7 +562,13 @@ async def test_invoke_checkpoint_aiosqlite(
         memory.at = checkpoint_at
         app = Pregel(
             nodes={"one": one},
-            channels={"total": BinaryOperatorAggregate(int, operator.add)},
+            channels={
+                "total": BinaryOperatorAggregate(int, operator.add),
+                "input": LastValue(int),
+                "output": LastValue(int),
+            },
+            input_channels="input",
+            output_channels="output",
             checkpointer=memory,
             debug=True,
         )
@@ -605,7 +663,13 @@ async def test_invoke_two_processes_two_in_join_two_out(mocker: MockerFixture) -
             "chain_three": chain_three,
             "chain_four": chain_four,
         },
-        channels={"inbox": Topic(int)},
+        channels={
+            "inbox": Topic(int),
+            "output": LastValue(int),
+            "input": LastValue(int),
+        },
+        input_channels="input",
+        output_channels="output",
     )
 
     # Then invoke app
@@ -619,14 +683,20 @@ async def test_invoke_two_processes_two_in_join_two_out(mocker: MockerFixture) -
     ]
 
 
-async def test_invoke_join_then_call_other_pubsub(mocker: MockerFixture) -> None:
+async def test_invoke_join_then_call_other_pregel(mocker: MockerFixture) -> None:
     add_one = mocker.Mock(side_effect=lambda x: x + 1)
     add_10_each = mocker.Mock(side_effect=lambda x: [y + 10 for y in x])
 
     inner_app = Pregel(
         nodes={
             "one": Channel.subscribe_to("input") | add_one | Channel.write_to("output")
-        }
+        },
+        channels={
+            "output": LastValue(int),
+            "input": LastValue(int),
+        },
+        input_channels="input",
+        output_channels="output",
     )
 
     one = (
@@ -651,7 +721,11 @@ async def test_invoke_join_then_call_other_pubsub(mocker: MockerFixture) -> None
         channels={
             "inbox_one": Topic(int),
             "outbox_one": LastValue(int),
+            "output": LastValue(int),
+            "input": LastValue(int),
         },
+        input_channels="input",
+        output_channels="output",
     )
 
     # Then invoke pubsub
@@ -673,7 +747,17 @@ async def test_invoke_two_processes_one_in_two_out(mocker: MockerFixture) -> Non
     )
     two = Channel.subscribe_to("between") | add_one | Channel.write_to("output")
 
-    app = Pregel(nodes={"one": one, "two": two}, stream_channels=["output", "between"])
+    app = Pregel(
+        nodes={"one": one, "two": two},
+        channels={
+            "input": LastValue(int),
+            "between": LastValue(int),
+            "output": LastValue(int),
+        },
+        stream_channels=["output", "between"],
+        input_channels="input",
+        output_channels="output",
+    )
 
     # Then invoke pubsub
     assert [c async for c in app.astream(2)] == [
@@ -687,7 +771,16 @@ async def test_invoke_two_processes_no_out(mocker: MockerFixture) -> None:
     one = Channel.subscribe_to("input") | add_one | Channel.write_to("between")
     two = Channel.subscribe_to("between") | add_one
 
-    app = Pregel(nodes={"one": one, "two": two})
+    app = Pregel(
+        nodes={"one": one, "two": two},
+        channels={
+            "input": LastValue(int),
+            "between": LastValue(int),
+            "output": LastValue(int),
+        },
+        input_channels="input",
+        output_channels="output",
+    )
 
     # It finishes executing (once no more messages being published)
     # but returns nothing, as nothing was published to "output" topic
@@ -727,9 +820,12 @@ async def test_channel_enter_exit_timing(mocker: MockerFixture) -> None:
     app = Pregel(
         nodes={"one": one, "two": two},
         channels={
+            "input": LastValue(int),
+            "output": LastValue(int),
             "inbox": Topic(int),
             "ctx": Context(an_int, an_int_async, typ=int),
         },
+        input_channels="input",
         output_channels=["inbox", "output"],
         stream_channels=["inbox", "output"],
     )
@@ -3563,3 +3659,36 @@ async def test_nested_graph(snapshot: SnapshotAssertion) -> None:
                 ]
             }
     assert times_called == 1
+
+    chain = app | RunnablePassthrough()
+
+    assert await chain.ainvoke(
+        {"my_key": "my value", "never_called": never_called}
+    ) == {
+        "my_key": "my value there and back again",
+        "never_called": never_called,
+    }
+    assert [
+        chunk
+        async for chunk in chain.astream(
+            {"my_key": "my value", "never_called": never_called}
+        )
+    ] == [
+        {"inner": {"my_key": "my value there"}},
+        {"side": {"my_key": "my value there and back again"}},
+    ]
+    times_called = 0
+    async for event in chain.astream_events(
+        {"my_key": "my value", "never_called": never_called},
+        version="v1",
+        config={"run_id": UUID(int=0)},
+    ):
+        if event["event"] == "on_chain_end" and event["run_id"] == str(UUID(int=0)):
+            times_called += 1
+            assert event["data"] == {
+                "output": [
+                    {"inner": {"my_key": "my value there"}},
+                    {"side": {"my_key": "my value there and back again"}},
+                ]
+            }
+    assert times_called == 1