feat(langgraph): apply input inserted into a running run at declared safe steps

libs/langgraph/langgraph/graph/state.py

@@ -1185,6 +1185,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
         debug: bool = False,
         name: str | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
+        insert_before: Sequence[str] | None = None,
     ) -> CompiledStateGraph[StateT, ContextT, InputT, OutputT]:
         """Compiles the `StateGraph` into a `CompiledStateGraph` object.
 
@@ -1224,6 +1225,12 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
                 the standard `StreamTransformer` constructor shape by
                 accepting `scope` as their first argument. Appended after the
                 built-in stream transformers.
+            insert_before: Nodes before which input from `RunControl.submit_input`
+                may be applied mid-run, e.g. an agent's model node.
+
+                !!! warning "Experimental"
+
+                    This parameter may change without notice.
 
         Returns:
             CompiledStateGraph: The compiled `StateGraph`.
@@ -1256,6 +1263,13 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
         # assign default values
         interrupt_before = interrupt_before or []
         interrupt_after = interrupt_after or []
+        if isinstance(insert_before, str):
+            msg = "`insert_before` takes a sequence of node names, not a string"
+            raise TypeError(msg)
+        insert_before = list(insert_before or ())
+        if unknown := [node for node in insert_before if node not in self.nodes]:
+            msg = f"`insert_before` names nodes not in the graph: {unknown}"
+            raise ValueError(msg)
 
         # validate the graph
         self.validate(
@@ -1360,6 +1374,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
             checkpointer=checkpointer,
             interrupt_before_nodes=interrupt_before,
             interrupt_after_nodes=interrupt_after,
+            insert_before_nodes=insert_before,
             auto_validate=False,
             debug=debug,
             store=store,
@@ -1441,6 +1456,15 @@ class CompiledStateGraph(
             name=self.get_name("Output"),
         )
 
+    def _inserted_input_channels(self) -> Sequence[str]:
+        """The graph's input keys: inserted input may write only these."""
+        keys = [
+            key
+            for key, value in self.builder.schemas[self.builder.input_schema].items()
+            if not is_managed_value(value)
+        ]
+        return () if keys == ["__root__"] else keys
+
     def attach_node(self, key: str, node: StateNodeSpec[Any, ContextT] | None) -> None:
         if key == START:
             output_keys = [

libs/langgraph/langgraph/pregel/_loop.py

@@ -122,6 +122,7 @@ from langgraph.pregel._io import (
     map_output_values,
     read_channels,
 )
+from langgraph.pregel._log import logger
 from langgraph.pregel._messages import ensure_message_ids
 from langgraph.pregel._read import PregelNode
 from langgraph.pregel._task_status import read_task_statuses
@@ -187,6 +188,8 @@ class PregelLoop:
     manager: None | AsyncParentRunManager | ParentRunManager
     interrupt_after: All | Sequence[str]
     interrupt_before: All | Sequence[str]
+    insert_before: Sequence[str]
+    insert_channels: Sequence[str]
     durability: Durability
     retry_policy: Sequence[RetryPolicy]
     cache_policy: CachePolicy | None
@@ -326,6 +329,8 @@ class PregelLoop:
         durability: Durability,
         interrupt_after: All | Sequence[str] = EMPTY_SEQ,
         interrupt_before: All | Sequence[str] = EMPTY_SEQ,
+        insert_before: Sequence[str] = EMPTY_SEQ,
+        insert_channels: Sequence[str] = EMPTY_SEQ,
         manager: None | AsyncParentRunManager | ParentRunManager = None,
         migrate_checkpoint: Callable[[Checkpoint], None] | None = None,
         retry_policy: Sequence[RetryPolicy] = (),
@@ -347,6 +352,8 @@ class PregelLoop:
         self.stream_keys = stream_keys
         self.interrupt_after = interrupt_after
         self.interrupt_before = interrupt_before
+        self.insert_before = insert_before
+        self.insert_channels = insert_channels
         self.manager = manager
         self.is_nested = CONFIG_KEY_TASK_ID in self.config.get(CONF, {})
         self.is_replaying = CONFIG_KEY_CHECKPOINT_ID in config[CONF]
@@ -707,6 +714,11 @@ class PregelLoop:
             self.status = "draining"
             return False
 
+        # apply input inserted into the running run, then plan the step again
+        # from the new state
+        if self._accepts_inserted_input() and self._apply_inserted_input():
+            return self.tick()
+
         # produce debug output
         self._emit("tasks", map_debug_tasks, self.tasks.values())
 
@@ -801,6 +813,105 @@ class PregelLoop:
 
     # private
 
+    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.
+        """
+        return (
+            self.control is not None
+            and self.control.has_inputs
+            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())
+        )
+
+    def _apply_inserted_input(self) -> bool:
+        """Apply queued input to the state and save it as an `input` checkpoint.
+
+        Each input gets its own write, so two that set the same plain key don't
+        raise; the later one wins. Returns whether any was applied, so `tick()`
+        plans the step again.
+        """
+        applied: list[str] = []
+        writes: list[tuple[str, Any]] = []
+        inserted_channels: set[str] = set()
+        for input_id, value in cast(RunControl, self.control).take_inputs():
+            if not (input_writes := self._inserted_writes(input_id, value)):
+                continue
+            inserted_channels |= apply_writes(
+                self.checkpoint,
+                self.channels,
+                [PregelTaskWrites((), INPUT, input_writes, [])],
+                self.checkpointer_get_next_version,
+                self.trigger_to_nodes,
+            )
+            writes.extend(input_writes)
+            applied.append(input_id)
+        if not applied:
+            return False
+        # snapshot delta channels this input wrote, so they read back without
+        # their writes stored separately
+        self._delta_channels_forced_snapshot.update(
+            channel
+            for channel, _ in writes
+            if isinstance(self.specs.get(channel), DeltaChannel)
+        )
+        # add to the last step's updated channels, which trigger the planned tasks
+        self.updated_channels = (self.updated_channels or set()) | inserted_channels
+        if not inserted_channels.isdisjoint(
+            (self.output_keys,)
+            if isinstance(self.output_keys, str)
+            else self.output_keys
+        ):
+            self._emit(
+                "values", map_output_values, self.output_keys, writes, self.channels
+            )
+        self._put_checkpoint(
+            cast(
+                CheckpointMetadata,
+                {"source": "input", "inserted_inputs": applied},
+            )
+        )
+        return True
+
+    def _inserted_writes(self, input_id: str, value: Any) -> list[tuple[str, Any]]:
+        """State updates for inserted input, or none if the state rejects it.
+
+        Written straight to the state channels: through `__start__`, the graph
+        would route back to its entry.
+        """
+        if not isinstance(value, dict):
+            logger.warning(
+                "Skipping inserted input %s: expected a dict of state keys, got %s",
+                input_id,
+                type(value).__name__,
+            )
+            return []
+        if ignored := [key for key in value if key not in self.insert_channels]:
+            logger.warning(
+                "Ignoring keys of inserted input %s that aren't graph input keys: %s",
+                input_id,
+                ignored,
+            )
+        writes = [(k, v) for k, v in value.items() if k in self.insert_channels]
+        try:
+            for key, val in writes:
+                self.channels[key].copy().update([val])
+        except Exception:
+            logger.warning(
+                "Skipping inserted input %s: the state rejected it",
+                input_id,
+                exc_info=True,
+            )
+            return []
+        return writes
+
     @property
     def _reapplies_pending_writes(self) -> bool:
         """Whether the writes loaded with the checkpoint go back to the tasks
@@ -1638,6 +1749,8 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
         manager: None | AsyncParentRunManager | ParentRunManager = None,
         interrupt_after: All | Sequence[str] = EMPTY_SEQ,
         interrupt_before: All | Sequence[str] = EMPTY_SEQ,
+        insert_before: Sequence[str] = EMPTY_SEQ,
+        insert_channels: Sequence[str] = EMPTY_SEQ,
         input_keys: str | Sequence[str] = EMPTY_SEQ,
         output_keys: str | Sequence[str] = EMPTY_SEQ,
         stream_keys: str | Sequence[str] = EMPTY_SEQ,
@@ -1660,6 +1773,8 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             stream_keys=stream_keys,
             interrupt_after=interrupt_after,
             interrupt_before=interrupt_before,
+            insert_before=insert_before,
+            insert_channels=insert_channels,
             manager=manager,
             migrate_checkpoint=migrate_checkpoint,
             trigger_to_nodes=trigger_to_nodes,
@@ -1893,6 +2008,8 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
         durability: Durability,
         interrupt_after: All | Sequence[str] = EMPTY_SEQ,
         interrupt_before: All | Sequence[str] = EMPTY_SEQ,
+        insert_before: Sequence[str] = EMPTY_SEQ,
+        insert_channels: Sequence[str] = EMPTY_SEQ,
         manager: None | AsyncParentRunManager | ParentRunManager = None,
         input_keys: str | Sequence[str] = EMPTY_SEQ,
         output_keys: str | Sequence[str] = EMPTY_SEQ,
@@ -1916,6 +2033,8 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             stream_keys=stream_keys,
             interrupt_after=interrupt_after,
             interrupt_before=interrupt_before,
+            insert_before=insert_before,
+            insert_channels=insert_channels,
             manager=manager,
             migrate_checkpoint=migrate_checkpoint,
             trigger_to_nodes=trigger_to_nodes,

libs/langgraph/langgraph/pregel/main.py

@@ -732,6 +732,14 @@ class Pregel(
 
     interrupt_before_nodes: All | Sequence[str]
 
+    insert_before_nodes: Sequence[str] = ()
+    """Nodes before which input inserted into a running run may be applied.
+
+    !!! warning "Experimental"
+
+        This attribute may change without notice.
+    """
+
     input_channels: str | Sequence[str]
 
     step_timeout: float | None = None
@@ -777,6 +785,7 @@ class Pregel(
         stream_channels: str | Sequence[str] | None = None,
         interrupt_after_nodes: All | Sequence[str] = (),
         interrupt_before_nodes: All | Sequence[str] = (),
+        insert_before_nodes: Sequence[str] = (),
         input_channels: str | Sequence[str],
         step_timeout: float | None = None,
         debug: bool | None = None,
@@ -823,6 +832,7 @@ class Pregel(
         self.stream_channels = stream_channels
         self.interrupt_after_nodes = interrupt_after_nodes
         self.interrupt_before_nodes = interrupt_before_nodes
+        self.insert_before_nodes = insert_before_nodes
         self.input_channels = input_channels
         self.step_timeout = step_timeout
         self.debug = debug if debug is not None else get_debug()
@@ -987,6 +997,10 @@ class Pregel(
             "image/png": self.get_graph().draw_mermaid_png(),
         }
 
+    def _inserted_input_channels(self) -> Sequence[str]:
+        """Channels inserted input may write; none for a plain `Pregel`."""
+        return ()
+
     def copy(self, update: dict[str, Any] | None = None) -> Self:
         attrs = {k: v for k, v in self.__dict__.items() if k != "__orig_class__"}
         attrs.update(update or {})
@@ -2802,7 +2816,8 @@ class Pregel(
                 - `"sync"`: Changes are persisted synchronously before the next step starts.
                 - `"async"`: Changes are persisted asynchronously while the next step executes.
                 - `"exit"`: Changes are persisted only when the graph exits.
-            control: Optional run control used to request cooperative drain.
+            control: Optional run control used to request cooperative drain or to
+                insert input into the run (see `RunControl`).
             subgraphs: Whether to stream events from inside subgraphs, defaults to `False`.
 
                 If `True`, the events will be emitted as tuples `(namespace, data)`,
@@ -3001,6 +3016,8 @@ class Pregel(
                 stream_keys=self.stream_channels_asis,
                 interrupt_before=interrupt_before_,
                 interrupt_after=interrupt_after_,
+                insert_before=self.insert_before_nodes,
+                insert_channels=self._inserted_input_channels(),
                 manager=run_manager,
                 durability=durability_,
                 trigger_to_nodes=self.trigger_to_nodes,
@@ -3208,7 +3225,8 @@ class Pregel(
                 - `"sync"`: Changes are persisted synchronously before the next step starts.
                 - `"async"`: Changes are persisted asynchronously while the next step executes.
                 - `"exit"`: Changes are persisted only when the graph exits.
-            control: Optional run control used to request cooperative drain.
+            control: Optional run control used to request cooperative drain or to
+                insert input into the run (see `RunControl`).
             subgraphs: Whether to stream events from inside subgraphs, defaults to `False`.
 
                 If `True`, the events will be emitted as tuples `(namespace, data)`,
@@ -3453,6 +3471,8 @@ class Pregel(
                 stream_keys=self.stream_channels_asis,
                 interrupt_before=interrupt_before_,
                 interrupt_after=interrupt_after_,
+                insert_before=self.insert_before_nodes,
+                insert_channels=self._inserted_input_channels(),
                 manager=run_manager,
                 durability=durability_,
                 trigger_to_nodes=self.trigger_to_nodes,
@@ -3958,7 +3978,8 @@ class Pregel(
                 - `"sync"`: Changes are persisted synchronously before the next step starts.
                 - `"async"`: Changes are persisted asynchronously while the next step executes.
                 - `"exit"`: Changes are persisted only when the graph exits.
-            control: Optional run control used to request cooperative drain.
+            control: Optional run control used to request cooperative drain or to
+                insert input into the run (see `RunControl`).
             version: The streaming format version. `"v1"` (default) returns the
                 traditional format, `"v2"` returns `StreamPart` typed dicts when
                 `stream_mode` is not `"values"`.
@@ -4135,7 +4156,8 @@ class Pregel(
                 - `"sync"`: Changes are persisted synchronously before the next step starts.
                 - `"async"`: Changes are persisted asynchronously while the next step executes.
                 - `"exit"`: Changes are persisted only when the graph exits.
-            control: Optional run control used to request cooperative drain.
+            control: Optional run control used to request cooperative drain or to
+                insert input into the run (see `RunControl`).
             version: The streaming format version. `"v1"` (default) returns the
                 traditional format, `"v2"` returns `StreamPart` typed dicts when
                 `stream_mode` is not `"values"`.

libs/langgraph/langgraph/runtime.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import threading
 from collections.abc import Callable
 from dataclasses import dataclass, field, replace
 from typing import Any, Generic, cast
@@ -77,24 +78,66 @@ class ServerInfo:
 
 
 class RunControl:
-    """Run-scoped control surface for cooperative draining.
+    """Run-scoped control surface for cooperative draining and inserted input.
 
     Intended for a single graph run. Create a fresh `RunControl` per run;
     reusing a control after `request_drain()` leaves it drained.
 
     Safe to call from any thread: the drain request is represented by a
-    single attribute write, so no lock is needed for this signal.
-    If more mutable state is added here, add synchronization.
+    single attribute write, so no lock is needed for this signal. Inserted
+    input is guarded by a lock.
     """
 
-    __slots__ = ("_drain_reason",)
+    __slots__ = ("_drain_reason", "_inputs", "_inputs_lock")
 
     def __init__(self) -> None:
         self._drain_reason: str | None = None
+        self._inputs: list[tuple[str, Any]] = []
+        self._inputs_lock = threading.Lock()
 
     def request_drain(self, reason: str = "shutdown") -> None:
         self._drain_reason = reason
 
+    def submit_input(self, input_id: str, value: Any) -> None:
+        """Queue input for the running graph to apply at its next safe step boundary.
+
+        !!! warning "Experimental"
+
+            This method may change without notice.
+
+        The graph applies queued input only right before the nodes in its
+        `insert_before`, the points it declares safe (for an agent, before the
+        next model call, so input never lands between a tool call and its
+        result). Only the top-level graph applies it, since a subgraph's state
+        isn't the caller's, and never on the step a resume continues, since the
+        resume answers the state it paused on. Input it can't apply (not a dict
+        of the graph's input keys, or a value the state rejects) is skipped and
+        logged rather than failing the run. Input the run never reaches stays
+        queued, so the caller can run it afterwards.
+
+        Args:
+            input_id: Recorded in the `inserted_inputs` metadata of the
+                checkpoint that applied it.
+            value: A dict of the graph's input keys.
+        """
+        with self._inputs_lock:
+            self._inputs.append((input_id, value))
+
+    @property
+    def has_inputs(self) -> bool:
+        """Whether input is queued and not yet applied."""
+        return bool(self._inputs)
+
+    def take_inputs(self) -> list[tuple[str, Any]]:
+        """Remove and return queued input in the order it was submitted.
+
+        Returns:
+            `(input_id, value)` pairs.
+        """
+        with self._inputs_lock:
+            inputs, self._inputs = self._inputs, []
+        return inputs
+
     @property
     def drain_requested(self) -> bool:
         return self._drain_reason is not None

libs/langgraph/tests/test_inserted_input.py

@@ -0,0 +1,295 @@
+import operator
+from typing import Annotated
+
+import pytest
+from langgraph.checkpoint.memory import InMemorySaver
+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
+
+
+class State(TypedDict, total=False):
+    log: Annotated[list[str], operator.add]
+    seen: list[str]
+
+
+def _tools_then_model(control: RunControl) -> StateGraph:
+    """`tools` queues input while it runs; `model` records what it sees."""
+
+    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"])}
+
+    graph = StateGraph(State)
+    graph.add_node("tools", tools)
+    graph.add_node("model", model)
+    graph.add_edge(START, "tools")
+    graph.add_edge("tools", "model")
+    graph.add_edge("model", END)
+    return graph
+
+
+def test_inserted_input_is_applied_before_an_allowed_node() -> None:
+    control = RunControl()
+    checkpointer = InMemorySaver()
+    graph = _tools_then_model(control).compile(
+        checkpointer=checkpointer, insert_before=["model"]
+    )
+    config = {"configurable": {"thread_id": "1"}}
+
+    result = graph.invoke({"log": ["user"]}, config, control=control)
+
+    assert result["seen"] == ["user", "tools", "inserted"]
+    assert not control.has_inputs
+    inserted = [
+        snapshot.metadata
+        for snapshot in graph.get_state_history(config)
+        if snapshot.metadata.get("inserted_inputs")
+    ]
+    assert len(inserted) == 1
+    assert inserted[0]["source"] == "input"
+    assert inserted[0]["inserted_inputs"] == ["insert-1"]
+
+
+@pytest.mark.anyio
+async def test_inserted_input_is_applied_before_an_allowed_node_async() -> None:
+    control = RunControl()
+    graph = _tools_then_model(control).compile(
+        checkpointer=InMemorySaver(), insert_before=["model"]
+    )
+
+    result = await graph.ainvoke(
+        {"log": ["user"]}, {"configurable": {"thread_id": "1"}}, control=control
+    )
+
+    assert result["seen"] == ["user", "tools", "inserted"]
+    assert not control.has_inputs
+
+
+def test_inserted_input_is_applied_without_a_checkpointer() -> None:
+    control = RunControl()
+    graph = _tools_then_model(control).compile(insert_before=["model"])
+
+    result = graph.invoke({"log": ["user"]}, control=control)
+
+    assert result["seen"] == ["user", "tools", "inserted"]
+
+
+@pytest.mark.parametrize("insert_before", [None, ["tools"]])
+def test_inserted_input_waits_when_the_next_node_is_not_allowed(
+    insert_before: list[str] | None,
+) -> None:
+    control = RunControl()
+    graph = _tools_then_model(control).compile(insert_before=insert_before)
+
+    result = graph.invoke({"log": ["user"]}, control=control)
+
+    assert result["seen"] == ["user", "tools"]
+    assert control.take_inputs() == [("insert-1", {"log": ["inserted"]})]
+
+
+def _outer(node: object) -> StateGraph:
+    """Runs `node` (which runs a subgraph), then `finish`."""
+
+    def finish(state: State) -> State:
+        return {"log": ["finished"]}
+
+    outer = StateGraph(State)
+    outer.add_node("outer", node)
+    outer.add_node("finish", finish)
+    outer.add_edge(START, "outer")
+    outer.add_edge("outer", "finish")
+    outer.add_edge("finish", END)
+    return outer
+
+
+def test_inserted_input_is_not_applied_inside_a_subgraph() -> None:
+    control = RunControl()
+    inner = _tools_then_model(control).compile(insert_before=["model"])
+
+    def outer_node(state: State) -> State:
+        return inner.invoke({"log": state["log"]})
+
+    result = (
+        _outer(outer_node)
+        .compile(insert_before=["finish"])
+        .invoke({"log": ["user"]}, control=control)
+    )
+
+    # the subgraph's model didn't see it; the parent applied it before `finish`
+    assert result["seen"] == ["user", "tools"]
+    assert result["log"][-2:] == ["inserted", "finished"]
+
+
+@pytest.mark.anyio
+async def test_inserted_input_is_not_applied_inside_a_subgraph_async() -> None:
+    control = RunControl()
+    inner = _tools_then_model(control).compile(insert_before=["model"])
+
+    async def outer_node(state: State) -> State:
+        return await inner.ainvoke({"log": state["log"]})
+
+    result = await (
+        _outer(outer_node)
+        .compile(insert_before=["finish"])
+        .ainvoke({"log": ["user"]}, control=control)
+    )
+
+    assert result["seen"] == ["user", "tools"]
+    assert result["log"][-2:] == ["inserted", "finished"]
+
+
+def test_inserted_input_is_streamed_as_values() -> None:
+    control = RunControl()
+    graph = _tools_then_model(control).compile(insert_before=["model"])
+
+    chunks = list(
+        graph.stream({"log": ["user"]}, control=control, stream_mode="values")
+    )
+
+    assert {"log": ["user", "tools", "inserted"]} in chunks
+
+
+@pytest.mark.parametrize(
+    "value", ["not a dict", {"seen": ["forged"]}, {"branch:to:model": None}]
+)
+def test_invalid_inserted_input_is_skipped(value: object) -> None:
+    class InputState(TypedDict):
+        log: Annotated[list[str], operator.add]
+
+    control = RunControl()
+
+    def tools(state: State) -> State:
+        control.submit_input("bad", value)
+        return {"log": ["tools"]}
+
+    def model(state: State) -> State:
+        return {"seen": list(state["log"])}
+
+    builder = StateGraph(State, input_schema=InputState)
+    builder.add_node("tools", tools)
+    builder.add_node("model", model)
+    builder.add_edge(START, "tools")
+    builder.add_edge("tools", "model")
+    builder.add_edge("model", END)
+    checkpointer = InMemorySaver()
+    graph = builder.compile(checkpointer=checkpointer, insert_before=["model"])
+    config = {"configurable": {"thread_id": "1"}}
+
+    result = graph.invoke({"log": ["user"]}, config, control=control)
+
+    assert result["seen"] == ["user", "tools"]
+    assert not any(
+        snapshot.metadata.get("inserted_inputs")
+        for snapshot in graph.get_state_history(config)
+    )
+
+
+@pytest.mark.parametrize("by_id", [False, True])
+def test_inserted_input_waits_while_resuming(by_id: bool) -> None:
+    def model(state: State) -> State:
+        answer = interrupt("approve?")
+        return {"log": [f"answered {answer}"], "seen": list(state["log"])}
+
+    builder = StateGraph(State)
+    builder.add_node("model", model)
+    builder.add_edge(START, "model")
+    builder.add_edge("model", END)
+    graph = builder.compile(checkpointer=InMemorySaver(), insert_before=["model"])
+    config = {"configurable": {"thread_id": "1"}}
+    paused = graph.invoke({"log": ["user"]}, config)
+    interrupt_id = paused["__interrupt__"][0].id
+    control = RunControl()
+    control.submit_input("insert-1", {"log": ["inserted"]})
+
+    result = graph.invoke(
+        Command(resume={interrupt_id: "yes"} if by_id else "yes"),
+        config,
+        control=control,
+    )
+
+    # the resume is answered as asked; the input waits for a later boundary
+    assert result["seen"] == ["user"]
+    assert result["log"] == ["user", "answered yes"]
+    assert control.take_inputs() == [("insert-1", {"log": ["inserted"]})]
+
+
+def test_insert_before_rejects_unknown_nodes() -> None:
+    graph = _tools_then_model(RunControl())
+
+    with pytest.raises(ValueError, match="missing"):
+        graph.compile(insert_before=["missing"])
+
+
+def test_inserts_are_applied_one_by_one_and_bad_values_skipped() -> None:
+    class InputState(TypedDict, total=False):
+        log: Annotated[list[str], operator.add]
+        note: str
+
+    class FullState(InputState, total=False):
+        seen: list[str]
+
+    control = RunControl()
+
+    def tools(state: FullState) -> FullState:
+        control.submit_input("first", {"note": "a", "log": ["one"]})
+        control.submit_input("bad", {"log": "not a list"})
+        control.submit_input("second", {"note": "b", "log": ["two"]})
+        return {"log": ["tools"]}
+
+    def model(state: FullState) -> FullState:
+        return {"seen": [*state["log"], state.get("note", "")]}
+
+    builder = StateGraph(FullState, input_schema=InputState)
+    builder.add_node("tools", tools)
+    builder.add_node("model", model)
+    builder.add_edge(START, "tools")
+    builder.add_edge("tools", "model")
+    builder.add_edge("model", END)
+    checkpointer = InMemorySaver()
+    graph = builder.compile(checkpointer=checkpointer, insert_before=["model"])
+    config = {"configurable": {"thread_id": "1"}}
+
+    result = graph.invoke({"log": ["user"]}, config, control=control)
+
+    assert result["seen"] == ["user", "tools", "one", "two", "b"]
+    inserted = [
+        snapshot.metadata["inserted_inputs"]
+        for snapshot in graph.get_state_history(config)
+        if snapshot.metadata.get("inserted_inputs")
+    ]
+    assert inserted == [["first", "second"]]
+
+
+def test_inserted_input_waits_while_resuming_from_a_breakpoint() -> None:
+    control = RunControl()
+    graph = _tools_then_model(control).compile(
+        checkpointer=InMemorySaver(),
+        insert_before=["model"],
+        interrupt_before=["model"],
+    )
+    config = {"configurable": {"thread_id": "1"}}
+    graph.invoke({"log": ["user"]}, config)
+    assert graph.get_state(config).next == ("model",)
+
+    result = graph.invoke(None, config, control=control)
+
+    # the breakpoint is passed once; the input waits for a later boundary
+    assert result["seen"] == ["user", "tools"]
+    assert graph.get_state(config).next == ()
+
+
+def test_inserted_input_is_not_applied_without_steps_left() -> None:
+    control = RunControl()
+    graph = _tools_then_model(control).compile(insert_before=["model"])
+
+    with pytest.raises(GraphRecursionError):
+        graph.invoke({"log": ["user"]}, {"recursion_limit": 1}, control=control)
+
+    assert control.take_inputs() == [("insert-1", {"log": ["inserted"]})]

libs/sdk-py/langgraph_sdk/_async/cron.py

@@ -104,7 +104,7 @@ class CronClient:
 
             webhook: Webhook to call after LangGraph API call is done.
             multitask_strategy: Multitask strategy to use.
-                Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
+                Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'.
             end_time: The time to stop running the cron job. If not provided, the cron job will run indefinitely.
             enabled: Whether the cron job is enabled or not.
             timezone: IANA timezone for the cron schedule. Accepts a string (e.g. 'America/New_York') or a ``datetime.tzinfo`` instance (e.g. ``ZoneInfo("America/New_York")``).
@@ -222,7 +222,7 @@ class CronClient:
                 after execution. 'keep' creates a new thread for each execution but does not
                 clean them up. Clients are responsible for cleaning up kept threads.
             multitask_strategy: Multitask strategy to use.
-                Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
+                Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'.
             end_time: The time to stop running the cron job. If not provided, the cron job will run indefinitely.
             enabled: Whether the cron job is enabled or not.
             timezone: IANA timezone for the cron schedule. Accepts a string (e.g. 'America/New_York') or a ``datetime.tzinfo`` instance (e.g. ``ZoneInfo("America/New_York")``).

libs/sdk-py/langgraph_sdk/_async/runs.py

@@ -252,7 +252,7 @@ class RunsClient:
                 Must be one of 'delete' or 'keep'.
             webhook: Webhook to call after LangGraph API call is done.
             multitask_strategy: Multitask strategy to use.
-                Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
+                Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'.
             if_not_exists: How to handle missing thread. Defaults to 'reject'.
                 Must be either 'reject' (raise error if missing), or 'create' (create new thread).
             after_seconds: The number of seconds to wait before starting the run.
@@ -468,7 +468,7 @@ class RunsClient:
             interrupt_after: Nodes to Nodes to interrupt immediately after they get executed.
             webhook: Webhook to call after LangGraph API call is done.
             multitask_strategy: Multitask strategy to use.
-                Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
+                Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'.
             on_completion: Whether to delete or keep the thread created for a stateless run.
                 Must be one of 'delete' or 'keep'.
             if_not_exists: How to handle missing thread. Defaults to 'reject'.
@@ -726,7 +726,7 @@ class RunsClient:
             on_completion: Whether to delete or keep the thread created for a stateless run.
                 Must be one of 'delete' or 'keep'.
             multitask_strategy: Multitask strategy to use.
-                Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
+                Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'.
             if_not_exists: How to handle missing thread. Defaults to 'reject'.
                 Must be either 'reject' (raise error if missing), or 'create' (create new thread).
             after_seconds: The number of seconds to wait before starting the run.

libs/sdk-py/langgraph_sdk/_sync/cron.py

@@ -96,7 +96,7 @@ class SyncCronClient:
             interrupt_after: Nodes to Nodes to interrupt immediately after they get executed.
             webhook: Webhook to call after LangGraph API call is done.
             multitask_strategy: Multitask strategy to use.
-                Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
+                Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'.
             end_time: The time to stop running the cron job. If not provided, the cron job will run indefinitely.
             enabled: Whether the cron job is enabled. By default, it is considered enabled.
             timezone: IANA timezone for the cron schedule. Accepts a string (e.g. 'America/New_York') or a ``datetime.tzinfo`` instance (e.g. ``ZoneInfo("America/New_York")``).
@@ -212,7 +212,7 @@ class SyncCronClient:
                 after execution. 'keep' creates a new thread for each execution but does not
                 clean them up. Clients are responsible for cleaning up kept threads.
             multitask_strategy: Multitask strategy to use.
-                Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
+                Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'.
             end_time: The time to stop running the cron job. If not provided, the cron job will run indefinitely.
             enabled: Whether the cron job is enabled. By default, it is considered enabled.
             timezone: IANA timezone for the cron schedule. Accepts a string (e.g. 'America/New_York') or a ``datetime.tzinfo`` instance (e.g. ``ZoneInfo("America/New_York")``).

libs/sdk-py/langgraph_sdk/_sync/runs.py

@@ -252,7 +252,7 @@ class SyncRunsClient:
                 Must be one of 'delete' or 'keep'.
             webhook: Webhook to call after LangGraph API call is done.
             multitask_strategy: Multitask strategy to use.
-                Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
+                Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'.
             if_not_exists: How to handle missing thread. Defaults to 'reject'.
                 Must be either 'reject' (raise error if missing), or 'create' (create new thread).
             after_seconds: The number of seconds to wait before starting the run.
@@ -463,7 +463,7 @@ class SyncRunsClient:
             interrupt_after: Nodes to Nodes to interrupt immediately after they get executed.
             webhook: Webhook to call after LangGraph API call is done.
             multitask_strategy: Multitask strategy to use.
-                Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
+                Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'.
             on_completion: Whether to delete or keep the thread created for a stateless run.
                 Must be one of 'delete' or 'keep'.
             if_not_exists: How to handle missing thread. Defaults to 'reject'.
@@ -721,7 +721,7 @@ class SyncRunsClient:
             on_completion: Whether to delete or keep the thread created for a stateless run.
                 Must be one of 'delete' or 'keep'.
             multitask_strategy: Multitask strategy to use.
-                Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
+                Must be one of 'reject', 'interrupt', 'rollback', 'enqueue', or 'insert'.
             if_not_exists: How to handle missing thread. Defaults to 'reject'.
                 Must be either 'reject' (raise error if missing), or 'create' (create new thread).
             after_seconds: The number of seconds to wait before starting the run.

libs/sdk-py/langgraph_sdk/auth/types.py

@@ -29,7 +29,9 @@ Values:
     - interrupted: Run was manually interrupted
 """
 
-MultitaskStrategy = typing.Literal["reject", "rollback", "interrupt", "enqueue"]
+MultitaskStrategy = typing.Literal[
+    "reject", "rollback", "interrupt", "enqueue", "insert"
+]
 """Strategy for handling multiple concurrent tasks.
 
 Values:
@@ -37,6 +39,7 @@ Values:
     - rollback: Cancel current task and start new one
     - interrupt: Interrupt current task and start new one
     - enqueue: Queue new tasks to run after current one
+    - insert: Add input to current task at its next safe step; held while interrupted
 """
 
 OnConflictBehavior = typing.Literal["raise", "do_nothing"]

libs/sdk-py/langgraph_sdk/schema.py

@@ -78,13 +78,14 @@ Specifies behavior on disconnection:
 - "continue": Continue the operation even if disconnected.
 """
 
-MultitaskStrategy = Literal["reject", "interrupt", "rollback", "enqueue"]
+MultitaskStrategy = Literal["reject", "interrupt", "rollback", "enqueue", "insert"]
 """
 Defines how to handle multiple tasks:
 - "reject": Reject new tasks when busy.
 - "interrupt": Interrupt current task for new ones.
 - "rollback": Roll back current task and start new one.
 - "enqueue": Queue new tasks for later execution.
+- "insert": Add input to the current task at its next safe step; held while interrupted.
 """
 
 OnConflictBehavior = Literal["raise", "do_nothing"]