Remove UntrackedValue channel

- This is incompatible with distributed execution modes, so needs to go

libs/langgraph/langgraph/channels/__init__.py

@@ -1,15 +1,14 @@
 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
+from langgraph.channels.last_value import LastValue, LastValueAfterFinish
 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

@@ -1,66 +0,0 @@
-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
@@ -1298,7 +1298,7 @@ def test_conditional_state_graph(
     from langchain_core.tools import tool
 
     class AgentState(TypedDict, total=False):
-        input: Annotated[str, UntrackedValue]
+        input: Annotated[str, EphemeralValue]
         agent_outcome: Optional[Union[AgentAction, AgentFinish]]
         intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]
 
@@ -1386,7 +1386,6 @@ 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(
@@ -1819,6 +1818,7 @@ 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.graph import Graph
@@ -1367,7 +1367,7 @@ async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver)
     from langchain_core.tools import tool
 
     class AgentState(TypedDict):
-        input: Annotated[str, UntrackedValue]
+        input: Annotated[str, EphemeralValue]
         agent_outcome: Optional[Union[AgentAction, AgentFinish]]
         intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]
 
@@ -1442,7 +1442,6 @@ 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(