release(cli): 0.4.36 (#9330)

## Summary

<!-- Write the summary yourself -->

## Notes for reviewers
- Includes [push through the active Docker
context](https://github.com/langchain-ai/langgraph/pull/9321) and
[accept serverless and dedicated deployment
aliases](https://github.com/langchain-ai/langgraph/pull/9325), the two
CLI changes merged since 0.4.35.
- Merging doesn't publish by itself: after it merges, run the `release`
workflow with `working-directory: libs/cli`, as was done for 0.4.35.

## Test Plan
- [x] Only `__version__` in `libs/cli/langgraph_cli/__init__.py` changes
(0.4.35 → 0.4.36), like the 0.4.35 release PR
- [ ] After merge, the `release` workflow for `libs/cli` succeeds and
PyPI shows `langgraph-cli` 0.4.36
- [ ] `uvx --from langgraph-cli==0.4.36 langgraph deploy` pushes on
Docker Desktop without `DOCKER_HOST` set

## Demo

<!-- Drag a screen recording or screenshot here -->

libs/cli/langgraph_cli/__init__.py

@@ -1 +1 @@
-__version__ = "0.4.35"
+__version__ = "0.4.36"

libs/cli/langgraph_cli/deploy.py

@@ -1209,6 +1209,33 @@ class DockerBuildCommand:
         return cls(("docker", "buildx", "build"), flags)
 
 
+def _active_docker_host(runner: CommandRunner) -> str | None:
+    """Return the daemon endpoint of the user's active Docker context.
+
+    None when `DOCKER_HOST` is set, which Docker honors with any config
+    directory. Also None when the lookup fails (for example, `DOCKER_CONTEXT`
+    names a context that doesn't exist) or the context has no Docker endpoint
+    (Docker prints `<no value>`); that keeps Docker's own default in place.
+    """
+    if os.environ.get("DOCKER_HOST"):
+        return None
+    try:
+        stdout, _ = runner.run(
+            subp_exec(
+                "docker",
+                "context",
+                "inspect",
+                "--format",
+                "{{.Endpoints.docker.Host}}",
+                collect=True,
+            )
+        )
+    except (click.exceptions.Exit, OSError):
+        return None
+    host = (stdout or "").strip()
+    return host if "://" in host else None
+
+
 def _docker_argv(docker_config_dir: str | None) -> tuple[str, ...]:
     if docker_config_dir is None:
         return ("docker",)
@@ -1242,17 +1269,13 @@ def _push_image(
     runner: CommandRunner,
     image: str,
     *,
-    docker_config_dir: str | None,
+    docker: tuple[str, ...],
     verbose: bool,
 ) -> None:
     for attempt in range(1, _PUSH_ATTEMPTS + 1):
         try:
             with Progress(message="Pushing...", elapsed=not verbose):
-                runner.run(
-                    subp_exec(
-                        *_docker_argv(docker_config_dir), "push", image, verbose=verbose
-                    )
-                )
+                runner.run(subp_exec(*docker, "push", image, verbose=verbose))
             return
         except click.exceptions.Exit:
             if attempt == _PUSH_ATTEMPTS:
@@ -1337,6 +1360,11 @@ def _run_local_build(
         registry_host = repository.split("/")[0]
 
         with _docker_config_for_token(registry_host, deployment_token, username) as cfg:
+            # A config directory has no current context, so Docker would fall
+            # back to /var/run/docker.sock; keep using the daemon the build used.
+            docker_host = _active_docker_host(runner)
+            host_args = ("--host", docker_host) if docker_host else ()
+            docker = (*_docker_argv(cfg), *host_args)
             _log_deploy_step(step, f"Logging into {registry_host}")
             token_input = (
                 deployment_token
@@ -1345,7 +1373,7 @@ def _run_local_build(
             )
             runner.run(
                 subp_exec(
-                    *_docker_argv(cfg),
+                    *docker,
                     "login",
                     "-u",
                     username,
@@ -1361,7 +1389,7 @@ def _run_local_build(
             runner.run(
                 subp_exec("docker", "tag", image_to_push, remote_image, verbose=verbose)
             )
-            _push_image(runner, remote_image, docker_config_dir=cfg, verbose=verbose)
+            _push_image(runner, remote_image, docker=docker, verbose=verbose)
         step += 1
 
         resolved_image = _resolve_pushed_image_digest(
@@ -1826,7 +1854,7 @@ class CustomerRegistrySource:
                 _build_image(runner, ctx.spec, image, verbose=ctx.verbose)
             step += 1
             _log_deploy_step(step, f"Pushing image {image}")
-            _push_image(runner, image, docker_config_dir=None, verbose=ctx.verbose)
+            _push_image(runner, image, docker=_docker_argv(None), verbose=ctx.verbose)
             step += 1
             digest = _resolve_pushed_image_digest(
                 runner, remote_image=image, docker_config_dir=None, verbose=ctx.verbose
@@ -2185,11 +2213,13 @@ def _deploy_base_options(
             ),
             click.option(
                 "--deployment-type",
-                type=click.Choice(["dev", "prod"]),
+                type=click.Choice(["serverless", "dedicated", "dev", "prod"]),
                 default="dev",
                 show_default=True,
                 help=(
-                    "Deployment type (used when creating a new deployment). "
+                    "Deployment type (used when creating a new deployment): "
+                    "serverless (scale to zero) or dedicated (prod); "
+                    "legacy dev and prod are also accepted. "
                     "Ignored with --push-to."
                 ),
             ),
@@ -2480,7 +2510,9 @@ def _deploy_cmd(
             ),
             verbose=verbose,
             selector=selector,
-            deployment_type=deployment_type,
+            deployment_type={"serverless": "dev_zero", "dedicated": "prod"}.get(
+                deployment_type, deployment_type
+            ),
             secrets=secrets,
             tracked_packages=tracked_packages,
         )

libs/cli/tests/unit_tests/cli/test_deploy_command.py

@@ -52,6 +52,8 @@ ARCHIVE = ("/tmp/src.tgz", 2048, "langgraph.json")
 OBJECT_PATH = "tarballs/src.tgz"
 PLATFORM_FORMAT = "{{.Os}}/{{.Architecture}}"
 DIGESTS_FORMAT = "{{json .RepoDigests}}"
+CONTEXT_HOST_FORMAT = "{{.Endpoints.docker.Host}}"
+CONTEXT_HOST = "unix:///Users/dev/.docker/run/docker.sock"
 NOT_A_CLI_DEPLOYMENT = (
     "push token is only available for 'internal_docker' source deployments"
 )
@@ -192,6 +194,8 @@ class DockerCommand:
 class DockerDouble:
     timeline: list[str]
     failing_pushes: int = 0
+    context_lookup_fails: bool = False
+    context_host: str = CONTEXT_HOST
     builds: list[dict] = field(default_factory=list)
     commands: list[DockerCommand] = field(default_factory=list)
 
@@ -235,6 +239,10 @@ class DockerDouble:
         if "push" in args and self.failing_pushes > 0:
             self.failing_pushes -= 1
             raise click.exceptions.Exit(1)
+        if CONTEXT_HOST_FORMAT in args:
+            if self.context_lookup_fails:
+                raise click.exceptions.Exit(1)
+            return f"{self.context_host}\n", None
         if PLATFORM_FORMAT in args:
             return "linux/amd64\n", None
         if DIGESTS_FORMAT in args:
@@ -244,6 +252,8 @@ class DockerDouble:
 
     @staticmethod
     def _verb(args: tuple[str, ...]) -> str:
+        if CONTEXT_HOST_FORMAT in args:
+            return "inspect-context"
         if PLATFORM_FORMAT in args:
             return "inspect-platform"
         if DIGESTS_FORMAT in args:
@@ -292,7 +302,12 @@ def deploy_project(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> DeployPro
         json.dumps({"graphs": {"agent": "agent.py:graph"}, "dependencies": ["."]})
     )
     monkeypatch.chdir(tmp_path)
-    for name in ("LANGSMITH_TENANT_ID", "LANGSMITH_ENDPOINT", "LANGGRAPH_HOST_URL"):
+    for name in (
+        "LANGSMITH_TENANT_ID",
+        "LANGSMITH_ENDPOINT",
+        "LANGGRAPH_HOST_URL",
+        "DOCKER_HOST",
+    ):
         monkeypatch.delenv(name, raising=False)
 
     timeline: list[str] = []
@@ -340,6 +355,7 @@ def test_first_local_deploy_creates_then_builds_pushes_and_updates_in_order(
         CREATE_DEPLOYMENT,
         "docker build",
         _push_token(CREATED_ID),
+        "docker inspect-context",
         "docker login",
         "docker tag",
         "docker push",
@@ -349,6 +365,27 @@ def test_first_local_deploy_creates_then_builds_pushes_and_updates_in_order(
     assert "Deployment updated" in result.output
 
 
+@pytest.mark.parametrize(
+    ("deployment_type", "wire_type"),
+    [
+        ("serverless", "dev_zero"),
+        ("dedicated", "prod"),
+        ("dev", "dev"),
+        ("prod", "prod"),
+    ],
+)
+@pytest.mark.parametrize("build_mode", ["--no-remote", "--remote"])
+def test_deployment_type_aliases(
+    deploy_project: DeployProject, deployment_type: str, wire_type: str, build_mode: str
+) -> None:
+    result = deploy_project.run(build_mode, "--deployment-type", deployment_type)
+
+    assert result.exit_code == 0, result.output
+    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
+        "deployment_type": wire_type
+    }
+
+
 def test_first_local_deploy_creates_an_internal_docker_deployment(
     deploy_project: DeployProject,
 ) -> None:
@@ -406,7 +443,7 @@ def test_local_deploy_logs_in_with_the_control_plane_push_token(
 
     login = deploy_project.docker.command("login")
     assert login.args[:2] == ("docker", "--config")
-    assert login.args[3:] == (
+    assert login.args[-5:] == (
         "login",
         "-u",
         "oauth2accesstoken",
@@ -416,6 +453,53 @@ def test_local_deploy_logs_in_with_the_control_plane_push_token(
     assert login.kwargs["input"] == f"{PUSH_TOKEN}\n"
 
 
+def test_local_deploy_logs_in_and_pushes_through_the_active_docker_context(
+    deploy_project: DeployProject,
+) -> None:
+    """The temporary config has no current context, so without the active
+    context's endpoint Docker would push through /var/run/docker.sock."""
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code == 0, result.output
+    for verb in ("login", "push"):
+        args = deploy_project.docker.command(verb).args
+        assert args[3:5] == ("--host", CONTEXT_HOST), verb
+
+
+def test_local_deploy_leaves_an_explicit_docker_host_to_docker(
+    deploy_project: DeployProject, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setenv("DOCKER_HOST", "tcp://127.0.0.1:2375")
+
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code == 0, result.output
+    for verb in ("login", "push"):
+        assert "--host" not in deploy_project.docker.command(verb).args, verb
+
+
+def test_local_deploy_keeps_dockers_default_when_the_context_lookup_fails(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.docker.context_lookup_fails = True
+
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code == 0, result.output
+    assert "--host" not in deploy_project.docker.command("push").args
+
+
+def test_local_deploy_ignores_a_context_without_a_docker_endpoint(
+    deploy_project: DeployProject,
+) -> None:
+    deploy_project.docker.context_host = "<no value>"
+
+    result = deploy_project.run("--no-remote")
+
+    assert result.exit_code == 0, result.output
+    assert "--host" not in deploy_project.docker.command("push").args
+
+
 def test_local_deploy_tells_the_control_plane_it_honors_registry_usernames(
     deploy_project: DeployProject,
 ) -> None:
@@ -438,7 +522,7 @@ def test_local_deploy_logs_in_with_the_username_the_control_plane_returns(
     deploy_project.run("--no-remote")
 
     login = deploy_project.docker.command("login")
-    assert login.args[3:] == (
+    assert login.args[-5:] == (
         "login",
         "-u",
         "AWS",
@@ -506,6 +590,7 @@ def test_prebuilt_image_is_validated_and_pushed_without_a_build(
     assert deploy_project.docker.builds == []
     assert deploy_project.docker.verbs() == [
         "docker inspect-platform",
+        "docker inspect-context",
         "docker login",
         "docker tag",
         "docker push",

libs/langgraph/langgraph/pregel/_loop.py

@@ -257,8 +257,10 @@ class PregelLoop:
     _resume_loaded_writes: Sequence[PendingWrite] = ()
     # Tasks `_reapply_writes_to_succeeded_nodes` handed loaded writes back to.
     _reapplied_task_ids: set[str]
-    # An update or fork checkpoint addressed by `checkpoint_id` that the thread
-    # has moved past, so it may already have children.
+    # Tasks that wrote during this run, the only ones exit mode saves at exit.
+    _written_task_ids: set[str]
+    # A checkpoint addressed by `checkpoint_id` that the thread has moved past,
+    # so it may already have children.
     _addressed_checkpoint_superseded: bool = False
 
     # The checkpoint_config that points at the parent loaded at `__enter__`
@@ -349,7 +351,6 @@ class PregelLoop:
         self.interrupt_before = interrupt_before
         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]
         self._migrate_checkpoint = migrate_checkpoint
         self.trigger_to_nodes = trigger_to_nodes
         self.retry_policy = retry_policy
@@ -407,6 +408,14 @@ class PregelLoop:
             else ()
         )
         self.prev_checkpoint_config = None
+        # Whether the run points this loop at a checkpoint: a truthy
+        # `checkpoint_id`, this graph's `checkpoint_map` entry, or a replaying
+        # parent. Every task config carries `checkpoint_id: None`, so a key
+        # check would count every subgraph.
+        self._addressed = bool(
+            self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID)
+            or self.config[CONF].get(CONFIG_KEY_REPLAY_STATE)
+        )
         runtime = self.config[CONF].get(CONFIG_KEY_RUNTIME)
         self.control = runtime.control if isinstance(runtime, Runtime) else None
 
@@ -455,6 +464,7 @@ class PregelLoop:
         """Put writes for a task, to be read by the next tick."""
         if not writes:
             return
+        self._written_task_ids.add(task_id)
         # deduplicate writes to special channels, last write wins
         if all(w[0] in WRITES_IDX_MAP for w in writes):
             writes = list({w[0]: w for w in writes}.values())
@@ -468,6 +478,11 @@ class PregelLoop:
                 for w in self.checkpoint_pending_writes
                 if w[0] != task_id or w[1] not in WRITES_IDX_MAP
             ]
+            # Savers key a regular write by its position in the list, and the
+            # special ones dropped above won't be there next time, so they go
+            # last: a resume saved ahead of an update would move it to a new
+            # slot once the list is saved again, and it would apply twice.
+            writes = sorted(writes, key=lambda w: w[0] in WRITES_IDX_MAP)
             writes_to_save: WritesT = [
                 w[1:] for w in self.checkpoint_pending_writes if w[0] == task_id
             ] + list(writes)
@@ -556,10 +571,12 @@ class PregelLoop:
                 CONFIG_KEY_CHECKPOINT_ID: self.checkpoint["id"],
             },
         )
-        # group by task id
+        # Writes loaded with the checkpoint are stored on it already. Saving them
+        # again can file them under new indexes, and then they apply twice.
         by_task = defaultdict(list)
         for task_id, channel, value in self.checkpoint_pending_writes:
-            by_task[task_id].append((channel, value))
+            if task_id in self._written_task_ids:
+                by_task[task_id].append((channel, value))
         # submit writes to checkpointer
         for task_id, writes in by_task.items():
             if self.checkpointer_put_writes_accepts_task_path and hasattr(
@@ -904,6 +921,7 @@ class PregelLoop:
             if tid != NULL_TASK_ID and isinstance(self.specs.get(ch), DeltaChannel)
         }
         self._exit_command_writes = []
+        self._written_task_ids = set()
         # Resuming from a previous checkpoint requires two things:
         # 1. A prior checkpoint exists (channel_versions is non-empty)
         # 2. The input signals continuation (not a fresh run with new input)
@@ -931,7 +949,7 @@ class PregelLoop:
         # writes so that interrupt() calls re-fire instead of returning
         # stale values. But if we're actively resuming, keep them —
         # multi-interrupt scenarios need previously resolved values preserved.
-        is_time_traveling = self.is_replaying and (
+        is_time_traveling = self._addressed and (
             # Time-travel to a subgraph checkpoint: the parent sets
             # RESUMING=True (it can't distinguish time-travel from resume),
             # so we check if this subgraph's own ns is in checkpoint_map.
@@ -950,6 +968,10 @@ class PregelLoop:
                 or configurable.get(CONFIG_KEY_RESUMING, False)
             )
         )
+        # Finished tasks rerun only on time travel, or on a resume at a
+        # checkpoint the thread has moved past. A resume addressed at the
+        # thread's latest checkpoint hands their writes back, as without an id.
+        self.is_replaying = is_time_traveling or self._addressed_checkpoint_superseded
         if is_time_traveling:
             self.checkpoint_pending_writes = [
                 w for w in self.checkpoint_pending_writes if w[1] != RESUME
@@ -1165,11 +1187,11 @@ class PregelLoop:
             # checkpoint ID since the fork was created after the subgraph's
             # checkpoints from the original execution.
             #
-            # Only gate on is_time_traveling (not is_replaying). When the
+            # Only gate on is_time_traveling (not _addressed). When the
             # client resumes with an explicit checkpoint_id that happens to
             # point at the current head (e.g. LangGraph Studio sending
             # `checkpoint: {checkpoint_id}` alongside Command(resume=...)),
-            # is_replaying is True but is_time_traveling is False. In that
+            # the run is addressed but not time traveling. In that
             # case subgraphs should load their latest checkpoint normally,
             # not go through ReplayState's before-bound lookup which would
             # miss subgraph checkpoints created during processing of the
@@ -1789,7 +1811,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             # This covers both normal replay and subgraphs resolved via
             # checkpoint_map during time-travel.
             saved = self.checkpointer.get_tuple(self.checkpoint_config)
-            if saved is not None and saved.metadata.get("source") in ("update", "fork"):
+            if saved is not None:
                 self._addressed_checkpoint_superseded = checkpoint_superseded(
                     self.checkpointer, self.checkpoint_config, saved
                 )
@@ -2045,7 +2067,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             # This covers both normal replay and subgraphs resolved via
             # checkpoint_map during time-travel.
             saved = await self.checkpointer.aget_tuple(self.checkpoint_config)
-            if saved is not None and saved.metadata.get("source") in ("update", "fork"):
+            if saved is not None:
                 self._addressed_checkpoint_superseded = await acheckpoint_superseded(
                     self.checkpointer, self.checkpoint_config, saved
                 )

libs/langgraph/tests/test_delta_channel_exit_mode.py

@@ -521,10 +521,11 @@ def test_addressed_resume_keeps_a_rerun_tasks_write_to_a_new_channel(
     graph = builder.compile(checkpointer=sync_checkpointer)
     config = {"configurable": {"thread_id": "t"}}
     graph.invoke(_both("in"), config, durability=durability)
-    head = graph.get_state(config).config
+    paused = graph.get_state(config).config
+    graph.update_state(config, None)
 
     live = graph.invoke(
-        Command(resume="yes", update={"flag": True}), head, durability=durability
+        Command(resume="yes", update={"flag": True}), paused, durability=durability
     )
 
     state = graph.get_state(config)

libs/langgraph/tests/test_interrupt_state.py

@@ -152,6 +152,83 @@ def test_finished_task_does_not_report_answered_interrupt(
         assert asked == [["A1", "B"]]
 
 
+def test_finished_task_output_applies_once(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    builder = StateGraph(State)
+    builder.add_node("a", lambda s: {"log": [f"a:{interrupt('A')}"]})
+    builder.add_node("b", lambda s: {"log": [f"b:{interrupt('B')}"]})
+    builder.add_node("c", lambda s: {"log": [f"c:{interrupt('C')}"]})
+    builder.add_edge(START, "a")
+    builder.add_edge(START, "b")
+    builder.add_edge(START, "c")
+    graph = builder.compile(checkpointer=sync_checkpointer)
+    config = _config()
+
+    result = graph.invoke({"log": []}, config, durability=durability)
+    while pending := result.get("__interrupt__"):
+        result = graph.invoke(
+            Command(resume={pending[0].id: "yes"}), config, durability=durability
+        )
+
+    assert sorted(result["log"]) == ["a:yes", "b:yes", "c:yes"]
+
+
+def test_resume_with_update_applies_the_update_once(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    def ask_three_times(state: State) -> State:
+        answers = [interrupt(str(question)) for question in range(1, 4)]
+        return {"log": ["a:" + "".join(answers)]}
+
+    graph = (
+        StateGraph(State)
+        .add_node("a", ask_three_times)
+        .add_edge(START, "a")
+        .compile(checkpointer=sync_checkpointer)
+    )
+    config = _config()
+
+    result = graph.invoke({"log": []}, config, durability=durability)
+    for i in range(1, 4):
+        result = graph.invoke(
+            Command(resume=str(i), update={"log": [f"u{i}"]}),
+            config,
+            durability=durability,
+        )
+
+    assert sorted(result["log"]) == ["a:123", "u1", "u2", "u3"]
+    assert sorted(graph.get_state(config).values["log"]) == sorted(result["log"])
+
+
+@pytest.mark.parametrize("checkpoint_id", ["none", "head"])
+def test_resume_at_the_latest_checkpoint_does_not_rerun_finished_task(
+    sync_checkpointer: BaseCheckpointSaver, checkpoint_id: str
+) -> None:
+    calls: Counter[str] = Counter()
+    graph = _build_parallel(sync_checkpointer, calls)
+    config = _config()
+    graph.invoke({"log": []}, config)
+
+    def resume(question: str, answer: str) -> Any:
+        snapshot = graph.get_state(config)
+        target = (
+            snapshot.config
+            if checkpoint_id == "head"
+            else {"configurable": {**config["configurable"], "checkpoint_id": None}}
+        )
+        return graph.invoke(
+            Command(resume={_interrupt_by_value(snapshot, question).id: answer}),
+            target,
+        )
+
+    resume("A1", "yes")
+    result = resume("B", "ok")
+
+    assert sorted(result["log"]) == ["a:yes", "b:ok"]
+    assert calls == {"a": 2, "b": 3}, "resuming B ran the finished A again"
+
+
 @NEEDS_CONTEXTVARS
 async def test_finished_task_does_not_report_answered_interrupt_async(
     async_checkpointer: BaseCheckpointSaver,
@@ -474,6 +551,64 @@ def test_functional_task_finished_with_none_is_not_rerun(
     assert calls == {"a": 2, "b": 3}
 
 
+def test_subgraph_finished_task_is_not_rerun(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    calls: Counter[str] = Counter()
+    builder = StateGraph(State)
+    builder.add_node("child", _build_parallel(None, calls))
+    builder.add_edge(START, "child")
+    graph = builder.compile(checkpointer=sync_checkpointer)
+    config = _config()
+
+    graph.invoke({"log": []}, config, durability=durability)
+    snapshot = graph.get_state(config)
+    graph.invoke(
+        Command(resume={_interrupt_by_value(snapshot, "A1").id: "yes"}),
+        config,
+        durability=durability,
+    )
+    snapshot = graph.get_state(config)
+    result = graph.invoke(
+        Command(resume={_interrupt_by_value(snapshot, "B").id: "ok"}),
+        config,
+        durability=durability,
+    )
+
+    assert sorted(result["log"]) == ["a:yes", "b:ok"]
+    assert calls == {"a": 2, "b": 3}, "resuming B ran the finished A again"
+
+
+def test_nested_entrypoint_finished_task_is_not_rerun(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    calls: Counter[str] = Counter()
+
+    @task
+    def step_1(query: str) -> str:
+        calls["step_1"] += 1
+        return f"{query} bar"
+
+    @task
+    def human_feedback(query: str) -> str:
+        calls["human_feedback"] += 1
+        return f"{query} {interrupt('feedback?')}"
+
+    @entrypoint()
+    def child(query: str) -> str:
+        return human_feedback(step_1(query).result()).result()
+
+    @entrypoint(checkpointer=sync_checkpointer)
+    def parent(query: str) -> str:
+        return child.invoke(query)
+
+    config = _config()
+    parent.invoke("foo", config)
+
+    assert parent.invoke(Command(resume="baz"), config) == "foo bar baz"
+    assert calls == {"step_1": 1, "human_feedback": 2}
+
+
 # --- Reading task status from recorded writes ---
 
 

libs/langgraph/tests/test_pregel.py

@@ -7737,6 +7737,8 @@ def test_parallel_interrupts(sync_checkpointer: BaseCheckpointSaver) -> None:
 def test_parallel_interrupts_double(sync_checkpointer: BaseCheckpointSaver) -> None:
     # --- CHILD GRAPH ---
 
+    answered: Counter[str] = Counter()
+
     class ChildState(BaseModel):
         prompt: str = Field(..., description="What is going to be asked to the user?")
         human_input: str | None = Field(None, description="What the human said")
@@ -7746,6 +7748,7 @@ def test_parallel_interrupts_double(sync_checkpointer: BaseCheckpointSaver) -> N
 
     def get_human_input(state: ChildState):
         human_input = interrupt(state.prompt)
+        answered["human"] += 1
 
         return dict(
             human_inputs=[human_input],  # update parent state
@@ -7753,6 +7756,7 @@ def test_parallel_interrupts_double(sync_checkpointer: BaseCheckpointSaver) -> N
 
     def get_dolphin_input(state: ChildState):
         human_input = interrupt(state.prompt)
+        answered["dolphin"] += 1
 
         return dict(
             human_inputs=[human_input],  # update parent state
@@ -7846,6 +7850,7 @@ def test_parallel_interrupts_double(sync_checkpointer: BaseCheckpointSaver) -> N
 
     assert invokes == 5
     assert len(events) == 5
+    assert answered == {"human": 2, "dolphin": 2}
 
 
 def test_pregel_loop_refcount():

libs/langgraph/tests/test_time_travel.py

@@ -1177,6 +1177,43 @@ def test_subgraph_interrupt_resume_with_explicit_head_checkpoint_id(
     assert result["value"] == ["sub_a", "human:answer", "sub_b"]
 
 
+def test_subgraph_replay_reasks_answered_parallel_interrupt(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """Replaying a subgraph checkpoint asks its step's questions again, including
+    one already answered. An update checkpoint replays in place instead of
+    forking, so its tasks keep their ids and its saved answers could be reused."""
+
+    subgraph = (
+        StateGraph(State)
+        .add_node("a", lambda s: {"value": [f"a:{interrupt('a')}"]})
+        .add_node("b", lambda s: {"value": [f"b:{interrupt('b')}"]})
+        .add_edge(START, "a")
+        .add_edge(START, "b")
+        .compile(checkpointer=True)
+    )
+    graph = (
+        StateGraph(State)
+        .add_node("subgraph_node", subgraph)
+        .add_edge(START, "subgraph_node")
+        .compile(checkpointer=sync_checkpointer)
+    )
+    config = {"configurable": {"thread_id": "1"}}
+
+    def subgraph_head() -> RunnableConfig:
+        return graph.get_state(config, subgraphs=True).tasks[0].state.config
+
+    graph.invoke({"value": []}, config)
+    graph.update_state(subgraph_head(), {"value": ["u"]}, as_node="__start__")
+    result = graph.invoke(None, config)
+    question_a = next(i for i in result["__interrupt__"] if i.value == "a")
+    graph.invoke(Command(resume={question_a.id: "yes"}), config)
+
+    result = graph.invoke(None, subgraph_head())
+
+    assert sorted(i.value for i in result["__interrupt__"]) == ["a", "b"]
+
+
 def test_subgraph_replay_loads_accumulated_state_then_resume(
     sync_checkpointer: BaseCheckpointSaver,
 ) -> None:

libs/langgraph/uv.lock

@@ -1835,7 +1835,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.1"
 source = { editable = "../prebuilt" }
 dependencies = [
     { name = "langchain-core" },

libs/prebuilt/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.1"
 description = "Library with high-level APIs for creating and executing LangGraph agents and tools."
 authors = []
 requires-python = ">=3.10"

libs/prebuilt/uv.lock

@@ -512,7 +512,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.1"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

libs/sdk-py/uv.lock

@@ -433,7 +433,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.1"
 source = { editable = "../prebuilt" }
 dependencies = [
     { name = "langchain-core" },