fix(langgraph): persist a failure record when recovery finds no checkpoint

libs/langgraph/langgraph/pregel/_loop.py

@@ -1032,7 +1032,12 @@ def _first(
             self.updated_channels = updated_channels
             self._put_checkpoint({"source": "input"})
         elif CONFIG_KEY_RESUMING not in configurable:
-            raise EmptyInputError(f"Received no input for {input_keys}")
+            err = EmptyInputError(f"Received no input for {input_keys}")
+            # Recovery of a known thread_id with no durable checkpoint
+            # (crash before the first put landed). Persist a failure
+            # record so fire-and-forget callers can observe the loss.
+            self._persist_empty_resume(repr(err))
+            raise err
         # Propagate resuming and replaying flags to subgraphs.
         if not self.is_nested:
             # Pass the resolved before-bound checkpoint ID so subgraphs can
@@ -1078,6 +1083,19 @@ def _first(
             self._push_graph_lifecycle_event("resume")
         return updated_channels
 
+    def _persist_empty_resume(self, error: str) -> None:
+        """Write a durable aborted record when resume finds no checkpoint.
+
+        `invoke(None, config)` is the recovery path. If the original run died
+        before its first `put()`, the thread has no checkpoint and no input to
+        replay. Without this write, recovery raises `EmptyInputError` and
+        leaves no durable evidence that the accepted run was lost.
+        """
+        if self.checkpointer is None or self._has_persisted_parent:
+            return
+        self._put_checkpoint({"source": "loop"})
+        self.put_writes(NULL_TASK_ID, [(ERROR, error)])
+
     def _put_checkpoint(self, metadata: CheckpointMetadata) -> None:
         # `is` (object identity) — not `==`. Three of four call sites pass a
         # fresh dict ({"source":"input"|"loop"|"fork"}); only
@@ -1700,12 +1718,26 @@ def __enter__(self) -> Self:
         self.step = self.checkpoint_metadata["step"] + 1
         self.stop = self.step + self.config["recursion_limit"] + 1
         self.checkpoint_previous_versions = self.checkpoint["channel_versions"].copy()
-        self.updated_channels = self._first(
-            input_keys=self.input_keys,
-            updated_channels=set(self.checkpoint.get("updated_channels"))  # type: ignore[arg-type]
-            if self.checkpoint.get("updated_channels")
-            else None,
-        )
+        try:
+            self.updated_channels = self._first(
+                input_keys=self.input_keys,
+                updated_channels=set(self.checkpoint.get("updated_channels"))  # type: ignore[arg-type]
+                if self.checkpoint.get("updated_channels")
+                else None,
+            )
+            # durability="sync": the input / accepted checkpoint must be
+            # durable before the first user node runs. The post-tick wait
+            # in Pregel.stream() is too late — a crash in that window
+            # leaves no checkpoint for recovery.
+            if self.durability == "sync" and (
+                fut := getattr(self, "_put_checkpoint_fut", None)
+            ):
+                fut.result()
+        except BaseException as exc:
+            # __enter__ raising skips __exit__; unwind so background puts
+            # (including an empty-resume failure record) finish.
+            self.stack.__exit__(type(exc), exc, exc.__traceback__)
+            raise
 
         return self
 
@@ -1960,12 +1992,24 @@ async def __aenter__(self) -> Self:
         self.step = self.checkpoint_metadata["step"] + 1
         self.stop = self.step + self.config["recursion_limit"] + 1
         self.checkpoint_previous_versions = self.checkpoint["channel_versions"].copy()
-        self.updated_channels = self._first(
-            input_keys=self.input_keys,
-            updated_channels=set(self.checkpoint.get("updated_channels"))  # type: ignore[arg-type]
-            if self.checkpoint.get("updated_channels")
-            else None,
-        )
+        try:
+            self.updated_channels = self._first(
+                input_keys=self.input_keys,
+                updated_channels=set(self.checkpoint.get("updated_channels"))  # type: ignore[arg-type]
+                if self.checkpoint.get("updated_channels")
+                else None,
+            )
+            # durability="sync": the input / accepted checkpoint must be
+            # durable before the first user node runs.
+            if self.durability == "sync" and (
+                fut := getattr(self, "_put_checkpoint_fut", None)
+            ):
+                await fut
+        except BaseException as exc:
+            # __aenter__ raising skips __aexit__; unwind so background puts
+            # (including an empty-resume failure record) finish.
+            await self.stack.__aexit__(type(exc), exc, exc.__traceback__)
+            raise
 
         return self
 

libs/langgraph/tests/test_pregel.py

@@ -62,7 +62,12 @@
 from langgraph.channels.topic import Topic
 from langgraph.channels.untracked_value import UntrackedValue
 from langgraph.config import get_stream_writer
-from langgraph.errors import GraphRecursionError, InvalidUpdateError, ParentCommand
+from langgraph.errors import (
+    EmptyInputError,
+    GraphRecursionError,
+    InvalidUpdateError,
+    ParentCommand,
+)
 from langgraph.func import entrypoint, task
 from langgraph.graph import END, START, StateGraph
 from langgraph.graph.message import MessagesState, _messages_delta_reducer, add_messages
@@ -5432,6 +5437,65 @@ def second_node(state: State):
     assert [*graph.get_state_history(config)] == []
 
 
+def test_crash_before_first_checkpoint_records_failure_on_recovery() -> None:
+    """If the first durable checkpoint never lands, recovery must persist a
+    failure record instead of raising EmptyInputError with an empty thread.
+
+    Regression for https://github.com/langchain-ai/langgraph/issues/8764.
+    """
+
+    class CrashBeforeFirstPut(InMemorySaver):
+        def __init__(self) -> None:
+            super().__init__()
+            self.allow_put = False
+
+        def put(
+            self,
+            config: RunnableConfig,
+            checkpoint: Checkpoint,
+            metadata: CheckpointMetadata,
+            new_versions: dict[str, str | int | float] | None = None,
+        ) -> RunnableConfig:
+            if not self.allow_put:
+                raise RuntimeError("crash before first checkpoint")
+            return super().put(config, checkpoint, metadata, new_versions)
+
+    class State(TypedDict):
+        done: bool
+
+    effects: list[str] = []
+
+    def node(state: State) -> State:
+        effects.append("effect")
+        return {"done": True}
+
+    builder = StateGraph(State)
+    builder.add_node("node", node)
+    builder.add_edge(START, "node")
+    saver = CrashBeforeFirstPut()
+    app = builder.compile(checkpointer=saver)
+    config: RunnableConfig = {"configurable": {"thread_id": "accepted-run"}}
+
+    with pytest.raises(RuntimeError, match="crash before first checkpoint"):
+        app.invoke({"done": False}, config, durability="sync")
+
+    assert effects == []
+    assert saver.get_tuple(config) is None
+
+    saver.allow_put = True
+    with pytest.raises(EmptyInputError, match="Received no input"):
+        app.invoke(None, config, durability="sync")
+
+    saved = saver.get_tuple(config)
+    assert saved is not None
+    assert saved.pending_writes
+    assert any(w[1] == ERROR for w in saved.pending_writes)
+
+    # A later invoke with input can still start a fresh run on this thread.
+    assert app.invoke({"done": False}, config, durability="sync") == {"done": True}
+    assert effects == ["effect"]
+
+
 def test_multiple_updates_root() -> None:
     def node_a(state):
         return [Command(update="a1"), Command(update="a2")]

libs/langgraph/tests/test_pregel_async.py

@@ -61,6 +61,7 @@
 from langgraph.channels.last_value import LastValue
 from langgraph.channels.topic import Topic
 from langgraph.errors import (
+    EmptyInputError,
     GraphRecursionError,
     InvalidUpdateError,
     NodeError,
@@ -6680,6 +6681,66 @@ async def second_node(state: State):
     assert "RuntimeError('Simulated failure')" in failed_checkpoint.tasks[0].error
 
 
+async def test_crash_before_first_checkpoint_records_failure_on_recovery() -> None:
+    """If the first durable checkpoint never lands, recovery must persist a
+    failure record instead of raising EmptyInputError with an empty thread.
+
+    Regression for https://github.com/langchain-ai/langgraph/issues/8764.
+    """
+
+    class CrashBeforeFirstPut(InMemorySaver):
+        def __init__(self) -> None:
+            super().__init__()
+            self.allow_put = False
+
+        def put(
+            self,
+            config: RunnableConfig,
+            checkpoint: Checkpoint,
+            metadata: CheckpointMetadata,
+            new_versions: ChannelVersions | None = None,
+        ) -> RunnableConfig:
+            if not self.allow_put:
+                raise RuntimeError("crash before first checkpoint")
+            return super().put(config, checkpoint, metadata, new_versions)
+
+    class State(TypedDict):
+        done: bool
+
+    effects: list[str] = []
+
+    async def node(state: State) -> State:
+        effects.append("effect")
+        return {"done": True}
+
+    builder = StateGraph(State)
+    builder.add_node("node", node)
+    builder.add_edge(START, "node")
+    saver = CrashBeforeFirstPut()
+    app = builder.compile(checkpointer=saver)
+    config: RunnableConfig = {"configurable": {"thread_id": "accepted-run"}}
+
+    with pytest.raises(RuntimeError, match="crash before first checkpoint"):
+        await app.ainvoke({"done": False}, config, durability="sync")
+
+    assert effects == []
+    assert await saver.aget_tuple(config) is None
+
+    saver.allow_put = True
+    with pytest.raises(EmptyInputError, match="Received no input"):
+        await app.ainvoke(None, config, durability="sync")
+
+    saved = await saver.aget_tuple(config)
+    assert saved is not None
+    assert saved.pending_writes
+    assert any(w[1] == ERROR for w in saved.pending_writes)
+
+    assert await app.ainvoke({"done": False}, config, durability="sync") == {
+        "done": True
+    }
+    assert effects == ["effect"]
+
+
 async def test_multiple_updates_root() -> None:
     def node_a(state):
         return [Command(update="a1"), Command(update="a2")]