refactor: extract _track helper for available_channels sync in apply_writes

Replace 5 repeated inline blocks that sync available_channels with
a local _track() helper that checks is_available() and updates the
set, returning the availability bool for callers that also need to
update updated_channels.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

libs/langgraph/langgraph/graph/message.py

@@ -191,14 +191,8 @@ def add_messages(
     if not isinstance(right, list):
         right = [right]  # type: ignore[assignment]
     # coerce to message
-    left = [
-        message_chunk_to_message(cast(BaseMessageChunk, m))
-        for m in convert_to_messages(left)
-    ]
-    right = [
-        message_chunk_to_message(cast(BaseMessageChunk, m))
-        for m in convert_to_messages(right)
-    ]
+    left = [message_chunk_to_message(m) for m in convert_to_messages(left)]
+    right = [message_chunk_to_message(m) for m in convert_to_messages(right)]
     # assign missing ids
     for m in left:
         if m.id is None:

libs/langgraph/langgraph/pregel/_algo.py

@@ -235,6 +235,7 @@ def apply_writes(
     tasks: Iterable[WritesProtocol],
     get_next_version: GetNextVersion | None,
     trigger_to_nodes: Mapping[str, Sequence[str]],
+    available_channels: set[str] | None = None,
 ) -> set[str]:
     """Apply writes from a set of tasks (usually the tasks from a Pregel step)
     to the checkpoint and channels, and return managed values writes to be applied
@@ -281,6 +282,18 @@ def apply_writes(
             None,
         )
 
+    # Sync available_channels with channel's actual availability state.
+    # Returns True if the channel is available (for callers that also need
+    # to update updated_channels).
+    def _track(chan: str) -> bool:
+        avail = channels[chan].is_available()
+        if available_channels is not None:
+            if avail:
+                available_channels.add(chan)
+            else:
+                available_channels.discard(chan)
+        return avail
+
     # Consume all channels that were read
     for chan in {
         chan
@@ -290,6 +303,7 @@ def apply_writes(
     }:
         if channels[chan].consume() and next_version is not None:
             checkpoint["channel_versions"][chan] = next_version
+        _track(chan)
 
     # Group writes by channel
     pending_writes_by_channel: dict[str, list[Any]] = defaultdict(list)
@@ -319,18 +333,28 @@ def apply_writes(
             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():
+                if _track(chan):
                     updated_channels.add(chan)
+            else:
+                _track(chan)
 
     # Channels that weren't updated in this step are notified of a new step
     if bump_step:
-        for chan in channels:
-            if channels[chan].is_available() and chan not in updated_channels:
-                if channels[chan].update(EMPTY_SEQ) 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)
+        candidates = (
+            available_channels - updated_channels
+            if available_channels is not None
+            else (
+                chan
+                for chan in channels
+                if channels[chan].is_available() and chan not in updated_channels
+            )
+        )
+        for chan in candidates:
+            if channels[chan].update(EMPTY_SEQ) and next_version is not None:
+                checkpoint["channel_versions"][chan] = next_version
+                # unavailable channels can't trigger tasks, so don't add them
+                if _track(chan):
+                    updated_channels.add(chan)
 
     # If this is (tentatively) the last superstep, notify all channels of finish
     if bump_step and updated_channels.isdisjoint(trigger_to_nodes):
@@ -338,8 +362,10 @@ def apply_writes(
             if channels[chan].finish() 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():
+                if _track(chan):
                     updated_channels.add(chan)
+            else:
+                _track(chan)
 
     # Return managed values writes to be applied externally
     return updated_channels
@@ -517,7 +543,7 @@ PUSH_TRIGGER = (PUSH,)
 
 
 class _TaskIDFn(Protocol):
-    def __call__(self, namespace: bytes, *parts: str | bytes) -> str:
+    def __call__(self, namespace: bytes, *parts: str) -> str:
         pass
 
 
@@ -1392,32 +1418,37 @@ def _proc_input(
     return val
 
 
-def _uuid5_str(namespace: bytes, *parts: str | bytes) -> str:
+def _uuid5_str(namespace: bytes, *parts: str) -> str:
     """Generate a UUID from the SHA-1 hash of a namespace and str parts."""
 
     sha = sha1(namespace, usedforsecurity=False)
-    sha.update(b"".join(p.encode() if isinstance(p, str) else p for p in parts))
+    sha.update(b"".join(p.encode() for p in parts))
     hex = sha.hexdigest()
     return f"{hex[:8]}-{hex[8:12]}-{hex[12:16]}-{hex[16:20]}-{hex[20:32]}"
 
 
-def _xxhash_str(namespace: bytes, *parts: str | bytes) -> str:
+def _xxhash_str(namespace: bytes, *parts: str) -> str:
     """Generate a UUID from the XXH3 hash of a namespace and str parts."""
-    hex = xxh3_128_hexdigest(
-        namespace + b"".join(p.encode() if isinstance(p, str) else p for p in parts)
-    )
+    hex = xxh3_128_hexdigest(namespace + b"".join(p.encode() for p in parts))
     return f"{hex[:8]}-{hex[8:12]}-{hex[12:16]}-{hex[16:20]}-{hex[20:32]}"
 
 
-def task_path_str(tup: str | int | tuple) -> str:
+def task_path_str(tup: str | int | tuple | list) -> str:
     """Generate a string representation of the task path."""
-    return (
-        f"~{', '.join(task_path_str(x) for x in tup)}"
-        if isinstance(tup, (tuple, list))
-        else f"{tup:010d}"
-        if isinstance(tup, int)
-        else str(tup)
-    )
+    if isinstance(tup, (tuple, list)):
+        parts: list[str] = []
+        for x in tup:
+            if isinstance(x, int):
+                parts.append(f"{x:010d}")
+            elif isinstance(x, (tuple, list)):
+                parts.append(task_path_str(x))
+            else:
+                parts.append(str(x))
+        return f"~{', '.join(parts)}"
+    elif isinstance(tup, int):
+        return f"{tup:010d}"
+    else:
+        return str(tup)
 
 
 LAZY_ATOMIC_COUNTER_LOCK = threading.Lock()

libs/langgraph/langgraph/pregel/_loop.py

@@ -198,6 +198,7 @@ class PregelLoop:
     _migrate_checkpoint: Callable[[Checkpoint], None] | None
     submit: Submit
     channels: Mapping[str, BaseChannel]
+    _available_channels: set[str]
     # Futures from `checkpointer.put_writes` calls that produced delta-channel
     # writes. `_checkpointer_put_after_previous` drains this list (swap to a
     # local `futs` then reset to `[]` and wait/gather) before putting the
@@ -695,6 +696,7 @@ class PregelLoop:
             self.tasks.values(),
             self.checkpointer_get_next_version,
             self.trigger_to_nodes,
+            available_channels=self._available_channels,
         )
         # produce values output
         if not self.updated_channels.isdisjoint(
@@ -939,6 +941,7 @@ class PregelLoop:
                 [PregelTaskWrites((), INPUT, null_writes, [])],
                 self.checkpointer_get_next_version,
                 self.trigger_to_nodes,
+                available_channels=self._available_channels,
             )
             if updated_channels is not None:
                 updated_channels.update(null_updated_channels)
@@ -1006,6 +1009,7 @@ class PregelLoop:
                 ],
                 self.checkpointer_get_next_version,
                 self.trigger_to_nodes,
+                available_channels=self._available_channels,
             )
             # Input writes go through `apply_writes` directly (above) — they
             # never enter `checkpoint_pending_writes`, so the after_tick
@@ -1349,6 +1353,7 @@ class PregelLoop:
                     self.tasks.values(),
                     self.checkpointer_get_next_version,
                     self.trigger_to_nodes,
+                    available_channels=self._available_channels,
                 )
                 if not updated_channels.isdisjoint(
                     (self.output_keys,)
@@ -1695,6 +1700,9 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             saver=self.checkpointer,
             config=self.checkpoint_config,
         )
+        self._available_channels: set[str] = {
+            k for k, v in self.channels.items() if v.is_available()
+        }
         self.stack.push(self._suppress_interrupt)
         self.status = "input"
         self.step = self.checkpoint_metadata["step"] + 1
@@ -1955,6 +1963,9 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             saver=self.checkpointer,
             config=self.checkpoint_config,
         )
+        self._available_channels: set[str] = {
+            k for k, v in self.channels.items() if v.is_available()
+        }
         self.stack.push(self._suppress_interrupt)
         self.status = "input"
         self.step = self.checkpoint_metadata["step"] + 1

libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py

@@ -638,15 +638,18 @@ def create_react_agent(
             messages = (
                 _get_state_value(state, "llm_input_messages")
             ) or _get_state_value(state, "messages")
-            error_msg = f"Expected input to call_model to have 'llm_input_messages' or 'messages' key, but got {state}"
         else:
             messages = _get_state_value(state, "messages")
-            error_msg = (
-                f"Expected input to call_model to have 'messages' key, but got {state}"
-            )
 
         if messages is None:
-            raise ValueError(error_msg)
+            if pre_model_hook is not None:
+                raise ValueError(
+                    f"Expected input to call_model to have 'llm_input_messages' or 'messages' key, but got {state}"
+                )
+            else:
+                raise ValueError(
+                    f"Expected input to call_model to have 'messages' key, but got {state}"
+                )
 
         _validate_chat_history(messages)
         # we're passing messages under `messages` key, as this is expected by the prompt