add overwrite command

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
-- LangSmith Debugger: {debugger_origin}/studio/?baseUrl={debugger_base_url_query}
+- LangGraph Studio: {debugger_origin}/studio/?baseUrl={debugger_base_url_query}
 """
                 )
                 sys.stdout.flush()
@@ -652,17 +652,11 @@ 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="(Deprecated: use --debugger-url instead) URL of the LangSmith Debugger instance to connect to.",
+    help="URL of the LangGraph Studio instance to connect to. Defaults to https://smith.langchain.com",
 )
 @click.option(
     "--allow-blocking",
@@ -698,21 +692,12 @@ 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:
@@ -776,7 +761,7 @@ def dev(
         http=config_json.get("http"),
         ui=config_json.get("ui"),
         ui_config=config_json.get("ui_config"),
-        studio_url=debugger_url,
+        studio_url=studio_url,
         allow_blocking=allow_blocking,
         tunnel=tunnel,
         server_level=server_log_level,

libs/langgraph/langgraph/pregel/__init__.py

@@ -1,3 +1,4 @@
+from langgraph.pregel._write import Overwrite, overwrite
 from langgraph.pregel.main import NodeBuilder, Pregel
 
-__all__ = ("Pregel", "NodeBuilder")
+__all__ = ("Pregel", "NodeBuilder", "overwrite", "Overwrite")

libs/langgraph/langgraph/pregel/_algo.py

@@ -71,6 +71,7 @@ from langgraph.pregel._call import get_runnable_for_task, identifier
 from langgraph.pregel._io import read_channels
 from langgraph.pregel._log import logger
 from langgraph.pregel._read import INPUT_CACHE_KEY_TYPE, PregelNode
+from langgraph.pregel._write import _Overwrite
 from langgraph.runtime import DEFAULT_RUNTIME, Runtime
 from langgraph.types import (
     All,
@@ -198,8 +199,16 @@ def local_read(
         # apply writes
         local_channels: dict[str, BaseChannel] = {}
         for k in channels:
-            cc = channels[k].copy()
-            cc.update(updated[k])
+            if updated[k]:
+                # If any overwrite is present for this channel, reflect it directly
+                ow = next((v for v in updated[k] if isinstance(v, _Overwrite)), None)
+                if ow is not None:
+                    cc = channels[k].from_checkpoint(ow.value)
+                else:
+                    cc = channels[k].copy()
+                    cc.update(updated[k])
+            else:
+                cc = channels[k].copy()
             local_channels[k] = cc
         # read fresh values
         values = read_channels(local_channels, select)
@@ -277,12 +286,16 @@ def apply_writes(
 
     # Group writes by channel
     pending_writes_by_channel: dict[str, list[Any]] = defaultdict(list)
+    overwrite_by_channel: dict[str, Any] = {}
     for task in tasks:
         for chan, val in task.writes:
             if chan in (NO_WRITES, PUSH, RESUME, INTERRUPT, RETURN, ERROR):
-                pass
-            elif chan in channels:
-                pending_writes_by_channel[chan].append(val)
+                continue
+            if chan in channels:
+                if isinstance(val, _Overwrite):
+                    overwrite_by_channel[chan] = val.value
+                else:
+                    pending_writes_by_channel[chan].append(val)
             else:
                 logger.warning(
                     f"Task {task.name} with path {task.path} wrote to unknown channel {chan}, ignoring it."
@@ -290,13 +303,26 @@ def apply_writes(
 
     # Apply writes to channels
     updated_channels: set[str] = set()
-    for chan, vals in pending_writes_by_channel.items():
+    for chan in set(pending_writes_by_channel.keys()) | set(
+        overwrite_by_channel.keys()
+    ):
         if chan in channels:
-            if channels[chan].update(vals) and next_version is not None:
-                checkpoint["channel_versions"][chan] = next_version
-                # unavailable channels can't trigger tasks, so don't add them
+            if chan in overwrite_by_channel:
+                # Overwrite the entire channel value, bypassing reducers.
+                channels[chan] = channels[chan].from_checkpoint(
+                    overwrite_by_channel[chan]
+                )
+                if next_version is not None:
+                    checkpoint["channel_versions"][chan] = next_version
                 if channels[chan].is_available():
                     updated_channels.add(chan)
+            else:
+                vals = pending_writes_by_channel.get(chan, [])
+                if channels[chan].update(vals) and next_version is not None:
+                    checkpoint["channel_versions"][chan] = next_version
+                    # unavailable channels can't trigger tasks, so don't add them
+                    if channels[chan].is_available():
+                        updated_channels.add(chan)
 
     # Channels that weren't updated in this step are notified of a new step
     if bump_step:

libs/langgraph/langgraph/pregel/_write.py

@@ -26,6 +26,32 @@ SKIP_WRITE = object()
 PASSTHROUGH = object()
 
 
+class _Overwrite:
+    """Marker wrapper indicating a direct channel overwrite.
+
+    Use via `overwrite(channel, value)` or `Overwrite(value)`.
+    """
+
+    __slots__ = ("value",)
+
+    def __init__(self, value: Any):
+        self.value = value
+
+
+def Overwrite(value: Any) -> _Overwrite:
+    """Wrap a value to force overwrite a channel, bypassing reducers."""
+    return _Overwrite(value)
+
+
+def overwrite(channel: str, value: Any) -> ChannelWriteEntry:
+    """Convenience factory for a write that overwrites the target channel.
+
+    Example:
+        NodeBuilder().write_to(overwrite("foo", 123))
+    """
+    return ChannelWriteEntry(channel, Overwrite(value))
+
+
 class ChannelWriteEntry(NamedTuple):
     channel: str
     """Channel name to write to."""