diff --git a/apps/website/content/docs/langgraph/api/api-docs.json b/apps/website/content/docs/langgraph/api/api-docs.json index 26947be55..060243860 100644 --- a/apps/website/content/docs/langgraph/api/api-docs.json +++ b/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 } ], diff --git a/libs/langgraph/src/lib/agent.provider.ts b/libs/langgraph/src/lib/agent.provider.ts index e68bcb47a..9d2bf7a57 100644 --- a/libs/langgraph/src/lib/agent.provider.ts +++ b/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; diff --git a/libs/langgraph/src/lib/agent.types.ts b/libs/langgraph/src/lib/agent.types.ts index 65e578a53..25eed31aa 100644 --- a/libs/langgraph/src/lib/agent.types.ts +++ b/libs/langgraph/src/lib/agent.types.ts @@ -277,7 +277,15 @@ export interface AgentOptions { 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; diff --git a/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts b/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts index af74638cd..2b99b05ae 100644 --- a/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts +++ b/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(); + const threadId$ = new BehaviorSubject(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(); + const threadId$ = new BehaviorSubject('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(); diff --git a/libs/langgraph/src/lib/internals/stream-manager.bridge.ts b/libs/langgraph/src/lib/internals/stream-manager.bridge.ts index 693faafa0..0e4b00318 100644 --- a/libs/langgraph/src/lib/internals/stream-manager.bridge.ts +++ b/libs/langgraph/src/lib/internals/stream-manager.bridge.ts @@ -140,9 +140,14 @@ export interface StreamManagerBridge { export function createStreamManagerBridge( { options, subjects, threadId$, destroy$, reportOperationFailure }: StreamManagerBridgeOptions ): 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(); const toolProgressMap = new Map(); // Message ids whose content is known-final (installed by a canonical @@ -422,10 +426,15 @@ export function createStreamManagerBridge { - const shouldReset = hasSeenThreadId && currentThreadId !== id; - hasSeenThreadId = true; + const shouldReset = currentThreadId !== null && currentThreadId !== id; setThreadId(id, shouldReset); });