From f91f5c837f32003dea8608add6eaff81d722e4ec Mon Sep 17 00:00:00 2001 From: Brian Love Date: Wed, 23 Sep 2026 10:24:03 -0700 Subject: [PATCH 1/3] fix(langgraph): honor cancellation for state writes Forward the existing AbortSignal to LangGraph state writes and cover cancellation through normal and protected transports with real HTTP regressions. Include the generated API documentation and parity inventory update. --- .../content/docs/langgraph/api/api-docs.json | 4 +- .../lib/transport/fetch-stream.transport.ts | 6 +- .../runtime/state-write-cancellation.spec.ts | 112 ++++++++++++++++++ scripts/react-parity/baseline.json | 10 +- 4 files changed, 123 insertions(+), 9 deletions(-) create mode 100644 libs/langgraph/src/runtime/state-write-cancellation.spec.ts diff --git a/apps/website/content/docs/langgraph/api/api-docs.json b/apps/website/content/docs/langgraph/api/api-docs.json index 17091a659..cf5b0820a 100644 --- a/apps/website/content/docs/langgraph/api/api-docs.json +++ b/apps/website/content/docs/langgraph/api/api-docs.json @@ -363,7 +363,7 @@ }, { "name": "updateState", - "signature": "updateState(threadId: string, values: Record, _signal: AbortSignal, options: object): Promise", + "signature": "updateState(threadId: string, values: Record, signal: AbortSignal, options: object): Promise", "description": "Update server-side thread state, e.g. to remove messages for regenerate rollback.", "params": [ { @@ -379,7 +379,7 @@ "optional": false }, { - "name": "_signal", + "name": "signal", "type": "AbortSignal", "description": "", "optional": false diff --git a/libs/langgraph/src/lib/transport/fetch-stream.transport.ts b/libs/langgraph/src/lib/transport/fetch-stream.transport.ts index 847a86da4..b88b976cf 100644 --- a/libs/langgraph/src/lib/transport/fetch-stream.transport.ts +++ b/libs/langgraph/src/lib/transport/fetch-stream.transport.ts @@ -193,17 +193,17 @@ export class FetchStreamTransport implements AgentTransport { async updateState( threadId: string, values: Record, - _signal: AbortSignal, + signal: AbortSignal, options?: { asNode?: string }, ): Promise { - const body: { values: Record; asNode?: string } = { values }; + const body: { values: Record; signal: AbortSignal; asNode?: string } = { values, signal }; if (options?.asNode !== undefined) { body.asNode = options.asNode; } try { await this.client.threads.updateState(threadId, body); } catch (error) { - this.rethrowOperationError(error, _signal); + this.rethrowOperationError(error, signal); } } diff --git a/libs/langgraph/src/runtime/state-write-cancellation.spec.ts b/libs/langgraph/src/runtime/state-write-cancellation.spec.ts new file mode 100644 index 000000000..034cfd172 --- /dev/null +++ b/libs/langgraph/src/runtime/state-write-cancellation.spec.ts @@ -0,0 +1,112 @@ +import { createServer, type ServerResponse } from 'node:http'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { FetchStreamTransport } from '../lib/transport/fetch-stream.transport'; +import { deferred } from './testing/deferred'; + +const cleanup: (() => Promise)[] = []; +afterEach(async () => { + await Promise.all(cleanup.splice(0).map((close) => close())); +}); + +async function endpoint(hold = false) { + const started = deferred(); + const requests: { method?: string; url?: string; body: unknown }[] = []; + const server = createServer(async (request, response) => { + let body = ''; + for await (const chunk of request) body += chunk; + requests.push({ + method: request.method, + url: request.url, + body: JSON.parse(body), + }); + started.resolve(response); + if (!hold) { + response.setHeader('content-type', 'application/json'); + response.end(JSON.stringify({ checkpoint_id: 'saved' })); + } + }); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + cleanup.push( + () => + new Promise((resolve, reject) => { + server.closeAllConnections(); + server.close((error) => (error ? reject(error) : resolve())); + }) + ); + const address = server.address(); + if (!address || typeof address === 'string') + throw new Error('Expected TCP server'); + return { + url: `http://127.0.0.1:${address.port}`, + started: started.promise, + requests, + }; +} + +describe.each([false, true])( + 'state-write HTTP cancellation (protected=%s)', + (protectedErrors) => { + function transport(url: string) { + const report = vi.fn(); + return { + client: new FetchStreamTransport( + url, + undefined, + { maxRetries: 0 }, + protectedErrors ? report : undefined + ), + report, + }; + } + + it('does not send a state write when already aborted', async () => { + const server = await endpoint(); + const { client, report } = transport(server.url); + const controller = new AbortController(); + controller.abort(); + await expect( + client.updateState('thread-a', {}, controller.signal) + ).rejects.toMatchObject({ name: 'AbortError' }); + expect(server.requests).toEqual([]); + expect(report).not.toHaveBeenCalled(); + }); + + it('aborts an in-flight state write and closes its HTTP response', async () => { + const server = await endpoint(true); + const { client, report } = transport(server.url); + const controller = new AbortController(); + const write = client.updateState('thread-a', {}, controller.signal); + const rejected = expect(write).rejects.toMatchObject({ + name: 'AbortError', + }); + const response = await server.started; + const closed = new Promise((resolve) => + response.once('close', resolve) + ); + controller.abort(); + await rejected; + await closed; + expect(server.requests).toHaveLength(1); + expect(report).not.toHaveBeenCalled(); + }); + + it('preserves the state body and asNode for successful writes', async () => { + const server = await endpoint(); + const { client } = transport(server.url); + const values = Object.freeze({ messages: Object.freeze([]) }); + await client.updateState( + 'thread-a', + values, + new AbortController().signal, + { asNode: '__start__' } + ); + expect(server.requests).toEqual([ + { + method: 'POST', + url: '/threads/thread-a/state', + body: { values, as_node: '__start__' }, + }, + ]); + }); + } +); diff --git a/scripts/react-parity/baseline.json b/scripts/react-parity/baseline.json index bebab6c3b..f79aac36d 100644 --- a/scripts/react-parity/baseline.json +++ b/scripts/react-parity/baseline.json @@ -1,8 +1,10 @@ { "schemaVersion": 1, - "baselineHead": "82c5469743a8fc3484d395f03372dc6c6975e9b2", + "baselineHead": "ea15ba3e48b156872bed66066462fc0a30f0425a", "sourceState": { - "modified": [], + "modified": [ + "libs/langgraph/src/lib/transport/fetch-stream.transport.ts" + ], "untracked": [] }, "scope": { @@ -8356,7 +8358,7 @@ "path": "libs/langgraph/src/lib/transport/fetch-stream.transport.ts", "symbol": "FetchStreamTransport", "syntaxKind": "ClassDeclaration", - "signature": "export class FetchStreamTransport implements AgentTransport {\n private client: Client;\n private onThreadId?: (id: string) => void;\n private readonly protectErrors: boolean;\n private readonly reportOperationFailure?: RuntimeOperationFailureReporter;\n readonly protectsOperationErrors: boolean;\n constructor(apiUrl: string, onThreadId?: (id: string) => void, clientOptions?: LangGraphClientOptions, reportOperationFailure?: RuntimeOperationFailureReporter) {\n this.protectErrors = clientOptions?.defaultHeaders !== undefined || reportOperationFailure !== undefined;\n this.protectsOperationErrors = this.protectErrors;\n this.reportOperationFailure = reportOperationFailure;\n this.client = this.protectErrors\n ? ɵcreateProtectedLangGraphClient(apiUrl, clientOptions, createLangGraphRuntimeFetch(reportOperationFailure))\n : createLangGraphClient(apiUrl, clientOptions);\n this.onThreadId = onThreadId;\n }\n async *stream(assistantId: string, threadId: string | null, payload: unknown, signal: AbortSignal, options?: LangGraphSubmitOptions): AsyncIterable {\n let thread = threadId;\n if (!thread) {\n try {\n const t = await this.client.threads.create();\n thread = t.thread_id;\n }\n catch (error) {\n this.rethrowOperationError(error, signal);\n }\n try {\n this.onThreadId?.(thread);\n }\n catch (error) {\n this.rethrowLocalError(error, signal);\n }\n }\n let runPayload: ReturnType;\n try {\n runPayload = buildRunPayload(payload, signal, options);\n }\n catch (error) {\n return this.rethrowLocalError(error, signal);\n }\n let run: ReturnType;\n try {\n run = this.client.runs.stream(thread, assistantId, runPayload);\n }\n catch (error) {\n this.rethrowOperationError(error, signal);\n }\n yield* this.iterateSdkRun(run, signal);\n }\n async *joinStream(threadId: string, runId: string, lastEventId: string | undefined, signal: AbortSignal): AsyncIterable {\n let run: ReturnType;\n try {\n run = this.client.runs.joinStream(threadId, runId, {\n signal,\n ...(lastEventId !== undefined ? { lastEventId } : {}),\n });\n }\n catch (error) {\n this.rethrowOperationError(error, signal);\n }\n yield* this.iterateSdkRun(run, signal);\n }\n async getRunStatus(threadId: string, runId: string, signal: AbortSignal): Promise {\n try {\n const run = await this.client.runs.get(threadId, runId, { signal });\n if (run.run_id !== runId || run.thread_id !== threadId ||\n !['pending', 'running', 'success', 'error', 'timeout', 'interrupted'].includes(run.status))\n throw new Error('Invalid LangGraph run status response.');\n return run.status;\n }\n catch (error) {\n return this.rethrowOperationError(error, signal);\n }\n }\n async createQueuedRun(assistantId: string, threadId: string, payload: unknown, signal: AbortSignal, options?: LangGraphSubmitOptions): Promise {\n let runPayload: ReturnType & {\n multitaskStrategy: 'enqueue';\n };\n try {\n runPayload = {\n ...buildRunPayload(payload, signal, options),\n multitaskStrategy: 'enqueue',\n };\n }\n catch (error) {\n return this.rethrowLocalError(error, signal);\n }\n let run: Awaited>;\n try {\n run = await this.client.runs.create(threadId, assistantId, runPayload);\n }\n catch (error) {\n return this.rethrowOperationError(error, signal);\n }\n try {\n return {\n id: run.run_id,\n threadId: run.thread_id ?? threadId,\n values: payload,\n options: { multitaskStrategy: 'enqueue', signal },\n createdAt: run.created_at ? new Date(run.created_at) : new Date(),\n };\n }\n catch (error) {\n return this.rethrowLocalError(error, signal);\n }\n }\n async cancelRun(threadId: string, runId: string, signal: AbortSignal): Promise {\n try {\n await this.client.runs.cancel(threadId, runId, false, 'interrupt', { signal });\n }\n catch (error) {\n this.rethrowOperationError(error, signal);\n }\n }\n async getHistory(threadId: string, signal: AbortSignal): Promise {\n try {\n return await this.client.threads.getHistory(threadId, { signal });\n }\n catch (error) {\n return this.rethrowOperationError(error, signal);\n }\n }\n async updateState(threadId: string, values: Record, _signal: AbortSignal, options?: {\n asNode?: string;\n }): Promise {\n const body: {\n values: Record;\n asNode?: string;\n } = { values };\n if (options?.asNode !== undefined) {\n body.asNode = options.asNode;\n }\n try {\n await this.client.threads.updateState(threadId, body);\n }\n catch (error) {\n this.rethrowOperationError(error, _signal);\n }\n }\n private rethrowOperationError(error: unknown, signal: AbortSignal): never {\n if (!this.protectErrors)\n throw error;\n return projectLangGraphOperationFailure(error, signal, this.reportOperationFailure);\n }\n private rethrowLocalError(error: unknown, signal: AbortSignal): never {\n if (!this.protectErrors)\n throw error;\n return projectLangGraphOperationFailure(error, signal, undefined);\n }\n private async *iterateSdkRun(run: ReturnType | ReturnType, signal: AbortSignal): AsyncIterable {\n let iterator: AsyncIterator<{\n event: string;\n data: unknown;\n id?: unknown;\n }>;\n try {\n iterator = run[Symbol.asyncIterator]();\n }\n catch (error) {\n return this.rethrowOperationError(error, signal);\n }\n let completed = false;\n let failed = false;\n try {\n while (true) {\n let next: IteratorResult<{\n event: string;\n data: unknown;\n id?: unknown;\n }>;\n try {\n next = await iterator.next();\n }\n catch (error) {\n return this.rethrowOperationError(error, signal);\n }\n if (next.done) {\n completed = true;\n return;\n }\n try {\n yield { ...normalizeSdkEvent(next.value.event as StreamEvent['type'], next.value.data), sseId: typeof next.value.id === 'string' ? next.value.id : undefined };\n }\n catch (error) {\n return this.rethrowLocalError(error, signal);\n }\n }\n }\n catch (error) {\n failed = true;\n throw error;\n }\n finally {\n if (!completed) {\n try {\n await iterator.return?.();\n }\n catch (error) {\n if (!failed)\n this.rethrowOperationError(error, signal);\n }\n }\n }\n }\n}" + "signature": "export class FetchStreamTransport implements AgentTransport {\n private client: Client;\n private onThreadId?: (id: string) => void;\n private readonly protectErrors: boolean;\n private readonly reportOperationFailure?: RuntimeOperationFailureReporter;\n readonly protectsOperationErrors: boolean;\n constructor(apiUrl: string, onThreadId?: (id: string) => void, clientOptions?: LangGraphClientOptions, reportOperationFailure?: RuntimeOperationFailureReporter) {\n this.protectErrors = clientOptions?.defaultHeaders !== undefined || reportOperationFailure !== undefined;\n this.protectsOperationErrors = this.protectErrors;\n this.reportOperationFailure = reportOperationFailure;\n this.client = this.protectErrors\n ? ɵcreateProtectedLangGraphClient(apiUrl, clientOptions, createLangGraphRuntimeFetch(reportOperationFailure))\n : createLangGraphClient(apiUrl, clientOptions);\n this.onThreadId = onThreadId;\n }\n async *stream(assistantId: string, threadId: string | null, payload: unknown, signal: AbortSignal, options?: LangGraphSubmitOptions): AsyncIterable {\n let thread = threadId;\n if (!thread) {\n try {\n const t = await this.client.threads.create();\n thread = t.thread_id;\n }\n catch (error) {\n this.rethrowOperationError(error, signal);\n }\n try {\n this.onThreadId?.(thread);\n }\n catch (error) {\n this.rethrowLocalError(error, signal);\n }\n }\n let runPayload: ReturnType;\n try {\n runPayload = buildRunPayload(payload, signal, options);\n }\n catch (error) {\n return this.rethrowLocalError(error, signal);\n }\n let run: ReturnType;\n try {\n run = this.client.runs.stream(thread, assistantId, runPayload);\n }\n catch (error) {\n this.rethrowOperationError(error, signal);\n }\n yield* this.iterateSdkRun(run, signal);\n }\n async *joinStream(threadId: string, runId: string, lastEventId: string | undefined, signal: AbortSignal): AsyncIterable {\n let run: ReturnType;\n try {\n run = this.client.runs.joinStream(threadId, runId, {\n signal,\n ...(lastEventId !== undefined ? { lastEventId } : {}),\n });\n }\n catch (error) {\n this.rethrowOperationError(error, signal);\n }\n yield* this.iterateSdkRun(run, signal);\n }\n async getRunStatus(threadId: string, runId: string, signal: AbortSignal): Promise {\n try {\n const run = await this.client.runs.get(threadId, runId, { signal });\n if (run.run_id !== runId || run.thread_id !== threadId ||\n !['pending', 'running', 'success', 'error', 'timeout', 'interrupted'].includes(run.status))\n throw new Error('Invalid LangGraph run status response.');\n return run.status;\n }\n catch (error) {\n return this.rethrowOperationError(error, signal);\n }\n }\n async createQueuedRun(assistantId: string, threadId: string, payload: unknown, signal: AbortSignal, options?: LangGraphSubmitOptions): Promise {\n let runPayload: ReturnType & {\n multitaskStrategy: 'enqueue';\n };\n try {\n runPayload = {\n ...buildRunPayload(payload, signal, options),\n multitaskStrategy: 'enqueue',\n };\n }\n catch (error) {\n return this.rethrowLocalError(error, signal);\n }\n let run: Awaited>;\n try {\n run = await this.client.runs.create(threadId, assistantId, runPayload);\n }\n catch (error) {\n return this.rethrowOperationError(error, signal);\n }\n try {\n return {\n id: run.run_id,\n threadId: run.thread_id ?? threadId,\n values: payload,\n options: { multitaskStrategy: 'enqueue', signal },\n createdAt: run.created_at ? new Date(run.created_at) : new Date(),\n };\n }\n catch (error) {\n return this.rethrowLocalError(error, signal);\n }\n }\n async cancelRun(threadId: string, runId: string, signal: AbortSignal): Promise {\n try {\n await this.client.runs.cancel(threadId, runId, false, 'interrupt', { signal });\n }\n catch (error) {\n this.rethrowOperationError(error, signal);\n }\n }\n async getHistory(threadId: string, signal: AbortSignal): Promise {\n try {\n return await this.client.threads.getHistory(threadId, { signal });\n }\n catch (error) {\n return this.rethrowOperationError(error, signal);\n }\n }\n async updateState(threadId: string, values: Record, signal: AbortSignal, options?: {\n asNode?: string;\n }): Promise {\n const body: {\n values: Record;\n signal: AbortSignal;\n asNode?: string;\n } = { values, signal };\n if (options?.asNode !== undefined) {\n body.asNode = options.asNode;\n }\n try {\n await this.client.threads.updateState(threadId, body);\n }\n catch (error) {\n this.rethrowOperationError(error, signal);\n }\n }\n private rethrowOperationError(error: unknown, signal: AbortSignal): never {\n if (!this.protectErrors)\n throw error;\n return projectLangGraphOperationFailure(error, signal, this.reportOperationFailure);\n }\n private rethrowLocalError(error: unknown, signal: AbortSignal): never {\n if (!this.protectErrors)\n throw error;\n return projectLangGraphOperationFailure(error, signal, undefined);\n }\n private async *iterateSdkRun(run: ReturnType | ReturnType, signal: AbortSignal): AsyncIterable {\n let iterator: AsyncIterator<{\n event: string;\n data: unknown;\n id?: unknown;\n }>;\n try {\n iterator = run[Symbol.asyncIterator]();\n }\n catch (error) {\n return this.rethrowOperationError(error, signal);\n }\n let completed = false;\n let failed = false;\n try {\n while (true) {\n let next: IteratorResult<{\n event: string;\n data: unknown;\n id?: unknown;\n }>;\n try {\n next = await iterator.next();\n }\n catch (error) {\n return this.rethrowOperationError(error, signal);\n }\n if (next.done) {\n completed = true;\n return;\n }\n try {\n yield { ...normalizeSdkEvent(next.value.event as StreamEvent['type'], next.value.data), sseId: typeof next.value.id === 'string' ? next.value.id : undefined };\n }\n catch (error) {\n return this.rethrowLocalError(error, signal);\n }\n }\n }\n catch (error) {\n failed = true;\n throw error;\n }\n finally {\n if (!completed) {\n try {\n await iterator.return?.();\n }\n catch (error) {\n if (!failed)\n this.rethrowOperationError(error, signal);\n }\n }\n }\n }\n}" } ] }, @@ -12534,7 +12536,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": "a4a66891600fdf1bed6cc769deca86b87a724f392568369ba006c987fc583f95" + "sha256": "2b899f3fb3758f8c57c500b902d16d826fbfd1090c3b160692f0b01456024981" }, { "id": "source:libs/langgraph/src/lib/transport/mock-stream.transport.ts", From 0e725f928bde26aebd2dec4b5bfb367703d468bd Mon Sep 17 00:00:00 2001 From: Brian Love Date: Wed, 23 Sep 2026 12:45:24 -0700 Subject: [PATCH 2/3] test(langgraph): characterize checkpoint replay and write uncertainty --- .../angular/e2e/checkpoint-protocol.spec.ts | 244 ++++++++++++++++++ .../e2e/fixtures/checkpoint-protocol.json | 13 + .../angular/e2e/global-setup-impl.ts | 12 +- 3 files changed, 268 insertions(+), 1 deletion(-) create mode 100644 cockpit/langgraph/client-tools/angular/e2e/checkpoint-protocol.spec.ts create mode 100644 cockpit/langgraph/client-tools/angular/e2e/fixtures/checkpoint-protocol.json diff --git a/cockpit/langgraph/client-tools/angular/e2e/checkpoint-protocol.spec.ts b/cockpit/langgraph/client-tools/angular/e2e/checkpoint-protocol.spec.ts new file mode 100644 index 000000000..63ca0f92e --- /dev/null +++ b/cockpit/langgraph/client-tools/angular/e2e/checkpoint-protocol.spec.ts @@ -0,0 +1,244 @@ +import { createServer } from 'node:http'; +import { test, expect } from '@playwright/test'; +import { Client, type Checkpoint, type ThreadState } from '@langchain/langgraph-sdk'; + +// This lane uses the published middleware from python/uv.lock (0.1.0), not a +// PYTHONPATH override to packages/threadplane-middleware. It proves the actual +// client-tools graph/API protocol, not browser tool execution or store claims. +const prompt = 'Checkpoint protocol weather in Paris'; +const input = { + messages: [{ type: 'human', content: prompt }], + client_tools: [{ + name: 'get_weather', + description: 'Get weather for a location', + parameters: { type: 'object', properties: { location: { type: 'string' } } }, + }], +}; +type ToolCall = { id: string; name: string; args: Record }; +type State = { messages: { type: string; content: unknown; tool_calls?: ToolCall[]; tool_call_id?: string }[] }; +type CheckpointEvent = { + id?: string; + event: string; + data: { config: { configurable?: Record }; values: State; next: string[]; metadata: Record }; +}; + +function backendUrl(): string { + const url = process.env['CLIENT_TOOLS_API_URL']; + if (!url) throw new Error('Global setup must expose the local client-tools API URL'); + return url; +} + +function client() { + return new Client({ apiUrl: backendUrl(), apiKey: null, callerOptions: { maxRetries: 0 }, timeoutMs: 20_000 }); +} + +// Copy only supported routing fields, never the complete event configurable +// object (which may also contain run identity or request-specific metadata). +function position(config: Record | undefined): Checkpoint & { checkpoint_id: string } { + if (!config) throw new Error('Checkpoint event must include routing configuration'); + expect(config['thread_id']).toEqual(expect.any(String)); + expect(config['checkpoint_ns']).toEqual(expect.any(String)); + expect(config['checkpoint_id']).toEqual(expect.any(String)); + return { + thread_id: config['thread_id'] as string, + checkpoint_ns: config['checkpoint_ns'] as string, + checkpoint_id: config['checkpoint_id'] as string, + checkpoint_map: undefined, + }; +} + +async function run(api: Client, threadId: string, checkpoint?: Checkpoint, values: Record | null = null) { + let final: CheckpointEvent | undefined; + for await (const event of api.runs.stream(threadId, 'client-tools', { + input: values, checkpoint, streamMode: ['values', 'checkpoints'], signal: AbortSignal.timeout(20_000), + })) { + expect(event.event).not.toBe('error'); + if (event.event === 'checkpoints') final = event as CheckpointEvent; + } + if (!final) throw new Error('Run must emit a checkpoint event'); + const saved = await api.threads.getState(threadId, position(final.data.config.configurable)); + expect(saved.values).toEqual(final.data.values); + expect(saved.next).toEqual(final.data.next); + return saved; +} + +function calls(state: ThreadState) { + return state.values.messages.flatMap((message) => message.tool_calls ?? []); +} + +async function modelRequests() { + const url = process.env['CLIENT_TOOLS_AIMOCK_URL']; + if (!url) throw new Error('Global setup must expose the local aimock journal URL'); + const response = await fetch(`${url}/__aimock/journal`, { signal: AbortSignal.timeout(5_000) }); + expect(response.ok).toBe(true); + const entries = await response.json() as { body?: { messages?: { role: string; content: unknown }[] } }[]; + return entries.filter((entry) => entry.body?.messages?.some((message) => message.role === 'user' && message.content === prompt)).length; +} + +test('checkpoint protocol: terminal replay preserves a pending call; pre-agent replay creates a new call', async () => { + const api = client(); + const { thread_id: threadId } = await api.threads.create(); + try { + const pending = await run(api, threadId, undefined, input); + expect(pending.next).toEqual([]); + expect(calls(pending)).toHaveLength(1); + const original = calls(pending)[0]; + expect(original).toMatchObject({ id: expect.any(String), name: 'get_weather', args: { location: 'Paris' } }); + const beforeAgent = (await api.threads.getHistory(threadId)).find((state) => state.next.includes('agent')); + if (!beforeAgent) throw new Error('History must include a checkpoint before the agent runs'); + expect(calls(beforeAgent)).toEqual([]); + + const count = await modelRequests(); + const replayed = await run(api, threadId, pending.checkpoint); + expect(replayed.next).toEqual([]); + expect(replayed.values).toEqual(pending.values); + expect(calls(replayed)).toEqual([original]); + expect(await modelRequests()).toBe(count); + + const regenerated = await run(api, threadId, beforeAgent.checkpoint); + expect(calls(regenerated)).toHaveLength(1); + expect(calls(regenerated)[0]).toMatchObject({ name: original.name, args: original.args }); + expect(calls(regenerated)[0].id).not.toBe(original.id); + expect(await modelRequests()).toBe(count + 1); + + const completed = await run(api, threadId, pending.checkpoint, { + messages: [{ type: 'tool', tool_call_id: original.id, content: '68°F' }], + }); + expect(completed.next).toEqual([]); + expect(completed.values.messages.filter((message) => message.type === 'tool')).toMatchObject([{ tool_call_id: original.id }]); + expect(completed.values.messages.at(-1)).toMatchObject({ type: 'ai', content: 'Protocol weather complete: Paris is 68°F.' }); + expect(await modelRequests()).toBe(count + 2); + const completedReplay = await run(api, threadId, completed.checkpoint); + expect(completedReplay.next).toEqual([]); + expect(completedReplay.values).toEqual(completed.values); + expect(await modelRequests()).toBe(count + 2); + } finally { + await api.threads.delete(threadId); + } +}); + +test('checkpoint protocol: reconnect retains the run and exact saved checkpoint identity', async () => { + const requests: { path: string; method: string }[] = []; + const api = new Client({ + apiUrl: backendUrl(), apiKey: null, callerOptions: { maxRetries: 0 }, timeoutMs: 20_000, + onRequest: (url, init) => { + requests.push({ path: url.pathname, method: init.method ?? 'GET' }); + return init; + }, + }); + const { thread_id: threadId } = await api.threads.create(); + const controller = new AbortController(); + try { + let runId: string | undefined; + let cursor: string | undefined; + let disconnectedCheckpoint: string | undefined; + try { + for await (const event of api.runs.stream(threadId, 'client-tools', { + input, streamMode: ['values', 'checkpoints'], streamResumable: true, + onDisconnect: 'continue', signal: AbortSignal.any([controller.signal, AbortSignal.timeout(20_000)]), + })) { + expect(event.event).not.toBe('error'); + if (event.event === 'metadata') runId = event.data.run_id; + if (event.event === 'checkpoints' && event.id && event.data.next.includes('agent')) { + cursor = event.id; + disconnectedCheckpoint = position(event.data.config.configurable).checkpoint_id; + controller.abort(); + break; + } + } + } catch (error) { + if (!controller.signal.aborted) throw error; + } + if (!runId || !cursor) throw new Error('Disconnected stream must provide a run ID and checkpoint cursor'); + const checkpoints: CheckpointEvent[] = []; + for await (const event of api.runs.joinStream(threadId, runId, { + lastEventId: cursor, cancelOnDisconnect: false, streamMode: ['values', 'checkpoints'], signal: AbortSignal.timeout(20_000), + })) { + expect(event.event).not.toBe('error'); + expect(event.id).not.toBe(cursor); + if (event.event === 'checkpoints') checkpoints.push(event as CheckpointEvent); + } + const final = checkpoints.at(-1); + if (!final) throw new Error('Reconnected stream must emit a checkpoint event'); + const config = final.data.config.configurable; + if (!config) throw new Error('Reconnected checkpoint must include routing configuration'); + expect(config['run_id']).toBe(runId); + // Event metadata does not carry run_id; the retained config and the saved + // checkpoint metadata are the two independently checked run identities. + expect(final.data.metadata['run_id']).toBeUndefined(); + const exactPosition = position(config); + expect(exactPosition.checkpoint_id).not.toBe(disconnectedCheckpoint); + const saved = await api.threads.getState(threadId, exactPosition); + expect(saved.checkpoint.checkpoint_id).toBe(exactPosition.checkpoint_id); + expect(saved.metadata?.['run_id']).toBe(runId); + expect(saved.values).toEqual(final.data.values); + expect(saved.next).toEqual([]); + expect((await api.runs.get(threadId, runId)).status).toBe('success'); + expect(requests.filter((request) => request.method === 'POST' && /^\/threads\/[^/]+\/runs(?:\/stream|\/wait)?$/.test(request.path))) + .toEqual([{ method: 'POST', path: `/threads/${threadId}/runs/stream` }]); + expect(requests).toContainEqual({ method: 'GET', path: `/threads/${threadId}/runs/${runId}/stream` }); + } finally { + controller.abort(); + await api.threads.delete(threadId); + } +}); + +test('checkpoint protocol: a lost write response is ambiguous even with retries disabled', async () => { + const api = client(); + const { thread_id: threadId } = await api.threads.create(); + const accepted: Checkpoint[] = []; + const proxyErrors: unknown[] = []; + let writeRequests = 0; + const proxy = createServer(async (request, response) => { + try { + expect(request.method).toBe('POST'); + expect(request.url).toBe(`/threads/${threadId}/state`); + writeRequests++; + const chunks: Buffer[] = []; + for await (const chunk of request) chunks.push(Buffer.from(chunk)); + const upstream = await fetch(`${backendUrl()}${request.url}`, { + method: 'POST', headers: { 'content-type': 'application/json' }, body: Buffer.concat(chunks), signal: AbortSignal.timeout(10_000), + }); + expect(upstream.status).toBe(200); + const result = await upstream.json() as { configurable: Record }; + accepted.push(position(result.configurable)); + // Upstream has accepted and returned the child position. Only this test + // proxy knows it: the caller receives no successful response to adopt. + response.destroy(); + } catch (error) { + proxyErrors.push(error); + response.writeHead(500).end(); + } + }); + try { + const parent = await run(api, threadId, undefined, input); + await new Promise((resolve, reject) => { + proxy.once('error', reject); + proxy.listen(0, '127.0.0.1', resolve); + }); + const address = proxy.address(); + if (!address || typeof address === 'string') throw new Error('Proxy did not bind a TCP port'); + const caller = new Client({ + apiUrl: `http://127.0.0.1:${address.port}`, apiKey: null, + // Mirrors the owned session default; standalone SDK defaults differ. + callerOptions: { maxRetries: 0 }, + }); + await expect(caller.threads.updateState(threadId, { + checkpoint: position(parent.checkpoint), values: { messages: [] }, asNode: 'agent', signal: AbortSignal.timeout(15_000), + })).rejects.toThrow(); + expect(proxyErrors).toEqual([]); + expect(writeRequests).toBe(1); + expect(accepted).toHaveLength(1); + expect(accepted[0].checkpoint_id).not.toBe(parent.checkpoint.checkpoint_id); + // Read the proxy-observed child, never the thread's global latest tip. + const child = await api.threads.getState(threadId, accepted[0]); + expect(child.checkpoint.checkpoint_id).toBe(accepted[0].checkpoint_id); + expect(child.parent_checkpoint?.checkpoint_id).toBe(parent.checkpoint.checkpoint_id); + expect(child.values).toEqual(parent.values); + expect(child.metadata?.['source']).toBe('update'); + } finally { + proxy.closeAllConnections(); + if (proxy.listening) await new Promise((resolve) => proxy.close(() => resolve())); + await api.threads.delete(threadId); + } +}); diff --git a/cockpit/langgraph/client-tools/angular/e2e/fixtures/checkpoint-protocol.json b/cockpit/langgraph/client-tools/angular/e2e/fixtures/checkpoint-protocol.json new file mode 100644 index 000000000..2164aa5c3 --- /dev/null +++ b/cockpit/langgraph/client-tools/angular/e2e/fixtures/checkpoint-protocol.json @@ -0,0 +1,13 @@ +{ + "fixtures": [ + { + "match": { "userMessage": "Checkpoint protocol weather in Paris", "hasToolResult": true }, + "response": { "content": "Protocol weather complete: Paris is 68°F." } + }, + { + "match": { "userMessage": "Checkpoint protocol weather in Paris" }, + "response": { "toolCalls": [{ "name": "get_weather", "arguments": { "location": "Paris" } }] }, + "latency": 250 + } + ] +} diff --git a/cockpit/langgraph/client-tools/angular/e2e/global-setup-impl.ts b/cockpit/langgraph/client-tools/angular/e2e/global-setup-impl.ts index 26c29df26..194c29c1f 100644 --- a/cockpit/langgraph/client-tools/angular/e2e/global-setup-impl.ts +++ b/cockpit/langgraph/client-tools/angular/e2e/global-setup-impl.ts @@ -4,10 +4,20 @@ import { createGlobalSetup } from '@threadplane-internal/e2e-harness'; const ports = portsFor('cockpit-langgraph-client-tools-angular'); -export default createGlobalSetup({ +const setup = createGlobalSetup({ langgraphCwd: 'cockpit/langgraph/client-tools/python', langgraphPort: ports.langgraph, angularProject: 'cockpit-langgraph-client-tools-angular', angularPort: ports.angular, fixturesDir: resolve(__dirname, 'fixtures'), }); + +export default async function globalSetup(): Promise { + await setup(); + const harness = globalThis.__AIMOCK_HARNESS_STATE__?.get('cockpit-langgraph-client-tools-angular'); + if (!harness) throw new Error('Client-tools harness did not start'); + // Playwright forwards setup environment changes to workers. Expose the local + // API and fixture journal URLs for protocol tests and model request counts. + process.env['CLIENT_TOOLS_API_URL'] = `http://127.0.0.1:${ports.langgraph}`; + process.env['CLIENT_TOOLS_AIMOCK_URL'] = `http://127.0.0.1:${harness.aimock.port}`; +} From 2002ed3ab23eb231ff030f03c6b0319445cc52c7 Mon Sep 17 00:00:00 2001 From: Brian Love Date: Wed, 23 Sep 2026 13:03:31 -0700 Subject: [PATCH 3/3] test(langgraph): require concrete replayed call identities --- .../client-tools/angular/e2e/checkpoint-protocol.spec.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/cockpit/langgraph/client-tools/angular/e2e/checkpoint-protocol.spec.ts b/cockpit/langgraph/client-tools/angular/e2e/checkpoint-protocol.spec.ts index 63ca0f92e..9c53e836c 100644 --- a/cockpit/langgraph/client-tools/angular/e2e/checkpoint-protocol.spec.ts +++ b/cockpit/langgraph/client-tools/angular/e2e/checkpoint-protocol.spec.ts @@ -83,7 +83,7 @@ test('checkpoint protocol: terminal replay preserves a pending call; pre-agent r expect(pending.next).toEqual([]); expect(calls(pending)).toHaveLength(1); const original = calls(pending)[0]; - expect(original).toMatchObject({ id: expect.any(String), name: 'get_weather', args: { location: 'Paris' } }); + expect(original).toMatchObject({ id: expect.stringMatching(/\S/), name: 'get_weather', args: { location: 'Paris' } }); const beforeAgent = (await api.threads.getHistory(threadId)).find((state) => state.next.includes('agent')); if (!beforeAgent) throw new Error('History must include a checkpoint before the agent runs'); expect(calls(beforeAgent)).toEqual([]); @@ -97,7 +97,7 @@ test('checkpoint protocol: terminal replay preserves a pending call; pre-agent r const regenerated = await run(api, threadId, beforeAgent.checkpoint); expect(calls(regenerated)).toHaveLength(1); - expect(calls(regenerated)[0]).toMatchObject({ name: original.name, args: original.args }); + expect(calls(regenerated)[0]).toMatchObject({ id: expect.stringMatching(/\S/), name: original.name, args: original.args }); expect(calls(regenerated)[0].id).not.toBe(original.id); expect(await modelRequests()).toBe(count + 1);