fix(sdk): show interrupts raised after a passive thread rejoin

.changeset/sdk-passive-rejoin-interrupts.md

@@ -0,0 +1,14 @@
+---
+"@langchain/langgraph-sdk": patch
+---
+
+fix(sdk): show interrupts raised after a passive thread rejoin
+
+After a page refresh mid-run, `useStream` filtered every interrupt it did not
+already know from the hydrated thread state as replayed history, and waited
+for a `checkpoints` event to lift that filter. Current runtimes never emit
+that event and the replay buffer trims it on long runs, so interrupts raised
+after the refresh never appeared until the next reload. Unknown interrupts are
+now settled against the server's thread state when the run reaches a terminal
+lifecycle: the ones the server lists as pending are shown, the rest are
+dropped as history.

libs/sdk/src/stream/controller.test.ts

@@ -206,10 +206,13 @@ function namespacedLifecycleEvent(
   } as Event;
 }
 
-async function waitForExpectation(assertion: () => void): Promise<void> {
+async function waitForExpectation(
+  assertion: () => void,
+  timeoutMs = 500
+): Promise<void> {
   const started = Date.now();
   let lastError: unknown;
-  while (Date.now() - started < 500) {
+  while (Date.now() - started < timeoutMs) {
     try {
       assertion();
       return;
@@ -844,6 +847,137 @@ describe("StreamController", () => {
     await controller.dispose();
   });
 
+  function passiveRejoinFixture(
+    states: Array<Record<string, unknown> | Error>
+  ): {
+    controller: StreamController<State, unknown>;
+    emit: (event: Event) => void;
+    getState: ReturnType<typeof vi.fn>;
+  } {
+    let onEvent: ((event: Event) => void) | undefined;
+    const thread = {
+      subscribe: vi.fn(async () => makeNeverEndingSubscription()),
+      onEvent: vi.fn((listener: (event: Event) => void) => {
+        onEvent = listener;
+        return vi.fn();
+      }),
+      close: vi.fn(async () => undefined),
+      interrupts: [],
+      startLifecycleWatcher: vi.fn(() => undefined),
+    } as unknown as ThreadStream;
+    let calls = 0;
+    const getState = vi.fn(async () => {
+      const state = states[Math.min(calls, states.length - 1)];
+      calls += 1;
+      if (state instanceof Error) throw state;
+      return state;
+    });
+    const client = {
+      threads: { getState, stream: vi.fn(() => thread) },
+    };
+    const controller = new StreamController<State, unknown>({
+      assistantId: "human-in-the-loop",
+      client: client as never,
+      threadId: "thread-passive-rejoin",
+    });
+    return { controller, emit: (event) => onEvent?.(event), getState };
+  }
+
+  const runningAtRefresh = {
+    values: {},
+    next: ["ask"],
+    tasks: [],
+    checkpoint: { checkpoint_id: "checkpoint-at-refresh" },
+  };
+  const interruptedOn = (id: string) => ({
+    values: {},
+    next: ["ask"],
+    tasks: [{ interrupts: [{ id, value: { question: "approve?" } }] }],
+    checkpoint: { checkpoint_id: "checkpoint-after-refresh" },
+  });
+  const finishedIdle = { values: {}, next: [], tasks: [] };
+
+  it("shows an interrupt raised after a passive rejoin without a checkpoints event", async () => {
+    const { controller, emit, getState } = passiveRejoinFixture([
+      runningAtRefresh,
+      interruptedOn("raised-after-refresh"),
+    ]);
+    await controller.hydrationPromise;
+
+    emit(inputRequestedEvent("raised-after-refresh", {}, [], 12));
+    expect(controller.rootStore.getSnapshot().interrupts).toEqual([]);
+
+    emit(lifecycleEvent("interrupted", 13));
+    await waitForExpectation(() => {
+      expect(
+        controller.rootStore.getSnapshot().interrupts.map((item) => item.id)
+      ).toEqual(["raised-after-refresh"]);
+    });
+    expect(getState).toHaveBeenCalledTimes(2);
+
+    await controller.dispose();
+  });
+
+  it("waits for the server to commit the interrupt before showing it", async () => {
+    const { controller, emit, getState } = passiveRejoinFixture([
+      runningAtRefresh,
+      runningAtRefresh,
+      interruptedOn("raised-after-refresh"),
+    ]);
+    await controller.hydrationPromise;
+
+    emit(inputRequestedEvent("raised-after-refresh", {}, [], 12));
+    emit(lifecycleEvent("interrupted", 13));
+    await waitForExpectation(() => {
+      expect(
+        controller.rootStore.getSnapshot().interrupts.map((item) => item.id)
+      ).toEqual(["raised-after-refresh"]);
+    }, 3_000);
+    expect(getState).toHaveBeenCalledTimes(3);
+
+    await controller.dispose();
+  });
+
+  it("retries a failed state fetch while settling parked interrupts", async () => {
+    const { controller, emit, getState } = passiveRejoinFixture([
+      runningAtRefresh,
+      new Error("502 from the load balancer"),
+      interruptedOn("raised-after-refresh"),
+    ]);
+    await controller.hydrationPromise;
+
+    emit(inputRequestedEvent("raised-after-refresh", {}, [], 12));
+    emit(lifecycleEvent("interrupted", 13));
+    await waitForExpectation(() => {
+      expect(
+        controller.rootStore.getSnapshot().interrupts.map((item) => item.id)
+      ).toEqual(["raised-after-refresh"]);
+    }, 3_000);
+    expect(getState).toHaveBeenCalledTimes(3);
+
+    await controller.dispose();
+  });
+
+  it("drops replayed interrupts the server no longer lists", async () => {
+    const { controller, emit, getState } = passiveRejoinFixture([
+      runningAtRefresh,
+      finishedIdle,
+    ]);
+    await controller.hydrationPromise;
+
+    emit(inputRequestedEvent("resolved-long-ago", {}, [], 3));
+    emit(lifecycleEvent("interrupted", 4));
+    await waitForExpectation(() => expect(getState).toHaveBeenCalledTimes(2));
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    expect(controller.rootStore.getSnapshot().interrupts).toEqual([]);
+
+    emit(lifecycleEvent("interrupted", 5));
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    expect(getState).toHaveBeenCalledTimes(2);
+
+    await controller.dispose();
+  });
+
   it("filters replayed interrupts while accepting newer interrupts after submit", async () => {
     const eventListeners = new Set<(event: Event) => void>();
     let resolveSubmit: (() => void) | undefined;

libs/sdk/src/stream/controller.ts

@@ -180,6 +180,12 @@ export const ROOT_PUMP_CHANNELS: readonly Channel[] = [
   "tools",
 ];
 
+// The server writes an interrupt into thread state shortly after it streams
+// the `input.requested` event, so a state fetch on the terminal lifecycle can
+// miss it; mirrors the server's own interrupt settle window.
+const PARKED_INTERRUPT_SETTLE_MS = 5_000;
+const PARKED_INTERRUPT_SETTLE_POLL_MS = 500;
+
 interface ResolvedInterrupt {
   interruptId: string;
   namespace: string[];
@@ -267,6 +273,9 @@ export class StreamController<
    * from the newly-started run and must be accepted.
    */
   #interruptReplayThroughSeq: number | undefined;
+  /** Unknown inputs waiting to be classified as live or replayed history. */
+  #pendingHydrationInterruptEvents: Event[] = [];
+  #settleParkedInterruptsPromise: Promise<void> | undefined;
   /**
    * Unknown interrupt events received between local command dispatch and its
    * response. The response supplies the sequence barrier needed to classify
@@ -443,7 +452,8 @@ export class StreamController<
         this.#submitGeneration += 1;
         this.#interruptCommandPending =
           this.#hydratedActiveInterruptIds != null;
-        this.#pendingInterruptEvents = [];
+        this.#pendingInterruptEvents = this.#pendingHydrationInterruptEvents;
+        this.#pendingHydrationInterruptEvents = [];
       },
       onRunStart: () => this.#markLocalRunStart(),
       onRunCreated: (runId) => {
@@ -1761,6 +1771,7 @@ export class StreamController<
     // will repopulate it from that thread's `state.tasks[].interrupts`.
     this.#hydratedActiveInterruptIds = null;
     this.#interruptReplayThroughSeq = undefined;
+    this.#pendingHydrationInterruptEvents = [];
     this.#discardPendingInterruptEvents();
     this.queueStore.setState(
       () => EMPTY_QUEUE as SubmissionQueueSnapshot<StateType>
@@ -1986,6 +1997,7 @@ export class StreamController<
      * resume targeting.
      */
     this.#recordInterrupt(event);
+    this.#settleParkedInterruptsOnTerminal(event);
   }
 
   /**
@@ -2380,10 +2392,16 @@ export class StreamController<
       this.#pendingInterruptEvents.push(event);
       return;
     }
+    const barrier = this.#interruptReplayThroughSeq;
+    if (isUnknownInterrupt && barrier == null) {
+      this.#pendingHydrationInterruptEvents.push(event);
+      return;
+    }
     const isHistoricalUnknownInterrupt =
       isUnknownInterrupt &&
-      (this.#interruptReplayThroughSeq == null ||
-        (eventSeq != null && eventSeq <= this.#interruptReplayThroughSeq));
+      barrier != null &&
+      eventSeq != null &&
+      eventSeq <= barrier;
     if (isHistoricalUnknownInterrupt) {
       return;
     }
@@ -2499,9 +2517,9 @@ export class StreamController<
    * drops the result if the thread swapped or a new submit/respond
    * started while the fetch was in flight.
    */
-  async #reconcilePendingInterruptsFromServer(): Promise<void> {
+  async #reconcilePendingInterruptsFromServer(): Promise<ThreadState<StateType> | null> {
     const threadId = this.#currentThreadId;
-    if (threadId == null || this.#disposed) return;
+    if (threadId == null || this.#disposed) return null;
     const generationAtFetch = this.#submitGeneration;
     try {
       const state = await this.#fetchHydrationState();
@@ -2510,9 +2528,9 @@ export class StreamController<
         this.#currentThreadId !== threadId ||
         this.#submitGeneration !== generationAtFetch
       ) {
-        return;
+        return null;
       }
-      if (!Array.isArray(state?.tasks)) return;
+      if (!Array.isArray(state?.tasks)) return null;
       const { activeInterrupts, activeIds } =
         collectActiveInterruptsFromTasks<InterruptType>(state.tasks);
       const pendingInterrupts = activeInterrupts.filter(
@@ -2525,9 +2543,82 @@ export class StreamController<
         interrupt: pendingInterrupts[0],
       }));
       this.#hydratedActiveInterruptIds = activeIds;
+      return state;
     } catch {
       /* best-effort — leave the optimistic interrupt list alone */
+      return null;
+    }
+  }
+
+  #settleParkedInterruptsOnTerminal(event: Event): void {
+    if (this.#pendingHydrationInterruptEvents.length === 0) return;
+    if (event.method !== "lifecycle") return;
+    if (!isRootNamespace(event.params.namespace)) return;
+    const status = (event as LifecycleEvent).params.data?.event;
+    if (
+      status !== "interrupted" &&
+      status !== "completed" &&
+      status !== "failed"
+    ) {
+      return;
     }
+    void this.#settleParkedInterrupts();
+  }
+
+  /**
+   * Decide parked interrupts with server state instead of a stream marker.
+   *
+   * After a passive rejoin the stream cannot tell a replayed, already
+   * resolved `input.requested` from one the current run just raised: the
+   * SSE replays the thread buffer on connect with no marker for where the
+   * replay ends. `state.tasks[].interrupts` is authoritative: parked ids it
+   * lists are live, the rest are history.
+   */
+  async #settleParkedInterrupts(): Promise<void> {
+    const makePromise = async () => {
+      try {
+        const threadId = this.#currentThreadId;
+        const generation = this.#submitGeneration;
+        const deadline = Date.now() + PARKED_INTERRUPT_SETTLE_MS;
+        for (;;) {
+          if (
+            this.#disposed ||
+            this.#currentThreadId !== threadId ||
+            this.#submitGeneration !== generation ||
+            this.#pendingHydrationInterruptEvents.length === 0
+          ) {
+            return;
+          }
+          const parkedIds = new Set(
+            this.#pendingHydrationInterruptEvents.map(
+              (event) =>
+                (event.params.data as { interrupt_id?: string }).interrupt_id
+            )
+          );
+          // A null state is a failed or unusable fetch; the terminal that
+          // started this loop will not come again, so keep trying until the
+          // deadline rather than stranding the parked interrupts.
+          const state = await this.#reconcilePendingInterruptsFromServer();
+          if (state != null) {
+            const settled = [...(this.#hydratedActiveInterruptIds ?? [])].some(
+              (id) => parkedIds.has(id)
+            );
+            if (settled || !isThreadStateActive(state)) {
+              this.#pendingHydrationInterruptEvents = [];
+              return;
+            }
+          }
+          if (Date.now() >= deadline) return;
+          await new Promise<void>((resolve) => {
+            setTimeout(resolve, PARKED_INTERRUPT_SETTLE_POLL_MS);
+          });
+        }
+      } finally {
+        this.#settleParkedInterruptsPromise = undefined;
+      }
+    };
+    this.#settleParkedInterruptsPromise ??= makePromise();
+    return this.#settleParkedInterruptsPromise;
   }
 
   /**