From 4d14bcf5aee0bdd79d11bc7d15af44cfffc0d2b3 Mon Sep 17 00:00:00 2001 From: william Date: Mon, 2 Mar 2026 17:19:04 +0100 Subject: [PATCH 1/3] feat(checkpoint-postgres)!: align version format and inline primitive storage with Python BREAKING: Align checkpoint version format and channel_values storage with the Python implementation. Breaking changes: 1. Version format: getNextVersion now produces zero-padded string versions (e.g. "00000000000000000000000000000001.0482910384729105") instead of integer versions (1, 2, 3). Migration impact: Existing checkpoints with integer versions will still be readable. New checkpoints will use string versions. 2. Inline primitives: Primitive channel values (string, number, boolean, null) are now stored inline in the checkpoint JSONB column instead of in checkpoint_blobs. Requires PR #1967 (read-side merge) to be deployed first, so existing readers can handle both formats. These changes enable cross-compatibility between Python and JS checkpoint implementations sharing the same database, required for hybrid Python/JS LangGraph deployments. --- libs/checkpoint-postgres/src/index.ts | 94 +++++++++++++++++++++------ 1 file changed, 74 insertions(+), 20 deletions(-) diff --git a/libs/checkpoint-postgres/src/index.ts b/libs/checkpoint-postgres/src/index.ts index 3c180e918b..77e0f24d50 100644 --- a/libs/checkpoint-postgres/src/index.ts +++ b/libs/checkpoint-postgres/src/index.ts @@ -81,7 +81,7 @@ const { Pool } = pg; * }, config); * ``` */ -export class PostgresSaver extends BaseCheckpointSaver { +export class PostgresSaver extends BaseCheckpointSaver { private readonly pool: pg.Pool; private readonly options: PostgresSaverOptions; @@ -175,13 +175,39 @@ export class PostgresSaver extends BaseCheckpointSaver { } } + /** + * Generate the next version ID for a channel. + * + * Uses zero-padded string versions with random suffix for compatibility + * with the Python checkpoint-postgres implementation. + */ + getNextVersion( + current: string | number | undefined + ): string | number { + if (current === undefined) { + return `${"0".repeat(31)}1.${Math.random().toString().slice(2, 18).padEnd(16, "0")}`; + } + const currentStr = String(current); + const currentV = parseInt(currentStr.split(".")[0], 10) || 0; + const nextV = currentV + 1; + const nextH = Math.random().toString().slice(2, 18).padEnd(16, "0"); + return `${String(nextV).padStart(32, "0")}.${nextH}`; + } + protected async _loadCheckpoint( - checkpoint: Omit, + checkpoint: Omit & { + channel_values?: Record; + }, channelValues: [Uint8Array, Uint8Array, Uint8Array][] ): Promise { return { ...checkpoint, - channel_values: await this._loadBlobs(channelValues), + channel_values: { + // Merge inline primitives from checkpoint JSONB + ...(checkpoint.channel_values || {}), + // Blob values override inline primitives + ...(await this._loadBlobs(channelValues)), + }, }; } @@ -235,27 +261,55 @@ export class PostgresSaver extends BaseCheckpointSaver { return []; } - return Promise.all( - Object.entries(versions).map(async ([k, ver]) => { - const [type, value] = - k in values - ? await this.serde.dumpsTyped(values[k]) - : ["empty", null]; - return [ - threadId, - checkpointNs, - k, - ver.toString(), - type, - value ? new Uint8Array(value) : undefined, - ]; - }) - ); + const results: [string, string, string, string, string, Uint8Array | undefined][] = []; + for (const [k, ver] of Object.entries(versions)) { + // Skip primitive values — they are stored inline in the checkpoint JSONB + if (k in values) { + const v = values[k]; + if ( + v === null || + typeof v === "string" || + typeof v === "number" || + typeof v === "boolean" + ) { + continue; + } + } + const [type, value] = + k in values + ? await this.serde.dumpsTyped(values[k]) + : ["empty", null]; + results.push([ + threadId, + checkpointNs, + k, + ver.toString(), + type, + value ? new Uint8Array(value) : undefined, + ]); + } + return results; } protected _dumpCheckpoint(checkpoint: Checkpoint) { const serialized: Record = { ...checkpoint }; - if ("channel_values" in serialized) delete serialized.channel_values; + // Extract inline primitives from channel_values and store them + // in the checkpoint JSONB column. Complex values are stored in + // checkpoint_blobs table separately. + const inlinePrimitives: Record = {}; + if (checkpoint.channel_values) { + for (const [key, value] of Object.entries(checkpoint.channel_values)) { + if ( + value === null || + typeof value === "string" || + typeof value === "number" || + typeof value === "boolean" + ) { + inlinePrimitives[key] = value; + } + } + } + serialized.channel_values = inlinePrimitives; return serialized; } From f646f7ad4f05463056e4fd82188e8c3d0cac79cb Mon Sep 17 00:00:00 2001 From: william Date: Tue, 24 Feb 2026 10:16:03 +0100 Subject: [PATCH 2/3] test(checkpoint-postgres): add unit tests for version format and inline primitive storage Add unit tests covering the changes introduced in this PR: - getNextVersion: format validation, counter increment, backward compat with integer versions, uniqueness, string sorting, Python format parity - _dumpCheckpoint: inline primitive extraction (string, number, boolean, null) vs complex value exclusion - _dumpBlobs: primitive skipping, complex value serialization, mixed value handling - _loadCheckpoint: inline + blob merge, blob-wins-on-collision, backward compat with missing inline values --- libs/checkpoint-postgres/src/index.ts | 22 +- .../src/tests/postgres-saver.test.ts | 401 ++++++++++++++++++ 2 files changed, 415 insertions(+), 8 deletions(-) create mode 100644 libs/checkpoint-postgres/src/tests/postgres-saver.test.ts diff --git a/libs/checkpoint-postgres/src/index.ts b/libs/checkpoint-postgres/src/index.ts index 77e0f24d50..e9fae4597b 100644 --- a/libs/checkpoint-postgres/src/index.ts +++ b/libs/checkpoint-postgres/src/index.ts @@ -181,11 +181,12 @@ export class PostgresSaver extends BaseCheckpointSaver { * Uses zero-padded string versions with random suffix for compatibility * with the Python checkpoint-postgres implementation. */ - getNextVersion( - current: string | number | undefined - ): string | number { + getNextVersion(current: string | number | undefined): string | number { if (current === undefined) { - return `${"0".repeat(31)}1.${Math.random().toString().slice(2, 18).padEnd(16, "0")}`; + return `${"0".repeat(31)}1.${Math.random() + .toString() + .slice(2, 18) + .padEnd(16, "0")}`; } const currentStr = String(current); const currentV = parseInt(currentStr.split(".")[0], 10) || 0; @@ -261,7 +262,14 @@ export class PostgresSaver extends BaseCheckpointSaver { return []; } - const results: [string, string, string, string, string, Uint8Array | undefined][] = []; + const results: [ + string, + string, + string, + string, + string, + Uint8Array | undefined + ][] = []; for (const [k, ver] of Object.entries(versions)) { // Skip primitive values — they are stored inline in the checkpoint JSONB if (k in values) { @@ -276,9 +284,7 @@ export class PostgresSaver extends BaseCheckpointSaver { } } const [type, value] = - k in values - ? await this.serde.dumpsTyped(values[k]) - : ["empty", null]; + k in values ? await this.serde.dumpsTyped(values[k]) : ["empty", null]; results.push([ threadId, checkpointNs, diff --git a/libs/checkpoint-postgres/src/tests/postgres-saver.test.ts b/libs/checkpoint-postgres/src/tests/postgres-saver.test.ts new file mode 100644 index 0000000000..6a207e2a11 --- /dev/null +++ b/libs/checkpoint-postgres/src/tests/postgres-saver.test.ts @@ -0,0 +1,401 @@ +import { describe, it, expect } from "vitest"; +import { PostgresSaver } from "../index.js"; + +/** + * Helper to create a PostgresSaver with a mocked pg.Pool. + * Unit tests don't need a real database connection. + */ +function createSaver() { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const mockPool = {} as any; + return new PostgresSaver(mockPool); +} + +/** + * Access protected methods for testing via a thin subclass. + */ +class TestableSaver extends PostgresSaver { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + constructor() { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + super({} as any); + } + + public testDumpCheckpoint( + ...args: Parameters + ) { + return this._dumpCheckpoint(...args); + } + + public testDumpBlobs(...args: Parameters) { + return this._dumpBlobs(...args); + } + + public testLoadCheckpoint( + ...args: Parameters + ) { + return this._loadCheckpoint(...args); + } +} + +// ─── getNextVersion ────────────────────────────────────────────────────────── + +describe("getNextVersion", () => { + const saver = createSaver(); + + it("should produce a string with 32-char zero-padded counter and 16-char hash", () => { + const version = saver.getNextVersion(undefined); + expect(typeof version).toBe("string"); + + const str = version as string; + const [counter, hash] = str.split("."); + expect(counter).toHaveLength(32); + expect(hash).toHaveLength(16); + }); + + it("should start at counter 1 when current is undefined", () => { + const version = saver.getNextVersion(undefined) as string; + const counter = version.split(".")[0]; + expect(parseInt(counter, 10)).toBe(1); + // Should be zero-padded + expect(counter).toMatch(/^0{31}1$/); + }); + + it("should increment the counter from a string version", () => { + const v1 = saver.getNextVersion(undefined) as string; + const v2 = saver.getNextVersion(v1) as string; + + const c1 = parseInt(v1.split(".")[0], 10); + const c2 = parseInt(v2.split(".")[0], 10); + expect(c2).toBe(c1 + 1); + }); + + it("should handle legacy integer versions (backward compat)", () => { + // Old checkpoints stored integer versions like 1, 2, 3 + const version = saver.getNextVersion(3) as string; + const [counter, hash] = version.split("."); + expect(parseInt(counter, 10)).toBe(4); + expect(counter).toHaveLength(32); + expect(hash).toHaveLength(16); + }); + + it("should produce unique versions (different random hash each call)", () => { + const versions = new Set(); + for (let i = 0; i < 100; i += 1) { + versions.add(saver.getNextVersion(undefined) as string); + } + // With random hashes, all 100 should be unique + expect(versions.size).toBe(100); + }); + + it("should produce monotonically increasing versions by counter", () => { + let current: string | number | undefined; + for (let i = 0; i < 10; i += 1) { + const next = saver.getNextVersion(current) as string; + const nextCounter = parseInt(next.split(".")[0], 10); + expect(nextCounter).toBe(i + 1); + current = next; + } + }); + + it("should produce versions that sort correctly as strings", () => { + // This is critical for Postgres ORDER BY to work correctly + const v1 = saver.getNextVersion(undefined) as string; + const v2 = saver.getNextVersion(v1) as string; + const v3 = saver.getNextVersion(v2) as string; + + expect(v1 < v2).toBe(true); + expect(v2 < v3).toBe(true); + }); + + it("should produce format compatible with Python checkpoint-postgres", () => { + // Python format: f"{next_v:032}.{next_h:016}" + // The JS version should match this pattern + const version = saver.getNextVersion(undefined) as string; + // Full format: 32 digits, dot, 16 digits + expect(version).toMatch(/^\d{32}\.\d{16}$/); + }); +}); + +// ─── _dumpCheckpoint ───────────────────────────────────────────────────────── + +describe("_dumpCheckpoint", () => { + const saver = new TestableSaver(); + + it("should inline primitive channel_values (string, number, boolean, null)", () => { + const checkpoint = { + v: 4, + id: "test-id", + ts: "2024-01-01T00:00:00Z", + channel_values: { + strVal: "hello", + numVal: 42, + boolVal: true, + nullVal: null, + }, + channel_versions: {}, + versions_seen: {}, + }; + + const result = saver.testDumpCheckpoint(checkpoint); + expect(result.channel_values).toEqual({ + strVal: "hello", + numVal: 42, + boolVal: true, + nullVal: null, + }); + }); + + it("should exclude complex values from inline channel_values", () => { + const checkpoint = { + v: 4, + id: "test-id", + ts: "2024-01-01T00:00:00Z", + channel_values: { + strVal: "hello", + arrayVal: [1, 2, 3], + objVal: { nested: true }, + }, + channel_versions: {}, + versions_seen: {}, + }; + + const result = saver.testDumpCheckpoint(checkpoint); + // Only the primitive is inlined + expect(result.channel_values).toEqual({ + strVal: "hello", + }); + // Complex values should NOT be in the serialized checkpoint + const cv = result.channel_values as Record; + expect(cv.arrayVal).toBeUndefined(); + expect(cv.objVal).toBeUndefined(); + }); + + it("should handle empty channel_values", () => { + const checkpoint = { + v: 4, + id: "test-id", + ts: "2024-01-01T00:00:00Z", + channel_values: {}, + channel_versions: {}, + versions_seen: {}, + }; + + const result = saver.testDumpCheckpoint(checkpoint); + expect(result.channel_values).toEqual({}); + }); + + it("should preserve other checkpoint fields", () => { + const checkpoint = { + v: 4, + id: "test-id", + ts: "2024-01-01T00:00:00Z", + channel_values: { key: "value" }, + channel_versions: { key: 1 }, + versions_seen: { node: { key: 1 } }, + }; + + const result = saver.testDumpCheckpoint(checkpoint); + expect(result.v).toBe(4); + expect(result.id).toBe("test-id"); + expect(result.ts).toBe("2024-01-01T00:00:00Z"); + expect(result.channel_versions).toEqual({ key: 1 }); + expect(result.versions_seen).toEqual({ node: { key: 1 } }); + }); +}); + +// ─── _dumpBlobs ────────────────────────────────────────────────────────────── + +describe("_dumpBlobs", () => { + const saver = new TestableSaver(); + + it("should skip primitive values (stored inline in checkpoint)", async () => { + const values = { + strKey: "hello", + numKey: 42, + boolKey: true, + nullKey: null, + }; + const versions = { + strKey: "00000000000000000000000000000001.1234567890123456", + numKey: "00000000000000000000000000000001.2345678901234567", + boolKey: "00000000000000000000000000000001.3456789012345678", + nullKey: "00000000000000000000000000000001.4567890123456789", + }; + + const result = await saver.testDumpBlobs("thread-1", "", values, versions); + + // All values are primitives, so nothing should be dumped to blobs + expect(result).toHaveLength(0); + }); + + it("should include complex values in blob output", async () => { + const values = { + arrayKey: [1, 2, 3], + objKey: { nested: true }, + }; + const versions = { + arrayKey: "00000000000000000000000000000001.1234567890123456", + objKey: "00000000000000000000000000000001.2345678901234567", + }; + + const result = await saver.testDumpBlobs("thread-1", "", values, versions); + + expect(result).toHaveLength(2); + // Each blob tuple: [threadId, checkpointNs, key, version, type, bytes] + const keys = result.map((r) => r[2]); + expect(keys).toContain("arrayKey"); + expect(keys).toContain("objKey"); + }); + + it("should handle mix of primitive and complex values", async () => { + const values = { + strKey: "hello", + arrayKey: [1, 2, 3], + numKey: 42, + objKey: { nested: true }, + }; + const versions = { + strKey: "v1", + arrayKey: "v1", + numKey: "v1", + objKey: "v1", + }; + + const result = await saver.testDumpBlobs("thread-1", "", values, versions); + + // Only complex values should be in blobs + expect(result).toHaveLength(2); + const keys = result.map((r) => r[2]); + expect(keys).toContain("arrayKey"); + expect(keys).toContain("objKey"); + expect(keys).not.toContain("strKey"); + expect(keys).not.toContain("numKey"); + }); + + it("should return empty array when versions is empty", async () => { + const result = await saver.testDumpBlobs( + "thread-1", + "", + { key: "val" }, + {} + ); + expect(result).toEqual([]); + }); + + it("should handle channels with version but missing from values (empty type)", async () => { + const values = {}; + const versions = { missingKey: "v1" }; + + const result = await saver.testDumpBlobs("thread-1", "", values, versions); + + expect(result).toHaveLength(1); + expect(result[0][2]).toBe("missingKey"); + expect(result[0][4]).toBe("empty"); + expect(result[0][5]).toBeUndefined(); + }); +}); + +// ─── _loadCheckpoint ───────────────────────────────────────────────────────── + +describe("_loadCheckpoint", () => { + const saver = new TestableSaver(); + + it("should merge inline primitives from checkpoint with blob values", async () => { + const checkpoint = { + v: 4, + id: "test-id", + ts: "2024-01-01T00:00:00Z", + channel_values: { + inlineKey: "inlineValue", + }, + channel_versions: {}, + versions_seen: {}, + }; + + const encoder = new TextEncoder(); + // Simulate a blob for a complex value — use JSON serde format + const channelValues: [Uint8Array, Uint8Array, Uint8Array][] = [ + [ + encoder.encode("blobKey"), + encoder.encode("json"), + encoder.encode(JSON.stringify({ complex: true })), + ], + ]; + + const result = await saver.testLoadCheckpoint(checkpoint, channelValues); + + // Both inline and blob values should be present + expect(result.channel_values.inlineKey).toBe("inlineValue"); + expect(result.channel_values.blobKey).toEqual({ complex: true }); + }); + + it("should let blob values override inline primitives on collision", async () => { + const checkpoint = { + v: 4, + id: "test-id", + ts: "2024-01-01T00:00:00Z", + channel_values: { + key: "inlineValue", + }, + channel_versions: {}, + versions_seen: {}, + }; + + const encoder = new TextEncoder(); + const channelValues: [Uint8Array, Uint8Array, Uint8Array][] = [ + [ + encoder.encode("key"), + encoder.encode("json"), + encoder.encode(JSON.stringify("blobValue")), + ], + ]; + + const result = await saver.testLoadCheckpoint(checkpoint, channelValues); + // Blob should win + expect(result.channel_values.key).toBe("blobValue"); + }); + + it("should work with empty blob values", async () => { + const checkpoint = { + v: 4, + id: "test-id", + ts: "2024-01-01T00:00:00Z", + channel_values: { + key: "value", + }, + channel_versions: {}, + versions_seen: {}, + }; + + const result = await saver.testLoadCheckpoint(checkpoint, []); + expect(result.channel_values.key).toBe("value"); + }); + + it("should work with no inline channel_values", async () => { + const checkpoint = { + v: 4, + id: "test-id", + ts: "2024-01-01T00:00:00Z", + // No channel_values field (pre-inline-primitive checkpoints) + channel_versions: {}, + versions_seen: {}, + }; + + const encoder = new TextEncoder(); + const channelValues: [Uint8Array, Uint8Array, Uint8Array][] = [ + [ + encoder.encode("blobKey"), + encoder.encode("json"), + encoder.encode(JSON.stringify([1, 2, 3])), + ], + ]; + + const result = await saver.testLoadCheckpoint( + checkpoint as Parameters[0], + channelValues + ); + + expect(result.channel_values.blobKey).toEqual([1, 2, 3]); + }); +}); From 7a85bb76711554aff6a6b51803481b6fac0928db Mon Sep 17 00:00:00 2001 From: william Date: Mon, 2 Mar 2026 17:18:28 +0100 Subject: [PATCH 3/3] chore: add changeset for version format alignment --- .changeset/align-version-format-python.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 .changeset/align-version-format-python.md diff --git a/.changeset/align-version-format-python.md b/.changeset/align-version-format-python.md new file mode 100644 index 0000000000..a977aa2a2c --- /dev/null +++ b/.changeset/align-version-format-python.md @@ -0,0 +1,13 @@ +--- +"@langchain/langgraph-checkpoint-postgres": minor +--- + +feat(checkpoint-postgres)!: align version format and inline primitive storage with Python + +Align checkpoint version format and channel_values storage with the Python implementation: + +1. **Version format:** `getNextVersion` now produces zero-padded string versions (e.g. `"00000000000000000000000000000001.0482910384729105"`) instead of integer versions (`1`, `2`, `3`). + +2. **Inline primitives:** Primitive channel values (`string`, `number`, `boolean`, `null`) are now stored inline in the checkpoint JSONB column instead of in `checkpoint_blobs`. + +These changes enable cross-compatibility between Python and JS checkpoint implementations sharing the same database, required for hybrid Python/JS LangGraph deployments.