fix(langgraph): keep the recursion budget and skip Send tasks when applying inserted input

libs/langgraph/langgraph/pregel/_loop.py

@@ -57,6 +57,7 @@ from langgraph._internal._constants import (
     NS_END,
     NS_SEP,
     NULL_TASK_ID,
+    PULL,
     PUSH,
     RESUME,
     TASKS,
@@ -816,9 +817,10 @@ class PregelLoop:
     def _accepts_inserted_input(self) -> bool:
         """Whether queued input can be applied before the next step.
 
-        Only in a top-level `StateGraph` with steps left, only when every next
-        task is a node declared in `insert_before`, and never on the step a
-        resume continues: the resume applies to the state it paused on.
+        Only in a top-level `StateGraph`, only when every next task is a node
+        declared in `insert_before` that reads the state (not a `Send`, which
+        gets its own argument), and never on the step a resume continues: the
+        resume applies to the state it paused on.
         """
         return (
             self.control is not None
@@ -826,9 +828,11 @@ class PregelLoop:
             and not self.is_nested
             and bool(self.insert_before)
             and bool(self.insert_channels)
-            and self.step < self.stop
             and not self.config[CONF].get(CONFIG_KEY_RESUMING, False)
-            and all(task.name in self.insert_before for task in self.tasks.values())
+            and all(
+                task.path[0] == PULL and task.name in self.insert_before
+                for task in self.tasks.values()
+            )
         )
 
     def _apply_inserted_input(self) -> bool:
@@ -878,6 +882,8 @@ class PregelLoop:
                 {"source": "input", "inserted_inputs": applied},
             )
         )
+        # the checkpoint took a step, but no node ran: keep the recursion budget
+        self.stop += 1
         return True
 
     def _inserted_writes(self, input_id: str, value: Any) -> list[tuple[str, Any]]:

libs/langgraph/tests/test_inserted_input.py

@@ -8,7 +8,7 @@ from typing_extensions import TypedDict
 from langgraph.errors import GraphRecursionError
 from langgraph.graph import END, START, StateGraph
 from langgraph.runtime import RunControl
-from langgraph.types import Command, interrupt
+from langgraph.types import Command, Send, interrupt
 
 
 class State(TypedDict, total=False):
@@ -293,3 +293,41 @@ def test_inserted_input_is_not_applied_without_steps_left() -> None:
         graph.invoke({"log": ["user"]}, {"recursion_limit": 1}, control=control)
 
     assert control.take_inputs() == [("insert-1", {"log": ["inserted"]})]
+
+
+def test_inserted_input_keeps_the_recursion_budget() -> None:
+    control = RunControl()
+    graph = _tools_then_model(control).compile(insert_before=["model"])
+    # the smallest limit the graph runs within without inserted input
+    config = {"recursion_limit": 3}
+    _tools_then_model(RunControl()).compile().invoke({"log": ["user"]}, config)
+
+    result = graph.invoke({"log": ["user"]}, config, control=control)
+
+    assert result["seen"] == ["user", "tools", "inserted"]
+
+
+def test_inserted_input_waits_for_a_node_started_by_send() -> None:
+    control = RunControl()
+
+    def tools(state: State) -> State:
+        control.submit_input("insert-1", {"log": ["inserted"]})
+        return {"log": ["tools"]}
+
+    def model(state: State) -> State:
+        return {"seen": list(state["log"])}
+
+    builder = StateGraph(State)
+    builder.add_node("tools", tools)
+    builder.add_node("model", model)
+    builder.add_edge(START, "tools")
+    builder.add_conditional_edges("tools", lambda _: [Send("model", {"log": ["sent"]})])
+    builder.add_edge("model", END)
+
+    result = builder.compile(insert_before=["model"]).invoke(
+        {"log": ["user"]}, control=control
+    )
+
+    # `model` reads its `Send` argument, so the input would never reach it
+    assert result["seen"] == ["sent"]
+    assert control.take_inputs() == [("insert-1", {"log": ["inserted"]})]