0.3.27

- This provides a new mode of execution where only the last checkpoint
is saved
- We save the last checkpoint no matter how the agent run is terminated
(success, error, interrupt, etc)
- This cuts down on cpu time spent on checkpointing, while not losing
any resilience benefits, given individual task writes are still saved
- If an error occurs and the run is retried, any tasks that completed
successfully before will be skipped (as currently)
- checkpoint_during=True is useful when you want to time-travel to inner
steps of a run
- The default value will remain the current behavior, ie.
checkpoint_during=True

libs/langgraph/bench/__main__.py

@@ -26,6 +26,7 @@ async def arun(graph: Pregel, input: dict):
                     "configurable": {"thread_id": str(uuid4())},
                     "recursion_limit": 1000000000,
                 },
+                checkpoint_during=False,
             )
         ]
     )
@@ -42,6 +43,7 @@ async def arun_first_event_latency(graph: Pregel, input: dict) -> None:
             "configurable": {"thread_id": str(uuid4())},
             "recursion_limit": 1000000000,
         },
+        checkpoint_during=False,
     )
 
     try:
@@ -61,6 +63,7 @@ def run(graph: Pregel, input: dict):
                     "configurable": {"thread_id": str(uuid4())},
                     "recursion_limit": 1000000000,
                 },
+                checkpoint_during=False,
             )
         ]
     )
@@ -77,6 +80,7 @@ def run_first_event_latency(graph: Pregel, input: dict) -> None:
             "configurable": {"thread_id": str(uuid4())},
             "recursion_limit": 1000000000,
         },
+        checkpoint_during=False,
     )
 
     try:

libs/langgraph/langgraph/constants.py

@@ -83,6 +83,8 @@ CONFIG_KEY_PREVIOUS = sys.intern("__pregel_previous")
 # holds the previous return value from a stateful Pregel graph.
 CONFIG_KEY_RUNNER_SUBMIT = sys.intern("__pregel_runner_submit")
 # holds a function that receives tasks from runner, executes them and returns results
+CONFIG_KEY_CHECKPOINT_DURING = sys.intern("__pregel_checkpoint_during")
+# holds a boolean indicating whether to checkpoint during the run (or only at the end)
 
 # --- Other constants ---
 PUSH = sys.intern("__pregel_push")

libs/langgraph/langgraph/graph/schema_utils.py

@@ -1,55 +1,210 @@
 import logging
 import weakref
+from inspect import isclass
 from typing import (
     Any,
     Callable,
     Optional,
     Type,
+    Union,
+    get_args,
+    get_origin,
+    get_type_hints,
 )
 
 from pydantic import BaseModel
 from pydantic.v1 import BaseModel as BaseModelV1
+from typing_extensions import Annotated
 
 logger = logging.getLogger(__name__)
 
 
-_cache: weakref.WeakKeyDictionary[Type[Any], "SchemaCoercionMapper"] = (
+_cache: weakref.WeakKeyDictionary[Type[Any], dict[int, "SchemaCoercionMapper"]] = (
     weakref.WeakKeyDictionary()
 )
 
 
 class SchemaCoercionMapper:
-    __slots__ = ("_inited", "schema", "_fields", "_construct", "_field_coercers")
-
     def __new__(
         cls,
         schema: Type[Any],
-        **kwargs: Any,
+        type_hints: Optional[dict[str, Any]] = None,
+        max_depth: int = 12,
     ) -> "SchemaCoercionMapper":
-        if schema in _cache:
-            return _cache[schema]
+        if schema not in _cache:
+            _cache[schema] = {}
+        if max_depth in _cache[schema]:
+            return _cache[schema][max_depth]
+
         inst = super().__new__(cls)
-        _cache[schema] = inst
+        _cache[schema][max_depth] = inst
         return inst
 
     def __init__(
         self,
         schema: Type[Any],
-        **kwargs: Any,
+        type_hints: Optional[dict[str, Any]] = None,
+        max_depth: int = 12,
     ):
         if hasattr(self, "_inited"):
             return
         self._inited = True
-        if issubclass(schema, BaseModelV1):
-            self._construct: Callable[..., Any] = schema.parse_obj
+        self.schema = schema
+        self.type_hints = (
+            type_hints
+            if type_hints is not None
+            else get_type_hints(schema, localns={schema.__name__: schema})
+        )
+        self.max_depth = max_depth
 
-        elif issubclass(schema, BaseModel):
-            self._construct = schema.model_validate
+        if issubclass(schema, BaseModel):
+            self._fields = {
+                n: self.type_hints.get(n, f.annotation)
+                for n, f in schema.model_fields.items()
+            }
+            self._construct: Callable[..., Any] = schema.model_construct
 
+        elif issubclass(schema, BaseModelV1):
+            self._fields = {
+                n: self.type_hints.get(n, f.annotation)
+                for n, f in schema.__fields__.items()
+            }
+            self._construct = schema.construct
         else:
             raise TypeError("Schema is neither valid Pydantic v1 nor v2 model.")
+        self._field_coercers: Optional[dict[str, Callable[[Any, Any], Any]]] = None
 
     def __call__(self, input_data: Any, depth: Optional[int] = None) -> Any:
-        if not isinstance(input_data, dict):
+        return self.coerce(input_data, depth)
+
+    def coerce(self, input_data: Any, depth: Optional[int] = None) -> Any:
+        if depth is None:
+            depth = self.max_depth
+        if not isinstance(input_data, dict) or depth <= 0:
             return input_data
-        return self._construct(input_data)
+        processed = {}
+        if self._field_coercers is None:
+            self._field_coercers = {
+                n: self._build_coercer(t, depth - 1) for n, t in self._fields.items()
+            }
+        for k, v in input_data.items():
+            fn = self._field_coercers.get(k)
+            processed[k] = fn(v, depth - 1) if fn else v
+        return self._construct(**processed)
+
+    def _build_coercer(
+        self, field_type: Any, depth: int, throw: bool = False
+    ) -> Callable[[Any, Any], Any]:
+        if depth == 0:
+            return self._passthrough
+        origin = get_origin(field_type)
+
+        if origin is Annotated:
+            real_type, *_ = get_args(field_type)
+            sub = self._build_coercer(real_type, depth - 1)
+            return lambda v, d: sub(v, d)
+        if isclass(field_type):
+            is_class_ = True
+            try:
+                is_base_model = issubclass(field_type, BaseModel)
+            except TypeError:
+                is_class_ = False
+                is_base_model = False
+
+            if is_base_model:
+                mapper = SchemaCoercionMapper(field_type, max_depth=depth - 1)
+                return lambda v, d: mapper.coerce(v, d) if isinstance(v, dict) else v
+            if is_class_ and issubclass(field_type, BaseModelV1):
+                mapper = SchemaCoercionMapper(field_type, max_depth=depth - 1)
+                return lambda v, d: mapper.coerce(v, d) if isinstance(v, dict) else v
+        if origin is list or field_type is list:
+            args = get_args(field_type)
+            if len(args) != 1:
+                return lambda v, d: v
+            sub = self._build_coercer(args[0], depth - 1)
+
+            def list_coercer(v: Any, d: Any) -> Any:
+                if not isinstance(v, (list, tuple)):
+                    return v
+                return [sub(x, d - 1) for x in v]
+
+            return list_coercer
+        if origin is set or field_type is set:
+            args = get_args(field_type)
+            if len(args) != 1:
+                return lambda v, d: v
+            sub = self._build_coercer(args[0], depth - 1)
+
+            def set_coercer(v: Any, d: Any) -> Any:
+                if not isinstance(v, (list, tuple, set)):
+                    return v
+                return {sub(x, d - 1) for x in v}
+
+            return set_coercer
+        if origin is dict or field_type is dict:
+            args = get_args(field_type)
+            if len(args) != 2:
+
+                def dict_coercer(v: Any, d: Any) -> Any:
+                    if not isinstance(v, dict):
+                        if throw:
+                            raise TypeError("Expected dict, got %s" % type(v))
+                    return v
+
+                return dict_coercer
+            k_sub = self._build_coercer(args[0], depth - 1)
+            v_sub = self._build_coercer(args[1], depth - 1)
+
+            def dict_coercer(v: Any, d: Any) -> Any:
+                if not isinstance(v, dict):
+                    if throw:
+                        raise TypeError("Expected dict, got %s" % type(v))
+                    return v
+                return {k_sub(k, d - 1): v_sub(val, d - 1) for k, val in v.items()}
+
+            return dict_coercer
+
+        if origin is tuple:
+            targs = get_args(field_type)
+            if not targs:
+                return lambda v, d: v
+            subs = [self._build_coercer(a, depth - 1) for a in targs]
+
+            def tuple_coercer(v: Any, d: Any) -> Any:
+                if not isinstance(v, (list, tuple)):
+                    return v
+                out = []
+                for i, sp in enumerate(subs):
+                    out.append(sp(v[i] if i < len(v) else None, d - 1))
+                return tuple(out)
+
+            return tuple_coercer
+        if origin is Union:
+            uargs = get_args(field_type)
+            subs, none_in_union = [], False
+            for ix, arg in enumerate(uargs):
+                if arg is type(None):
+                    none_in_union = True
+                else:
+                    subs.append(
+                        self._build_coercer(arg, depth - 1, throw=ix < len(uargs) - 1)
+                    )
+
+            def union_coercer(v: Any, d: Any) -> Any:
+                if v is None and none_in_union:
+                    return None
+                err = None
+                for sp in subs:
+                    try:
+                        return sp(v, d - 1)
+                    except TypeError as e:
+                        err = e
+                if err:
+                    raise err
+                return v
+
+            return union_coercer
+        return self._passthrough
+
+    def _passthrough(self, v: Any, d: Any) -> Any:
+        return v

libs/langgraph/langgraph/graph/state.py

@@ -1060,7 +1060,7 @@ def _pick_mapper(
         if issubclass(schema, dict):
             return None
         if issubclass(schema, (BaseModel, BaseModelV1)):
-            return SchemaCoercionMapper(schema)
+            return SchemaCoercionMapper(schema, type_hints)
     return partial(_coerce_state, schema)
 
 

libs/langgraph/langgraph/pregel/__init__.py

@@ -53,6 +53,7 @@ from langgraph.checkpoint.base import (
 )
 from langgraph.constants import (
     CONF,
+    CONFIG_KEY_CHECKPOINT_DURING,
     CONFIG_KEY_CHECKPOINT_ID,
     CONFIG_KEY_CHECKPOINT_NS,
     CONFIG_KEY_CHECKPOINTER,
@@ -2098,6 +2099,7 @@ class Pregel(PregelProtocol):
         output_keys: Optional[Union[str, Sequence[str]]] = None,
         interrupt_before: Optional[Union[All, Sequence[str]]] = None,
         interrupt_after: Optional[Union[All, Sequence[str]]] = None,
+        checkpoint_during: Optional[bool] = None,
         debug: Optional[bool] = None,
         subgraphs: bool = False,
     ) -> Iterator[Union[dict[str, Any], Any]]:
@@ -2119,6 +2121,7 @@ class Pregel(PregelProtocol):
             output_keys: The keys to stream, defaults to all non-context channels.
             interrupt_before: Nodes to interrupt before, defaults to all nodes in the graph.
             interrupt_after: Nodes to interrupt after, defaults to all nodes in the graph.
+            checkpoint_during: Whether to checkpoint intermediate steps, defaults to True. If False, only the final checkpoint is saved.
             debug: Whether to print debug information during execution, defaults to False.
             subgraphs: Whether to stream subgraphs, defaults to False.
 
@@ -2280,6 +2283,9 @@ class Pregel(PregelProtocol):
                 config[CONF][CONFIG_KEY_STREAM_WRITER] = lambda c: stream.put(
                     ((), "custom", c)
                 )
+            # set checkpointing mode for subgraphs
+            if checkpoint_during is not None:
+                config[CONF][CONFIG_KEY_CHECKPOINT_DURING] = checkpoint_during
             with SyncPregelLoop(
                 input,
                 input_model=self.input_model,
@@ -2295,6 +2301,9 @@ class Pregel(PregelProtocol):
                 interrupt_after=interrupt_after_,
                 manager=run_manager,
                 debug=debug,
+                checkpoint_during=checkpoint_during
+                if checkpoint_during is not None
+                else config[CONF].get(CONFIG_KEY_CHECKPOINT_DURING, True),
                 trigger_to_nodes=self.trigger_to_nodes,
                 migrate_checkpoint=self._migrate_checkpoint,
             ) as loop:
@@ -2377,6 +2386,7 @@ class Pregel(PregelProtocol):
         output_keys: Optional[Union[str, Sequence[str]]] = None,
         interrupt_before: Optional[Union[All, Sequence[str]]] = None,
         interrupt_after: Optional[Union[All, Sequence[str]]] = None,
+        checkpoint_during: Optional[bool] = None,
         debug: Optional[bool] = None,
         subgraphs: bool = False,
     ) -> AsyncIterator[Union[dict[str, Any], Any]]:
@@ -2398,6 +2408,7 @@ class Pregel(PregelProtocol):
             output_keys: The keys to stream, defaults to all non-context channels.
             interrupt_before: Nodes to interrupt before, defaults to all nodes in the graph.
             interrupt_after: Nodes to interrupt after, defaults to all nodes in the graph.
+            checkpoint_during: Whether to checkpoint intermediate steps, defaults to True. If False, only the final checkpoint is saved.
             debug: Whether to print debug information during execution, defaults to False.
             subgraphs: Whether to stream subgraphs, defaults to False.
 
@@ -2579,6 +2590,9 @@ class Pregel(PregelProtocol):
                         stream.put_nowait, ((), "custom", c)
                     )
                 )
+            # set checkpointing mode for subgraphs
+            if checkpoint_during is not None:
+                config[CONF][CONFIG_KEY_CHECKPOINT_DURING] = checkpoint_during
             async with AsyncPregelLoop(
                 input,
                 input_model=self.input_model,
@@ -2594,6 +2608,9 @@ class Pregel(PregelProtocol):
                 interrupt_after=interrupt_after_,
                 manager=run_manager,
                 debug=debug,
+                checkpoint_during=checkpoint_during
+                if checkpoint_during is not None
+                else config[CONF].get(CONFIG_KEY_CHECKPOINT_DURING, True),
                 trigger_to_nodes=self.trigger_to_nodes,
                 migrate_checkpoint=self._migrate_checkpoint,
             ) as loop:
@@ -2669,6 +2686,7 @@ class Pregel(PregelProtocol):
         output_keys: Optional[Union[str, Sequence[str]]] = None,
         interrupt_before: Optional[Union[All, Sequence[str]]] = None,
         interrupt_after: Optional[Union[All, Sequence[str]]] = None,
+        checkpoint_during: Optional[bool] = None,
         debug: Optional[bool] = None,
         **kwargs: Any,
     ) -> Union[dict[str, Any], Any]:
@@ -2700,6 +2718,7 @@ class Pregel(PregelProtocol):
             output_keys=output_keys,
             interrupt_before=interrupt_before,
             interrupt_after=interrupt_after,
+            checkpoint_during=checkpoint_during,
             debug=debug,
             **kwargs,
         ):
@@ -2721,6 +2740,7 @@ class Pregel(PregelProtocol):
         output_keys: Optional[Union[str, Sequence[str]]] = None,
         interrupt_before: Optional[Union[All, Sequence[str]]] = None,
         interrupt_after: Optional[Union[All, Sequence[str]]] = None,
+        checkpoint_during: Optional[bool] = None,
         debug: Optional[bool] = None,
         **kwargs: Any,
     ) -> Union[dict[str, Any], Any]:
@@ -2753,6 +2773,7 @@ class Pregel(PregelProtocol):
             output_keys=output_keys,
             interrupt_before=interrupt_before,
             interrupt_after=interrupt_after,
+            checkpoint_during=checkpoint_during,
             debug=debug,
             **kwargs,
         ):

libs/langgraph/langgraph/pregel/loop.py

@@ -63,6 +63,7 @@ from langgraph.constants import (
     RESUME,
     SCHEDULED,
     TAG_HIDDEN,
+    TASKS,
 )
 from langgraph.errors import (
     CheckpointNotLatest,
@@ -155,7 +156,7 @@ class PregelLoop(LoopProtocol):
     manager: Union[None, AsyncParentRunManager, ParentRunManager]
     interrupt_after: Union[All, Sequence[str]]
     interrupt_before: Union[All, Sequence[str]]
-    checkpoint_every_step: bool
+    checkpoint_during: bool
     debug: bool
 
     checkpointer_get_next_version: GetNextVersion
@@ -180,6 +181,7 @@ class PregelLoop(LoopProtocol):
     channels: Mapping[str, BaseChannel]
     managed: ManagedValueMapping
     checkpoint: Checkpoint
+    checkpoint_id_saved: str
     checkpoint_ns: tuple[str, ...]
     checkpoint_config: RunnableConfig
     checkpoint_metadata: CheckpointMetadata
@@ -215,7 +217,7 @@ class PregelLoop(LoopProtocol):
         debug: bool = False,
         migrate_checkpoint: Optional[Callable[[Checkpoint], None]] = None,
         trigger_to_nodes: Optional[Mapping[str, Sequence[str]]] = None,
-        checkpoint_every_step: bool = True,
+        checkpoint_during: bool = True,
     ) -> None:
         super().__init__(
             step=0,
@@ -241,7 +243,7 @@ class PregelLoop(LoopProtocol):
         )
         self._migrate_checkpoint = migrate_checkpoint
         self.trigger_to_nodes = trigger_to_nodes
-        self.checkpoint_every_step = checkpoint_every_step
+        self.checkpoint_during = checkpoint_during
         self.debug = debug
         if self.stream is not None and CONFIG_KEY_STREAM in config[CONF]:
             self.stream = DuplexStream(self.stream, config[CONF][CONFIG_KEY_STREAM])
@@ -294,29 +296,19 @@ class PregelLoop(LoopProtocol):
         """Put writes for a task, to be read by the next tick."""
         if not writes:
             return
+        # always checkpoint writes containing Send, as they are fetched from the
+        # parent checkpoint, not the current one
+        checkpoint_during = self.checkpoint_during or any(w[0] == TASKS for w in writes)
         # deduplicate writes to special channels, last write wins
         if all(w[0] in WRITES_IDX_MAP for w in writes):
             writes = list({w[0]: w for w in writes}.values())
+        # remove existing writes for this task
+        self.checkpoint_pending_writes = [
+            w for w in self.checkpoint_pending_writes if w[0] != task_id
+        ]
         # save writes
-        for c, v in writes:
-            if (
-                c in WRITES_IDX_MAP
-                and (
-                    idx := next(
-                        (
-                            i
-                            for i, w in enumerate(self.checkpoint_pending_writes)
-                            if w[0] == task_id and w[1] == c
-                        ),
-                        None,
-                    )
-                )
-                is not None
-            ):
-                self.checkpoint_pending_writes[idx] = (task_id, c, v)
-            else:
-                self.checkpoint_pending_writes.append((task_id, c, v))
-        if self.checkpointer_put_writes is not None:
+        self.checkpoint_pending_writes.extend((task_id, c, v) for c, v in writes)
+        if checkpoint_during and self.checkpointer_put_writes is not None:
             config = patch_configurable(
                 self.checkpoint_config,
                 {
@@ -349,6 +341,46 @@ class PregelLoop(LoopProtocol):
         if hasattr(self, "tasks"):
             self._output_writes(task_id, writes)
 
+    def _put_pending_writes(self) -> None:
+        if self.checkpointer_put_writes is None:
+            return
+        if not self.checkpoint_pending_writes:
+            return
+        # patch config
+        config = patch_configurable(
+            self.checkpoint_config,
+            {
+                CONFIG_KEY_CHECKPOINT_NS: self.config[CONF].get(
+                    CONFIG_KEY_CHECKPOINT_NS, ""
+                ),
+                CONFIG_KEY_CHECKPOINT_ID: self.checkpoint["id"],
+            },
+        )
+        # group by task id
+        by_task = defaultdict(list)
+        for task_id, channel, value in self.checkpoint_pending_writes:
+            by_task[task_id].append((channel, value))
+        # submit writes to checkpointer
+        for task_id, writes in by_task.items():
+            if self.checkpointer_put_writes_accepts_task_path and hasattr(
+                self, "tasks"
+            ):
+                task = self.tasks.get(task_id)
+                self.submit(
+                    self.checkpointer_put_writes,
+                    config,
+                    writes,
+                    task_id,
+                    task_path_str(task.path) if task else "",
+                )
+            else:
+                self.submit(
+                    self.checkpointer_put_writes,
+                    config,
+                    writes,
+                    task_id,
+                )
+
     def accept_push(
         self, task: PregelExecutableTask, write_idx: int, call: Optional[Call] = None
     ) -> Optional[PregelExecutableTask]:
@@ -711,32 +743,44 @@ class PregelLoop(LoopProtocol):
 
     def _put_checkpoint(self, metadata: CheckpointMetadata) -> None:
         # assign step and parents
-        metadata["step"] = self.step
-        metadata["parents"] = self.config[CONF].get(CONFIG_KEY_CHECKPOINT_MAP, {})
-        # debug flag
-        if self.debug:
-            print_step_checkpoint(
-                metadata,
-                self.channels,
-                (
-                    [self.stream_keys]
-                    if isinstance(self.stream_keys, str)
-                    else self.stream_keys
-                ),
-            )
+        exiting = metadata is self.checkpoint_metadata
+        if exiting and self.checkpoint["id"] == self.checkpoint_id_saved:
+            # checkpoint already saved
+            return
+        if not exiting:
+            metadata["step"] = self.step
+            metadata["parents"] = self.config[CONF].get(CONFIG_KEY_CHECKPOINT_MAP, {})
+            self.checkpoint_metadata = metadata
+            # debug flag
+            if self.debug:
+                print_step_checkpoint(
+                    metadata,
+                    self.channels,
+                    (
+                        [self.stream_keys]
+                        if isinstance(self.stream_keys, str)
+                        else self.stream_keys
+                    ),
+                )
+            self.checkpoint_id_prev = self.checkpoint["id"] if self.step > -1 else None
+        # do checkpoint?
+        do_checkpoint = self._checkpointer_put_after_previous is not None and (
+            exiting or self.checkpoint_during
+        )
+        # create new checkpoint
+        self.checkpoint = create_checkpoint(
+            self.checkpoint,
+            self.channels if do_checkpoint else None,
+            self.step,
+            id=self.checkpoint["id"] if exiting else None,
+        )
         # bail if no checkpointer
-        if self._checkpointer_put_after_previous is not None:
+        if do_checkpoint and self._checkpointer_put_after_previous is not None:
             for k, v in self.config["metadata"].items():
                 if k in EXCLUDED_METADATA_KEYS:
                     continue
                 metadata.setdefault(k, v)  # type: ignore
 
-            # create new checkpoint
-            self.checkpoint = create_checkpoint(
-                self.checkpoint, self.channels, self.step
-            )
-            self.checkpoint_metadata = metadata
-
             self.prev_checkpoint_config = (
                 self.checkpoint_config
                 if CONFIG_KEY_CHECKPOINT_ID in self.checkpoint_config[CONF]
@@ -747,6 +791,8 @@ class PregelLoop(LoopProtocol):
                 **self.checkpoint_config,
                 CONF: {
                     **self.checkpoint_config[CONF],
+                    # this is guaranteed to be set by code above
+                    CONFIG_KEY_CHECKPOINT_ID: self.checkpoint_id_prev,
                     CONFIG_KEY_CHECKPOINT_NS: self.config[CONF].get(
                         CONFIG_KEY_CHECKPOINT_NS, ""
                     ),
@@ -777,8 +823,9 @@ class PregelLoop(LoopProtocol):
                     CONFIG_KEY_CHECKPOINT_ID: self.checkpoint["id"],
                 },
             }
-        # increment step
-        self.step += 1
+        if not exiting:
+            # increment step
+            self.step += 1
 
     def _update_mv(self, key: str, values: Sequence[Any]) -> None:
         raise NotImplementedError
@@ -789,6 +836,10 @@ class PregelLoop(LoopProtocol):
         exc_value: Optional[BaseException],
         traceback: Optional[TracebackType],
     ) -> Optional[bool]:
+        # persist current checkpoint and writes
+        if not self.checkpoint_during:
+            self._put_checkpoint(self.checkpoint_metadata)
+            self._put_pending_writes()
         # suppress interrupt
         suppress = isinstance(exc_value, GraphInterrupt) and not self.is_nested
         if suppress:
@@ -907,6 +958,7 @@ class SyncPregelLoop(PregelLoop, ContextManager):
         debug: bool = False,
         migrate_checkpoint: Optional[Callable[[Checkpoint], None]] = None,
         trigger_to_nodes: Optional[Mapping[str, Sequence[str]]] = None,
+        checkpoint_during: bool = True,
     ) -> None:
         super().__init__(
             input,
@@ -925,6 +977,7 @@ class SyncPregelLoop(PregelLoop, ContextManager):
             debug=debug,
             migrate_checkpoint=migrate_checkpoint,
             trigger_to_nodes=trigger_to_nodes,
+            checkpoint_during=checkpoint_during,
         )
         self.stack = ExitStack()
         if checkpointer:
@@ -1004,6 +1057,7 @@ class SyncPregelLoop(PregelLoop, ContextManager):
             },
         }
         self.prev_checkpoint_config = saved.parent_config
+        self.checkpoint_id_saved = saved.checkpoint["id"]
         self.checkpoint = saved.checkpoint
         self.checkpoint_metadata = saved.metadata
         self.checkpoint_pending_writes = (
@@ -1054,6 +1108,7 @@ class AsyncPregelLoop(PregelLoop, AsyncContextManager):
         debug: bool = False,
         migrate_checkpoint: Optional[Callable[[Checkpoint], None]] = None,
         trigger_to_nodes: Optional[Mapping[str, Sequence[str]]] = None,
+        checkpoint_during: bool = True,
     ) -> None:
         super().__init__(
             input,
@@ -1072,6 +1127,7 @@ class AsyncPregelLoop(PregelLoop, AsyncContextManager):
             debug=debug,
             migrate_checkpoint=migrate_checkpoint,
             trigger_to_nodes=trigger_to_nodes,
+            checkpoint_during=checkpoint_during,
         )
         self.stack = AsyncExitStack()
         if checkpointer:
@@ -1151,6 +1207,7 @@ class AsyncPregelLoop(PregelLoop, AsyncContextManager):
             },
         }
         self.prev_checkpoint_config = saved.parent_config
+        self.checkpoint_id_saved = saved.checkpoint["id"]
         self.checkpoint = saved.checkpoint
         self.checkpoint_metadata = saved.metadata
         self.checkpoint_pending_writes = (

libs/langgraph/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "langgraph"
-version = "0.3.26"
+version = "0.3.27"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 license = "MIT"

libs/langgraph/tests/test_interruption.py

@@ -1,20 +1,20 @@
 import pytest
-from pytest_mock import MockerFixture
 from typing_extensions import TypedDict
 
 from langgraph.graph import END, START, StateGraph
 from tests.conftest import (
-    ALL_CHECKPOINTERS_ASYNC,
-    ALL_CHECKPOINTERS_SYNC,
+    REGULAR_CHECKPOINTERS_ASYNC,
+    REGULAR_CHECKPOINTERS_SYNC,
     awith_checkpointer,
 )
 
 pytestmark = pytest.mark.anyio
 
 
-@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
+@pytest.mark.parametrize("checkpoint_during", [True, False])
+@pytest.mark.parametrize("checkpointer_name", REGULAR_CHECKPOINTERS_SYNC)
 def test_interruption_without_state_updates(
-    request: pytest.FixtureRequest, checkpointer_name: str, mocker: MockerFixture
+    request: pytest.FixtureRequest, checkpointer_name: str, checkpoint_during: bool
 ) -> None:
     """Test interruption without state updates. This test confirms that
     interrupting doesn't require a state key having been updated in the prev step"""
@@ -40,20 +40,27 @@ def test_interruption_without_state_updates(
     initial_input = {"input": "hello world"}
     thread = {"configurable": {"thread_id": "1"}}
 
-    graph.invoke(initial_input, thread, debug=True)
+    graph.invoke(initial_input, thread, checkpoint_during=checkpoint_during)
     assert graph.get_state(thread).next == ("step_2",)
+    n_checkpoints = len([c for c in graph.get_state_history(thread)])
+    assert n_checkpoints == (3 if checkpoint_during else 1)
 
-    graph.invoke(None, thread, debug=True)
+    graph.invoke(None, thread, checkpoint_during=checkpoint_during)
     assert graph.get_state(thread).next == ("step_3",)
+    n_checkpoints = len([c for c in graph.get_state_history(thread)])
+    assert n_checkpoints == (4 if checkpoint_during else 2)
 
-    graph.invoke(None, thread, debug=True)
+    graph.invoke(None, thread, checkpoint_during=checkpoint_during)
     assert graph.get_state(thread).next == ()
+    n_checkpoints = len([c for c in graph.get_state_history(thread)])
+    assert n_checkpoints == (5 if checkpoint_during else 3)
 
 
-@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC)
+@pytest.mark.parametrize("checkpoint_during", [True, False])
+@pytest.mark.parametrize("checkpointer_name", REGULAR_CHECKPOINTERS_ASYNC)
 async def test_interruption_without_state_updates_async(
-    checkpointer_name: str, mocker: MockerFixture
-):
+    checkpointer_name: str, checkpoint_during: bool
+) -> None:
     """Test interruption without state updates. This test confirms that
     interrupting doesn't require a state key having been updated in the prev step"""
 
@@ -78,11 +85,17 @@ async def test_interruption_without_state_updates_async(
         initial_input = {"input": "hello world"}
         thread = {"configurable": {"thread_id": "1"}}
 
-        await graph.ainvoke(initial_input, thread, debug=True)
+        await graph.ainvoke(initial_input, thread, checkpoint_during=checkpoint_during)
         assert (await graph.aget_state(thread)).next == ("step_2",)
+        n_checkpoints = len([c async for c in graph.aget_state_history(thread)])
+        assert n_checkpoints == (3 if checkpoint_during else 1)
 
-        await graph.ainvoke(None, thread, debug=True)
+        await graph.ainvoke(None, thread, checkpoint_during=checkpoint_during)
         assert (await graph.aget_state(thread)).next == ("step_3",)
+        n_checkpoints = len([c async for c in graph.aget_state_history(thread)])
+        assert n_checkpoints == (4 if checkpoint_during else 2)
 
-        await graph.ainvoke(None, thread, debug=True)
+        await graph.ainvoke(None, thread, checkpoint_during=checkpoint_during)
         assert (await graph.aget_state(thread)).next == ()
+        n_checkpoints = len([c async for c in graph.aget_state_history(thread)])
+        assert n_checkpoints == (5 if checkpoint_during else 3)

libs/langgraph/tests/test_large_cases.py

@@ -7258,9 +7258,10 @@ def test_branch_then(
     )
 
 
-@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
+@pytest.mark.parametrize("checkpoint_during", [True, False])
+@pytest.mark.parametrize("checkpointer_name", REGULAR_CHECKPOINTERS_SYNC)
 def test_send_dedupe_on_resume(
-    request: pytest.FixtureRequest, checkpointer_name: str
+    request: pytest.FixtureRequest, checkpointer_name: str, checkpoint_during: bool
 ) -> None:
     checkpointer = request.getfixturevalue(f"checkpointer_{checkpointer_name}")
 
@@ -7316,7 +7317,7 @@ def test_send_dedupe_on_resume(
 
     graph = builder.compile(checkpointer=checkpointer)
     thread1 = {"configurable": {"thread_id": "1"}}
-    assert graph.invoke(["0"], thread1, debug=1) == [
+    assert graph.invoke(["0"], thread1, checkpoint_during=checkpoint_during) == [
         "0",
         "1",
         "3.1",
@@ -7333,12 +7334,11 @@ def test_send_dedupe_on_resume(
         pytest.xfail("TODO: shallow checkpointer reports wrong next set")
     assert state.next == ("flaky",)
     # check history
-    if "shallow" not in checkpointer_name:
-        history = [c for c in graph.get_state_history(thread1)]
-        assert len(history) == 4
+    history = [c for c in graph.get_state_history(thread1)]
+    assert len(history) == (4 if checkpoint_during else 1)
 
     # resume execution
-    assert graph.invoke(None, thread1, debug=1) == [
+    assert graph.invoke(None, thread1, checkpoint_during=checkpoint_during) == [
         "0",
         "1",
         "3.1",
@@ -7358,6 +7358,7 @@ def test_send_dedupe_on_resume(
     assert state.next == ()
     # check history
     history = [c for c in graph.get_state_history(thread1)]
+    assert len(history) == (6 if checkpoint_during else 2)
     expected_history = [
         StateSnapshot(
             values=[
@@ -7494,13 +7495,9 @@ def test_send_dedupe_on_resume(
                     name="flaky",
                     path=("__pregel_push", 1),
                     error=None,
-                    interrupts=(
-                        Interrupt(
-                            value="Bahh", resumable=False, ns=None, when="during"
-                        ),
-                    ),
+                    interrupts=(Interrupt(value="Bahh", resumable=False, ns=None),),
                     state=None,
-                    result=["flaky|4"],
+                    result=["flaky|4"] if checkpoint_during else None,
                 ),
                 PregelTask(
                     id=AnyStr(),
@@ -7637,10 +7634,11 @@ def test_send_dedupe_on_resume(
             ),
         ),
     ]
-    if "shallow" in checkpointer_name:
-        expected_history = expected_history[:1]
-
-    assert history == expected_history
+    if checkpoint_during:
+        assert history == expected_history
+    else:
+        assert history[0] == expected_history[0]
+        assert history[1] == expected_history[2]
 
 
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)

libs/langgraph/tests/test_pregel.py

@@ -1115,10 +1115,14 @@ def test_invoke_checkpoint_two(
     assert checkpoint["channel_values"].get("total") == 5
 
 
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
 def test_pending_writes_resume(
-    request: pytest.FixtureRequest, checkpointer_name: str
+    request: pytest.FixtureRequest, checkpointer_name: str, checkpoint_during: bool
 ) -> None:
+    if not checkpoint_during and "shallow" in checkpointer_name:
+        pytest.skip("Checkpointing during execution not supported")
+
     checkpointer: BaseCheckpointSaver = request.getfixturevalue(
         f"checkpointer_{checkpointer_name}"
     )
@@ -1144,17 +1148,19 @@ def test_pending_writes_resume(
             self.calls = 0
 
     one = AwhileMaker(0.1, {"value": 2})
-    two = AwhileMaker(0.3, ConnectionError("I'm not good"))
+    two = AwhileMaker(0.2, ConnectionError("I'm not good"))
     builder = StateGraph(State)
     builder.add_node("one", one)
-    builder.add_node("two", two, retry=RetryPolicy(max_attempts=2))
+    builder.add_node(
+        "two", two, retry=RetryPolicy(max_attempts=2, initial_interval=0, jitter=False)
+    )
     builder.add_edge(START, "one")
     builder.add_edge(START, "two")
     graph = builder.compile(checkpointer=checkpointer)
 
     thread1: RunnableConfig = {"configurable": {"thread_id": "1"}}
     with pytest.raises(ConnectionError, match="I'm not good"):
-        graph.invoke({"value": 1}, thread1)
+        graph.invoke({"value": 1}, thread1, checkpoint_during=checkpoint_during)
 
     # both nodes should have been called once
     assert one.calls == 1
@@ -1200,7 +1206,7 @@ def test_pending_writes_resume(
 
     # resume execution
     with pytest.raises(ConnectionError, match="I'm not good"):
-        graph.invoke(None, thread1)
+        graph.invoke(None, thread1, checkpoint_during=checkpoint_during)
 
     # node "one" succeeded previously, so shouldn't be called again
     assert one.calls == 1
@@ -1214,7 +1220,9 @@ def test_pending_writes_resume(
     # resume execution, without exception
     two.rtn = {"value": 3}
     # both the pending write and the new write were applied, 1 + 2 + 3 = 6
-    assert graph.invoke(None, thread1) == {"value": 6}
+    assert graph.invoke(None, thread1, checkpoint_during=checkpoint_during) == {
+        "value": 6
+    }
 
     if "shallow" in checkpointer_name:
         assert len(list(checkpointer.list(thread1))) == 1
@@ -1223,7 +1231,7 @@ def test_pending_writes_resume(
     # check all final checkpoints
     checkpoints = [c for c in checkpointer.list(thread1)]
     # we should have 3
-    assert len(checkpoints) == 3
+    assert len(checkpoints) == (3 if checkpoint_during else 2)
     # the last one not too interesting for this test
     assert checkpoints[0] == CheckpointTuple(
         config={
@@ -1325,15 +1333,26 @@ def test_pending_writes_resume(
             "configurable": {
                 "thread_id": "1",
                 "checkpoint_ns": "",
-                "checkpoint_id": checkpoints[2].config["configurable"]["checkpoint_id"],
+                "checkpoint_id": checkpoints[2].config["configurable"]["checkpoint_id"]
+                if checkpoint_during
+                else AnyStr(),
             }
         },
         pending_writes=UnsortedSequence(
             (AnyStr(), "value", 2),
             (AnyStr(), "__error__", 'ConnectionError("I\'m not good")'),
             (AnyStr(), "value", 3),
+        )
+        if checkpoint_during
+        else UnsortedSequence(
+            (AnyStr(), "value", 2),
+            (AnyStr(), "__error__", 'ConnectionError("I\'m not good")'),
+            # the write against the previous checkpoint is not saved, as it is
+            # produced in a run where only the next checkpoint (the last) is saved
         ),
     )
+    if not checkpoint_during:
+        return
     assert checkpoints[2] == CheckpointTuple(
         config={
             "configurable": {
@@ -1491,8 +1510,14 @@ def test_send_sequences() -> None:
     ]
 
 
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
-def test_imp_task(request: pytest.FixtureRequest, checkpointer_name: str) -> None:
+def test_imp_task(
+    request: pytest.FixtureRequest, checkpointer_name: str, checkpoint_during: bool
+) -> None:
+    if not checkpoint_during and "shallow" in checkpointer_name:
+        pytest.skip("Checkpointing during execution not supported")
+
     checkpointer = request.getfixturevalue(f"checkpointer_{checkpointer_name}")
     mapper_calls = 0
 
@@ -1558,7 +1583,7 @@ def test_imp_task(request: pytest.FixtureRequest, checkpointer_name: str) -> Non
     }
 
     thread1 = {"configurable": {"thread_id": "1"}}
-    assert [*graph.stream([0, 1], thread1)] == [
+    assert [*graph.stream([0, 1], thread1, checkpoint_during=checkpoint_during)] == [
         {"mapper": "00"},
         {"mapper": "11"},
         {
@@ -1574,17 +1599,23 @@ def test_imp_task(request: pytest.FixtureRequest, checkpointer_name: str) -> Non
     ]
     assert mapper_calls == 2
 
-    assert graph.invoke(Command(resume="answer"), thread1) == [
+    assert graph.invoke(
+        Command(resume="answer"), thread1, checkpoint_during=checkpoint_during
+    ) == [
         "00answer",
         "11answer",
     ]
     assert mapper_calls == 2
 
 
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
 def test_imp_nested(
-    request: pytest.FixtureRequest, checkpointer_name: str, snapshot: SnapshotAssertion
+    request: pytest.FixtureRequest, checkpointer_name: str, checkpoint_during: bool
 ) -> None:
+    if not checkpoint_during and "shallow" in checkpointer_name:
+        pytest.skip("Checkpointing during execution not supported")
+
     checkpointer = request.getfixturevalue(f"checkpointer_{checkpointer_name}")
 
     def mynode(input: list[str]) -> list[str]:
@@ -1626,7 +1657,7 @@ def test_imp_nested(
     }
 
     thread1 = {"configurable": {"thread_id": "1"}}
-    assert [*graph.stream([0, 1], thread1)] == [
+    assert [*graph.stream([0, 1], thread1, checkpoint_during=checkpoint_during)] == [
         {"submapper": "0"},
         {"mapper": "00"},
         {"submapper": "1"},
@@ -1643,16 +1674,22 @@ def test_imp_nested(
         },
     ]
 
-    assert graph.invoke(Command(resume="answer"), thread1) == [
+    assert graph.invoke(
+        Command(resume="answer"), thread1, checkpoint_during=checkpoint_during
+    ) == [
         "00answera",
         "11answera",
     ]
 
 
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
 def test_imp_stream_order(
-    request: pytest.FixtureRequest, checkpointer_name: str, snapshot: SnapshotAssertion
+    request: pytest.FixtureRequest, checkpointer_name: str, checkpoint_during: bool
 ) -> None:
+    if not checkpoint_during and "shallow" in checkpointer_name:
+        pytest.skip("Checkpointing during execution not supported")
+
     checkpointer = request.getfixturevalue(f"checkpointer_{checkpointer_name}")
 
     @task()
@@ -1675,7 +1712,10 @@ def test_imp_stream_order(
         return fut_baz.result()
 
     thread1 = {"configurable": {"thread_id": "1"}}
-    assert [c for c in graph.stream({"a": "0"}, thread1)] == [
+    assert [
+        c
+        for c in graph.stream({"a": "0"}, thread1, checkpoint_during=checkpoint_during)
+    ] == [
         {
             "foo": (
                 "0foo",
@@ -3643,10 +3683,14 @@ def test_nested_graph(snapshot: SnapshotAssertion) -> None:
     ]
 
 
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
 def test_subgraph_checkpoint_true(
-    request: pytest.FixtureRequest, checkpointer_name: str
+    request: pytest.FixtureRequest, checkpointer_name: str, checkpoint_during: bool
 ) -> None:
+    if not checkpoint_during and "shallow" in checkpointer_name:
+        pytest.skip("Unsupported combo")
+
     checkpointer = request.getfixturevalue("checkpointer_" + checkpointer_name)
 
     class InnerState(TypedDict):
@@ -3678,7 +3722,12 @@ def test_subgraph_checkpoint_true(
     app = graph.compile(checkpointer=checkpointer)
 
     config = {"configurable": {"thread_id": "2"}}
-    assert [c for c in app.stream({"my_key": ""}, config, subgraphs=True)] == [
+    assert [
+        c
+        for c in app.stream(
+            {"my_key": ""}, config, subgraphs=True, checkpoint_during=checkpoint_during
+        )
+    ] == [
         (("inner",), {"inner_1": {"my_key": " got here", "my_other_key": ""}}),
         (("inner",), {"inner_2": {"my_key": " and there"}}),
         ((), {"inner": {"my_key": " got here and there"}}),
@@ -3703,10 +3752,14 @@ def test_subgraph_checkpoint_true(
     ]
 
 
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
 def test_subgraph_checkpoint_true_interrupt(
-    request: pytest.FixtureRequest, checkpointer_name: str
+    request: pytest.FixtureRequest, checkpointer_name: str, checkpoint_during: bool
 ) -> None:
+    if not checkpoint_during and "shallow" in checkpointer_name:
+        pytest.skip("Unsupported combo")
+
     checkpointer = request.getfixturevalue("checkpointer_" + checkpointer_name)
 
     # Define subgraph
@@ -3745,15 +3798,18 @@ def test_subgraph_checkpoint_true_interrupt(
     builder.add_edge(START, "node_1")
     builder.add_edge("node_1", "node_2")
 
-    checkpointer = MemorySaver()
     graph = builder.compile(checkpointer=checkpointer)
     config = {"configurable": {"thread_id": "1"}}
 
-    assert graph.invoke({"foo": "foo"}, config) == {"foo": "hi! foo"}
+    assert graph.invoke(
+        {"foo": "foo"}, config, checkpoint_during=checkpoint_during
+    ) == {"foo": "hi! foo"}
     assert graph.get_state(config, subgraphs=True).tasks[0].state.values == {
         "bar": "hi! foo"
     }
-    assert graph.invoke(Command(resume="baz"), config) == {"foo": "hi! foobaz"}
+    assert graph.invoke(
+        Command(resume="baz"), config, checkpoint_during=checkpoint_during
+    ) == {"foo": "hi! foobaz"}
 
 
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
@@ -3869,10 +3925,14 @@ def test_stream_buffering_single_node(
     ]
 
 
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
 def test_nested_graph_interrupts_parallel(
-    request: pytest.FixtureRequest, checkpointer_name: str
+    request: pytest.FixtureRequest, checkpointer_name: str, checkpoint_during: bool
 ) -> None:
+    if not checkpoint_during and "shallow" in checkpointer_name:
+        pytest.skip("Unsupported combo")
+
     checkpointer = request.getfixturevalue("checkpointer_" + checkpointer_name)
 
     class InnerState(TypedDict):
@@ -3919,11 +3979,11 @@ def test_nested_graph_interrupts_parallel(
 
     # test invoke w/ nested interrupt
     config = {"configurable": {"thread_id": "1"}}
-    assert app.invoke({"my_key": ""}, config, debug=True) == {
+    assert app.invoke({"my_key": ""}, config, checkpoint_during=checkpoint_during) == {
         "my_key": " and parallel",
     }
 
-    assert app.invoke(None, config, debug=True) == {
+    assert app.invoke(None, config, checkpoint_during=checkpoint_during) == {
         "my_key": "got here and there and parallel and back again",
     }
 
@@ -3932,13 +3992,17 @@ def test_nested_graph_interrupts_parallel(
     # - the writes of outer are persisted in 1st call and used in 2nd call, ie outer isn't called again (because we dont see outer_1 output again in 2nd stream)
     # test stream updates w/ nested interrupt
     config = {"configurable": {"thread_id": "2"}}
-    assert [*app.stream({"my_key": ""}, config, subgraphs=True)] == [
+    assert [
+        *app.stream(
+            {"my_key": ""}, config, subgraphs=True, checkpoint_during=checkpoint_during
+        )
+    ] == [
         # we got to parallel node first
         ((), {"outer_1": {"my_key": " and parallel"}}),
         ((AnyStr("inner:"),), {"inner_1": {"my_key": "got here", "my_other_key": ""}}),
         ((), {"__interrupt__": ()}),
     ]
-    assert [*app.stream(None, config)] == [
+    assert [*app.stream(None, config, checkpoint_during=checkpoint_during)] == [
         {"outer_1": {"my_key": " and parallel"}, "__metadata__": {"cached": True}},
         {"inner": {"my_key": "got here and there"}},
         {"outer_2": {"my_key": " and back again"}},
@@ -3946,11 +4010,22 @@ def test_nested_graph_interrupts_parallel(
 
     # test stream values w/ nested interrupt
     config = {"configurable": {"thread_id": "3"}}
-    assert [*app.stream({"my_key": ""}, config, stream_mode="values")] == [
+    assert [
+        *app.stream(
+            {"my_key": ""},
+            config,
+            stream_mode="values",
+            checkpoint_during=checkpoint_during,
+        )
+    ] == [
         {"my_key": ""},
         {"my_key": " and parallel"},
     ]
-    assert [*app.stream(None, config, stream_mode="values")] == [
+    assert [
+        *app.stream(
+            None, config, stream_mode="values", checkpoint_during=checkpoint_during
+        )
+    ] == [
         {"my_key": ""},
         {"my_key": "got here and there and parallel"},
         {"my_key": "got here and there and parallel and back again"},
@@ -3959,15 +4034,28 @@ def test_nested_graph_interrupts_parallel(
     # test interrupts BEFORE the parallel node
     app = graph.compile(checkpointer=checkpointer, interrupt_before=["outer_1"])
     config = {"configurable": {"thread_id": "4"}}
-    assert [*app.stream({"my_key": ""}, config, stream_mode="values")] == [
-        {"my_key": ""}
-    ]
+    assert [
+        *app.stream(
+            {"my_key": ""},
+            config,
+            stream_mode="values",
+            checkpoint_during=checkpoint_during,
+        )
+    ] == [{"my_key": ""}]
     # while we're waiting for the node w/ interrupt inside to finish
-    assert [*app.stream(None, config, stream_mode="values")] == [
+    assert [
+        *app.stream(
+            None, config, stream_mode="values", checkpoint_during=checkpoint_during
+        )
+    ] == [
         {"my_key": ""},
         {"my_key": " and parallel"},
     ]
-    assert [*app.stream(None, config, stream_mode="values")] == [
+    assert [
+        *app.stream(
+            None, config, stream_mode="values", checkpoint_during=checkpoint_during
+        )
+    ] == [
         {"my_key": ""},
         {"my_key": "got here and there and parallel"},
         {"my_key": "got here and there and parallel and back again"},
@@ -3976,24 +4064,43 @@ def test_nested_graph_interrupts_parallel(
     # test interrupts AFTER the parallel node
     app = graph.compile(checkpointer=checkpointer, interrupt_after=["outer_1"])
     config = {"configurable": {"thread_id": "5"}}
-    assert [*app.stream({"my_key": ""}, config, stream_mode="values")] == [
+    assert [
+        *app.stream(
+            {"my_key": ""},
+            config,
+            stream_mode="values",
+            checkpoint_during=checkpoint_during,
+        )
+    ] == [
         {"my_key": ""},
         {"my_key": " and parallel"},
     ]
-    assert [*app.stream(None, config, stream_mode="values")] == [
+    assert [
+        *app.stream(
+            None, config, stream_mode="values", checkpoint_during=checkpoint_during
+        )
+    ] == [
         {"my_key": ""},
         {"my_key": "got here and there and parallel"},
     ]
-    assert [*app.stream(None, config, stream_mode="values")] == [
+    assert [
+        *app.stream(
+            None, config, stream_mode="values", checkpoint_during=checkpoint_during
+        )
+    ] == [
         {"my_key": "got here and there and parallel"},
         {"my_key": "got here and there and parallel and back again"},
     ]
 
 
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
 def test_doubly_nested_graph_interrupts(
-    request: pytest.FixtureRequest, checkpointer_name: str
+    request: pytest.FixtureRequest, checkpointer_name: str, checkpoint_during: bool
 ) -> None:
+    if not checkpoint_during and "shallow" in checkpointer_name:
+        pytest.skip("Unsupported combo")
+
     checkpointer = request.getfixturevalue("checkpointer_" + checkpointer_name)
 
     class State(TypedDict):
@@ -4047,11 +4154,13 @@ def test_doubly_nested_graph_interrupts(
 
     # test invoke w/ nested interrupt
     config = {"configurable": {"thread_id": "1"}}
-    assert app.invoke({"my_key": "my value"}, config, debug=True) == {
+    assert app.invoke(
+        {"my_key": "my value"}, config, checkpoint_during=checkpoint_during
+    ) == {
         "my_key": "hi my value",
     }
 
-    assert app.invoke(None, config, debug=True) == {
+    assert app.invoke(None, config, checkpoint_during=checkpoint_during) == {
         "my_key": "hi my value here and there and back again",
     }
 
@@ -4060,12 +4169,14 @@ def test_doubly_nested_graph_interrupts(
     config = {
         "configurable": {"thread_id": "2", CONFIG_KEY_NODE_FINISHED: nodes.append}
     }
-    assert [*app.stream({"my_key": "my value"}, config)] == [
+    assert [
+        *app.stream({"my_key": "my value"}, config, checkpoint_during=checkpoint_during)
+    ] == [
         {"parent_1": {"my_key": "hi my value"}},
         {"__interrupt__": ()},
     ]
     assert nodes == ["parent_1", "grandchild_1"]
-    assert [*app.stream(None, config)] == [
+    assert [*app.stream(None, config, checkpoint_during=checkpoint_during)] == [
         {"child": {"my_key": "hi my value here and there"}},
         {"parent_2": {"my_key": "hi my value here and there and back again"}},
     ]
@@ -4080,11 +4191,22 @@ def test_doubly_nested_graph_interrupts(
 
     # test stream values w/ nested interrupt
     config = {"configurable": {"thread_id": "3"}}
-    assert [*app.stream({"my_key": "my value"}, config, stream_mode="values")] == [
+    assert [
+        *app.stream(
+            {"my_key": "my value"},
+            config,
+            stream_mode="values",
+            checkpoint_during=checkpoint_during,
+        )
+    ] == [
         {"my_key": "my value"},
         {"my_key": "hi my value"},
     ]
-    assert [*app.stream(None, config, stream_mode="values")] == [
+    assert [
+        *app.stream(
+            None, config, stream_mode="values", checkpoint_during=checkpoint_during
+        )
+    ] == [
         {"my_key": "hi my value"},
         {"my_key": "hi my value here and there"},
         {"my_key": "hi my value here and there and back again"},

libs/langgraph/tests/test_pregel_async.py

@@ -1947,10 +1947,14 @@ async def test_invoke_checkpoint(mocker: MockerFixture, checkpointer_name: str)
         assert checkpoint["channel_values"].get("total") == 5
 
 
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC)
 async def test_pending_writes_resume(
-    request: pytest.FixtureRequest, checkpointer_name: str
+    checkpointer_name: str, checkpoint_during: bool
 ) -> None:
+    if not checkpoint_during and "shallow" in checkpointer_name:
+        pytest.skip("Checkpointing during execution not supported")
+
     class State(TypedDict):
         value: Annotated[int, operator.add]
 
@@ -1972,10 +1976,12 @@ async def test_pending_writes_resume(
             self.calls = 0
 
     one = AwhileMaker(0.1, {"value": 2})
-    two = AwhileMaker(0.3, ConnectionError("I'm not good"))
+    two = AwhileMaker(0.2, ConnectionError("I'm not good"))
     builder = StateGraph(State)
     builder.add_node("one", one)
-    builder.add_node("two", two, retry=RetryPolicy(max_attempts=2))
+    builder.add_node(
+        "two", two, retry=RetryPolicy(max_attempts=2, initial_interval=0, jitter=False)
+    )
     builder.add_edge(START, "one")
     builder.add_edge(START, "two")
     async with awith_checkpointer(checkpointer_name) as checkpointer:
@@ -1983,7 +1989,9 @@ async def test_pending_writes_resume(
 
         thread1: RunnableConfig = {"configurable": {"thread_id": "1"}}
         with pytest.raises(ConnectionError, match="I'm not good"):
-            await graph.ainvoke({"value": 1}, thread1)
+            await graph.ainvoke(
+                {"value": 1}, thread1, checkpoint_during=checkpoint_during
+            )
 
         # both nodes should have been called once
         assert one.calls == 1
@@ -2034,7 +2042,7 @@ async def test_pending_writes_resume(
 
         # resume execution
         with pytest.raises(ConnectionError, match="I'm not good"):
-            await graph.ainvoke(None, thread1)
+            await graph.ainvoke(None, thread1, checkpoint_during=checkpoint_during)
 
         # node "one" succeeded previously, so shouldn't be called again
         assert one.calls == 1
@@ -2048,7 +2056,9 @@ async def test_pending_writes_resume(
         # resume execution, without exception
         two.rtn = {"value": 3}
         # both the pending write and the new write were applied, 1 + 2 + 3 = 6
-        assert await graph.ainvoke(None, thread1) == {"value": 6}
+        assert await graph.ainvoke(
+            None, thread1, checkpoint_during=checkpoint_during
+        ) == {"value": 6}
 
         if "shallow" in checkpointer_name:
             assert len([c async for c in checkpointer.alist(thread1)]) == 1
@@ -2057,7 +2067,7 @@ async def test_pending_writes_resume(
         # check all final checkpoints
         checkpoints = [c async for c in checkpointer.alist(thread1)]
         # we should have 3
-        assert len(checkpoints) == 3
+        assert len(checkpoints) == (3 if checkpoint_during else 2)
         # the last one not too interesting for this test
         assert checkpoints[0] == CheckpointTuple(
             config={
@@ -2163,15 +2173,26 @@ async def test_pending_writes_resume(
                     "checkpoint_ns": "",
                     "checkpoint_id": checkpoints[2].config["configurable"][
                         "checkpoint_id"
-                    ],
+                    ]
+                    if checkpoint_during
+                    else AnyStr(),
                 }
             },
             pending_writes=UnsortedSequence(
                 (AnyStr(), "value", 2),
                 (AnyStr(), "__error__", 'ConnectionError("I\'m not good")'),
                 (AnyStr(), "value", 3),
+            )
+            if checkpoint_during
+            else UnsortedSequence(
+                (AnyStr(), "value", 2),
+                (AnyStr(), "__error__", 'ConnectionError("I\'m not good")'),
+                # the write against the previous checkpoint is not saved, as it is
+                # produced in a run where only the next checkpoint (the last) is saved
             ),
         )
+        if not checkpoint_during:
+            return
         assert checkpoints[2] == CheckpointTuple(
             config={
                 "configurable": {
@@ -2209,7 +2230,7 @@ async def test_pending_writes_resume(
 
 @pytest.mark.parametrize("checkpointer_name", REGULAR_CHECKPOINTERS_ASYNC)
 async def test_run_from_checkpoint_id_retains_previous_writes(
-    request: pytest.FixtureRequest, checkpointer_name: str, mocker: MockerFixture
+    checkpointer_name: str,
 ) -> None:
     class MyState(TypedDict):
         myval: Annotated[int, operator.add]
@@ -2254,8 +2275,8 @@ async def test_run_from_checkpoint_id_retains_previous_writes(
         history = [c async for c in graph.aget_state_history(thread1)]
 
         assert len(history) == 4
-        assert history[-1].values == {"myval": 0}
         assert history[0].values == {"myval": 4, "otherval": False}
+        assert history[-1].values == {"myval": 0}
 
         second_run_config = {
             **thread1,
@@ -2432,8 +2453,12 @@ async def test_send_sequences(checkpointer_name: str) -> None:
 
 
 @NEEDS_CONTEXTVARS
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC)
-async def test_imp_task(checkpointer_name: str) -> None:
+async def test_imp_task(checkpointer_name: str, checkpoint_during: bool) -> None:
+    if not checkpoint_during and "shallow" in checkpointer_name:
+        pytest.skip("Checkpointing during execution not supported")
+
     async with awith_checkpointer(checkpointer_name) as checkpointer:
         mapper_calls = 0
 
@@ -2453,7 +2478,12 @@ async def test_imp_task(checkpointer_name: str) -> None:
 
         tracer = FakeTracer()
         thread1 = {"configurable": {"thread_id": "1"}, "callbacks": [tracer]}
-        assert [c async for c in graph.astream([0, 1], thread1)] == [
+        assert [
+            c
+            async for c in graph.astream(
+                [0, 1], thread1, checkpoint_during=checkpoint_during
+            )
+        ] == [
             {"mapper": "00"},
             {"mapper": "11"},
             {
@@ -2477,7 +2507,9 @@ async def test_imp_task(checkpointer_name: str) -> None:
         assert any(r.inputs == {"input": 0} for r in mapper_runs)
         assert any(r.inputs == {"input": 1} for r in mapper_runs)
 
-        assert await graph.ainvoke(Command(resume="answer"), thread1) == [
+        assert await graph.ainvoke(
+            Command(resume="answer"), thread1, checkpoint_during=checkpoint_during
+        ) == [
             "00answer",
             "11answer",
         ]
@@ -2485,8 +2517,12 @@ async def test_imp_task(checkpointer_name: str) -> None:
 
 
 @NEEDS_CONTEXTVARS
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC)
-async def test_imp_nested(checkpointer_name: str) -> None:
+async def test_imp_nested(checkpointer_name: str, checkpoint_during: bool) -> None:
+    if not checkpoint_during and "shallow" in checkpointer_name:
+        pytest.skip("Checkpointing during execution not supported")
+
     async def mynode(input: list[str]) -> list[str]:
         return [it + "a" for it in input]
 
@@ -2526,7 +2562,12 @@ async def test_imp_nested(checkpointer_name: str) -> None:
         }
 
         thread1 = {"configurable": {"thread_id": "1"}}
-        assert [c async for c in graph.astream([0, 1], thread1)] == [
+        assert [
+            c
+            async for c in graph.astream(
+                [0, 1], thread1, checkpoint_during=checkpoint_during
+            )
+        ] == [
             {"submapper": "0"},
             {"mapper": "00"},
             {"submapper": "1"},
@@ -2543,15 +2584,21 @@ async def test_imp_nested(checkpointer_name: str) -> None:
             },
         ]
 
-        assert await graph.ainvoke(Command(resume="answer"), thread1) == [
+        assert await graph.ainvoke(
+            Command(resume="answer"), thread1, checkpoint_during=checkpoint_during
+        ) == [
             "00answera",
             "11answera",
         ]
 
 
 @NEEDS_CONTEXTVARS
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC)
-async def test_imp_task_cancel(checkpointer_name: str) -> None:
+async def test_imp_task_cancel(checkpointer_name: str, checkpoint_during: bool) -> None:
+    if not checkpoint_during and "shallow" in checkpointer_name:
+        pytest.skip("Checkpointing during execution not supported")
+
     async with awith_checkpointer(checkpointer_name) as checkpointer:
         mapper_calls = 0
         mapper_cancels = 0
@@ -2577,7 +2624,12 @@ async def test_imp_task_cancel(checkpointer_name: str) -> None:
             return [m + answer for m in mapped]
 
         thread1 = {"configurable": {"thread_id": "1"}}
-        assert [c async for c in graph.astream([0, 1], thread1)] == [
+        assert [
+            c
+            async for c in graph.astream(
+                [0, 1], thread1, checkpoint_during=checkpoint_during
+            )
+        ] == [
             {"mapper": "00"},
             {
                 "__interrupt__": (
@@ -2593,7 +2645,9 @@ async def test_imp_task_cancel(checkpointer_name: str) -> None:
         assert mapper_calls == 2
         assert mapper_cancels == 1
 
-        assert await graph.ainvoke(Command(resume="answer"), thread1) == [
+        assert await graph.ainvoke(
+            Command(resume="answer"), thread1, checkpoint_during=checkpoint_during
+        ) == [
             "00answer",
         ]
         assert mapper_calls == 3
@@ -2601,8 +2655,14 @@ async def test_imp_task_cancel(checkpointer_name: str) -> None:
 
 
 @NEEDS_CONTEXTVARS
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC)
-async def test_imp_sync_from_async(checkpointer_name: str) -> None:
+async def test_imp_sync_from_async(
+    checkpointer_name: str, checkpoint_during: bool
+) -> None:
+    if not checkpoint_during and "shallow" in checkpointer_name:
+        pytest.skip("Checkpointing during execution not supported")
+
     async with awith_checkpointer(checkpointer_name) as checkpointer:
 
         @task()
@@ -2625,7 +2685,12 @@ async def test_imp_sync_from_async(checkpointer_name: str) -> None:
             return fut_baz.result()
 
         thread1 = {"configurable": {"thread_id": "1"}}
-        assert [c async for c in graph.astream({"a": "0"}, thread1)] == [
+        assert [
+            c
+            async for c in graph.astream(
+                {"a": "0"}, thread1, checkpoint_during=checkpoint_during
+            )
+        ] == [
             {"foo": {"a": "0foo", "b": "bar"}},
             {"bar": {"a": "0foobar", "c": "bark"}},
             {"baz": {"a": "0foobarbaz", "c": "something else"}},
@@ -2634,8 +2699,14 @@ async def test_imp_sync_from_async(checkpointer_name: str) -> None:
 
 
 @NEEDS_CONTEXTVARS
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC)
-async def test_imp_stream_order(checkpointer_name: str) -> None:
+async def test_imp_stream_order(
+    checkpointer_name: str, checkpoint_during: bool
+) -> None:
+    if not checkpoint_during and "shallow" in checkpointer_name:
+        pytest.skip("Checkpointing during execution not supported")
+
     async with awith_checkpointer(checkpointer_name) as checkpointer:
 
         @task()
@@ -2659,7 +2730,12 @@ async def test_imp_stream_order(checkpointer_name: str) -> None:
             return await fut_baz
 
         thread1 = {"configurable": {"thread_id": "1"}}
-        assert [c async for c in graph.astream({"a": "0"}, thread1)] == [
+        assert [
+            c
+            async for c in graph.astream(
+                {"a": "0"}, thread1, checkpoint_during=checkpoint_during
+            )
+        ] == [
             {"foo": {"a": "0foo", "b": "bar"}},
             {"bar": {"a": "0foobar", "c": "bark"}},
             {"baz": {"a": "0foobarbaz", "c": "something else"}},
@@ -2667,8 +2743,11 @@ async def test_imp_stream_order(checkpointer_name: str) -> None:
         ]
 
 
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", REGULAR_CHECKPOINTERS_ASYNC)
-async def test_send_dedupe_on_resume(checkpointer_name: str) -> None:
+async def test_send_dedupe_on_resume(
+    checkpointer_name: str, checkpoint_during: bool
+) -> None:
     class InterruptOnce:
         ticks: int = 0
 
@@ -2719,7 +2798,9 @@ async def test_send_dedupe_on_resume(checkpointer_name: str) -> None:
     async with awith_checkpointer(checkpointer_name) as checkpointer:
         graph = builder.compile(checkpointer=checkpointer)
         thread1 = {"configurable": {"thread_id": "1"}}
-        assert await graph.ainvoke(["0"], thread1, debug=1) == [
+        assert await graph.ainvoke(
+            ["0"], thread1, checkpoint_during=checkpoint_during
+        ) == [
             "0",
             "1",
             "3.1",
@@ -2731,7 +2812,9 @@ async def test_send_dedupe_on_resume(checkpointer_name: str) -> None:
         assert builder.nodes["2"].runnable.func.ticks == 3
         assert builder.nodes["flaky"].runnable.func.ticks == 1
         # resume execution
-        assert await graph.ainvoke(None, thread1, debug=1) == [
+        assert await graph.ainvoke(
+            None, thread1, checkpoint_during=checkpoint_during
+        ) == [
             "0",
             "1",
             "3.1",
@@ -2748,7 +2831,8 @@ async def test_send_dedupe_on_resume(checkpointer_name: str) -> None:
         assert builder.nodes["flaky"].runnable.func.ticks == 2
         # check history
         history = [c async for c in graph.aget_state_history(thread1)]
-        assert history == [
+        assert len(history) == (6 if checkpoint_during else 2)
+        expected_history = [
             StateSnapshot(
                 values=[
                     "0",
@@ -2884,13 +2968,9 @@ async def test_send_dedupe_on_resume(checkpointer_name: str) -> None:
                         name="flaky",
                         path=("__pregel_push", 1),
                         error=None,
-                        interrupts=(
-                            Interrupt(
-                                value="Bahh", resumable=False, ns=None, when="during"
-                            ),
-                        ),
+                        interrupts=(Interrupt(value="Bahh", resumable=False, ns=None),),
                         state=None,
-                        result=["flaky|4"],
+                        result=["flaky|4"] if checkpoint_during else None,
                     ),
                     PregelTask(
                         id=AnyStr(),
@@ -3027,6 +3107,11 @@ async def test_send_dedupe_on_resume(checkpointer_name: str) -> None:
                 ),
             ),
         ]
+        if checkpoint_during:
+            assert history == expected_history
+        else:
+            assert history[0] == expected_history[0]
+            assert history[1] == expected_history[2]
 
 
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC)
@@ -5348,6 +5433,132 @@ async def test_nested_graph(snapshot: SnapshotAssertion) -> None:
     assert times_called == 1
 
 
+@pytest.mark.parametrize("checkpoint_during", [True, False])
+@pytest.mark.parametrize("checkpointer_name", REGULAR_CHECKPOINTERS_ASYNC)
+async def test_subgraph_checkpoint_true(
+    checkpointer_name: str, checkpoint_during: bool
+) -> None:
+    class InnerState(TypedDict):
+        my_key: Annotated[str, operator.add]
+        my_other_key: str
+
+    def inner_1(state: InnerState):
+        return {"my_key": " got here", "my_other_key": state["my_key"]}
+
+    def inner_2(state: InnerState):
+        return {"my_key": " and there"}
+
+    inner = StateGraph(InnerState)
+    inner.add_node("inner_1", inner_1)
+    inner.add_node("inner_2", inner_2)
+    inner.add_edge("inner_1", "inner_2")
+    inner.set_entry_point("inner_1")
+    inner.set_finish_point("inner_2")
+
+    class State(TypedDict):
+        my_key: str
+
+    graph = StateGraph(State)
+    graph.add_node("inner", inner.compile(checkpointer=True))
+    graph.add_edge(START, "inner")
+    graph.add_conditional_edges(
+        "inner", lambda s: "inner" if s["my_key"].count("there") < 2 else END
+    )
+
+    async with awith_checkpointer(checkpointer_name) as checkpointer:
+        app = graph.compile(checkpointer=checkpointer)
+
+        config = {"configurable": {"thread_id": "2"}}
+        assert [
+            c
+            async for c in app.astream(
+                {"my_key": ""},
+                config,
+                subgraphs=True,
+                checkpoint_during=checkpoint_during,
+            )
+        ] == [
+            (("inner",), {"inner_1": {"my_key": " got here", "my_other_key": ""}}),
+            (("inner",), {"inner_2": {"my_key": " and there"}}),
+            ((), {"inner": {"my_key": " got here and there"}}),
+            (
+                ("inner",),
+                {
+                    "inner_1": {
+                        "my_key": " got here",
+                        "my_other_key": " got here and there got here and there",
+                    }
+                },
+            ),
+            (("inner",), {"inner_2": {"my_key": " and there"}}),
+            (
+                (),
+                {
+                    "inner": {
+                        "my_key": " got here and there got here and there got here and there"
+                    }
+                },
+            ),
+        ]
+
+
+@NEEDS_CONTEXTVARS
+@pytest.mark.parametrize("checkpoint_during", [True, False])
+@pytest.mark.parametrize("checkpointer_name", REGULAR_CHECKPOINTERS_ASYNC)
+async def test_subgraph_checkpoint_true_interrupt(
+    checkpointer_name: str, checkpoint_during: bool
+) -> None:
+    # Define subgraph
+    class SubgraphState(TypedDict):
+        # note that none of these keys are shared with the parent graph state
+        bar: str
+        baz: str
+
+    def subgraph_node_1(state: SubgraphState):
+        baz_value = interrupt("Provide baz value")
+        return {"baz": baz_value}
+
+    def subgraph_node_2(state: SubgraphState):
+        return {"bar": state["bar"] + state["baz"]}
+
+    subgraph_builder = StateGraph(SubgraphState)
+    subgraph_builder.add_node(subgraph_node_1)
+    subgraph_builder.add_node(subgraph_node_2)
+    subgraph_builder.add_edge(START, "subgraph_node_1")
+    subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
+    subgraph = subgraph_builder.compile(checkpointer=True)
+
+    class ParentState(TypedDict):
+        foo: str
+
+    def node_1(state: ParentState):
+        return {"foo": "hi! " + state["foo"]}
+
+    async def node_2(state: ParentState, config: RunnableConfig):
+        response = await subgraph.ainvoke({"bar": state["foo"]})
+        return {"foo": response["bar"]}
+
+    builder = StateGraph(ParentState)
+    builder.add_node("node_1", node_1)
+    builder.add_node("node_2", node_2)
+    builder.add_edge(START, "node_1")
+    builder.add_edge("node_1", "node_2")
+
+    async with awith_checkpointer(checkpointer_name) as checkpointer:
+        graph = builder.compile(checkpointer=checkpointer)
+        config = {"configurable": {"thread_id": "1"}}
+
+        assert await graph.ainvoke(
+            {"foo": "foo"}, config, checkpoint_during=checkpoint_during
+        ) == {"foo": "hi! foo"}
+        assert (await graph.aget_state(config, subgraphs=True)).tasks[
+            0
+        ].state.values == {"bar": "hi! foo"}
+        assert await graph.ainvoke(
+            Command(resume="baz"), config, checkpoint_during=checkpoint_during
+        ) == {"foo": "hi! foobaz"}
+
+
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC)
 async def test_stream_subgraphs_during_execution(checkpointer_name: str) -> None:
     class InnerState(TypedDict):
@@ -5456,8 +5667,11 @@ async def test_stream_buffering_single_node(checkpointer_name: str) -> None:
         ]
 
 
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC)
-async def test_nested_graph_interrupts_parallel(checkpointer_name: str) -> None:
+async def test_nested_graph_interrupts_parallel(
+    checkpointer_name: str, checkpoint_during: bool
+) -> None:
     class InnerState(TypedDict):
         my_key: Annotated[str, operator.add]
         my_other_key: str
@@ -5506,11 +5720,13 @@ async def test_nested_graph_interrupts_parallel(checkpointer_name: str) -> None:
 
         # test invoke w/ nested interrupt
         config = {"configurable": {"thread_id": "1"}}
-        assert await app.ainvoke({"my_key": ""}, config, debug=True) == {
+        assert await app.ainvoke(
+            {"my_key": ""}, config, checkpoint_during=checkpoint_during
+        ) == {
             "my_key": " and parallel",
         }
 
-        assert await app.ainvoke(None, config, debug=True) == {
+        assert await app.ainvoke(None, config, checkpoint_during=checkpoint_during) == {
             "my_key": "got here and there and parallel and back again",
         }
 
@@ -5520,7 +5736,13 @@ async def test_nested_graph_interrupts_parallel(checkpointer_name: str) -> None:
         # test stream updates w/ nested interrupt
         config = {"configurable": {"thread_id": "2"}}
         assert [
-            c async for c in app.astream({"my_key": ""}, config, subgraphs=True)
+            c
+            async for c in app.astream(
+                {"my_key": ""},
+                config,
+                subgraphs=True,
+                checkpoint_during=checkpoint_during,
+            )
         ] == [
             # we got to parallel node first
             ((), {"outer_1": {"my_key": " and parallel"}}),
@@ -5530,7 +5752,12 @@ async def test_nested_graph_interrupts_parallel(checkpointer_name: str) -> None:
             ),
             ((), {"__interrupt__": ()}),
         ]
-        assert [c async for c in app.astream(None, config)] == [
+        assert [
+            c
+            async for c in app.astream(
+                None, config, checkpoint_during=checkpoint_during
+            )
+        ] == [
             {"outer_1": {"my_key": " and parallel"}, "__metadata__": {"cached": True}},
             {"inner": {"my_key": "got here and there"}},
             {"outer_2": {"my_key": " and back again"}},
@@ -5539,12 +5766,23 @@ async def test_nested_graph_interrupts_parallel(checkpointer_name: str) -> None:
         # test stream values w/ nested interrupt
         config = {"configurable": {"thread_id": "3"}}
         assert [
-            c async for c in app.astream({"my_key": ""}, config, stream_mode="values")
+            c
+            async for c in app.astream(
+                {"my_key": ""},
+                config,
+                stream_mode="values",
+                checkpoint_during=checkpoint_during,
+            )
         ] == [
             {"my_key": ""},
             {"my_key": " and parallel"},
         ]
-        assert [c async for c in app.astream(None, config, stream_mode="values")] == [
+        assert [
+            c
+            async for c in app.astream(
+                None, config, stream_mode="values", checkpoint_during=checkpoint_during
+            )
+        ] == [
             {"my_key": ""},
             {"my_key": "got here and there and parallel"},
             {"my_key": "got here and there and parallel and back again"},
@@ -5554,16 +5792,32 @@ async def test_nested_graph_interrupts_parallel(checkpointer_name: str) -> None:
         app = graph.compile(checkpointer=checkpointer, interrupt_before=["outer_1"])
         config = {"configurable": {"thread_id": "4"}}
         assert [
-            c async for c in app.astream({"my_key": ""}, config, stream_mode="values")
+            c
+            async for c in app.astream(
+                {"my_key": ""},
+                config,
+                stream_mode="values",
+                checkpoint_during=checkpoint_during,
+            )
         ] == [
             {"my_key": ""},
         ]
         # while we're waiting for the node w/ interrupt inside to finish
-        assert [c async for c in app.astream(None, config, stream_mode="values")] == [
+        assert [
+            c
+            async for c in app.astream(
+                None, config, stream_mode="values", checkpoint_during=checkpoint_during
+            )
+        ] == [
             {"my_key": ""},
             {"my_key": " and parallel"},
         ]
-        assert [c async for c in app.astream(None, config, stream_mode="values")] == [
+        assert [
+            c
+            async for c in app.astream(
+                None, config, stream_mode="values", checkpoint_during=checkpoint_during
+            )
+        ] == [
             {"my_key": ""},
             {"my_key": "got here and there and parallel"},
             {"my_key": "got here and there and parallel and back again"},
@@ -5573,23 +5827,42 @@ async def test_nested_graph_interrupts_parallel(checkpointer_name: str) -> None:
         app = graph.compile(checkpointer=checkpointer, interrupt_after=["outer_1"])
         config = {"configurable": {"thread_id": "5"}}
         assert [
-            c async for c in app.astream({"my_key": ""}, config, stream_mode="values")
+            c
+            async for c in app.astream(
+                {"my_key": ""},
+                config,
+                stream_mode="values",
+                checkpoint_during=checkpoint_during,
+            )
         ] == [
             {"my_key": ""},
             {"my_key": " and parallel"},
         ]
-        assert [c async for c in app.astream(None, config, stream_mode="values")] == [
+        assert [
+            c
+            async for c in app.astream(
+                None, config, stream_mode="values", checkpoint_during=checkpoint_during
+            )
+        ] == [
             {"my_key": ""},
             {"my_key": "got here and there and parallel"},
         ]
-        assert [c async for c in app.astream(None, config, stream_mode="values")] == [
+        assert [
+            c
+            async for c in app.astream(
+                None, config, stream_mode="values", checkpoint_during=checkpoint_during
+            )
+        ] == [
             {"my_key": "got here and there and parallel"},
             {"my_key": "got here and there and parallel and back again"},
         ]
 
 
+@pytest.mark.parametrize("checkpoint_during", [True, False])
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC)
-async def test_doubly_nested_graph_interrupts(checkpointer_name: str) -> None:
+async def test_doubly_nested_graph_interrupts(
+    checkpointer_name: str, checkpoint_during: bool
+) -> None:
     class State(TypedDict):
         my_key: str
 
@@ -5642,11 +5915,13 @@ async def test_doubly_nested_graph_interrupts(checkpointer_name: str) -> None:
 
         # test invoke w/ nested interrupt
         config = {"configurable": {"thread_id": "1"}}
-        assert await app.ainvoke({"my_key": "my value"}, config, debug=True) == {
+        assert await app.ainvoke(
+            {"my_key": "my value"}, config, checkpoint_during=checkpoint_during
+        ) == {
             "my_key": "hi my value",
         }
 
-        assert await app.ainvoke(None, config, debug=True) == {
+        assert await app.ainvoke(None, config, checkpoint_during=checkpoint_during) == {
             "my_key": "hi my value here and there and back again",
         }
 
@@ -5655,12 +5930,22 @@ async def test_doubly_nested_graph_interrupts(checkpointer_name: str) -> None:
         config = {
             "configurable": {"thread_id": "2", CONFIG_KEY_NODE_FINISHED: nodes.append}
         }
-        assert [c async for c in app.astream({"my_key": "my value"}, config)] == [
+        assert [
+            c
+            async for c in app.astream(
+                {"my_key": "my value"}, config, checkpoint_during=checkpoint_during
+            )
+        ] == [
             {"parent_1": {"my_key": "hi my value"}},
             {"__interrupt__": ()},
         ]
         assert nodes == ["parent_1", "grandchild_1"]
-        assert [c async for c in app.astream(None, config)] == [
+        assert [
+            c
+            async for c in app.astream(
+                None, config, checkpoint_during=checkpoint_during
+            )
+        ] == [
             {"child": {"my_key": "hi my value here and there"}},
             {"parent_2": {"my_key": "hi my value here and there and back again"}},
         ]
@@ -5678,13 +5963,21 @@ async def test_doubly_nested_graph_interrupts(checkpointer_name: str) -> None:
         assert [
             c
             async for c in app.astream(
-                {"my_key": "my value"}, config, stream_mode="values"
+                {"my_key": "my value"},
+                config,
+                stream_mode="values",
+                checkpoint_during=checkpoint_during,
             )
         ] == [
             {"my_key": "my value"},
             {"my_key": "hi my value"},
         ]
-        assert [c async for c in app.astream(None, config, stream_mode="values")] == [
+        assert [
+            c
+            async for c in app.astream(
+                None, config, stream_mode="values", checkpoint_during=checkpoint_during
+            )
+        ] == [
             {"my_key": "hi my value"},
             {"my_key": "hi my value here and there"},
             {"my_key": "hi my value here and there and back again"},

libs/sdk-js/.gitignore

@@ -6,6 +6,10 @@ client.cjs
 client.js
 client.d.ts
 client.d.cts
+auth.cjs
+auth.js
+auth.d.ts
+auth.d.cts
 react.cjs
 react.js
 react.d.ts

libs/sdk-js/langchain.config.js

@@ -14,6 +14,7 @@ export const config = {
   entrypoints: {
     index: "index",
     client: "client",
+    auth: "auth/index",
     react: "react/index",
     "react-ui": "react-ui/index",
     "react-ui/server": "react-ui/server/index",

libs/sdk-js/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@langchain/langgraph-sdk",
-  "version": "0.0.63",
+  "version": "0.0.64",
   "description": "Client library for interacting with the LangGraph API",
   "type": "module",
   "packageManager": "yarn@1.22.19",
@@ -72,6 +72,15 @@
       "import": "./client.js",
       "require": "./client.cjs"
     },
+    "./auth": {
+      "types": {
+        "import": "./auth.d.ts",
+        "require": "./auth.d.cts",
+        "default": "./auth.d.ts"
+      },
+      "import": "./auth.js",
+      "require": "./auth.cjs"
+    },
     "./react": {
       "types": {
         "import": "./react.d.ts",
@@ -111,6 +120,10 @@
     "client.js",
     "client.d.ts",
     "client.d.cts",
+    "auth.cjs",
+    "auth.js",
+    "auth.d.ts",
+    "auth.d.cts",
     "react.cjs",
     "react.js",
     "react.d.ts",

libs/sdk-js/src/auth/error.ts

@@ -0,0 +1,80 @@
+const HTTP_STATUS_MAPPING: { [key: number]: string } = {
+  100: "Continue",
+  101: "Switching Protocols",
+  102: "Processing",
+  103: "Early Hints",
+  200: "OK",
+  201: "Created",
+  202: "Accepted",
+  203: "Non-Authoritative Information",
+  204: "No Content",
+  205: "Reset Content",
+  206: "Partial Content",
+  207: "Multi-Status",
+  208: "Already Reported",
+  226: "IM Used",
+  300: "Multiple Choices",
+  301: "Moved Permanently",
+  302: "Found",
+  303: "See Other",
+  304: "Not Modified",
+  305: "Use Proxy",
+  307: "Temporary Redirect",
+  308: "Permanent Redirect",
+  400: "Bad Request",
+  401: "Unauthorized",
+  402: "Payment Required",
+  403: "Forbidden",
+  404: "Not Found",
+  405: "Method Not Allowed",
+  406: "Not Acceptable",
+  407: "Proxy Authentication Required",
+  408: "Request Timeout",
+  409: "Conflict",
+  410: "Gone",
+  411: "Length Required",
+  412: "Precondition Failed",
+  413: "Request Entity Too Large",
+  414: "Request-URI Too Long",
+  415: "Unsupported Media Type",
+  416: "Requested Range Not Satisfiable",
+  417: "Expectation Failed",
+  418: "I'm a Teapot",
+  421: "Misdirected Request",
+  422: "Unprocessable Entity",
+  423: "Locked",
+  424: "Failed Dependency",
+  425: "Too Early",
+  426: "Upgrade Required",
+  428: "Precondition Required",
+  429: "Too Many Requests",
+  431: "Request Header Fields Too Large",
+  451: "Unavailable For Legal Reasons",
+  500: "Internal Server Error",
+  501: "Not Implemented",
+  502: "Bad Gateway",
+  503: "Service Unavailable",
+  504: "Gateway Timeout",
+  505: "HTTP Version Not Supported",
+  506: "Variant Also Negotiates",
+  507: "Insufficient Storage",
+  508: "Loop Detected",
+  510: "Not Extended",
+  511: "Network Authentication Required",
+};
+
+export class HTTPException extends Error {
+  status: number;
+  headers: HeadersInit;
+
+  constructor(
+    status: number,
+    options?: { message?: string; headers?: HeadersInit; cause?: unknown },
+  ) {
+    super(options?.message ?? HTTP_STATUS_MAPPING[status] ?? "Unknown error", {
+      cause: options?.cause,
+    });
+    this.status = status;
+    this.headers = options?.headers ?? {};
+  }
+}

libs/sdk-js/src/auth/index.ts

@@ -0,0 +1,36 @@
+import type {
+  AuthenticateCallback,
+  AnyCallback,
+  CallbackEvent,
+  OnCallback,
+  BaseAuthReturn,
+  ToUserLike,
+  BaseUser,
+} from "./types.js";
+
+export class Auth<
+  TExtra = {},
+  TAuthReturn extends BaseAuthReturn = BaseAuthReturn,
+  TUser extends BaseUser = ToUserLike<TAuthReturn>,
+> {
+  "~handlerCache": {
+    authenticate?: AuthenticateCallback<BaseAuthReturn>;
+    callbacks?: Record<string, AnyCallback>;
+  } = {};
+
+  authenticate<T extends BaseAuthReturn>(
+    cb: AuthenticateCallback<T>,
+  ): Auth<TExtra, T> {
+    this["~handlerCache"].authenticate = cb;
+    return this as unknown as Auth<TExtra, T>;
+  }
+
+  on<T extends CallbackEvent>(event: T, callback: OnCallback<T, TUser>): this {
+    this["~handlerCache"].callbacks ??= {};
+    this["~handlerCache"].callbacks[event as string] = callback as AnyCallback;
+    return this;
+  }
+}
+
+export type { Filters, ResourceActionType } from "./types.js";
+export { HTTPException } from "./error.js";

libs/sdk-js/src/auth/types.ts

@@ -0,0 +1,345 @@
+type Maybe<T> = T | null | undefined;
+type PromiseMaybe<T> = Promise<T> | T;
+
+interface AssistantConfig {
+  tags?: Maybe<string[]>;
+  recursion_limit?: Maybe<number>;
+  configurable?: Maybe<{
+    thread_id?: Maybe<string>;
+    thread_ts?: Maybe<string>;
+    [key: string]: unknown;
+  }>;
+}
+
+interface AssistantCreate {
+  assistant_id?: Maybe<string>;
+  metadata?: Maybe<Record<string, unknown>>;
+  config?: Maybe<AssistantConfig>;
+  if_exists?: Maybe<"raise" | "do_nothing">;
+  name?: Maybe<string>;
+  graph_id: string;
+}
+
+interface AssistantRead {
+  assistant_id: string;
+  metadata?: Maybe<Record<string, unknown>>;
+}
+
+interface AssistantUpdate {
+  assistant_id: string;
+  metadata?: Maybe<Record<string, unknown>>;
+  config?: Maybe<AssistantConfig>;
+  graph_id?: Maybe<string>;
+  name?: Maybe<string>;
+  version?: Maybe<number>;
+}
+
+interface AssistantDelete {
+  assistant_id: string;
+}
+
+interface AssistantSearch {
+  graph_id?: Maybe<string>;
+  metadata?: Maybe<Record<string, unknown>>;
+  limit?: Maybe<number>;
+  offset?: Maybe<number>;
+}
+
+interface ThreadCreate {
+  thread_id?: Maybe<string>;
+  metadata?: Maybe<Record<string, unknown>>;
+  if_exists?: Maybe<"raise" | "do_nothing">;
+}
+
+interface ThreadRead {
+  thread_id?: Maybe<string>;
+}
+
+interface ThreadUpdate {
+  thread_id?: Maybe<string>;
+  metadata?: Maybe<Record<string, unknown>>;
+  action?: Maybe<"interrupt" | "rollback">;
+}
+
+interface ThreadDelete {
+  thread_id?: Maybe<string>;
+  run_id?: Maybe<string>;
+}
+
+interface ThreadSearch {
+  thread_id?: Maybe<string>;
+  status?: Maybe<"idle" | "busy" | "interrupted" | "error" | (string & {})>;
+  metadata?: Maybe<Record<string, unknown>>;
+  values?: Maybe<Record<string, unknown>>;
+  limit?: Maybe<number>;
+  offset?: Maybe<number>;
+}
+
+interface CronCreate {
+  payload?: Maybe<Record<string, unknown>>;
+  schedule: string;
+  cron_id?: Maybe<string>;
+  thread_id?: Maybe<string>;
+  user_id?: Maybe<string>;
+  end_time?: Maybe<string>;
+}
+
+interface CronRead {
+  cron_id: string;
+}
+
+interface CronUpdate {
+  cron_id: string;
+  payload?: Maybe<Record<string, unknown>>;
+  schedule?: Maybe<string>;
+}
+
+interface CronDelete {
+  cron_id: string;
+}
+
+interface CronSearch {
+  assistant_id?: Maybe<string>;
+  thread_id?: Maybe<string>;
+  limit?: Maybe<number>;
+  offset?: Maybe<number>;
+}
+
+interface StorePut {
+  namespace: string[];
+  key: string;
+  value: Record<string, unknown>;
+}
+
+interface StoreGet {
+  namespace: Maybe<string[]>;
+  key: string;
+}
+
+interface StoreSearch {
+  namespace?: Maybe<string[]>;
+  filter?: Maybe<Record<string, unknown>>;
+  limit?: Maybe<number>;
+  offset?: Maybe<number>;
+  query?: Maybe<string>;
+}
+
+interface StoreListNamespaces {
+  namespace?: Maybe<string[]>;
+  suffix?: Maybe<string[]>;
+  max_depth?: Maybe<number>;
+  limit?: Maybe<number>;
+  offset?: Maybe<number>;
+}
+
+interface StoreDelete {
+  namespace?: Maybe<string[]>;
+  key: string;
+}
+
+interface RunsCreate {
+  thread_id?: Maybe<string>;
+  assistant_id: string;
+  run_id: string;
+  status: Maybe<
+    "pending" | "running" | "error" | "success" | "timeout" | "interrupted"
+  >;
+  metadata?: Maybe<Record<string, unknown>>;
+  prevent_insert_if_inflight?: Maybe<boolean>;
+  multitask_strategy?: Maybe<"interrupt" | "rollback" | "reject" | "enqueue">;
+  if_not_exists?: Maybe<"reject" | "create">;
+  after_seconds?: Maybe<number>;
+  kwargs: Record<string, unknown>;
+}
+
+export interface ResourceActionType {
+  ["threads:create"]: ThreadCreate;
+  ["threads:read"]: ThreadRead;
+  ["threads:update"]: ThreadUpdate;
+  ["threads:delete"]: ThreadDelete;
+  ["threads:search"]: ThreadSearch;
+  ["threads:create_run"]: RunsCreate;
+
+  ["assistants:create"]: AssistantCreate;
+  ["assistants:read"]: AssistantRead;
+  ["assistants:update"]: AssistantUpdate;
+  ["assistants:delete"]: AssistantDelete;
+  ["assistants:search"]: AssistantSearch;
+
+  ["crons:create"]: CronCreate;
+  ["crons:read"]: CronRead;
+  ["crons:update"]: CronUpdate;
+  ["crons:delete"]: CronDelete;
+  ["crons:search"]: CronSearch;
+
+  ["store:put"]: StorePut;
+  ["store:get"]: StoreGet;
+  ["store:search"]: StoreSearch;
+  ["store:list_namespaces"]: StoreListNamespaces;
+  ["store:delete"]: StoreDelete;
+}
+interface ResourceType {
+  threads:
+    | "threads:create"
+    | "threads:read"
+    | "threads:update"
+    | "threads:delete"
+    | "threads:search"
+    | "threads:create_run";
+
+  assistants:
+    | "assistants:create"
+    | "assistants:read"
+    | "assistants:update"
+    | "assistants:delete"
+    | "assistants:search";
+  crons:
+    | "crons:create"
+    | "crons:read"
+    | "crons:update"
+    | "crons:delete"
+    | "crons:search";
+
+  store:
+    | "store:put"
+    | "store:get"
+    | "store:search"
+    | "store:list_namespaces"
+    | "store:delete";
+}
+interface ActionType {
+  "*:create": "threads:create" | "assistants:create" | "crons:create";
+
+  "*:read": "threads:read" | "assistants:read" | "crons:read";
+
+  "*:update": "threads:update" | "assistants:update" | "crons:update";
+
+  "*:delete":
+    | "threads:delete"
+    | "assistants:delete"
+    | "crons:delete"
+    | "store:delete";
+
+  "*:search":
+    | "threads:search"
+    | "assistants:search"
+    | "crons:search"
+    | "store:search";
+
+  "*:create_run": "threads:create_run";
+
+  "*:put": "store:put";
+
+  "*:get": "store:get";
+
+  "*:list_namespaces": "store:list_namespaces";
+}
+
+export type BaseAuthReturn =
+  | {
+      is_authenticated?: boolean;
+      display_name?: string;
+      identity: string;
+      permissions: string[];
+    }
+  | string;
+
+export interface BaseUser {
+  is_authenticated: boolean;
+  display_name: string;
+  identity: string;
+  permissions: string[];
+}
+
+export type ToUserLike<T extends BaseAuthReturn> = T extends string
+  ? {
+      is_authenticated: boolean;
+      display_name: string;
+      identity: string;
+      permissions: string[];
+    }
+  : Omit<T, "is_authenticated" | "display_name"> & {
+      is_authenticated: boolean;
+      display_name: string;
+    };
+
+type CallbackParameter<
+  Resource extends string = string,
+  Action extends string = string,
+  Value extends unknown = unknown,
+  TUser extends BaseUser = BaseUser,
+> = {
+  resource: Resource;
+  action: Action;
+  value: Value;
+  user: TUser;
+  permissions: string[];
+};
+
+type ContextMap = {
+  [ActionType in keyof ResourceActionType]: CallbackParameter<
+    ActionType extends `${infer Resource}:${string}` ? Resource : never,
+    ActionType,
+    ResourceActionType[ActionType],
+    BaseUser
+  >;
+};
+
+type ActionCallbackParameter<
+  T extends keyof ActionType,
+  TUser extends BaseUser = BaseUser,
+> = ContextMap[ActionType[T]] & { user: TUser };
+type AuthCallbackParameter<
+  T extends keyof ResourceActionType,
+  TUser extends BaseUser = BaseUser,
+> = ContextMap[T] & { user: TUser };
+type ResourceCallbackParameter<
+  T extends keyof ResourceType,
+  TUser extends BaseUser = BaseUser,
+> = ContextMap[ResourceType[T]] & { user: TUser };
+
+export type Filters<TKey extends string | number | symbol> = {
+  [key in TKey]: string | { [op in "$contains" | "$eq"]?: string };
+};
+
+export interface AuthenticateCallback<T extends BaseAuthReturn> {
+  (request: Request): PromiseMaybe<T>;
+}
+
+type OnKey = keyof ResourceType | keyof ActionType | keyof ResourceActionType;
+
+type OnSingleParameter<
+  T extends OnKey,
+  TUser extends BaseUser = BaseUser,
+> = T extends keyof ResourceType
+  ? ResourceCallbackParameter<T, TUser>
+  : T extends keyof ActionType
+    ? ActionCallbackParameter<T, TUser>
+    : T extends keyof ResourceActionType
+      ? AuthCallbackParameter<T, TUser>
+      : never;
+
+type OnParameter<
+  T extends "*" | OnKey | OnKey[],
+  TUser extends BaseUser = BaseUser,
+> = T extends OnKey[]
+  ? OnSingleParameter<T[number], TUser>
+  : T extends "*"
+    ? AuthCallbackParameter<keyof ResourceActionType, TUser>
+    : T extends OnKey
+      ? OnSingleParameter<T, TUser>
+      : never;
+
+export type AnyCallback = (
+  request: CallbackParameter,
+) => void | boolean | Filters<string>;
+
+export type CallbackEvent = "*" | OnKey | OnKey[];
+
+export type OnCallback<
+  T extends CallbackEvent,
+  TUser extends BaseUser = BaseUser,
+  TMetadata extends Record<string, unknown> = Record<string, unknown>,
+> = (
+  request: OnParameter<T, TUser>,
+) => void | boolean | Filters<keyof TMetadata>;

libs/sdk-js/tsconfig.json

@@ -2,11 +2,7 @@
   "extends": "@tsconfig/recommended",
   "compilerOptions": {
     "target": "ES2021",
-    "lib": [
-      "ES2021",
-      "ES2022.Object",
-      "DOM"
-    ],
+    "lib": ["ES2021", "ES2022.Object", "ES2022.Error", "DOM"],
     "module": "NodeNext",
     "moduleResolution": "nodenext",
     "esModuleInterop": true,
@@ -22,24 +18,14 @@
     "jsx": "react-jsx",
     "outDir": "dist"
   },
-  "include": [
-    "src/**/*"
-  ],
-  "exclude": [
-    "node_modules",
-    "dist",
-    "coverage"
-  ],
+  "include": ["src/**/*"],
+  "exclude": ["node_modules", "dist", "coverage"],
   "includeVersion": true,
   "typedocOptions": {
-    "entryPoints": [
-      "src/client.ts"
-    ],
+    "entryPoints": ["src/client.ts"],
     "readme": "none",
     "out": "docs",
-    "plugin": [
-      "typedoc-plugin-markdown"
-    ],
+    "plugin": ["typedoc-plugin-markdown"],
     "excludePrivate": true,
     "excludeProtected": true,
     "excludeExternals": false