clean up

libs/langgraph/langgraph/graph/__init__.py

@@ -1,5 +1,10 @@
 from langgraph.constants import END, START
-from langgraph.graph.message import MessageGraph, MessagesState, add_messages
+from langgraph.graph.message import (
+    MessageGraph,
+    MessagesState,
+    add_messages,
+    validate_messages_append_only,
+)
 from langgraph.graph.state import StateGraph
 
 __all__ = (
@@ -9,4 +14,5 @@ __all__ = (
     "add_messages",
     "MessagesState",
     "MessageGraph",
+    "validate_messages_append_only",
 )

libs/langgraph/langgraph/graph/message.py

@@ -31,6 +31,7 @@ __all__ = (
     "MessagesState",
     "MessageGraph",
     "REMOVE_ALL_MESSAGES",
+    "validate_messages_append_only",
 )
 
 Messages = list[MessageLikeRepresentation] | MessageLikeRepresentation
@@ -244,6 +245,89 @@ def add_messages(
     return merged
 
 
+def validate_messages_append_only(
+    input: dict[str, Any], current_state: dict[str, Any]
+) -> None:
+    """Validates that incoming messages are append-only (no updates or removals).
+
+    This validator can be passed to `compile(validate_input=...)` to enforce that
+    external inputs only append new messages and never update or remove existing ones.
+
+    Internal node updates bypass this validation, allowing nodes to perform operations
+    like message compaction or cleanup without restriction.
+
+    Args:
+        input: The raw input dictionary being provided externally
+        current_state: The current state loaded from the checkpoint
+
+    Raises:
+        ValueError: If the input attempts to update or remove existing messages
+
+    Example:
+        ```python
+        from langgraph.graph import StateGraph, MessagesState
+        from langgraph.graph.message import validate_messages_append_only
+
+        builder = StateGraph(MessagesState)
+        builder.add_node("chatbot", my_chatbot_node)
+        builder.set_entry_point("chatbot")
+        builder.set_finish_point("chatbot")
+
+        # Compile with validator to enforce append-only at the boundary
+        graph = builder.compile(validate_input=validate_messages_append_only)
+
+        # This works - adding new messages
+        graph.invoke({"messages": [("user", "Hello")]})
+
+        # This raises ValueError - trying to update existing message
+        graph.invoke({"messages": [HumanMessage(content="Modified", id="existing-id")]})
+        ```
+    """
+    # Only validate if input contains messages
+    if "messages" not in input:
+        return
+
+    # Get current messages from state
+    current_messages = current_state.get("messages", [])
+
+    # Build a set of existing message IDs
+    existing_ids: set[str] = set()
+    for msg in current_messages:
+        if hasattr(msg, "id") and msg.id is not None:
+            existing_ids.add(msg.id)
+
+    # Coerce input messages to list
+    input_messages = input["messages"]
+    if not isinstance(input_messages, list):
+        input_messages = [input_messages]
+
+    # Convert to messages to get proper IDs (let conversion errors propagate)
+    converted_messages = convert_to_messages(input_messages)
+
+    # Check each input message
+    for msg in converted_messages:
+        # Check for REMOVE_ALL_MESSAGES
+        if isinstance(msg, RemoveMessage) and msg.id == REMOVE_ALL_MESSAGES:
+            raise ValueError(
+                "Cannot remove all messages in append_only mode. "
+                "External inputs must only append new messages."
+            )
+
+        # Check for removals of existing messages
+        if isinstance(msg, RemoveMessage) and msg.id in existing_ids:
+            raise ValueError(
+                f"Cannot remove existing message with ID '{msg.id}' in append_only mode. "
+                "External inputs must only append new messages."
+            )
+
+        # Check for updates of existing messages
+        if not isinstance(msg, RemoveMessage) and msg.id and msg.id in existing_ids:
+            raise ValueError(
+                f"Cannot update existing message with ID '{msg.id}' in append_only mode. "
+                "External inputs must only append new messages."
+            )
+
+
 @deprecated(
     "MessageGraph is deprecated in langgraph 1.0.0, to be removed in 2.0.0. Please use StateGraph with a `messages` key instead.",
     category=None,

libs/langgraph/langgraph/graph/state.py

@@ -1042,6 +1042,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
         interrupt_after: All | list[str] | None = None,
         debug: bool = False,
         name: str | None = None,
+        validate_input: Callable[[Any, dict[str, Any]], None] | None = None,
     ) -> CompiledStateGraph[StateT, ContextT, InputT, OutputT]:
         """Compiles the `StateGraph` into a `CompiledStateGraph` object.
 
@@ -1074,6 +1075,34 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
             interrupt_after: An optional list of node names to interrupt after.
             debug: A flag indicating whether to enable debug mode.
             name: The name to use for the compiled graph.
+            validate_input: Optional validation function for external inputs.
+                The function receives `(input, current_state)` and should raise an
+                exception if the input is invalid, or return normally if valid.
+
+                This validation only applies to external updates (from `invoke()`, `stream()`,
+                `update_state()`, or `Command.update`) - internal node updates bypass this
+                validation, allowing trusted nodes to perform privileged operations.
+
+                Example:
+                    ```python
+                    from langgraph.graph import StateGraph
+                    from langgraph.graph.message import validate_messages_append_only
+
+                    builder = StateGraph(MessagesState)
+                    # ... add nodes ...
+
+                    # Use built-in validator
+                    graph = builder.compile(
+                        validate_input=validate_messages_append_only
+                    )
+
+                    # Or custom validator
+                    def my_validator(input: dict, current_state: dict) -> None:
+                        if "forbidden_key" in input:
+                            raise ValueError("Forbidden key detected")
+
+                    graph = builder.compile(validate_input=my_validator)
+                    ```
 
         Returns:
             CompiledStateGraph: The compiled `StateGraph`.
@@ -1134,6 +1163,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
             store=store,
             cache=cache,
             name=name or "LangGraph",
+            validate_input=validate_input,
         )
 
         compiled.attach_node(START, None)

libs/langgraph/langgraph/pregel/_loop.py

@@ -226,6 +226,7 @@ class PregelLoop:
         migrate_checkpoint: Callable[[Checkpoint], None] | None = None,
         retry_policy: Sequence[RetryPolicy] = (),
         cache_policy: CachePolicy | None = None,
+        validate_input: Callable[[Any, dict[str, Any]], None] | None = None,
     ) -> None:
         self.stream = stream
         self.config = config
@@ -250,6 +251,7 @@ class PregelLoop:
         self.retry_policy = retry_policy
         self.cache_policy = cache_policy
         self.durability = durability
+        self.validate_input = validate_input
         if self.stream is not None and CONFIG_KEY_STREAM in config[CONF]:
             self.stream = DuplexStream(self.stream, config[CONF][CONFIG_KEY_STREAM])
         scratchpad: PregelScratchpad | None = config[CONF].get(CONFIG_KEY_SCRATCHPAD)
@@ -637,6 +639,14 @@ class PregelLoop:
 
         # map command to writes
         if isinstance(self.input, Command):
+            # Validate command update if validator is provided
+            if self.validate_input is not None and self.input.update:
+                current_state = read_channels(self.channels, self.output_keys)
+                try:
+                    self.validate_input(self.input.update, current_state)
+                except Exception as e:
+                    raise ValueError(f"Input validation failed: {str(e)}") from e
+
             if (resume := self.input.resume) is not None:
                 if not self.checkpointer:
                     raise RuntimeError(
@@ -691,6 +701,13 @@ class PregelLoop:
             )
         # map inputs to channel updates
         elif input_writes := deque(map_input(input_keys, self.input)):
+            # Validate input if validator is provided
+            if self.validate_input is not None:
+                current_state = read_channels(self.channels, self.output_keys)
+                try:
+                    self.validate_input(self.input, current_state)
+                except Exception as e:
+                    raise ValueError(f"Input validation failed: {str(e)}") from e
             # discard any unfinished tasks from previous checkpoint
             discard_tasks = prepare_next_tasks(
                 self.checkpoint,
@@ -985,6 +1002,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
         migrate_checkpoint: Callable[[Checkpoint], None] | None = None,
         retry_policy: Sequence[RetryPolicy] = (),
         cache_policy: CachePolicy | None = None,
+        validate_input: Callable[[Any, dict[str, Any]], None] | None = None,
     ) -> None:
         super().__init__(
             input,
@@ -1006,6 +1024,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             retry_policy=retry_policy,
             cache_policy=cache_policy,
             durability=durability,
+            validate_input=validate_input,
         )
         self.stack = ExitStack()
         if checkpointer:
@@ -1161,6 +1180,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
         migrate_checkpoint: Callable[[Checkpoint], None] | None = None,
         retry_policy: Sequence[RetryPolicy] = (),
         cache_policy: CachePolicy | None = None,
+        validate_input: Callable[[Any, dict[str, Any]], None] | None = None,
     ) -> None:
         super().__init__(
             input,
@@ -1182,6 +1202,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             retry_policy=retry_policy,
             cache_policy=cache_policy,
             durability=durability,
+            validate_input=validate_input,
         )
         self.stack = AsyncExitStack()
         if checkpointer:

libs/langgraph/langgraph/pregel/main.py

@@ -652,6 +652,7 @@ class Pregel(
         config: RunnableConfig | None = None,
         trigger_to_nodes: Mapping[str, Sequence[str]] | None = None,
         name: str = "LangGraph",
+        validate_input: Callable[[Any, dict[str, Any]], None] | None = None,
         **deprecated_kwargs: Unpack[DeprecatedKwargs],
     ) -> None:
         if (
@@ -698,6 +699,7 @@ class Pregel(
         self.config = config
         self.trigger_to_nodes = trigger_to_nodes or {}
         self.name = name
+        self.validate_input = validate_input
         if auto_validate:
             self.validate()
 
@@ -2599,6 +2601,7 @@ class Pregel(
                 migrate_checkpoint=self._migrate_checkpoint,
                 retry_policy=self.retry_policy,
                 cache_policy=self.cache_policy,
+                validate_input=self.validate_input,
             ) as loop:
                 # create runner
                 runner = PregelRunner(
@@ -2908,6 +2911,7 @@ class Pregel(
                 migrate_checkpoint=self._migrate_checkpoint,
                 retry_policy=self.retry_policy,
                 cache_policy=self.cache_policy,
+                validate_input=self.validate_input,
             ) as loop:
                 # create runner
                 runner = PregelRunner(

libs/langgraph/tests/test_messages_state.py

@@ -20,7 +20,7 @@ from langgraph.graph.message import REMOVE_ALL_MESSAGES, MessagesState, push_mes
 from langgraph.graph.state import StateGraph
 from tests.messages import _AnyIdHumanMessage
 
-_, CORE_MINOR, CORE_PATCH = (
+CORE_MAJOR, CORE_MINOR, CORE_PATCH = (
     int("".join(c for c in v if c.isdigit()))
     for v in langchain_core.__version__.split(".")
 )
@@ -205,7 +205,11 @@ def test_messages_state(state_schema):
 
 
 @pytest.mark.skipif(
-    condition=not ((CORE_MINOR == 3 and CORE_PATCH >= 11) or CORE_MINOR > 3),
+    condition=not (
+        (CORE_MAJOR == 0 and CORE_MINOR == 3 and CORE_PATCH >= 11)
+        or (CORE_MAJOR == 0 and CORE_MINOR > 3)
+        or CORE_MAJOR > 0
+    ),
     reason="Requires langchain_core>=0.3.11.",
 )
 def test_messages_state_format_openai():

libs/langgraph/tests/test_validate_input_callable.py

@@ -0,0 +1,173 @@
+"""Tests for validate_input with callable approach."""
+
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
+from langgraph.checkpoint.memory import MemorySaver
+
+from langgraph.graph import StateGraph
+from langgraph.graph.message import MessagesState, validate_messages_append_only
+from langgraph.types import Command
+
+
+def test_validate_messages_append_only_allows_new():
+    """Test that validate_messages_append_only allows adding new messages."""
+
+    def chatbot(state: MessagesState) -> MessagesState:
+        return {"messages": [AIMessage(content="Response")]}
+
+    builder = StateGraph(MessagesState)
+    builder.add_node("chatbot", chatbot)
+    builder.set_entry_point("chatbot")
+    builder.set_finish_point("chatbot")
+
+    graph = builder.compile(validate_input=validate_messages_append_only)
+
+    result = graph.invoke({"messages": [HumanMessage(content="Hello", id="1")]})
+    assert len(result["messages"]) == 2
+
+
+def test_validate_messages_append_only_blocks_updates():
+    """Test that validate_messages_append_only blocks message updates."""
+
+    def chatbot(state: MessagesState) -> MessagesState:
+        return {"messages": [AIMessage(content="Response")]}
+
+    builder = StateGraph(MessagesState)
+    builder.add_node("chatbot", chatbot)
+    builder.set_entry_point("chatbot")
+    builder.set_finish_point("chatbot")
+
+    checkpointer = MemorySaver()
+    graph = builder.compile(
+        checkpointer=checkpointer, validate_input=validate_messages_append_only
+    )
+
+    config = {"configurable": {"thread_id": "test"}}
+    graph.invoke({"messages": [HumanMessage(content="Hello", id="1")]}, config)
+
+    # Try to update - should fail
+    with pytest.raises(ValueError, match="Cannot update existing message"):
+        graph.invoke({"messages": [HumanMessage(content="Modified", id="1")]}, config)
+
+
+def test_validate_messages_append_only_blocks_removals():
+    """Test that validate_messages_append_only blocks removals."""
+
+    def chatbot(state: MessagesState) -> MessagesState:
+        return {"messages": []}
+
+    builder = StateGraph(MessagesState)
+    builder.add_node("chatbot", chatbot)
+    builder.set_entry_point("chatbot")
+    builder.set_finish_point("chatbot")
+
+    checkpointer = MemorySaver()
+    graph = builder.compile(
+        checkpointer=checkpointer, validate_input=validate_messages_append_only
+    )
+
+    config = {"configurable": {"thread_id": "test"}}
+    graph.invoke({"messages": [HumanMessage(content="Hello", id="1")]}, config)
+
+    # Try to remove - should fail
+    with pytest.raises(ValueError, match="Cannot remove existing message"):
+        graph.invoke({"messages": [RemoveMessage(id="1")]}, config)
+
+
+def test_validate_with_command_update():
+    """Test that Command.update is also validated."""
+
+    def chatbot(state: MessagesState) -> MessagesState:
+        return {"messages": [AIMessage(content="Response")]}
+
+    builder = StateGraph(MessagesState)
+    builder.add_node("chatbot", chatbot)
+    builder.set_entry_point("chatbot")
+    builder.set_finish_point("chatbot")
+
+    checkpointer = MemorySaver()
+    graph = builder.compile(
+        checkpointer=checkpointer, validate_input=validate_messages_append_only
+    )
+
+    config = {"configurable": {"thread_id": "test"}}
+    graph.invoke({"messages": [HumanMessage(content="Hello", id="1")]}, config)
+
+    # Try to update via Command - should fail
+    with pytest.raises(ValueError, match="Cannot update existing message"):
+        graph.invoke(
+            Command(update={"messages": [HumanMessage(content="Modified", id="1")]}),
+            config,
+        )
+
+
+def test_custom_validator():
+    """Test with custom validation function."""
+
+    def my_validator(input: dict, current_state: dict) -> None:
+        if "forbidden" in str(input):
+            raise ValueError("Forbidden content")
+
+    def node(state: MessagesState) -> MessagesState:
+        return {"messages": [AIMessage(content="OK")]}
+
+    builder = StateGraph(MessagesState)
+    builder.add_node("node", node)
+    builder.set_entry_point("node")
+    builder.set_finish_point("node")
+
+    graph = builder.compile(validate_input=my_validator)
+
+    # Should work
+    graph.invoke({"messages": [("user", "Hello")]})
+
+    # Should fail
+    with pytest.raises(ValueError, match="Forbidden content"):
+        graph.invoke({"messages": [("user", "forbidden word")]})
+
+
+def test_node_bypasses_validation():
+    """Test that internal node operations bypass validation."""
+
+    def node_that_modifies(state: MessagesState) -> MessagesState:
+        # Node modifies existing message - should work even with strict validator
+        messages = state["messages"]
+        if messages:
+            updated = messages[0].model_copy(update={"content": "Modified by node"})
+            return {"messages": [updated]}
+        return {"messages": []}
+
+    builder = StateGraph(MessagesState)
+    builder.add_node("modifier", node_that_modifies)
+    builder.set_entry_point("modifier")
+    builder.set_finish_point("modifier")
+
+    graph = builder.compile(validate_input=validate_messages_append_only)
+
+    # Node can modify - validation only applies to external input
+    result = graph.invoke({"messages": [HumanMessage(content="Original", id="1")]})
+    assert any("Modified by node" in m.content for m in result["messages"])
+
+
+def test_without_validator_allows_everything():
+    """Test backwards compatibility - without validator, everything is allowed."""
+
+    def chatbot(state: MessagesState) -> MessagesState:
+        return {"messages": []}
+
+    builder = StateGraph(MessagesState)
+    builder.add_node("chatbot", chatbot)
+    builder.set_entry_point("chatbot")
+    builder.set_finish_point("chatbot")
+
+    checkpointer = MemorySaver()
+    graph = builder.compile(checkpointer=checkpointer)  # No validator
+
+    config = {"configurable": {"thread_id": "test"}}
+    graph.invoke({"messages": [HumanMessage(content="Hello", id="1")]}, config)
+
+    # Should allow update (old behavior)
+    result = graph.invoke(
+        {"messages": [HumanMessage(content="Modified", id="1")]}, config
+    )
+    assert result["messages"][0].content == "Modified"