fix(langgraph): adopt a transport-created thread id instead of aborting the run (#1005)

* fix(langgraph): adopt a transport-created thread id instead of aborting the run

The bridge treated the first non-null thread id as a thread SWITCH whenever
its own currentThreadId was still null, which aborted the in-flight run that
had just created that thread. Only the fallback FetchStreamTransport got the
wrappedOnThreadId interceptor, so a consumer-supplied transport left
currentThreadId at null and every adoption looked like a switch.

shouldReset now requires a KNOWN current thread: a null currentThreadId is
adoption, never a switch. hasSeenThreadId is subsumed by that check and is
removed. A genuine switch (known id -> different id, or -> null) still aborts
and resets exactly as before.

Also corrects the wrappedOnThreadId comment, which claimed a guarantee it only
provides on the default transport, and documents on AgentConfig.transport /
AgentOptions.transport that a custom transport must report created thread ids
through onThreadId or the threadId signal.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* chore(docs): regenerate api docs

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Author: 3 people

apps/website/content/docs/langgraph/api/api-docs.json

@@ -949,7 +949,7 @@
       {
         "name": "transport",
         "type": "AgentTransport",
-        "description": "Custom transport. Defaults to FetchStreamTransport.",
+        "description": "Custom transport. Defaults to FetchStreamTransport.\n\nA custom transport owns its own thread creation, so the runtime cannot\nobserve it directly: report every thread id the transport creates through\nAgentConfig.onThreadId or through the `threadId` signal, otherwise\nthe runtime keeps sending `null` and a new thread is created on each submit.",
         "optional": true
       }
     ],
@@ -1091,7 +1091,7 @@
       {
         "name": "transport",
         "type": "AgentTransport",
-        "description": "Custom transport. Defaults to FetchStreamTransport.",
+        "description": "Custom transport. Defaults to FetchStreamTransport.\n\nA custom transport owns its own thread creation, so the runtime cannot\nobserve it directly: report every thread id the transport creates through\nthis config's AgentOptions.onThreadId or through the `threadId`\nsignal, otherwise the runtime keeps sending `null` and a new thread is\ncreated on each submit.",
         "optional": true
       }
     ],

libs/langgraph/src/lib/agent.provider.ts

@@ -36,7 +36,14 @@ export interface AgentConfig<
   throttle?: number | false;
   /** Custom message deserializer for non-standard message formats. */
   toMessage?: (msg: unknown) => BaseMessage;
-  /** Custom transport. Defaults to {@link FetchStreamTransport}. */
+  /**
+   * Custom transport. Defaults to {@link FetchStreamTransport}.
+   *
+   * A custom transport owns its own thread creation, so the runtime cannot
+   * observe it directly: report every thread id the transport creates through
+   * {@link AgentConfig.onThreadId} or through the `threadId` signal, otherwise
+   * the runtime keeps sending `null` and a new thread is created on each submit.
+   */
   transport?: AgentTransport;
   /** Tuning options for the default transport's LangGraph SDK client (e.g. retry budget). */
   clientOptions?: LangGraphClientOptions;

libs/langgraph/src/lib/agent.types.ts

@@ -277,7 +277,15 @@ export interface AgentOptions<T, _ResolvedBag extends BagTemplate> {
   throttle?: number | false;
   /** Custom message deserializer for non-standard message formats. */
   toMessage?: (msg: unknown) => BaseMessage;
-  /** Custom transport. Defaults to FetchStreamTransport. */
+  /**
+   * Custom transport. Defaults to FetchStreamTransport.
+   *
+   * A custom transport owns its own thread creation, so the runtime cannot
+   * observe it directly: report every thread id the transport creates through
+   * this config's {@link AgentOptions.onThreadId} or through the `threadId`
+   * signal, otherwise the runtime keeps sending `null` and a new thread is
+   * created on each submit.
+   */
   transport?: AgentTransport;
   /** Tuning options for the default transport's LangGraph SDK client (e.g. retry budget). */
   clientOptions?: LangGraphClientOptions;

libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts

@@ -2501,6 +2501,71 @@ describe('createStreamManagerBridge', () => {
     destroy$.next();
   });
 
+  it('adopts a thread id created mid-stream by a custom transport instead of aborting the run', async () => {
+    // A consumer-supplied transport creates the thread itself and reports the
+    // id back through the configured thread-id signal while the first run is
+    // still streaming. The bridge has no thread of its own yet, so this is an
+    // adoption — it must NOT be mistaken for a thread switch and must not
+    // abort the run that just started.
+    const transport = new MockAgentTransport();
+    const streamSpy = vi.spyOn(transport, 'stream');
+    const subjects = makeSubjects();
+    const destroy$ = new Subject<void>();
+    const threadId$ = new BehaviorSubject<string | null>(null);
+    const bridge = createStreamManagerBridge({
+      options: { apiUrl: '', assistantId: 'test', transport },
+      subjects,
+      threadId$: threadId$.asObservable(),
+      destroy$: destroy$.asObservable(),
+    });
+
+    const submitted = bridge.submit({});
+    await new Promise(r => setTimeout(r, 10));
+
+    threadId$.next('thread-created-by-transport');
+
+    transport.emit([{ type: 'values', values: { count: 1 } }]);
+    await new Promise(r => setTimeout(r, 10));
+
+    const signal = streamSpy.mock.calls[0][3] as AbortSignal;
+    expect(signal.aborted).toBe(false);
+    expect(subjects.values$.value).toEqual({ count: 1 });
+    expect(subjects.status$.value).not.toBe(ResourceStatus.Error);
+
+    transport.close();
+    await submitted;
+    destroy$.next();
+  });
+
+  it('still aborts and resets when a known thread id switches to a different one mid-stream', async () => {
+    const transport = new MockAgentTransport();
+    const streamSpy = vi.spyOn(transport, 'stream');
+    const subjects = makeSubjects();
+    const destroy$ = new Subject<void>();
+    const threadId$ = new BehaviorSubject<string | null>('thread-1');
+    const bridge = createStreamManagerBridge({
+      options: { apiUrl: '', assistantId: 'test', transport },
+      subjects,
+      threadId$: threadId$.asObservable(),
+      destroy$: destroy$.asObservable(),
+    });
+
+    bridge.submit({});
+    await new Promise(r => setTimeout(r, 10));
+    transport.emit([{ type: 'values', values: { count: 1 } }]);
+    await new Promise(r => setTimeout(r, 10));
+    expect(subjects.values$.value).toEqual({ count: 1 });
+
+    threadId$.next('thread-2');
+    await new Promise(r => setTimeout(r, 10));
+
+    const signal = streamSpy.mock.calls[0][3] as AbortSignal;
+    expect(signal.aborted).toBe(true);
+    expect(subjects.values$.value).toEqual({});
+    expect(subjects.messages$.value).toEqual([]);
+    destroy$.next();
+  });
+
   it('stop() aborts the active stream and sets status to Idle (user-stop is not an error)', async () => {
     const transport = new MockAgentTransport();
     const subjects = makeSubjects();

libs/langgraph/src/lib/internals/stream-manager.bridge.ts

@@ -140,9 +140,14 @@ export interface StreamManagerBridge {
 export function createStreamManagerBridge<T, ResolvedBag extends BagTemplate = BagTemplate>(
   { options, subjects, threadId$, destroy$, reportOperationFailure }: StreamManagerBridgeOptions<T, ResolvedBag>
 ): StreamManagerBridge {
-  // Intercept onThreadId to update currentThreadId when the transport
-  // auto-creates a thread. Without this, each submit() creates a new thread
-  // because currentThreadId stays null.
+  // Intercept onThreadId so currentThreadId tracks a thread the DEFAULT
+  // transport auto-creates. Without this, each submit() would create a new
+  // thread because currentThreadId stays null. This wrapper only reaches the
+  // transport the bridge constructs below — a consumer-supplied transport owns
+  // its own creation callback, so it must report created ids through the
+  // configured `onThreadId` or the thread-id signal (see AgentConfig.transport).
+  // Either route is handled: the thread-id subscription treats a null
+  // currentThreadId as adoption rather than a switch.
   const userOnThreadId = options.onThreadId;
   const wrappedOnThreadId = (id: string) => {
     currentThreadId = id;
@@ -161,7 +166,6 @@ export function createStreamManagerBridge<T, ResolvedBag extends BagTemplate = B
   let lastOptions: LangGraphSubmitOptions | undefined;
   let abortController: AbortController | null = null;
   let historyAbortController: AbortController | null = null;
-  let hasSeenThreadId = false;
   const userAbortedControllers = new WeakSet<AbortController>();
   const toolProgressMap = new Map<string, ToolProgress>();
   // Message ids whose content is known-final (installed by a canonical
@@ -422,10 +426,15 @@ export function createStreamManagerBridge<T, ResolvedBag extends BagTemplate = B
     void refreshHistory();
   }
 
-  // Track threadId changes
+  // Track threadId changes.
+  //
+  // A null currentThreadId means the bridge is not on a thread yet, so an
+  // incoming id is an ADOPTION — typically the thread a transport just created
+  // for the run that is streaming right now. Resetting there would abort the
+  // bridge's own in-flight run. Only a KNOWN id changing to a different id (or
+  // to null) is a genuine switch, and only that resets.
   threadId$.pipe(takeUntil(destroy$)).subscribe(id => {
-    const shouldReset = hasSeenThreadId && currentThreadId !== id;
-    hasSeenThreadId = true;
+    const shouldReset = currentThreadId !== null && currentThreadId !== id;
     setThreadId(id, shouldReset);
   });