Create a RunnableConfig with mock Runtime for testing ToolNode. Returns: RunnableConfig with __pregel_runtime in configurable dict.

Create a small project structure for testing.

Create a database cursor as a context manager. Args: transaction (bool): whether to use transaction for the DB operations

Convert namespace string to tuple.

Create a temporary Docker config with only the push token. Yields the path to a temporary config directory that can be passed to `docker --co

Return the exception from a future, without raising CancelledError.

Backdate a row's expires_at into the past without deleting it (unswept).

Generate error message content based on exception handling configuration. This function centralizes error message generation logic, supporting di

Graph with a subgraph that raises.

Record that an inbox has an active consumer. If the handle is already closed (status != 'started'), immediately enqueue a sentinel so

Convert namespace tuple to text string.

Restore successful channel writes from checkpoint to in-memory tasks. Skips control signals (ERROR, ERROR_SOURCE_NODE, INTERRUPT, RESUME)

Return WHERE clause predicates for alist() given config, filter, before. This method returns a tuple of a string and a tuple of values. The s

Collect `(ns, payload)` for every `tools`-mode chunk.

Strip push stamps from a StreamChannel's internal buffer.

Validate that a handler function has the correct signature. Args: fn: The handler function to validate handler_type: Description

Validate that a reconnect Location URL is same-origin as the base URL. Raises ValueError if the Location header points to a different origin

List checkpoints from the database asynchronously. This method retrieves a list of checkpoint tuples from the Postgres database based

Store intermediate writes linked to a checkpoint asynchronously. This method saves intermediate writes associated with a checkpoint to the da

Async-stream events from this remote graph. For `version="v3"`, awaits to an `_AsyncRemoteGraphRunStream`, matching the local `Pregel

Convert an HTTP base URL plus API path into a WebSocket URL.

Delete the cached values for the given namespaces. If no namespaces are provided, clear all cached values.

Tear down the thread stream. Idempotent.

Delete an item. Args: key: The unique identifier for the item. namespace: Optional list of strings representing the n

Return the set of DeltaChannel names that should snapshot now. A channel snapshots when EITHER its accumulated update count reaches `snapshot

Fetch a checkpoint using the given configuration. Args: config: Configuration specifying which checkpoint to retrieve. R

Get a thread by ID. Args: thread_id: The ID of the thread to get. include: Additional fields to include in the respon

Get checkpoint metadata in a backwards-compatible manner.

Fast-path override of `BaseCheckpointSaver.get_delta_channel_history`. Two-stage query: * Stage 1 (paged): newest-first slice of `ch

Get the state history of a thread. Args: thread_id: The ID of the thread to get the state history for. checkpoint: Re

Get checkpoint metadata in a backwards-compatible manner.

Extract text from an object using a path expression or pre-tokenized path. Args: obj: The object to extract text from path: Eithe

Register the JSON encryption handler. Example: ```python @encryption.encrypt.json async def encrypt_json(

Map input chunk to a sequence of pending writes in the form (channel, value).

Save a checkpoint to the in-memory storage. This method saves a checkpoint to the in-memory storage. The checkpoint is associated wit

Store or update an item. Args: namespace: A list of strings representing the namespace path. key: The unique identifi

Adds the specified channels to read from, without subscribing to them.

Set per-open stream scripts consumed in order by /stream/events.

Add channels to subscribe to. Node will be invoked when any of these channels are updated, with a dict of the channel values as input

Build the canonical error for using `timeout` with a sync target.

Update the state of a thread. Args: thread_id: The ID of the thread to update. values: The values to update the state

Run the validation suite against a registered checkpointer. Args: registered: A RegisteredCheckpointer (from @checkpointer_test decorator

Return a new awaitable that raises `asyncio.TimeoutError` after `timeout` seconds. Bounds the wait for the lifecycle terminal (and only that

Call the async pump until it returns False.

Pre-subscribe every projection in the mux so synthetic pushes accumulate. Real consumer code subscribes by iterating the projection; tests in

Wait for the current run.start to commit the thread server-side. No-op when no run.start is in flight. Re-raises if run.start failed.

Build an InMemorySaver with a 3-checkpoint chain and per-step writes for a `messages` channel. Returns `(saver, thread_id, namespace)

Graph with two delta channels A and B. The node only writes to channel A; B is never written by the node. `n_loops` controls how many superst

Return the checkpoint namespace for the parent graph. The checkpoint namespace is a `|`-separated path. Each segment is usually of the form `