Revert "Remove UntrackedValue channel"

This reverts commit 05f3904d09.

libs/cli/uv.lock

@@ -501,7 +501,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-cli"
-version = "0.3.2"
+version = "0.3.3"
 source = { editable = "." }
 dependencies = [
     { name = "click" },

libs/langgraph/langgraph/channels/__init__.py

@@ -1,14 +1,15 @@
 from langgraph.channels.any_value import AnyValue
 from langgraph.channels.binop import BinaryOperatorAggregate
 from langgraph.channels.ephemeral_value import EphemeralValue
-from langgraph.channels.last_value import LastValue, LastValueAfterFinish
+from langgraph.channels.last_value import LastValue
 from langgraph.channels.topic import Topic
+from langgraph.channels.untracked_value import UntrackedValue
 
 __all__ = [
     "LastValue",
-    "LastValueAfterFinish",
     "Topic",
     "BinaryOperatorAggregate",
+    "UntrackedValue",
     "EphemeralValue",
     "AnyValue",
 ]

libs/langgraph/langgraph/channels/untracked_value.py

@@ -0,0 +1,66 @@
+from collections.abc import Sequence
+from typing import Generic
+
+from typing_extensions import Self
+
+from langgraph.channels.base import BaseChannel, Value
+from langgraph.constants import MISSING
+from langgraph.errors import EmptyChannelError, InvalidUpdateError
+
+
+class UntrackedValue(Generic[Value], BaseChannel[Value, Value, Value]):
+    """Stores the last value received, never checkpointed."""
+
+    __slots__ = ("value", "guard")
+
+    def __init__(self, typ: type[Value], guard: bool = True) -> None:
+        super().__init__(typ)
+        self.guard = guard
+        self.value = MISSING
+
+    def __eq__(self, value: object) -> bool:
+        return isinstance(value, UntrackedValue) and value.guard == self.guard
+
+    @property
+    def ValueType(self) -> type[Value]:
+        """The type of the value stored in the channel."""
+        return self.typ
+
+    @property
+    def UpdateType(self) -> type[Value]:
+        """The type of the update received by the channel."""
+        return self.typ
+
+    def copy(self) -> Self:
+        """Return a copy of the channel."""
+        empty = self.__class__(self.typ, self.guard)
+        empty.key = self.key
+        empty.value = self.value
+        return empty
+
+    def checkpoint(self) -> Value:
+        return MISSING
+
+    def from_checkpoint(self, checkpoint: Value) -> Self:
+        empty = self.__class__(self.typ, self.guard)
+        empty.key = self.key
+        return empty
+
+    def update(self, values: Sequence[Value]) -> bool:
+        if len(values) == 0:
+            return False
+        if len(values) != 1 and self.guard:
+            raise InvalidUpdateError(
+                f"At key '{self.key}': UntrackedValue(guard=True) can receive only one value per step. Use guard=False if you want to store any one of multiple values."
+            )
+
+        self.value = values[-1]
+        return True
+
+    def get(self) -> Value:
+        if self.value is MISSING:
+            raise EmptyChannelError()
+        return self.value
+
+    def is_available(self) -> bool:
+        return self.value is not MISSING

libs/langgraph/tests/test_large_cases.py

@@ -11,8 +11,8 @@ from pytest_mock import MockerFixture
 from syrupy import SnapshotAssertion
 from typing_extensions import TypedDict
 
-from langgraph.channels.ephemeral_value import EphemeralValue
 from langgraph.channels.last_value import LastValue
+from langgraph.channels.untracked_value import UntrackedValue
 from langgraph.checkpoint.base import BaseCheckpointSaver
 from langgraph.checkpoint.memory import InMemorySaver
 from langgraph.constants import END, PULL, PUSH, START
@@ -502,7 +502,7 @@ def test_conditional_state_graph(
     from langchain_core.tools import tool
 
     class AgentState(TypedDict, total=False):
-        input: Annotated[str, EphemeralValue]
+        input: Annotated[str, UntrackedValue]
         agent_outcome: Optional[Union[AgentAction, AgentFinish]]
         intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]
 
@@ -590,6 +590,7 @@ def test_conditional_state_graph(
         assert app.get_graph().draw_mermaid(with_styles=False) == snapshot
 
     assert app.invoke({"input": "what is weather in sf"}) == {
+        "input": "what is weather in sf",
         "intermediate_steps": [
             [
                 AgentAction(
@@ -1022,7 +1023,6 @@ def test_conditional_state_graph(
 
     assert app_w_interrupt.get_state(config) == StateSnapshot(
         values={
-            "input": "what is weather in sf",
             "intermediate_steps": [],
         },
         tasks=(PregelTask(AnyStr(), "agent", (PULL, "agent")),),

libs/langgraph/tests/test_large_cases_async.py

@@ -16,8 +16,8 @@ from langchain_core.runnables import RunnableConfig, RunnablePick
 from pytest_mock import MockerFixture
 from typing_extensions import TypedDict
 
-from langgraph.channels.ephemeral_value import EphemeralValue
 from langgraph.channels.last_value import LastValue
+from langgraph.channels.untracked_value import UntrackedValue
 from langgraph.checkpoint.base import BaseCheckpointSaver
 from langgraph.constants import END, PULL, PUSH, START
 from langgraph.graph.message import MessageGraph, add_messages
@@ -500,7 +500,7 @@ async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver)
     from langchain_core.tools import tool
 
     class AgentState(TypedDict):
-        input: Annotated[str, EphemeralValue]
+        input: Annotated[str, UntrackedValue]
         agent_outcome: Optional[Union[AgentAction, AgentFinish]]
         intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]
 
@@ -575,6 +575,7 @@ async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver)
     app = workflow.compile()
 
     assert await app.ainvoke({"input": "what is weather in sf"}) == {
+        "input": "what is weather in sf",
         "intermediate_steps": [
             [
                 AgentAction(