style(langgraph): fml

Reverts langchain-ai/langgraph#6158

libs/cli/langgraph_cli/cli.py

@@ -271,7 +271,7 @@ For production use, requires a license key in env var LANGGRAPH_CLOUD_LICENSE_KE
                     f"""Ready!
 - API: http://localhost:{port}
 - Docs: http://localhost:{port}/docs
-- LangGraph Studio: {debugger_origin}/studio/?baseUrl={debugger_base_url_query}
+- LangSmith Debugger: {debugger_origin}/studio/?baseUrl={debugger_base_url_query}
 """
                 )
                 sys.stdout.flush()
@@ -652,11 +652,17 @@ def dockerfile(
     help="Wait for a debugger client to connect to the debug port before starting the server",
     default=False,
 )
+@click.option(
+    "--debugger-url",
+    type=str,
+    default=None,
+    help="URL of the LangSmith Debugger instance to connect to. Defaults to https://smith.langchain.com",
+)
 @click.option(
     "--studio-url",
     type=str,
     default=None,
-    help="URL of the LangGraph Studio instance to connect to. Defaults to https://smith.langchain.com",
+    help="(Deprecated: use --debugger-url instead) URL of the LangSmith Debugger instance to connect to.",
 )
 @click.option(
     "--allow-blocking",
@@ -692,12 +698,21 @@ def dev(
     no_browser: bool,
     debug_port: Optional[int],
     wait_for_client: bool,
+    debugger_url: Optional[str],
     studio_url: Optional[str],
     allow_blocking: bool,
     tunnel: bool,
     server_log_level: str,
 ):
     """CLI entrypoint for running the LangGraph API server."""
+    if studio_url is not None:
+        click.secho(
+            "Warning: --studio-url is deprecated and will be removed in a future version. "
+            "Please use --debugger-url instead.",
+            fg="yellow",
+        )
+        if debugger_url is None:
+            debugger_url = studio_url
     try:
         from langgraph_api.cli import run_server  # type: ignore
     except ImportError:
@@ -761,7 +776,7 @@ def dev(
         http=config_json.get("http"),
         ui=config_json.get("ui"),
         ui_config=config_json.get("ui_config"),
-        studio_url=studio_url,
+        studio_url=debugger_url,
         allow_blocking=allow_blocking,
         tunnel=tunnel,
         server_level=server_log_level,

libs/langgraph/langgraph/pregel/_loop.py

@@ -114,7 +114,6 @@ from langgraph.types import (
     CachePolicy,
     Command,
     Durability,
-    Interrupt,
     PregelExecutableTask,
     RetryPolicy,
     StreamMode,
@@ -317,16 +316,31 @@ class PregelLoop:
             ]
             writes_to_save: WritesT = [
                 w[1:] for w in self.checkpoint_pending_writes if w[0] == task_id
-            ] + list(writes)
+            ] + [(c, v) for c, v in writes if c != RESUME]
             self.checkpoint_pending_writes.extend((task_id, c, v) for c, v in writes)
         else:
-            writes_to_save = [
-                # aggregate existing interrupts for this task
-                (ch, self._merge_interrupts(task_id, v) if ch == INTERRUPT else v)
-                for ch, v in writes
-            ]
+            # build map of existing interrupts for this task: interrupt id -> list of interrupts
+            existing_interrupts_by_id: dict[str, list[Any]] = {
+                v[0].id: v
+                for tid, ch, v in self.checkpoint_pending_writes
+                if tid == task_id and ch == INTERRUPT
+            }
+            writes_to_save = []
+            for ch, v in writes:
+                if ch == INTERRUPT:
+                    # we merge new interrupt writes with existing interrupts writes if they
+                    # occurred within the same task (which means they have the same interrupt id)
+                    new_interrupts = v if isinstance(v, list) else list(v)
+                    if new_interrupts and (
+                        existing := existing_interrupts_by_id.get(new_interrupts[0].id)
+                    ):
+                        v = existing + new_interrupts
+                    writes_to_save.append((ch, v))
+                else:
+                    # we add non-interrupt writes as-is
+                    writes_to_save.append((ch, v))
 
-            # replace all writes for this task_id in one shot
+            # replace all writes for this task_id with the merged writes
             self.checkpoint_pending_writes = [
                 w for w in self.checkpoint_pending_writes if w[0] != task_id
             ] + [(task_id, c, v) for c, v in writes_to_save]
@@ -479,7 +493,9 @@ class PregelLoop:
         )
 
         resume_map = self.config.get(CONF, {}).get(CONFIG_KEY_RESUME_MAP, {})
-        if resume_map:
+        if resume_map or self.input is None:
+            # do not re-execute tasks that have unresumable interrupts
+            # i.e. when the graph is invoked with None, or the interrupt id is not in the resume map
             skipped_interrupt_ids = self._pending_interrupts() - set(resume_map)
             self.skipped_task_ids = {
                 task_id
@@ -538,7 +554,7 @@ class PregelLoop:
                 self.output_writes(task.id, task.writes, cached=True)
 
         if self.skipped_task_ids:
-            # remove tasks with writes that may have been matched from previous loop
+            # remove tasks with writes that have been matched with previous pending writes
             self.skipped_task_ids = {
                 task_id
                 for task_id in self.skipped_task_ids
@@ -566,8 +582,8 @@ class PregelLoop:
     def after_tick(self) -> None:
         if self.skipped_task_ids:
             # raise early GraphInterrupt for skipped tasks.
-            # since we know len(resumes) != len(interrupts) for these tasks, we
-            # can prevent unnecessary node re-execution by raising preemptively
+            # since we know len(resumes) < len(interrupts) for these tasks, we
+            # can prevent unnecessary node re-execution by raising early
             interrupts = []
             for task_id, channel, value in self.checkpoint_pending_writes:
                 if channel == INTERRUPT and task_id in self.skipped_task_ids:
@@ -651,29 +667,6 @@ class PregelLoop:
 
         return hanging_interrupts
 
-    def _merge_interrupts(
-        self, task_id: str, value: Sequence[Interrupt]
-    ) -> Sequence[Interrupt]:
-        """Normalize interrupt value to list and merge with existing interrupts.
-
-        If the interrupt ID matches existing, append; otherwise replace.
-
-        Returns list of Interrupt objects for this task.
-        """
-        new = value if isinstance(value, list) else list(value)
-        existing = next(
-            (
-                v
-                for tid, ch, v in self.checkpoint_pending_writes
-                if tid == task_id and ch == INTERRUPT
-            ),
-            None,
-        )
-        if existing is None:
-            return new
-        old = existing if isinstance(existing, list) else list(existing)
-        return old + new if old and new and old[0].id == new[0].id else new
-
     def _first(
         self, *, input_keys: str | Sequence[str], updated_channels: set[str] | None
     ) -> set[str] | None:

libs/langgraph/tests/test_interruption.py

@@ -479,12 +479,12 @@ def test_node_with_multiple_interrupts_requires_full_resume(
     assert remaining_interrupts[0].value == "second"
     assert node_counter == 2
 
-    # invoke with None resume. this should execute the node
+    # invoke with None resume. this should NOT execute the node
     partial = graph.invoke(None, config=config)
     remaining_interrupts = partial.get("__interrupt__", [])
     assert len(remaining_interrupts) == 1
     assert remaining_interrupts[0].value == "second"
-    assert node_counter == 3
+    assert node_counter == 2
 
     # invoke with nonspecific resume. this should execute the node
     partial = graph.invoke(Command(resume="human_second"), config=config)
@@ -492,14 +492,14 @@ def test_node_with_multiple_interrupts_requires_full_resume(
     assert len(remaining_interrupts) == 1
     print("REMAINING INTERRUPTS: ", remaining_interrupts)
     assert remaining_interrupts[0].value == "third"
-    assert node_counter == 4
+    assert node_counter == 3
 
     # finally, invoke with an interrupt map that matches double_interrupt_node.
     # this should execute the node and all interrupts should be resolved
     final_result = graph.invoke(Command(resume="human_third"), config=config)
     assert "input" in final_result
     assert final_result["input"] == "human_first-human_second-human_third"
-    assert node_counter == 5
+    assert node_counter == 4
 
 
 @NEEDS_CONTEXTVARS
@@ -561,12 +561,12 @@ async def test_node_with_multiple_interrupts_requires_full_resume_async(
     assert remaining_interrupts[0].value == "second"
     assert node_counter == 2
 
-    # invoke with None resume. this should execute the node
+    # invoke with None resume. this should NOT execute the node
     partial = await graph.ainvoke(None, config=config)
     remaining_interrupts = partial.get("__interrupt__", [])
     assert len(remaining_interrupts) == 1
     assert remaining_interrupts[0].value == "second"
-    assert node_counter == 3
+    assert node_counter == 2
 
     # invoke with nonspecific resume. this should execute the node
     partial = await graph.ainvoke(Command(resume="human_second"), config=config)
@@ -574,11 +574,92 @@ async def test_node_with_multiple_interrupts_requires_full_resume_async(
     assert len(remaining_interrupts) == 1
     print("REMAINING INTERRUPTS: ", remaining_interrupts)
     assert remaining_interrupts[0].value == "third"
-    assert node_counter == 4
+    assert node_counter == 3
 
     # finally, invoke with an interrupt map that matches double_interrupt_node.
     # this should execute the node and all interrupts should be resolved
     final_result = await graph.ainvoke(Command(resume="human_third"), config=config)
     assert "input" in final_result
     assert final_result["input"] == "human_first-human_second-human_third"
-    assert node_counter == 5
+    assert node_counter == 4
+
+
+def test_invoke_interrupted_graph_with_none(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """Test that invoking an interrupted graph with None does not duplicate interrupt writes"""
+
+    node_counter = 0
+
+    class State(TypedDict):
+        input: str
+
+    def double_interrupt_node(state: State):
+        nonlocal node_counter
+        node_counter += 1
+        first = interrupt("first")
+        second = interrupt("second")
+        return {"input": f"{first}-{second}"}
+
+    builder = StateGraph(State)
+    builder.add_node("double_interrupt", double_interrupt_node)
+    builder.add_edge(START, "double_interrupt")
+    builder.add_edge("double_interrupt", END)
+
+    graph = builder.compile(checkpointer=sync_checkpointer)
+
+    config = {"configurable": {"thread_id": "test_none_resume"}}
+
+    result = graph.invoke({"input": "start"}, config=config)
+    first_history = list(graph.get_state_history(config))
+    interrupts = result.get("__interrupt__", [])
+    assert len(interrupts) == 1
+    assert node_counter == 1
+
+    # invoke with None. this should NOT execute the node and the history should
+    # look the same as the first run
+    partial = graph.invoke(None, config=config)
+    second_history = list(graph.get_state_history(config))
+    remaining_interrupts = partial.get("__interrupt__", [])
+    assert len(remaining_interrupts) == 1
+    assert remaining_interrupts[0].value == "first"
+    assert node_counter == 1
+
+    # history should look the same for tasks and interrupts
+    print("first_history[0].interrupts: ", first_history[0].interrupts)
+    print("second_history[0].interrupts: ", second_history[0].interrupts)
+    print("first_history[0].tasks: ", first_history[0].tasks)
+    print("second_history[0].tasks: ", second_history[0].tasks)
+    assert first_history[0].interrupts == second_history[0].interrupts
+    assert first_history[0].tasks == second_history[0].tasks
+
+    # now resume the first interrupt with some value
+    partial = graph.invoke(Command(resume="weet"), config=config)
+    print("partial 3", partial)
+    third_history = list(graph.get_state_history(config))
+    remaining_interrupts = partial.get("__interrupt__", [])
+    assert len(remaining_interrupts) == 1
+    assert remaining_interrupts[0].value == "second"
+    assert node_counter == 2
+
+    # invoke with None again. the history should look the same as
+    # the third run
+    partial = graph.invoke(None, config=config)
+    print("partial 4", partial)
+    fourth_history = list(graph.get_state_history(config))
+    remaining_interrupts = partial.get("__interrupt__", [])
+    assert len(remaining_interrupts) == 1
+    assert node_counter == 2
+
+    print("\nthird_history[0].interrupts: ", third_history[0].interrupts)
+    print("fourth_history[0].interrupts: ", fourth_history[0].interrupts)
+    print("third_history[0].tasks: ", third_history[0].tasks)
+    print("fourth_history[0].tasks: ", fourth_history[0].tasks)
+    assert third_history[0].interrupts == fourth_history[0].interrupts
+    assert third_history[0].tasks == fourth_history[0].tasks
+
+    # resume the graph once more with a real value
+    partial = graph.invoke(Command(resume="bix"), config=config)
+    remaining_interrupts = partial.get("__interrupt__", [])
+    assert len(remaining_interrupts) == 0
+    assert node_counter == 3

libs/langgraph/tests/test_large_cases.py

@@ -4726,7 +4726,7 @@ def test_send_dedupe_on_resume(
     assert len(history) == (4 if durability != "exit" else 1)
 
     # resume execution
-    assert graph.invoke(None, thread1, durability=durability) == [
+    assert graph.invoke(Command(resume=""), thread1, durability=durability) == [
         "0",
         "1",
         "3.1",

libs/langgraph/tests/test_pregel_async.py

@@ -2545,7 +2545,7 @@ async def test_send_dedupe_on_resume(
     assert builder.nodes["2"].runnable.func.ticks == 3
     assert builder.nodes["flaky"].runnable.func.ticks == 1
     # resume execution
-    assert await graph.ainvoke(None, thread1, durability=durability) == [
+    assert await graph.ainvoke(Command(resume=""), thread1, durability=durability) == [
         "0",
         "1",
         "3.1",