feat(langgraph): deferred nodes

libs/langgraph/src/channels/any_value.ts

@@ -52,4 +52,8 @@ export class AnyValue<Value> extends BaseChannel<Value, Value, Value> {
     }
     return this.value[0];
   }
+
+  isAvailable(): boolean {
+    return this.value.length !== 0;
+  }
 }

libs/langgraph/src/channels/base.ts

@@ -70,12 +70,42 @@ export abstract class BaseChannel<
 
   /**
    * Mark the current value of the channel as consumed. By default, no-op.
-   * This is called by Pregel before the start of the next step, for all
-   * channels that triggered a node. If the channel was updated, return true.
+   * A channel can use this method to modify its state, preventing the value
+   * from being consumed again.
+   *
+   * Returns True if the channel was updated, False otherwise.
    */
   consume(): boolean {
     return false;
   }
+
+  /**
+   * Notify the channel that the Pregel run is finishing. By default, no-op.
+   * A channel can use this method to modify its state, preventing finish.
+   *
+   * Returns True if the channel was updated, False otherwise.
+   */
+  finish(): boolean {
+    return false;
+  }
+
+  /**
+   * Return True if the channel is available (not empty), False otherwise.
+   * Subclasses should override this method to provide a more efficient
+   * implementation than calling get() and catching EmptyChannelError.
+   */
+  isAvailable(): boolean {
+    try {
+      this.get();
+      return true;
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    } catch (error: any) {
+      if (error.name === EmptyChannelError.unminifiable_name) {
+        return false;
+      }
+      throw error;
+    }
+  }
 }
 
 export function emptyChannels<Cc extends Record<string, BaseChannel>>(

libs/langgraph/src/channels/binop.ts

@@ -73,4 +73,8 @@ export class BinaryOperatorAggregate<
     }
     return this.value;
   }
+
+  isAvailable(): boolean {
+    return this.value !== undefined;
+  }
 }

libs/langgraph/src/channels/dynamic_barrier_value.ts

@@ -66,15 +66,9 @@ export class DynamicBarrierValue<Value> extends BaseChannel<
             "Assertion Error: Received unexpected WaitForNames instance."
           );
         }
-        if (this.names.has(value)) {
-          if (!this.seen.has(value)) {
-            this.seen.add(value);
-            updated = true;
-          }
-        } else {
-          throw new InvalidUpdateError(
-            `Value ${value} not in ${[...this.names]}`
-          );
+        if (this.names.has(value) && !this.seen.has(value)) {
+          this.seen.add(value);
+          updated = true;
         }
       }
       return updated;
@@ -103,4 +97,120 @@ export class DynamicBarrierValue<Value> extends BaseChannel<
   checkpoint(): [Value[] | undefined, Value[]] {
     return [this.names ? [...this.names] : undefined, [...this.seen]];
   }
+
+  isAvailable(): boolean {
+    return !!this.names && areSetsEqual(this.names, this.seen);
+  }
+}
+
+/**
+ * A channel that switches between two states with an additional finished flag
+ *
+ * - in the "priming" state it can't be read from.
+ *     - if it receives a WaitForNames update, it switches to the "waiting" state.
+ * - in the "waiting" state it collects named values until all are received.
+ *     - once all named values are received, and the finished flag is set, it can be read once, and it switches
+ *       back to the "priming" state.
+ * @internal
+ */
+export class DynamicBarrierValueAfterFinish<Value> extends BaseChannel<
+  void,
+  Value | WaitForNames<Value>,
+  [Value[] | undefined, Value[], boolean]
+> {
+  lc_graph_name = "DynamicBarrierValueAfterFinish";
+
+  names?: Set<Value>; // Names of nodes that we want to wait for.
+
+  seen: Set<Value>;
+
+  finished: boolean;
+
+  constructor() {
+    super();
+    this.names = undefined;
+    this.seen = new Set<Value>();
+    this.finished = false;
+  }
+
+  fromCheckpoint(checkpoint?: [Value[] | undefined, Value[], boolean]) {
+    const empty = new DynamicBarrierValueAfterFinish<Value>();
+    if (typeof checkpoint !== "undefined") {
+      const [names, seen, finished] = checkpoint;
+      empty.names = names ? new Set(names) : undefined;
+      empty.seen = new Set(seen);
+      empty.finished = finished;
+    }
+    return empty as this;
+  }
+
+  update(values: (Value | WaitForNames<Value>)[]): boolean {
+    const waitForNames = values.filter(isWaitForNames);
+    if (waitForNames.length > 0) {
+      if (waitForNames.length > 1) {
+        throw new InvalidUpdateError(
+          "Received multiple WaitForNames updates in the same step."
+        );
+      }
+      this.names = new Set(waitForNames[0].__names);
+      return true;
+    } else if (this.names !== undefined) {
+      let updated = false;
+      for (const value of values) {
+        if (isWaitForNames(value)) {
+          throw new Error(
+            "Assertion Error: Received unexpected WaitForNames instance."
+          );
+        }
+        if (this.names.has(value) && !this.seen.has(value)) {
+          this.seen.add(value);
+          updated = true;
+        }
+      }
+      return updated;
+    }
+    return false;
+  }
+
+  consume(): boolean {
+    if (
+      this.finished &&
+      this.seen &&
+      this.names &&
+      areSetsEqual(this.seen, this.names)
+    ) {
+      this.seen = new Set<Value>();
+      this.names = undefined;
+      this.finished = false;
+      return true;
+    }
+    return false;
+  }
+
+  finish(): boolean {
+    if (!this.finished && this.names && areSetsEqual(this.names, this.seen)) {
+      this.finished = true;
+      return true;
+    }
+    return false;
+  }
+
+  get(): void {
+    if (!this.finished || !this.names || !areSetsEqual(this.names, this.seen)) {
+      throw new EmptyChannelError();
+    }
+    return undefined;
+  }
+
+  checkpoint(): [Value[] | undefined, Value[], boolean] {
+    return [
+      this.names ? [...this.names] : undefined,
+      [...this.seen],
+      this.finished,
+    ];
+  }
+
+  isAvailable(): boolean {
+    return this.finished && !!this.names && areSetsEqual(this.names, this.seen);
+  }
 }

libs/langgraph/src/channels/ephemeral_value.ts

@@ -57,4 +57,8 @@ export class EphemeralValue<Value> extends BaseChannel<Value, Value, Value> {
     }
     return this.value[0];
   }
+
+  isAvailable(): boolean {
+    return this.value.length !== 0;
+  }
 }

libs/langgraph/src/channels/last_value.ts

@@ -31,9 +31,7 @@ export class LastValue<Value> extends BaseChannel<Value, Value, Value> {
     if (values.length !== 1) {
       throw new InvalidUpdateError(
         "LastValue can only receive one value per step.",
-        {
-          lc_error_code: "INVALID_CONCURRENT_GRAPH_UPDATE",
-        }
+        { lc_error_code: "INVALID_CONCURRENT_GRAPH_UPDATE" }
       );
     }
 
@@ -55,4 +53,80 @@ export class LastValue<Value> extends BaseChannel<Value, Value, Value> {
     }
     return this.value[0];
   }
+
+  isAvailable(): boolean {
+    return this.value.length !== 0;
+  }
+}
+
+/**
+ * Stores the last value received, but only made available after finish().
+ * Once made available, clears the value.
+ * @internal
+ */
+export class LastValueAfterFinish<Value> extends BaseChannel<
+  Value,
+  Value,
+  [Value, boolean]
+> {
+  lc_graph_name = "LastValueAfterFinish";
+
+  // value is an array so we don't misinterpret an update to undefined as no write
+  value: [Value] | [] = [];
+
+  finished: boolean = false;
+
+  fromCheckpoint(checkpoint?: [Value, boolean]) {
+    const empty = new LastValueAfterFinish<Value>();
+    if (typeof checkpoint !== "undefined") {
+      const [value, finished] = checkpoint;
+      empty.value = [value];
+      empty.finished = finished;
+    }
+    return empty as this;
+  }
+
+  update(values: Value[]): boolean {
+    if (values.length === 0) {
+      return false;
+    }
+
+    this.finished = false;
+    // eslint-disable-next-line prefer-destructuring
+    this.value = [values[values.length - 1]];
+    return true;
+  }
+
+  get(): Value {
+    if (this.value.length === 0 || !this.finished) {
+      throw new EmptyChannelError();
+    }
+    return this.value[0];
+  }
+
+  checkpoint(): [Value, boolean] | undefined {
+    if (this.value.length === 0) return undefined;
+    return [this.value[0], this.finished];
+  }
+
+  consume(): boolean {
+    if (this.finished) {
+      this.finished = false;
+      this.value = [];
+      return true;
+    }
+    return false;
+  }
+
+  finish(): boolean {
+    if (!this.finished && this.value.length > 0) {
+      this.finished = true;
+      return true;
+    }
+    return false;
+  }
+
+  isAvailable(): boolean {
+    return this.value.length !== 0 && this.finished;
+  }
 }