diff --git a/apps/website/content/docs/langgraph/guides/time-travel.mdx b/apps/website/content/docs/langgraph/guides/time-travel.mdx index c73b7ccf4..7012e1ba7 100644 --- a/apps/website/content/docs/langgraph/guides/time-travel.mdx +++ b/apps/website/content/docs/langgraph/guides/time-travel.mdx @@ -36,7 +36,7 @@ The example resolves its connection details at runtime from the host that serves ### Reading the checkpoint history -The component injects the agent and derives the timeline from it. `langGraphHistory()` is LangGraph's own view of the thread: an array of `ThreadState` snapshots. `selectedIndex` is local interface state, holding the row the user clicked last. +The component injects the agent and derives the timeline from it. `langGraphHistory()` is LangGraph's own view of the thread: an array of `ThreadState` snapshots. `selectedCheckpointId` is local interface state, holding the checkpoint the user clicked last. New checkpoints can move existing rows, so selection follows the identifier across history refreshes. diff --git a/cockpit/langgraph/time-travel/angular/e2e/fixtures/time-travel.json b/cockpit/langgraph/time-travel/angular/e2e/fixtures/time-travel.json index b72a2018a..dcc32ca70 100644 --- a/cockpit/langgraph/time-travel/angular/e2e/fixtures/time-travel.json +++ b/cockpit/langgraph/time-travel/angular/e2e/fixtures/time-travel.json @@ -7,6 +7,14 @@ "response": { "content": "Hi! How can I help you today?\n\nQuick note: every response here is saved as a checkpoint snapshot. If you want to explore a different path later, you can inspect the conversation history and branch from any previous checkpoint using stream.setBranch(checkpointId) to create an alternate timeline." } + }, + { + "match": { "userMessage": "Continue the original path." }, + "response": { "content": "This reply belongs only to the original path." } + }, + { + "match": { "userMessage": "Try a different approach from here." }, + "response": { "content": "This reply belongs to the checkpoint fork." } } ] -} \ No newline at end of file +} diff --git a/cockpit/langgraph/time-travel/angular/e2e/time-travel.spec.ts b/cockpit/langgraph/time-travel/angular/e2e/time-travel.spec.ts index 51303817d..09bf1da67 100644 --- a/cockpit/langgraph/time-travel/angular/e2e/time-travel.spec.ts +++ b/cockpit/langgraph/time-travel/angular/e2e/time-travel.spec.ts @@ -1,5 +1,6 @@ import { test, expect } from '@playwright/test'; import { submitAndWaitForResponse } from '@threadplane-internal/e2e-harness'; +import type { ThreadState } from '@langchain/langgraph-sdk'; test('time-travel: hello prompt produces assistant turn', async ({ page }) => { const bubble = await submitAndWaitForResponse(page, 'Hello'); @@ -7,3 +8,80 @@ test('time-travel: hello prompt produces assistant turn', async ({ page }) => { // finalized (data-streaming="false") and is present in the DOM. await expect(bubble).toBeVisible(); }); + +test('time-travel: fork starts at the selected checkpoint, not the thread tip', async ({ + page, +}) => { + const firstRun = page.waitForRequest( + (request) => + request.method() === 'POST' && + new URL(request.url()).pathname.endsWith('/runs/stream') + ); + await submitAndWaitForResponse(page, 'Hello'); + const stateUrl = (await firstRun).url().replace(/\/runs\/stream$/, '/state'); + const readState = async (): Promise< + ThreadState<{ + messages: { id: string; content: string }[]; + }> + > => { + const response = await page.request.get(stateUrl); + expect(response.ok()).toBe(true); + return response.json(); + }; + await expect + .poll(async () => (await readState()).values.messages.length) + .toBe(2); + const original = await readState(); + const checkpointId = original.checkpoint.checkpoint_id; + if (!checkpointId) + throw new Error('Completed first turn did not produce a checkpoint'); + + await page + .getByRole('textbox', { name: /message|prompt/i }) + .fill('Continue the original path.'); + await page.getByRole('button', { name: /send message/i }).click(); + await expect( + page + .locator('chat-message[data-role="assistant"][data-streaming="false"]') + .last() + ).toContainText('This reply belongs only to the original path.'); + await expect + .poll(async () => (await readState()).values.messages.length) + .toBe(4); + + const forkRun = page.waitForRequest( + (request) => + request.method() === 'POST' && + new URL(request.url()).pathname.endsWith('/runs/stream') + ); + await page + .locator('.row') + .filter({ hasText: checkpointId }) + .getByRole('button', { name: 'Fork', exact: true }) + .click(); + const body = (await forkRun).postDataJSON(); + expect(body.checkpoint).toEqual({ + checkpoint_id: checkpointId, + checkpoint_ns: '', + checkpoint_map: {}, + }); + expect(body).not.toHaveProperty('checkpoint_id'); + await expect( + page + .locator('chat-message[data-role="assistant"][data-streaming="false"]') + .last() + ).toContainText('This reply belongs to the checkpoint fork.'); + + // The real server must preserve the first turn and exclude the later original + // turn. Merely observing a completed reply would pass with the broken alias. + await expect + .poll(async () => + (await readState()).values.messages.map((message) => message.content) + ) + .toEqual([ + ...original.values.messages.map((message) => message.content), + 'Try a different approach from here.', + 'This reply belongs to the checkpoint fork.', + ]); + await expect(page.locator('.row--active .id')).toHaveText(checkpointId); +}); diff --git a/cockpit/langgraph/time-travel/angular/src/app/time-travel.component.ts b/cockpit/langgraph/time-travel/angular/src/app/time-travel.component.ts index 7bc007547..c5bf6877b 100644 --- a/cockpit/langgraph/time-travel/angular/src/app/time-travel.component.ts +++ b/cockpit/langgraph/time-travel/angular/src/app/time-travel.component.ts @@ -155,10 +155,16 @@ import { ExampleChatLayoutComponent } from '@threadplane/example-layouts';

No checkpoints yet. Send a message to begin.

} - @for (state of checkpoints(); track $index; let i = $index) { -
+ @for (state of checkpoints(); track state.checkpoint?.checkpoint_id ?? $index; let i = $index) { +
- + {{ i + 1 }} @@ -175,14 +181,14 @@ import { ExampleChatLayoutComponent } from '@threadplane/example-layouts'; @@ -199,8 +205,8 @@ export class TimeTravelComponent { // #region history protected readonly agent = injectAgent(); - /** Index of the currently selected checkpoint in the sidebar. */ - protected readonly selectedIndex = signal(-1); + /** Selection survives new checkpoints being prepended to the history. */ + protected readonly selectedCheckpointId = signal(null); /** Checkpoint history derived from the agent. */ protected readonly checkpoints = computed( @@ -226,9 +232,9 @@ export class TimeTravelComponent { * `setBranch()` records the identifier in the agent's `branch()` signal. * It starts no run: it is a pointer the interface can read back. */ - protected select(state: ThreadState, index: number): void { + protected select(state: ThreadState): void { if (state.checkpoint?.checkpoint_id) { - this.selectedIndex.set(index); + this.selectedCheckpointId.set(state.checkpoint.checkpoint_id); this.agent.setBranch(state.checkpoint.checkpoint_id); } } @@ -240,10 +246,10 @@ export class TimeTravelComponent { * rather than at its tip, so the new run hangs off the chosen checkpoint * and the original path stays intact. */ - protected fork(state: ThreadState, index: number): void { + protected fork(state: ThreadState): void { const checkpointId = state.checkpoint?.checkpoint_id; if (!checkpointId) return; - this.selectedIndex.set(index); + this.selectedCheckpointId.set(checkpointId); void this.agent.submit( { message: 'Try a different approach from here.' }, { checkpointId }, diff --git a/libs/langgraph/src/lib/transport/fetch-stream.transport.ts b/libs/langgraph/src/lib/transport/fetch-stream.transport.ts index 0741d8398..847a86da4 100644 --- a/libs/langgraph/src/lib/transport/fetch-stream.transport.ts +++ b/libs/langgraph/src/lib/transport/fetch-stream.transport.ts @@ -272,15 +272,24 @@ function buildRunPayload( streamMode: StreamMode[]; streamSubgraphs: boolean; signal: AbortSignal; -} & Omit { +} & Omit { const runOptions = { ...(options ?? {}) }; - const hasCheckpoint = Object.prototype.hasOwnProperty.call(runOptions, 'checkpoint'); - const checkpoint = runOptions.checkpoint; + const hasCheckpoint = Object.prototype.hasOwnProperty.call(runOptions, 'checkpoint') || + runOptions.checkpointId !== undefined; + // SDK stream serializes only `checkpoint`, unlike create/wait. Normalize the + // documented ID alias for both paths; an explicit object or null wins. + // Older API versions require checkpoint_map to be an object, not null. + const checkpoint: LangGraphSubmitOptions['checkpoint'] = runOptions.checkpoint !== undefined + ? runOptions.checkpoint + : runOptions.checkpointId !== undefined + ? { checkpoint_id: runOptions.checkpointId, checkpoint_ns: '', checkpoint_map: {} } + : undefined; const streamMode = runOptions.streamMode; const streamSubgraphs = runOptions.streamSubgraphs; delete runOptions.signal; delete runOptions.resume; delete runOptions.checkpoint; + delete runOptions.checkpointId; delete runOptions.streamMode; delete runOptions.streamSubgraphs; diff --git a/libs/langgraph/src/runtime/checkpoint-routing.spec.ts b/libs/langgraph/src/runtime/checkpoint-routing.spec.ts new file mode 100644 index 000000000..cd5752159 --- /dev/null +++ b/libs/langgraph/src/runtime/checkpoint-routing.spec.ts @@ -0,0 +1,116 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { FetchStreamTransport } from '../lib/transport/fetch-stream.transport'; +import type { LangGraphSubmitOptions } from './transport.types'; + +afterEach(() => vi.unstubAllGlobals()); + +const checkpoint = Object.freeze({ + checkpoint_id: 'full-checkpoint', + checkpoint_ns: 'child:task', + checkpoint_map: null, +}); +const cases: { + name: string; + options?: LangGraphSubmitOptions; + expected?: unknown; +}[] = [ + { + name: 'ID alias', + options: { checkpointId: 'chosen-checkpoint' }, + expected: { + checkpoint_id: 'chosen-checkpoint', + checkpoint_ns: '', + checkpoint_map: {}, + }, + }, + { + name: 'full object over alias', + options: { checkpoint, checkpointId: 'ignored' }, + expected: checkpoint, + }, + { + name: 'explicit null over alias', + options: { checkpoint: null, checkpointId: 'ignored' }, + expected: null, + }, + { + name: 'undefined object with alias', + options: { checkpoint: undefined, checkpointId: 'chosen-checkpoint' }, + expected: { + checkpoint_id: 'chosen-checkpoint', + checkpoint_ns: '', + checkpoint_map: {}, + }, + }, + { name: 'no checkpoint', options: undefined }, +]; + +describe.each(['stream', 'queue'] as const)( + 'SDK %s checkpoint routing', + (mode) => { + it.each(cases)( + 'serializes $name without mutating caller options', + async ({ options, expected }) => { + const bodies: unknown[] = []; + const request = vi.fn(async (url, init) => { + expect(String(url)).toBe( + `https://runtime.example/threads/thread-a/runs${ + mode === 'stream' ? '/stream' : '' + }` + ); + expect(init?.method).toBe('POST'); + bodies.push(JSON.parse(String(init?.body))); + return mode === 'stream' + ? new Response('event: values\ndata: {"messages":[]}\n\n', { + headers: { 'content-type': 'text/event-stream' }, + }) + : Response.json({ + run_id: 'queued', + thread_id: 'thread-a', + status: 'pending', + }); + }); + vi.stubGlobal('fetch', request); + const transport = new FetchStreamTransport( + 'https://runtime.example', + undefined, + { maxRetries: 0 } + ); + const before = options && { ...options }; + if (options) Object.freeze(options); + const signal = new AbortController().signal; + if (mode === 'stream') { + for await (const event of transport.stream( + 'assistant', + 'thread-a', + null, + signal, + options + )) { + expect(event.type).toBe('values'); + } + } else { + await transport.createQueuedRun( + 'assistant', + 'thread-a', + null, + signal, + options + ); + } + expect(bodies).toEqual([ + { + input: null, + assistant_id: 'assistant', + stream_mode: ['values', 'messages-tuple', 'updates', 'custom'], + stream_subgraphs: true, + ...(mode === 'queue' ? { multitask_strategy: 'enqueue' } : {}), + ...(expected === undefined ? {} : { checkpoint: expected }), + }, + ]); + expect(options).toEqual(before); + expect(request).toHaveBeenCalledTimes(1); + } + ); + } +); diff --git a/scripts/react-parity/baseline.json b/scripts/react-parity/baseline.json index a9341eddb..bebab6c3b 100644 --- a/scripts/react-parity/baseline.json +++ b/scripts/react-parity/baseline.json @@ -1,13 +1,9 @@ { "schemaVersion": 1, - "baselineHead": "56cd7b799ee98c544ff670f868762dd87a8b6787", + "baselineHead": "82c5469743a8fc3484d395f03372dc6c6975e9b2", "sourceState": { - "modified": [ - "libs/langgraph/src/runtime/create-session.ts" - ], - "untracked": [ - "libs/langgraph/src/runtime/run-options.ts" - ] + "modified": [], + "untracked": [] }, "scope": { "libraries": [ @@ -2540,7 +2536,7 @@ "id": "doc:apps/website/content/docs/langgraph/guides/time-travel.mdx", "kind": "doc", "path": "apps/website/content/docs/langgraph/guides/time-travel.mdx", - "sha256": "34ea8b1fbb52db5ba0a49e679104c8ca78c4ef476edcd058b201cac301d88022" + "sha256": "ddd916b992331e56abe4bf72b0f074dd7649bd521e1702c12a078bb1a6c228ba" }, { "id": "doc:apps/website/content/docs/middleware/api/client-tool-helpers.mdx", @@ -12538,7 +12534,7 @@ "id": "source:libs/langgraph/src/lib/transport/fetch-stream.transport.ts", "kind": "source", "path": "libs/langgraph/src/lib/transport/fetch-stream.transport.ts", - "sha256": "e3bca3ff3a3959c4d61cb41ddb5cf4087b3d55ceb55bdca4afda4d815e1ed44d" + "sha256": "a4a66891600fdf1bed6cc769deca86b87a724f392568369ba006c987fc583f95" }, { "id": "source:libs/langgraph/src/lib/transport/mock-stream.transport.ts",