algo.ts

/* eslint-disable no-param-reassign */
import {
  mergeConfigs,
  patchConfig,
  RunnableConfig,
} from "@langchain/core/runnables";
import { CallbackManagerForChainRun } from "@langchain/core/callbacks/manager";
import {
  All,
  BaseCheckpointSaver,
  Checkpoint,
  ReadonlyCheckpoint,
  copyCheckpoint,
  type PendingWrite,
  type PendingWriteValue,
  type WritesSortKey,
  uuid5,
  maxChannelVersion,
  BaseStore,
  CheckpointPendingWrite,
  compareWritesSortKeys,
  writesSortKey,
  SendProtocol,
} from "@langchain/langgraph-checkpoint";
import {
  BaseChannel,
  createCheckpoint,
  emptyChannels,
  isDeltaChannel,
  getOnlyChannels,
} from "../channels/base.js";
import { PregelNode } from "./read.js";
import { readChannel, readChannels } from "./io.js";
import {
  _isSend,
  _isSendInterface,
  CONFIG_KEY_CHECKPOINT_MAP,
  CHECKPOINT_NAMESPACE_SEPARATOR,
  CONFIG_KEY_CHECKPOINTER,
  CONFIG_KEY_READ,
  CONFIG_KEY_TASK_ID,
  CONFIG_KEY_SEND,
  INTERRUPT,
  RESERVED,
  Send,
  TAG_HIDDEN,
  TASKS,
  CHECKPOINT_NAMESPACE_END,
  PUSH,
  PULL,
  RESUME,
  NULL_TASK_ID,
  CONFIG_KEY_SCRATCHPAD,
  RETURN,
  ERROR,
  ERROR_SOURCE_NODE,
  NO_WRITES,
  CONFIG_KEY_PREVIOUS_STATE,
  PREVIOUS,
  CACHE_NS_WRITES,
  CONFIG_KEY_RESUME_MAP,
  CONFIG_KEY_NODE_ERROR,
  START,
} from "../constants.js";
import {
  Call,
  isCall,
  PregelExecutableTask,
  PregelScratchpad,
  PregelTaskDescription,
  SimpleTaskPath,
  TaskPath,
  VariadicTaskPath,
} from "./types.js";
import { EmptyChannelError, InvalidUpdateError, NodeError } from "../errors.js";
import { getNullChannelVersion } from "./utils/index.js";
import { ExecutionInfo, LangGraphRunnableConfig } from "./runnable_types.js";
import { getRunnableForFunc } from "./call.js";
import { IterableReadableWritableStream } from "./stream.js";
import { XXH3 } from "../hash.js";
import { interruptId } from "../interrupt.js";
import { Topic } from "../channels/topic.js";

/**
 * Construct a type with a set of properties K of type T
 */
export type StrRecord<K extends string, T> = {
  [P in K]: T;
};

export type WritesProtocol<C = string> = {
  name: string;
  writes: PendingWrite<C>[];
  triggers: string[];
  path?: TaskPath;
};

export const increment = (current?: number) => {
  return current !== undefined ? current + 1 : 1;
};

/**
 * Serialize a task path to the string stored with its writes, matching
 * Python's `task_path_str` byte for byte:
 *
 * - arrays (nested paths included) become `"~"` + their serialized elements
 *   joined by `", "` — so the empty path is `"~"`;
 * - booleans format as ints (Python's `isinstance(x, int)` is true for
 *   `bool` there): `true` → `"0000000001"`, `false` → `"0000000000"`;
 * - ints are zero-padded to 10 digits, sign-aware: `-1` → `"-000000001"`;
 * - anything else (node names) is used as-is.
 *
 * This is the `taskPath` passed to `BaseCheckpointSaver.putWrites` and the
 * first component of `writesSortKey`, so the stored string and its sort
 * order are identical across the JS and Python runtimes.
 */
export function taskPathStr(path: unknown): string {
  if (Array.isArray(path)) {
    return `~${path.map(taskPathStr).join(", ")}`;
  }
  if (typeof path === "boolean") {
    return path ? "0000000001" : "0000000000";
  }
  if (typeof path === "number") {
    return path < 0
      ? `-${String(-path).padStart(9, "0")}`
      : String(path).padStart(10, "0");
  }
  if (path === undefined || path === null) {
    // Python serializes the empty tuple `()` — the path of input/null-task
    // writes — to `"~"`.
    return "~";
  }
  return String(path);
}

function triggersNextStep(
  updatedChannels: Set<string>,
  triggerToNodes: Record<string, string[]> | undefined
) {
  if (triggerToNodes == null) return false;

  for (const chan of updatedChannels) {
    if (triggerToNodes[chan]) return true;
  }

  return false;
}

// Avoids unnecessary double iteration
function maxChannelMapVersion(
  channelVersions: Record<string, number | string>
): number | string | undefined {
  let maxVersion: number | string | undefined;
  for (const chan in channelVersions) {
    if (!Object.prototype.hasOwnProperty.call(channelVersions, chan)) continue;
    if (maxVersion == null) {
      maxVersion = channelVersions[chan];
    } else {
      maxVersion = maxChannelVersion(maxVersion, channelVersions[chan]);
    }
  }
  return maxVersion;
}

export function shouldInterrupt<N extends PropertyKey, C extends PropertyKey>(
  checkpoint: Checkpoint,
  interruptNodes: All | N[],
  tasks: PregelExecutableTask<N, C>[]
): boolean {
  const nullVersion = getNullChannelVersion(checkpoint.channel_versions);
  const seen = checkpoint.versions_seen[INTERRUPT] ?? {};

  let anyChannelUpdated = false;

  if (
    (checkpoint.channel_versions[START] ?? nullVersion) >
    (seen[START] ?? nullVersion)
  ) {
    anyChannelUpdated = true;
  } else {
    for (const chan in checkpoint.channel_versions) {
      if (
        !Object.prototype.hasOwnProperty.call(checkpoint.channel_versions, chan)
      )
        continue;

      if (checkpoint.channel_versions[chan] > (seen[chan] ?? nullVersion)) {
        anyChannelUpdated = true;
        break;
      }
    }
  }

  const anyTriggeredNodeInInterruptNodes = tasks.some((task) =>
    interruptNodes === "*"
      ? !task.config?.tags?.includes(TAG_HIDDEN)
      : interruptNodes.includes(task.name)
  );

  return anyChannelUpdated && anyTriggeredNodeInInterruptNodes;
}

export function _localRead<Cc extends Record<string, BaseChannel>>(
  checkpoint: ReadonlyCheckpoint,
  channels: Cc,
  task: WritesProtocol<keyof Cc>,
  select: Array<keyof Cc> | keyof Cc,
  fresh: boolean = false
): Record<string, unknown> | unknown {
  let updated = new Set<keyof Cc>();

  if (!Array.isArray(select)) {
    for (const [c] of task.writes) {
      if (c === select) {
        updated = new Set([c]);
        break;
      }
    }
    updated = updated || new Set();
  } else {
    updated = new Set(
      select.filter((c) => task.writes.some(([key, _]) => key === c))
    );
  }

  let values: Record<string, unknown>;

  if (fresh && updated.size > 0) {
    const localChannels = Object.fromEntries(
      Object.entries(channels).filter(([k, _]) => updated.has(k as keyof Cc))
    ) as Partial<Cc>;

    // DeltaChannels are omitted from `channel_values` by default (their state
    // is reconstructed from ancestor writes via the saver). A local read has
    // no saver, so snapshot every delta channel inline to round-trip its full
    // live value through `emptyChannels`.
    const channelsToSnapshot = new Set<string>();
    for (const k in localChannels) {
      if (!Object.prototype.hasOwnProperty.call(localChannels, k)) continue;
      const ch = (localChannels as Record<string, BaseChannel>)[k];
      if (isDeltaChannel(ch) && ch.isAvailable()) channelsToSnapshot.add(k);
    }

    const newCheckpoint = createCheckpoint(
      checkpoint,
      localChannels as Cc,
      -1,
      {
        channelsToSnapshot,
      }
    );
    const newChannels = emptyChannels(localChannels as Cc, newCheckpoint);

    _applyWrites(
      copyCheckpoint(newCheckpoint),
      newChannels,
      [task],
      undefined,
      undefined
    );
    values = readChannels({ ...channels, ...newChannels }, select);
  } else {
    values = readChannels(channels, select);
  }

  return values;
}

export function _localWrite(
  // eslint-disable-next-line @typescript-eslint/no-explicit-any
  commit: (writes: [string, any][]) => any,
  processes: Record<string, PregelNode>,
  // eslint-disable-next-line @typescript-eslint/no-explicit-any
  writes: [string, any][]
) {
  for (const [chan, value] of writes) {
    if ([PUSH, TASKS].includes(chan) && value != null) {
      if (!_isSend(value)) {
        throw new InvalidUpdateError(
          `Invalid packet type, expected SendProtocol, got ${JSON.stringify(
            value
          )}`
        );
      }
      if (!(value.node in processes)) {
        throw new InvalidUpdateError(
          `Invalid node name "${value.node}" in Send packet`
        );
      }
    }
  }
  commit(writes);
}

const IGNORE = new Set<string | number | symbol>([
  NO_WRITES,
  PUSH,
  RESUME,
  INTERRUPT,
  RETURN,
  ERROR,
  ERROR_SOURCE_NODE,
]);

const RESERVED_SET: ReadonlySet<string> = new Set(RESERVED);

export function _applyWrites<Cc extends Record<string, BaseChannel>>(
  checkpoint: Checkpoint,
  channels: Cc,
  tasks: WritesProtocol<keyof Cc>[],
  // eslint-disable-next-line @typescript-eslint/no-explicit-any
  getNextVersion: ((version: any) => any) | undefined,
  triggerToNodes: Record<string, string[]> | undefined
): Set<string> {
  // Sort tasks on the serialized full task path, for deterministic order of
  // update application. This is the same key every saver orders stored
  // writes by (`writesSortKey`), so live execution and checkpointer replay
  // cannot drift apart. Paths reaching here carry no task ids (call input
  // paths are cut to `slice(0, 3) + true/false` in `_prepareSingleTask` and
  // `acceptPush`), so this changes the live order only where the old
  // 3-element array-coercion comparator disagreed with Python's serialized
  // order (nested parent indices: it ordered `10` before `2`). Ties keep
  // insertion order: the sort is stable and the key carries the default
  // task-id/idx tie components, mirroring Python's `sorted`.
  // Serialize each path once (taskPathStr allocates), then sort on the
  // cached keys — the same precomputation the old comparator's pathCache did.
  const sortKeys = new Map<WritesProtocol<keyof Cc>, WritesSortKey>();
  for (const task of tasks) {
    sortKeys.set(task, writesSortKey(taskPathStr(task.path)));
  }
  tasks.sort((a, b) =>
    compareWritesSortKeys(sortKeys.get(a)!, sortKeys.get(b)!)
  );

  // Filter out non instances of BaseChannel
  const onlyChannels = getOnlyChannels(channels);

  // Single pass: update seen versions, check for triggers, collect channels to consume
  let bumpStep = false;
  const channelsToConsume = new Set<string>();
  for (const task of tasks) {
    if (task.triggers.length > 0) bumpStep = true;
    checkpoint.versions_seen[task.name] ??= {};
    for (const chan of task.triggers) {
      if (chan in checkpoint.channel_versions) {
        checkpoint.versions_seen[task.name][chan] =
          checkpoint.channel_versions[chan];
      }
      if (!RESERVED_SET.has(chan)) {
        channelsToConsume.add(chan);
      }
    }
  }

  // Find the highest version of all channels
  let maxVersion = maxChannelMapVersion(checkpoint.channel_versions);

  // Each version is minted once and shared by every channel it stamps, as in
  // Python: a saver that adds a random fraction would otherwise give channels
  // written together different versions, and updateState's "Ambiguous update"
  // check compares them.
  let consumedVersion: number | string | undefined;
  for (const chan of channelsToConsume) {
    if (chan in onlyChannels && onlyChannels[chan].consume()) {
      if (getNextVersion !== undefined) {
        consumedVersion ??= getNextVersion(maxVersion);
        checkpoint.channel_versions[chan] = consumedVersion!;
      }
    }
  }

  // Group writes by channel
  const pendingWritesByChannel = {} as Record<keyof Cc, PendingWriteValue[]>;
  for (const task of tasks) {
    for (const [chan, val] of task.writes) {
      if (IGNORE.has(chan)) {
        // do nothing
      } else if (chan in onlyChannels) {
        pendingWritesByChannel[chan] ??= [];
        pendingWritesByChannel[chan].push(val);
      }
    }
  }

  // Note: no DeltaChannel-specific reordering here. Concurrent writes to a
  // DeltaChannel apply in task-path order, exactly like every other channel
  // and exactly like the savers replay them (`writesSortKey`); the task-id
  // re-sort this replaces (langgraphjs#2544) made live order match a task-id
  // replay instead, which permuted parallel writers checkpoint by checkpoint.

  // Find the highest version of all channels
  if (maxVersion != null && consumedVersion !== undefined) {
    maxVersion = consumedVersion;
  }
  const nextVersion = getNextVersion?.(maxVersion);

  const updatedChannels: Set<string> = new Set();
  // Apply writes to channels
  for (const [chan, vals] of Object.entries(pendingWritesByChannel)) {
    if (chan in onlyChannels) {
      const channel = onlyChannels[chan];
      let updated;
      try {
        updated = channel.update(vals);
        // eslint-disable-next-line @typescript-eslint/no-explicit-any
      } catch (e: any) {
        if (e.name === InvalidUpdateError.unminifiable_name) {
          const wrappedError = new InvalidUpdateError(
            `Invalid update for channel "${chan}" with values ${JSON.stringify(
              vals
            )}: ${e.message}`
          );
          wrappedError.lc_error_code = e.lc_error_code;
          throw wrappedError;
        } else {
          throw e;
        }
      }
      if (updated && getNextVersion !== undefined) {
        checkpoint.channel_versions[chan] = nextVersion;

        // unavailable channels can't trigger tasks, so don't add them
        if (channel.isAvailable()) updatedChannels.add(chan);
      }
    }
  }

  // Channels that weren't updated in this step are notified of a new step
  if (bumpStep) {
    for (const chan in onlyChannels) {
      if (!Object.prototype.hasOwnProperty.call(onlyChannels, chan)) continue;

      const channel = onlyChannels[chan];
      if (channel.isAvailable() && !updatedChannels.has(chan)) {
        const updated = channel.update([]);

        if (updated && getNextVersion !== undefined) {
          checkpoint.channel_versions[chan] = nextVersion;

          // unavailable channels can't trigger tasks, so don't add them
          if (channel.isAvailable()) updatedChannels.add(chan);
        }
      }
    }
  }

  // If this is (tentatively) the last superstep, notify all channels of finish
  if (bumpStep && !triggersNextStep(updatedChannels, triggerToNodes)) {
    for (const chan in onlyChannels) {
      if (!Object.prototype.hasOwnProperty.call(onlyChannels, chan)) continue;

      const channel = onlyChannels[chan];
      if (channel.finish() && getNextVersion !== undefined) {
        checkpoint.channel_versions[chan] = nextVersion;

        // unavailable channels can't trigger tasks, so don't add them
        if (channel.isAvailable()) updatedChannels.add(chan);
      }
    }
  }

  return updatedChannels;
}

function* candidateNodes(
  checkpoint: ReadonlyCheckpoint,
  processes: StrRecord<string, PregelNode>,
  extra: NextTaskExtraFields
) {
  // This section is an optimization that allows which
  // nodes will be active during the next step.
  // When there's information about:
  // 1. Which channels were updated in the previous step
  // 2. Which nodes are triggered by which channels
  // Then we can determine which nodes should be triggered
  // in the next step without having to cycle through all nodes.
  if (extra.updatedChannels != null && extra.triggerToNodes != null) {
    const triggeredNodes = new Set<string>();

    // Get all nodes that have triggers associated with an updated channel
    for (const channel of extra.updatedChannels) {
      const nodeIds = extra.triggerToNodes[channel];
      for (const id of nodeIds ?? []) triggeredNodes.add(id);
    }

    // Sort the nodes to ensure deterministic order
    yield* [...triggeredNodes].sort();
    return;
  }

  // If there are no values in checkpoint, no need to run
  // through all the PULL candidates
  const isEmptyChannelVersions = (() => {
    for (const chan in checkpoint.channel_versions) {
      if (checkpoint.channel_versions[chan] !== null) return false;
    }
    return true;
  })();

  if (isEmptyChannelVersions) return;
  for (const name in processes) {
    if (!Object.prototype.hasOwnProperty.call(processes, name)) continue;
    yield name;
  }
}

/**
 * Pre-indexed pending writes for O(1) lookups, avoiding repeated
 * linear scans in _prepareSingleTask and _scratchpad.
 */
export type PendingWritesIndex = {
  nullResume: unknown | undefined;
  resumeByTaskId: Map<string, unknown[]>;
  successfulWriteTaskIds: Set<string>;
};

/**
 * Build an index over pendingWrites for O(1) lookups.
 *
 * @internal Exported for benchmarks and regression tests only.
 */
export function _indexPendingWrites(
  pendingWrites: [string, string, unknown][] | undefined
): PendingWritesIndex {
  let nullResume: unknown | undefined;
  const resumeByTaskId = new Map<string, unknown[]>();
  const successfulWriteTaskIds = new Set<string>();
  if (pendingWrites) {
    for (const [tid, chan, val] of pendingWrites) {
      if (tid === NULL_TASK_ID && chan === RESUME && nullResume === undefined) {
        nullResume = val;
      }
      if (chan === RESUME && tid !== NULL_TASK_ID) {
        let arr = resumeByTaskId.get(tid);
        if (!arr) {
          arr = [];
          resumeByTaskId.set(tid, arr);
        }
        arr.push(val);
      }
      if (chan !== ERROR) {
        successfulWriteTaskIds.add(tid);
      }
    }
  }
  return { nullResume, resumeByTaskId, successfulWriteTaskIds };
}

export type NextTaskExtraFields = {
  step: number;
  isResuming?: boolean;
  checkpointer?: BaseCheckpointSaver;
  manager?: CallbackManagerForChainRun;
  store?: BaseStore;
  stream?: IterableReadableWritableStream;
  triggerToNodes?: Record<string, string[]>;
  updatedChannels?: Set<string>;
  pendingWritesIndex?: PendingWritesIndex;
};

export type NextTaskExtraFieldsWithStore = NextTaskExtraFields & {
  store?: BaseStore;
};

export type NextTaskExtraFieldsWithoutStore = NextTaskExtraFields & {
  store?: never;
};

export function _prepareNextTasks<
  Nn extends StrRecord<string, PregelNode>,
  Cc extends StrRecord<string, BaseChannel>,
>(
  checkpoint: ReadonlyCheckpoint,
  pendingWrites: [string, string, unknown][] | undefined,
  processes: Nn,
  channels: Cc,
  config: RunnableConfig,
  forExecution: false,
  extra: NextTaskExtraFieldsWithoutStore
): Record<string, PregelTaskDescription>;

export function _prepareNextTasks<
  Nn extends StrRecord<string, PregelNode>,
  Cc extends StrRecord<string, BaseChannel>,
>(
  checkpoint: ReadonlyCheckpoint,
  pendingWrites: [string, string, unknown][] | undefined,
  processes: Nn,
  channels: Cc,
  config: RunnableConfig,
  forExecution: true,
  extra: NextTaskExtraFieldsWithStore
): Record<string, PregelExecutableTask<keyof Nn, keyof Cc>>;

/**
 * Prepare the set of tasks that will make up the next Pregel step.
 * This is the union of all PUSH tasks (Sends) and PULL tasks (nodes triggered
 * by edges).
 */
export function _prepareNextTasks<
  Nn extends StrRecord<string, PregelNode>,
  Cc extends StrRecord<string, BaseChannel>,
>(
  checkpoint: ReadonlyCheckpoint,
  pendingWrites: [string, string, unknown][] | undefined,
  processes: Nn,
  channels: Cc,
  config: RunnableConfig,
  forExecution: boolean,
  extra: NextTaskExtraFieldsWithStore | NextTaskExtraFieldsWithoutStore
):
  | Record<string, PregelTaskDescription>
  | Record<string, PregelExecutableTask<keyof Nn, keyof Cc>> {
  const tasks:
    | Record<string, PregelExecutableTask<keyof Nn, keyof Cc>>
    | Record<string, PregelTaskDescription> = {};

  // Pre-index pendingWrites once for O(1) lookups in _prepareSingleTask/_scratchpad
  const indexedExtra: typeof extra = extra.pendingWritesIndex
    ? extra
    : { ...extra, pendingWritesIndex: _indexPendingWrites(pendingWrites) };

  // Consume pending tasks
  const tasksChannel = channels[TASKS] as Topic<SendProtocol> | undefined;

  if (tasksChannel?.isAvailable()) {
    const len = tasksChannel.get().length;
    for (let i = 0; i < len; i += 1) {
      const task = _prepareSingleTask(
        [PUSH, i],
        checkpoint,
        pendingWrites,
        processes,
        channels,
        config,
        forExecution,
        indexedExtra
      );
      if (task !== undefined) {
        tasks[task.id] = task;
      }
    }
  }

  // Check if any processes should be run in next step
  // If so, prepare the values to be passed to them
  for (const name of candidateNodes(checkpoint, processes, indexedExtra)) {
    const task = _prepareSingleTask(
      [PULL, name],
      checkpoint,
      pendingWrites,
      processes,
      channels,
      config,
      forExecution,
      indexedExtra
    );
    if (task !== undefined) {
      tasks[task.id] = task;
    }
  }
  return tasks;
}

export function _prepareSingleTask<
  Nn extends StrRecord<string, PregelNode>,
  Cc extends StrRecord<string, BaseChannel>,
>(
  taskPath: SimpleTaskPath,
  checkpoint: ReadonlyCheckpoint,
  pendingWrites: CheckpointPendingWrite[] | undefined,
  processes: Nn,
  channels: Cc,
  config: RunnableConfig,
  forExecution: false,
  extra: NextTaskExtraFields
): PregelTaskDescription | undefined;

export function _prepareSingleTask<
  Nn extends StrRecord<string, PregelNode>,
  Cc extends StrRecord<string, BaseChannel>,
>(
  taskPath: TaskPath,
  checkpoint: ReadonlyCheckpoint,
  pendingWrites: CheckpointPendingWrite[] | undefined,
  processes: Nn,
  channels: Cc,
  config: RunnableConfig,
  forExecution: true,
  extra: NextTaskExtraFields
): PregelExecutableTask<keyof Nn, keyof Cc> | undefined;

export function _prepareSingleTask<
  Nn extends StrRecord<string, PregelNode>,
  Cc extends StrRecord<string, BaseChannel>,
>(
  taskPath: TaskPath,
  checkpoint: ReadonlyCheckpoint,
  pendingWrites: CheckpointPendingWrite[] | undefined,
  processes: Nn,
  channels: Cc,
  config: RunnableConfig,
  forExecution: boolean,
  extra: NextTaskExtraFieldsWithStore
): PregelTaskDescription | PregelExecutableTask<keyof Nn, keyof Cc> | undefined;

/**
 * Prepares a single task for the next Pregel step, given a task path, which
 * uniquely identifies a PUSH or PULL task within the graph.
 */
export function _prepareSingleTask<
  Nn extends StrRecord<string, PregelNode>,
  Cc extends StrRecord<string, BaseChannel>,
>(
  taskPath: TaskPath,
  checkpoint: ReadonlyCheckpoint,
  pendingWrites: CheckpointPendingWrite[] | undefined,
  processes: Nn,
  channels: Cc,
  config: LangGraphRunnableConfig,
  forExecution: boolean,
  extra: NextTaskExtraFields
):
  | PregelTaskDescription
  | PregelExecutableTask<keyof Nn, keyof Cc>
  | undefined {
  const { step, checkpointer, manager } = extra;
  const configurable = config.configurable ?? {};
  const parentNamespace = configurable.checkpoint_ns ?? "";

  if (taskPath[0] === PUSH && isCall(taskPath[taskPath.length - 1])) {
    const call = taskPath[taskPath.length - 1] as Call;
    const proc = getRunnableForFunc(call.name, call.func);
    const triggers = [PUSH];
    const checkpointNamespace =
      parentNamespace === ""
        ? call.name
        : `${parentNamespace}${CHECKPOINT_NAMESPACE_SEPARATOR}${call.name}`;
    const id = uuid5(
      JSON.stringify([
        checkpointNamespace,
        step.toString(),
        call.name,
        PUSH,
        taskPath[1],
        taskPath[2],
      ]),
      checkpoint.id
    );
    const taskCheckpointNamespace = `${checkpointNamespace}${CHECKPOINT_NAMESPACE_END}${id}`;

    // we append `true` to the task path to indicate that a call is being made
    // so we should not return interrupts from this task (responsibility lies with the parent)
    const outputTaskPath = [...taskPath.slice(0, 3), true] as VariadicTaskPath;
    const metadata = {
      langgraph_step: step,
      langgraph_node: call.name,
      langgraph_triggers: triggers,
      langgraph_path: outputTaskPath,
      langgraph_checkpoint_ns: taskCheckpointNamespace,
      checkpoint_ns: taskCheckpointNamespace,
    };
    if (forExecution) {
      const writes: [keyof Cc, unknown][] = [];
      const executionInfo: ExecutionInfo = {
        checkpointId: checkpoint.id,
        checkpointNs: taskCheckpointNamespace,
        taskId: id,
        threadId: configurable.thread_id as string | undefined,
        runId: config.runId != null ? String(config.runId) : undefined,
        nodeAttempt: 1,
      };
      const task = {
        name: call.name,
        input: call.input,
        proc,
        writes,
        config: {
          ...patchConfig(
            mergeConfigs(config, {
              metadata,
              store: extra.store ?? config.store,
            }),
            {
              runName: call.name,
              callbacks: manager?.getChild(`graph:step:${step}`),
              configurable: {
                [CONFIG_KEY_TASK_ID]: id,
                [CONFIG_KEY_SEND]: (writes_: PendingWrite[]) =>
                  _localWrite(
                    (items: PendingWrite<keyof Cc>[]) => writes.push(...items),
                    processes,
                    writes_
                  ),
                [CONFIG_KEY_READ]: (
                  select_: Array<keyof Cc> | keyof Cc,
                  fresh_: boolean = false
                ) =>
                  _localRead(
                    checkpoint,
                    channels,
                    {
                      name: call.name,
                      writes: writes as PendingWrite[],
                      triggers,
                      path: outputTaskPath,
                    },
                    select_,
                    fresh_
                  ),
                [CONFIG_KEY_CHECKPOINTER]:
                  checkpointer ?? configurable[CONFIG_KEY_CHECKPOINTER],
                [CONFIG_KEY_CHECKPOINT_MAP]: {
                  ...configurable[CONFIG_KEY_CHECKPOINT_MAP],
                  [parentNamespace]: checkpoint.id,
                },
                [CONFIG_KEY_SCRATCHPAD]: _scratchpad({
                  pendingWrites: pendingWrites ?? [],
                  taskId: id,
                  currentTaskInput: call.input,
                  resumeMap: config.configurable?.[CONFIG_KEY_RESUME_MAP],
                  taskCheckpointNamespace,
                  pendingWritesIndex: extra.pendingWritesIndex,
                }),
                [CONFIG_KEY_PREVIOUS_STATE]:
                  checkpoint.channel_values[PREVIOUS],
                checkpoint_id: undefined,
                checkpoint_ns: taskCheckpointNamespace,
              },
            }
          ),
          executionInfo,
        },
        triggers,
        retry_policy: call.retry,
        cache_key: call.cache
          ? {
              key: XXH3((call.cache.keyFunc ?? JSON.stringify)([call.input])),
              ns: [CACHE_NS_WRITES, call.name ?? "__dynamic__"],
              ttl: call.cache.ttl,
            }
          : undefined,
        id,
        path: outputTaskPath,
        writers: [],
        timeout: call.timeout,
      } satisfies PregelExecutableTask<keyof Nn, keyof Cc>;
      return task;
    } else {
      return {
        id,
        name: call.name,
        interrupts: [],
        path: outputTaskPath,
      };
    }
  } else if (taskPath[0] === PUSH) {
    const index =
      typeof taskPath[1] === "number"
        ? taskPath[1]
        : parseInt(taskPath[1] as string, 10);

    if (!channels[TASKS]?.isAvailable()) {
      return undefined;
    }

    const sends = channels[TASKS].get() as SendProtocol[];
    if (index < 0 || index >= sends.length) {
      return undefined;
    }

    const packet =
      _isSendInterface(sends[index]) && !_isSend(sends[index])
        ? new Send(
            sends[index].node,
            sends[index].args,
            sends[index].timeout !== undefined
              ? { timeout: sends[index].timeout }
              : undefined
          )
        : sends[index];

    if (!_isSendInterface(packet)) {
      console.warn(
        `Ignoring invalid packet ${JSON.stringify(packet)} in pending sends.`
      );
      return undefined;
    }
    if (!(packet.node in processes)) {
      console.warn(
        `Ignoring unknown node name ${packet.node} in pending sends.`
      );
      return undefined;
    }
    const triggers = [PUSH];
    const checkpointNamespace =
      parentNamespace === ""
        ? packet.node
        : `${parentNamespace}${CHECKPOINT_NAMESPACE_SEPARATOR}${packet.node}`;
    const taskId = uuid5(
      JSON.stringify([
        checkpointNamespace,
        step.toString(),
        packet.node,
        PUSH,
        index.toString(),
      ]),
      checkpoint.id
    );
    const taskCheckpointNamespace = `${checkpointNamespace}${CHECKPOINT_NAMESPACE_END}${taskId}`;
    // We append `false` to the task path to indicate that a call is not being
    // made, so we should return interrupts from this task — matching Python's
    // `(PUSH, idx, False)`. One translated path serves the executable task,
    // the description, the metadata, and the local-read view.
    const translatedTaskPath = [
      ...taskPath.slice(0, 3),
      false,
    ] as VariadicTaskPath;
    let metadata = {
      langgraph_step: step,
      langgraph_node: packet.node,
      langgraph_triggers: triggers,
      langgraph_path: translatedTaskPath,
      langgraph_checkpoint_ns: taskCheckpointNamespace,
      checkpoint_ns: taskCheckpointNamespace,
    };
    if (forExecution) {
      const proc = processes[packet.node];
      const node = proc.getNode();
      if (node !== undefined) {
        if (proc.metadata !== undefined) {
          metadata = { ...metadata, ...proc.metadata };
        }
        const writes: [keyof Cc, unknown][] = [];
        const executionInfo: ExecutionInfo = {
          checkpointId: checkpoint.id,
          checkpointNs: taskCheckpointNamespace,
          taskId,
          threadId: configurable.thread_id as string | undefined,
          runId: config.runId != null ? String(config.runId) : undefined,
          nodeAttempt: 1,
        };
        return {
          name: packet.node,
          input: packet.args,
          proc: node,
          subgraphs: proc.subgraphs,
          writes,
          config: {
            ...patchConfig(
              mergeConfigs(config, {
                metadata,
                tags: proc.tags,
                store: extra.store ?? config.store,
              }),
              {
                runName: packet.node,
                callbacks: manager?.getChild(`graph:step:${step}`),
                configurable: {
                  [CONFIG_KEY_TASK_ID]: taskId,
                  [CONFIG_KEY_SEND]: (writes_: PendingWrite[]) =>
                    _localWrite(
                      (items: PendingWrite<keyof Cc>[]) =>
                        writes.push(...items),
                      processes,
                      writes_
                    ),
                  [CONFIG_KEY_READ]: (
                    select_: Array<keyof Cc> | keyof Cc,
                    fresh_: boolean = false
                  ) =>
                    _localRead(
                      checkpoint,
                      channels,
                      {
                        name: packet.node,
                        writes: writes as PendingWrite[],
                        triggers,
                        path: translatedTaskPath,
                      },
                      select_,
                      fresh_
                    ),
                  [CONFIG_KEY_CHECKPOINTER]:
                    checkpointer ?? configurable[CONFIG_KEY_CHECKPOINTER],
                  [CONFIG_KEY_CHECKPOINT_MAP]: {
                    ...configurable[CONFIG_KEY_CHECKPOINT_MAP],
                    [parentNamespace]: checkpoint.id,
                  },
                  [CONFIG_KEY_SCRATCHPAD]: _scratchpad({
                    pendingWrites: pendingWrites ?? [],
                    taskId,
                    currentTaskInput: packet.args,
                    resumeMap: config.configurable?.[CONFIG_KEY_RESUME_MAP],
                    taskCheckpointNamespace,
                    pendingWritesIndex: extra.pendingWritesIndex,
                  }),
                  [CONFIG_KEY_PREVIOUS_STATE]:
                    checkpoint.channel_values[PREVIOUS],
                  checkpoint_id: undefined,
                  checkpoint_ns: taskCheckpointNamespace,
                },
              }
            ),
            executionInfo,
          },
          triggers,
          retry_policy: proc.retryPolicy,
          cache_key: proc.cachePolicy
            ? {
                key: XXH3(
                  (proc.cachePolicy.keyFunc ?? JSON.stringify)([packet.args])
                ),
                ns: [CACHE_NS_WRITES, proc.name ?? "__dynamic__", packet.node],
                ttl: proc.cachePolicy.ttl,
              }
            : undefined,
          id: taskId,
          path: translatedTaskPath,
          writers: proc.getWriters(),
          // a per-Send timeout overrides the target node's configured timeout
          timeout: packet.timeout ?? proc.timeout,
        } satisfies PregelExecutableTask<keyof Nn, keyof Cc>;
      }
    } else {
      return {
        id: taskId,
        name: packet.node,
        interrupts: [],
        path: translatedTaskPath,
      } satisfies PregelTaskDescription;
    }
  } else if (taskPath[0] === PULL) {
    const name = taskPath[1].toString();
    const proc = processes[name];
    if (proc === undefined) {
      return undefined;
    }

    // Check if this task already has successful writes in the pending writes
    if (pendingWrites?.length) {
      // Find the task ID for this node/path
      const checkpointNamespace =
        parentNamespace === ""
          ? name
          : `${parentNamespace}${CHECKPOINT_NAMESPACE_SEPARATOR}${name}`;

      const taskId = uuid5(
        JSON.stringify([
          checkpointNamespace,
          step.toString(),
          name,
          PULL,
          name,
        ]),
        checkpoint.id
      );

      const hasSuccessfulWrites = extra.pendingWritesIndex
        ? extra.pendingWritesIndex.successfulWriteTaskIds.has(taskId)
        : pendingWrites.some((w) => w[0] === taskId && w[1] !== ERROR);

      if (hasSuccessfulWrites) {
        return undefined;
      }
    }

    const nullVersion = getNullChannelVersion(checkpoint.channel_versions);
    if (nullVersion === undefined) {
      return undefined;
    }
    const seen = checkpoint.versions_seen[name] ?? {};

    // Find the first trigger that is available and has a new version
    const trigger = proc.triggers.find((chan) => {
      if (!channels[chan].isAvailable()) return false;

      return (
        (checkpoint.channel_versions[chan] ?? nullVersion) >
        (seen[chan] ?? nullVersion)
      );
    });

    // If any of the channels read by this process were updated
    if (trigger !== undefined) {
      const val = _procInput(proc, channels, forExecution);
      if (val === undefined) {
        return undefined;
      }
      const checkpointNamespace =
        parentNamespace === ""
          ? name
          : `${parentNamespace}${CHECKPOINT_NAMESPACE_SEPARATOR}${name}`;
      const taskId = uuid5(
        JSON.stringify([
          checkpointNamespace,
          step.toString(),
          name,
          PULL,
          [trigger],
        ]),
        checkpoint.id
      );
      const taskCheckpointNamespace = `${checkpointNamespace}${CHECKPOINT_NAMESPACE_END}${taskId}`;
      let metadata = {
        langgraph_step: step,
        langgraph_node: name,
        langgraph_triggers: [trigger],
        langgraph_path: taskPath,
        langgraph_checkpoint_ns: taskCheckpointNamespace,
        checkpoint_ns: taskCheckpointNamespace,
      };
      if (forExecution) {
        const node = proc.getNode();
        if (node !== undefined) {
          if (proc.metadata !== undefined) {
            metadata = { ...metadata, ...proc.metadata };
          }
          const writes: [keyof Cc, unknown][] = [];
          const executionInfo: ExecutionInfo = {
            checkpointId: checkpoint.id,
            checkpointNs: taskCheckpointNamespace,
            taskId,
            threadId: configurable.thread_id as string | undefined,
            runId: config.runId != null ? String(config.runId) : undefined,
            nodeAttempt: 1,
          };
          return {
            name,
            input: val,
            proc: node,
            subgraphs: proc.subgraphs,
            writes,
            config: {
              ...patchConfig(
                mergeConfigs(config, {
                  metadata,
                  tags: proc.tags,
                  store: extra.store ?? config.store,
                }),
                {
                  runName: name,
                  callbacks: manager?.getChild(`graph:step:${step}`),
                  configurable: {
                    [CONFIG_KEY_TASK_ID]: taskId,
                    [CONFIG_KEY_SEND]: (writes_: PendingWrite[]) =>
                      _localWrite(
                        (items: PendingWrite<keyof Cc>[]) => {
                          writes.push(...items);
                        },
                        processes,
                        writes_
                      ),
                    [CONFIG_KEY_READ]: (
                      select_: Array<keyof Cc> | keyof Cc,
                      fresh_: boolean = false
                    ) =>
                      _localRead(
                        checkpoint,
                        channels,
                        {
                          name,
                          writes: writes as PendingWrite[],
                          triggers: [trigger],
                          path: taskPath,
                        },
                        select_,
                        fresh_
                      ),
                    [CONFIG_KEY_CHECKPOINTER]:
                      checkpointer ?? configurable[CONFIG_KEY_CHECKPOINTER],
                    [CONFIG_KEY_CHECKPOINT_MAP]: {
                      ...configurable[CONFIG_KEY_CHECKPOINT_MAP],
                      [parentNamespace]: checkpoint.id,
                    },
                    [CONFIG_KEY_SCRATCHPAD]: _scratchpad({
                      pendingWrites: pendingWrites ?? [],
                      taskId,
                      currentTaskInput: val,
                      resumeMap: config.configurable?.[CONFIG_KEY_RESUME_MAP],
                      taskCheckpointNamespace,
                      pendingWritesIndex: extra.pendingWritesIndex,
                    }),
                    [CONFIG_KEY_PREVIOUS_STATE]:
                      checkpoint.channel_values[PREVIOUS],
                    checkpoint_id: undefined,
                    checkpoint_ns: taskCheckpointNamespace,
                  },
                }
              ),
              executionInfo,
            },
            triggers: [trigger],
            retry_policy: proc.retryPolicy,
            cache_key: proc.cachePolicy
              ? {
                  key: XXH3(
                    (proc.cachePolicy.keyFunc ?? JSON.stringify)([val])
                  ),
                  ns: [CACHE_NS_WRITES, proc.name ?? "__dynamic__", name],
                  ttl: proc.cachePolicy.ttl,
                }
              : undefined,
            id: taskId,
            path: taskPath,
            writers: proc.getWriters(),
            timeout: proc.timeout,
          } satisfies PregelExecutableTask<keyof Nn, keyof Cc>;
        }
      } else {
        return {
          id: taskId,
          name,
          interrupts: [],
          path: taskPath,
        } satisfies PregelTaskDescription;
      }
    }
  }
  return undefined;
}

/**
 * Prepare an immediate node-level error handler task for a failed task.
 *
 * The handler runs only after the failed node's retry policy is exhausted (the
 * runner schedules it once a non-bubble-up error settles). It is prepared like
 * a PUSH task targeting the auto-generated handler node, receives the failed
 * node's input, and is injected with a {@link NodeError} under
 * {@link CONFIG_KEY_NODE_ERROR} so the handler can inspect the failure
 * provenance (and route via `Command({ goto })`).
 *
 * @internal
 */
export function _prepareNodeErrorHandlerTask<
  Nn extends StrRecord<string, PregelNode>,
  Cc extends StrRecord<string, BaseChannel>,
>(
  failedTask: PregelExecutableTask<keyof Nn, keyof Cc>,
  handlerNodeName: string,
  error: Error,
  checkpoint: ReadonlyCheckpoint,
  pendingWrites: CheckpointPendingWrite[] | undefined,
  processes: Nn,
  channels: Cc,
  config: LangGraphRunnableConfig,
  extra: NextTaskExtraFieldsWithStore
): PregelExecutableTask<keyof Nn, keyof Cc> | undefined {
  const { step, checkpointer, manager } = extra;
  const proc = processes[handlerNodeName as keyof Nn];
  if (proc === undefined) {
    return undefined;
  }
  const node = proc.getNode();
  if (node === undefined) {
    return undefined;
  }

  const configurable = config.configurable ?? {};
  const parentNamespace = configurable.checkpoint_ns ?? "";
  const triggers = [PUSH];
  const checkpointNamespace =
    parentNamespace === ""
      ? handlerNodeName
      : `${parentNamespace}${CHECKPOINT_NAMESPACE_SEPARATOR}${handlerNodeName}`;
  // Deterministic task id (includes the failed task id) so resuming from a
  // checkpoint reconstructs the same handler task.
  const taskId = uuid5(
    JSON.stringify([
      checkpointNamespace,
      step.toString(),
      handlerNodeName,
      PUSH,
      "node_error_handler",
      failedTask.id,
    ]),
    checkpoint.id
  );
  const taskCheckpointNamespace = `${checkpointNamespace}${CHECKPOINT_NAMESPACE_END}${taskId}`;
  // The path is the failed task's own path plus "node_error_handler" and a
  // trailing `false` (not `true`, so interrupts raised by the
  // handler are surfaced normally rather than deferred to a parent call) —
  // matching Python's `(*failed_task.path[:3], "node_error_handler", False)`,
  // so the two runtimes order a handler's writes the same way relative to its
  // siblings.
  const failedPath = failedTask.path?.slice(0, 3) ?? [];
  const taskPath = [
    failedPath[0] ?? PUSH,
    ...failedPath.slice(1),
    "node_error_handler",
    false,
  ] as VariadicTaskPath;

  let metadata = {
    langgraph_step: step,
    langgraph_node: handlerNodeName,
    langgraph_triggers: triggers,
    langgraph_path: taskPath,
    langgraph_checkpoint_ns: taskCheckpointNamespace,
    checkpoint_ns: taskCheckpointNamespace,
  };
  if (proc.metadata !== undefined) {
    metadata = { ...metadata, ...proc.metadata };
  }

  const writes: [keyof Cc, unknown][] = [];
  const executionInfo: ExecutionInfo = {
    checkpointId: checkpoint.id,
    checkpointNs: taskCheckpointNamespace,
    taskId,
    threadId: configurable.thread_id as string | undefined,
    runId: config.runId != null ? String(config.runId) : undefined,
    nodeAttempt: 1,
  };

  return {
    name: handlerNodeName as keyof Nn,
    input: failedTask.input,
    proc: node,
    subgraphs: proc.subgraphs,
    writes,
    config: {
      ...patchConfig(
        mergeConfigs(config, {
          metadata,
          tags: proc.tags,
          store: extra.store ?? config.store,
        }),
        {
          runName: handlerNodeName,
          callbacks: manager?.getChild(`graph:step:${step}`),
          configurable: {
            [CONFIG_KEY_TASK_ID]: taskId,
            [CONFIG_KEY_SEND]: (writes_: PendingWrite[]) =>
              _localWrite(
                (items: PendingWrite<keyof Cc>[]) => writes.push(...items),
                processes,
                writes_
              ),
            [CONFIG_KEY_READ]: (
              select_: Array<keyof Cc> | keyof Cc,
              fresh_: boolean = false
            ) =>
              _localRead(
                checkpoint,
                channels,
                {
                  name: handlerNodeName,
                  writes: writes as PendingWrite[],
                  triggers,
                  path: taskPath,
                },
                select_,
                fresh_
              ),
            [CONFIG_KEY_CHECKPOINTER]:
              checkpointer ?? configurable[CONFIG_KEY_CHECKPOINTER],
            [CONFIG_KEY_CHECKPOINT_MAP]: {
              ...configurable[CONFIG_KEY_CHECKPOINT_MAP],
              [parentNamespace]: checkpoint.id,
            },
            [CONFIG_KEY_SCRATCHPAD]: _scratchpad({
              pendingWrites: pendingWrites ?? [],
              taskId,
              currentTaskInput: failedTask.input,
              resumeMap: config.configurable?.[CONFIG_KEY_RESUME_MAP],
              taskCheckpointNamespace,
            }),
            [CONFIG_KEY_PREVIOUS_STATE]: checkpoint.channel_values[PREVIOUS],
            [CONFIG_KEY_NODE_ERROR]: new NodeError(
              String(failedTask.name),
              error
            ),
            checkpoint_id: undefined,
            checkpoint_ns: taskCheckpointNamespace,
          },
        }
      ),
      executionInfo,
    },
    triggers,
    retry_policy: proc.retryPolicy,
    cache_key: undefined,
    id: taskId,
    path: taskPath,
    writers: proc.getWriters(),
  } satisfies PregelExecutableTask<keyof Nn, keyof Cc>;
}

/**
 *  Function injected under CONFIG_KEY_READ in task config, to read current state.
 *  Used by conditional edges to read a copy of the state with reflecting the writes
 *  from that node only.
 *
 * @internal
 */
function _procInput(
  proc: PregelNode,
  channels: StrRecord<string, BaseChannel>,
  forExecution: boolean
) {
  // eslint-disable-next-line @typescript-eslint/no-explicit-any
  let val: any;

  if (typeof proc.channels === "object" && !Array.isArray(proc.channels)) {
    val = {};
    for (const [k, chan] of Object.entries(proc.channels)) {
      if (proc.triggers.includes(chan)) {
        try {
          val[k] = readChannel(channels, chan, false);
          // eslint-disable-next-line @typescript-eslint/no-explicit-any
        } catch (e: any) {
          if (e.name === EmptyChannelError.unminifiable_name) {
            return undefined;
          } else {
            throw e;
          }
        }
      } else if (chan in channels) {
        try {
          val[k] = readChannel(channels, chan, false);
          // eslint-disable-next-line @typescript-eslint/no-explicit-any
        } catch (e: any) {
          if (e.name === EmptyChannelError.unminifiable_name) {
            continue;
          } else {
            throw e;
          }
        }
      }
    }
  } else if (Array.isArray(proc.channels)) {
    let successfulRead = false;
    for (const chan of proc.channels) {
      try {
        val = readChannel(channels, chan, false);
        successfulRead = true;
        break;
        // eslint-disable-next-line @typescript-eslint/no-explicit-any
      } catch (e: any) {
        if (e.name === EmptyChannelError.unminifiable_name) {
          continue;
        } else {
          throw e;
        }
      }
    }
    if (!successfulRead) {
      return undefined;
    }
  } else {
    throw new Error(
      `Invalid channels type, expected list or dict, got ${proc.channels}`
    );
  }

  // If the process has a mapper, apply it to the value
  if (forExecution && proc.mapper !== undefined) {
    val = proc.mapper(val);
  }

  return val;
}

/**
 * Remove any values belonging to UntrackedValue channels from a Send packet
 * before checkpointing.
 *
 * Send is often called with state to be passed to the destination node,
 * which may contain UntrackedValues at the top level.
 *
 * @internal
 */
export function sanitizeUntrackedValuesInSend(
  packet: Send,
  channels: StrRecord<string, BaseChannel>
): Send {
  if (typeof packet.args !== "object" || packet.args === null) {
    // Not a dict-like arg
    return packet;
  }

  // Top-level keys should be channel names
  const sanitizedArg: Record<string, unknown> = {};
  for (const [key, value] of Object.entries(packet.args)) {
    const channel = channels[key];
    // Check if channel is an UntrackedValue by its lc_graph_name
    if (!channel || channel.lc_graph_name !== "UntrackedValue") {
      sanitizedArg[key] = value;
    }
  }

  return new Send(packet.node, sanitizedArg);
}

function _scratchpad({
  pendingWrites,
  taskId,
  currentTaskInput,
  resumeMap,
  taskCheckpointNamespace,
  pendingWritesIndex,
}: {
  pendingWrites: CheckpointPendingWrite[];
  taskId: string;
  currentTaskInput: unknown;
  resumeMap: Record<string, unknown> | undefined;
  taskCheckpointNamespace: string;
  pendingWritesIndex?: PendingWritesIndex;
}): PregelScratchpad {
  const nullResume = pendingWritesIndex
    ? pendingWritesIndex.nullResume
    : pendingWrites.find(
        ([writeTaskId, chan]) => writeTaskId === NULL_TASK_ID && chan === RESUME
      )?.[2];

  const resume = (() => {
    // flatMap flattens array resume values one level; mirror that with .flat()
    const result: unknown[] = pendingWritesIndex
      ? (pendingWritesIndex.resumeByTaskId.get(taskId) ?? []).flat()
      : pendingWrites
          .filter(
            ([writeTaskId, chan]) => writeTaskId === taskId && chan === RESUME
          )
          .flatMap(([_writeTaskId, _chan, resume]) => resume);

    if (resumeMap != null) {
      const nextId = interruptId(taskCheckpointNamespace, result.length);
      const firstId = interruptId(taskCheckpointNamespace, 0);
      if (nextId in resumeMap) {
        result.push(resumeMap[nextId]);
      } else if (firstId in resumeMap) {
        // Older versions gave all of a task's interrupts the first id.
        result.push(resumeMap[firstId]);
      }
    }

    return result;
  })();

  const scratchpad = {
    callCounter: 0,
    interruptCounter: -1,
    resume,
    nullResume,
    subgraphCounter: 0,
    currentTaskInput,
    consumeNullResume: () => {
      if (scratchpad.nullResume != null) {
        delete scratchpad.nullResume;
        pendingWrites.splice(
          pendingWrites.findIndex(
            ([writeTaskId, chan]) =>
              writeTaskId === NULL_TASK_ID && chan === RESUME
          ),
          1
        );
        return nullResume;
      }

      return undefined;
    },
  };
  return scratchpad;
}