test(langgraph): provide dedicated benchmark snapshot builder

Build secret-free snapshots from a pinned public revision and align file transfers with the LangSmith sandbox SDK.

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

libs/langgraph/bench/delta-distributed.md

@@ -0,0 +1,153 @@
+# Distributed DeltaChannel benchmark
+
+The coordinator (`delta_distributed.py`) creates one disposable LangSmith sandbox
+per `(frequency, payload)` pair: 40 shards for the default 25–500 step-25 grid.
+Each shard owns a local Postgres process and grows one graph to 10000 updates,
+recording the five prefix lengths. `GenericFakeChatModel` scripts are deterministic;
+no provider API is used. Parallelism changes wall time, not total benchmark work.
+
+## Snapshot contract
+
+Supply an existing, immutable LangSmith snapshot with:
+
+- Linux, Python 3.14, git, PostgreSQL 14 binaries (`initdb`, `pg_ctl`) on PATH;
+- a checkout at `/workspace/langgraph`, pinned to the intended revision;
+- dependencies installed using `uv sync --frozen --group dev` in
+  `/workspace/langgraph/libs/langgraph`;
+- `/workspace/langgraph/libs/langgraph/.venv/bin/python` available;
+- a writable `/tmp`; a non-root user able to run Postgres, or root plus `runuser`
+  and the `postgres` OS account;
+- no Postgres listener already occupying loopback port 5432;
+- **no secrets baked into the snapshot**. Credentials are needed only by the
+  coordinator, not the shard. Use a dedicated snapshot, not a personal workspace
+  image with unrelated processes or credentials.
+
+The coordinator uploads only the benchmark source and shard worker. Libraries,
+interpreter and dependency versions come from the supplied snapshot. Every job
+uses that same snapshot ID and explicit CPU/memory/disk configuration. Snapshot
+construction/provisioning is intentionally separate; no image or paid fleet has
+been created by this change. This is a normal standalone benchmark project using
+the LangSmith SDK, not a change to the Open SWE workspace image.
+
+## Create the snapshot
+
+`delta_snapshot.py` builds the required image from the dedicated
+`delta-snapshot/Dockerfile` context. Only that Dockerfile is uploaded, never your
+working directory, `.env`, git credentials or LangSmith key. It uses Postgres
+14.7, uv 0.13.4 and Python 3.14.3, checks out a full public LangGraph commit SHA,
+and syncs the frozen dev dependencies. OS packages and version-tagged base images
+are not digest/apt-repository pinned; reuse the resulting immutable snapshot ID
+for comparable measurements rather than rebuilding between shards.
+
+From `libs/langgraph`, after checking out this PR and syncing dependencies:
+
+```bash
+uv run python bench/delta_snapshot.py \
+  --name delta-bench --output ./snapshot-plan.json
+```
+
+This dry run creates no resources. By default, the revision is local HEAD: it must
+already be pushed and reachable in the public repository. Uncommitted local code
+is **not** included. For a paid build, use a fresh receipt path:
+
+```bash
+uv run python bench/delta_snapshot.py \
+  --name delta-bench --output ./snapshot.json --execute
+```
+
+Optional flags: `--revision FULL_40_CHAR_SHA --vcpus 2 --memory-gib 8 --disk-gib 30
+--timeout 1800`. The defaults are builder sizing, not fleet concurrency. The SDK
+creates one temporary billable builder sandbox and a persistent snapshot, using
+`LANGSMITH_API_KEY`/`LANGSMITH_ENDPOINT` only on the coordinator. It cleans up the
+builder when the call exits; verify cleanup in LangSmith after a crash or failed
+build. The snapshot remains until you explicitly delete it. This command does not
+launch benchmark shards or alter an Open SWE workspace image.
+
+On success, `snapshot.json` contains `snapshot_id`, revision and build parameters:
+
+```bash
+export SNAPSHOT_ID="$(python -c 'import json; print(json.load(open("snapshot.json"))["snapshot_id"])')"
+```
+
+Then use `--snapshot-id "$SNAPSHOT_ID"` in the pilot command below. Build failures
+leave a `building` receipt: inspect LangSmith before retrying with a new name/path;
+do not assume a timeout means no snapshot or builder was created. Remote image
+build and pilot have not yet been executed; local SDK-contract tests are not a
+substitute for that end-to-end verification.
+
+## Plan first
+
+From `libs/langgraph`:
+
+```bash
+uv run python bench/delta_distributed.py \
+  --snapshot-id YOUR_SNAPSHOT_ID --concurrency 4 --timeout 14400 \
+  --vcpus 2 --memory-gib 8 --disk-gib 30 --output /tmp/delta-fleet
+```
+
+Without `--execute`, this only writes `plan.json`; it does not instantiate an SDK
+client, require credentials or provision resources. Values above are **examples,
+not selected budgets**. Agree on them before executing. With 40 shards and a
+4-hour benchmark timeout, the benchmark allowance alone is 160 sandbox-hours;
+readiness and result collection add overhead. The printed allowance is an
+estimate, not a monetary billing cap. Concurrency is capped at 40 and each shard
+benchmark timeout at 24 hours. There is no automatic retry storm.
+
+Start with a small pilot (new output directory):
+
+```bash
+uv run python bench/delta_distributed.py \
+  --snapshot-id YOUR_SNAPSHOT_ID --concurrency 2 --timeout 900 \
+  --vcpus 2 --memory-gib 8 --disk-gib 30 --output /tmp/delta-pilot \
+  --frequencies 25 300 --lengths 50 200 --profiles mock700
+```
+
+Review the generated plan, then repeat the exact command with `--execute` to
+create billable resources. The SDK reads `LANGSMITH_API_KEY` and
+`LANGSMITH_ENDPOINT` from the coordinator's environment. Never pass keys in CLI
+arguments, commit them, or bake them into the snapshot.
+
+## Lifecycle, recovery and outputs
+
+- Bounded thread-pool concurrency; each worker has its own SDK client.
+- Readiness timeout 300 seconds, benchmark subprocess timeout from the plan,
+  and remote execution timeout 120 seconds longer for startup/cleanup.
+- Completed CSV and metadata are fetched before sandbox deletion. A shard is
+  marked successful only after its sandbox has been deleted.
+- A shard's `attempt.json` records its sandbox name **before** creation. On a
+  process crash, ambiguous create or failed deletion, inspect that named resource
+  in LangSmith and delete it if it exists. Do not bulk-delete other sandboxes.
+  After verifying it is absent, set this ledger's state to `deleted` to allow
+  retry. An unresolved ledger blocks retry rather than leaking more resources.
+- Idle stop is 10 minutes and deletion after stop is 10 minutes, providing a
+  fallback for coordinator crashes. This is **not** a strict hard lifetime cap:
+  verify sandbox cleanup after interruption. Active exec/SDK lease behavior can
+  affect idle detection.
+- A shard failure stops admission of further jobs; already-running jobs finish
+  and clean up. Exceptions shown by workers use only their type, avoiding
+  inadvertent credential exposure. Failed command stdout/stderr are not copied
+  into shared artifacts. Inspect the sandbox before cleanup separately if deep
+  diagnostics are needed; automatic cleanup prioritizes preventing spend.
+- Rerun the identical plan/output directory to skip successful shards. Failed
+  shards restart from zero; partial remote CSVs are not recovered. Each completed
+  shard is an independent resume unit.
+- Launch **only one coordinator per output directory**; no cross-process lock is
+  implemented. Do not modify benchmark/worker source during a run. Source hashes
+  and plan equality reject changed-source or changed-budget resume.
+- Merge requires every expected frequency/profile/length exactly once. Missing,
+  duplicate, foreign and nonfinite rows fail validation. `combined.csv` and SVG
+  plots are emitted only for a complete plan. Per-shard CSV/metadata stay available.
+
+## Interpretation and verification limits
+
+See [measurement semantics](delta-results/README.md) before interpreting results.
+The normalized Pareto compromise is a heuristic, not a universal default. Full
+state hydration and mock payload compression can dominate the apparent tradeoff.
+Keep the end-to-end graph as the reference; direct checkpointer seeding is not
+implemented because equivalence has not been demonstrated. Read sampling remains
+one read per update to avoid treating assumed snapshot phase as measured truth.
+
+The coordinator lifecycle and failure paths are locally tested with a fake SDK;
+the benchmark itself has run against local Postgres. A real remote pilot still
+needs a prepared snapshot and an agreed concurrency/runtime budget. Do not treat
+mocked orchestration tests as evidence that a remote fleet has run successfully.

libs/langgraph/bench/delta-results/README.md

@@ -0,0 +1,124 @@
+# DeltaChannel snapshot-frequency sweep
+
+## Status and recommendation
+
+The sequential Postgres grid was stopped in favor of the distributed coordinator.
+`postgres.csv` contains four completed prefixes from frequency 25 / mock700, not a
+complete grid. No default change is justified by these partial measurements alone.
+Until the long-thread grid finishes, retain existing defaults rather than selecting a
+frequency from short threads. Frequency 300 remains a candidate for both LangGraph
+and deepagents based on the supplied fleet evidence, not a conclusion of this run.
+The capped grid does not benchmark LangGraph's current 1000 setting.
+
+`pilot.csv` contains the initial 25/300/500-frequency, 50/200-update exploratory run.
+Its first read includes initialization overhead; do not use that maximum as a steady
+state result. `scaling-pilot.csv` contains 1000-update runs at frequency 300 with
+warmup. These are explicitly partial grids. `postgres.csv` contains completed points
+from the requested full grid as it progresses; missing rows are not measurements.
+
+## Reproduce
+
+From `libs/langgraph`, sync declared dependencies with `uv sync --frozen --group dev`.
+Use a **dedicated disposable Postgres database**, not a production database. For
+example, start `postgres:14.7` with a loopback-only published port and configure
+`LANGGRAPH_BENCH_POSTGRES_URI` in the environment. The benchmark creates checkpointer
+tables and deletes only its UUID-scoped threads when each scenario finishes.
+
+```bash
+uv run python tests/test_delta_channel_benchmark.py --snapshot-sweep \
+  --backend postgres --output bench/delta-results/postgres.csv
+uv run python tests/test_delta_channel_benchmark.py --snapshot-sweep \
+  --backend memory --output bench/delta-results/memory.csv
+```
+
+Default grid: frequencies 25, 50, ..., 500; lengths 50, 200, 1000, 5000,
+10000; profiles `mock700` and `tool_calls`. Select a smoke grid with
+`--frequencies 25 300 500 --lengths 50 200`. The original multi-channel benchmark
+remains available by omitting `--snapshot-sweep`.
+
+Every frequency/profile builds one growing thread; shorter lengths are prefixes.
+Thus each backend performs 20 × 2 × 10000 = 400000 graph invocations and the
+same number of measured reads, plus warmups and separate heap measurements.
+There are 200 output rows per complete backend grid. Resume with identical
+arguments, source, revision and environment metadata: completed rows are skipped.
+An interrupted scenario is rebuilt from its beginning up to remaining lengths;
+resuming does not restore a partially written graph. Abrupt process termination
+may leave the current thread in the disposable database.
+
+## Workload and measurement semantics
+
+- Each invocation runs a deterministic `GenericFakeChatModel` script. There are
+  **no provider requests**, sleeps or stochastic outputs. Message and tool-call
+  IDs derive from the update index. Checkpoint IDs and timing vary between runs.
+- `mock700`: 350 ASCII bytes of human content plus 350 of assistant content.
+  This is **700 content bytes**, not 700 serialized bytes: message envelopes,
+  checkpoint metadata and a file update add overhead.
+- `tool_calls`: human request, scripted assistant `read_file` call, approximately
+  3.7 KB tool result, then an approximately 600-byte scripted response. It is a
+  synthetic deepagents-style workload, not captured production traffic.
+- Both `messages` and `files` receive one update in the same node per invocation.
+  Files use deterministic content lines and timestamps, adding a new report path
+  each time. This deliberately models growing file history, not overwriting a
+  fixed set of paths. The reducer merges batches of file dictionaries.
+- An update is **one graph invocation**, not one individual message. Each
+  invocation has input/start/node checkpoint supersteps; only the node writes
+  the two delta channels. The 5000-superstep bound is retained and recorded via
+  `LANGGRAPH_DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT`; the grid does not vary it.
+- Read statistics describe one warm-cache `get_state` after **every** invocation
+  from update 1 through the reported prefix. They are not repeated final-state
+  reads or a fixed-thread-length phase distribution. This captures pre-snapshot
+  phases as well as snapshots; the maximum is the **observed** slowest read and
+  may reflect system noise, not necessarily the deepest replay chain. Its update
+  and checkpoint ID are recorded. The measured checkpoint is later deleted.
+- Write statistics time end-to-end `graph.invoke`, including hydration, scripted
+  model CPU work and checkpoint writes; they are **not saver-only SQL latency**.
+- Heap peak comes from a separate traced read of the observed slowest checkpoint.
+  `tracemalloc` is off for all latency samples. This is Python heap, not RSS or
+  database memory. Reads and writes are sequential, with no concurrent clients.
+- Memory storage counts serialized checkpoint, metadata, blob and pending-write
+  payload bytes, not Python-object/key overhead. Postgres storage is the sum of
+  `pg_column_size(row)` across checkpoints, blobs and writes scoped to the thread.
+  It is an in-row/TOAST-sensitive estimate, **not total relation/index/WAL/disk
+  usage**, and is not directly interchangeable with the memory payload count.
+  Repeated mock text is highly compressible, especially in Postgres: absolute
+  storage numbers should not be compared with production chat without matching
+  entropy, serialization and storage accounting.
+- Metadata records interpreter, platform, git revision, script SHA-256, grid and
+  sampling semantics. Initial pilot artifacts predate the final script hash and
+  warmup additions; they are preserved as exploratory evidence, not pooled with
+  the full-grid results. Timings need repeat runs to assess variance.
+
+## Charts and selecting a frequency
+
+Each CSV produces an SVG per backend/profile/length. The left panel plots storage,
+p50 and maximum read latency against frequency, independently divided by each
+metric's maximum (0–1). These normalized curves **do not have shared units**;
+their crossing is not a physical optimum. The right panel plots storage bytes
+against actual p50/max milliseconds.
+
+A marked compromise point is selected separately for p50 and max: discard
+storage/latency-dominated points, min-max normalize each objective over that group,
+and choose the nondominated point nearest the ideal origin in Euclidean distance.
+This is an equal-weight heuristic, not a universal knee or recommended default.
+Changing the candidate range, latency statistic or storage/latency weighting can
+change it. Partial groups display their actual number of frequencies and must not
+be mistaken for completed sweeps. Frequency points above a short thread's update
+count can have identical snapshot behavior, making any apparent short-thread
+optimum particularly noisy.
+
+For a final recommendation, compare the compromise points across all five thread
+lengths and both workloads, apply an explicit latency budget, inspect tail noise,
+and repeat promising frequencies. Keep separate choices for growing message and
+file workloads if their access/update patterns differ. This suite currently varies
+both cadences together and cannot identify their independent optimum.
+
+## Fleet sanity check
+
+The supplied 10000-turn fleet simulation reports frequency 50: 2.9 GB / 44 ms;
+300: 488 MB / 244 ms; 500: 307 MB / 404 ms; 1000: 164 MB / 804 ms. Its delta DB time
+was about 35 ms and about 1% of fleet DB time. These are externally supplied
+context, **not measurements reproduced here**. Our local Postgres, full graph
+hydration, files, compressible scripted content and row-size estimate differ.
+Do not calibrate the results to force a match or equate local end-to-end latency
+with fleet database time. Storage reduction with less frequent snapshots is a
+useful directional sanity check, not evidence that absolute fleet values match.

libs/langgraph/bench/delta-results/pilot-postgres-mock700-200.svg

@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" width="1100" height="420" viewBox="0 0 1100 420"><rect width="1100" height="420" fill="white"/><text x="20" y="25" font-family="sans-serif" font-size="16">postgres / mock700 / 200 updates (n=3 frequencies)</text><path d="M 60 60 V 340 H 510" fill="none" stroke="black"/><text x="60" y="50" font-family="sans-serif" font-size="12">Frequency vs independently normalized metrics</text><polyline points="60.00,80.00 320.53,86.47 510.00,86.57" fill="none" stroke="#176baf" stroke-width="2"/><text x="60" y="365" fill="#176baf" font-family="sans-serif" font-size="11">storage_bytes</text><polyline points="60.00,161.64 320.53,123.69 510.00,80.00" fill="none" stroke="#c14b14" stroke-width="2"/><text x="60" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="60.00,80.00 320.53,293.43 510.00,289.76" fill="none" stroke="#6a378b" stroke-width="2"/><text x="60" y="395" fill="#6a378b" font-family="sans-serif" font-size="11">read_max_ms</text><text x="60" y="355" font-family="sans-serif" font-size="10">frequency: 25 to 500; y: 0 to 1.00 normalized</text><path d="M 610 60 V 340 H 1060" fill="none" stroke="black"/><text x="610" y="50" font-family="sans-serif" font-size="12">Storage vs latency (Pareto tradeoff)</text><polyline points="1060.00,327.56 616.92,324.92 610.00,321.87" fill="none" stroke="#176baf" stroke-width="2"/><circle cx="616.9150288807352" cy="324.91598843036314" r="5" fill="#176baf"/><text x="622.9150288807352" y="316.91598843036314" font-family="sans-serif" font-size="11">f=300</text><text x="610" y="365" fill="#176baf" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="1060.00,80.00 616.92,293.43 610.00,289.76" fill="none" stroke="#c14b14" stroke-width="2"/><circle cx="616.9150288807352" cy="293.42895568807285" r="5" fill="#c14b14"/><text x="622.9150288807352" y="285.42895568807285" font-family="sans-serif" font-size="11">f=300</text><text x="610" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_max_ms</text><text x="610" y="355" font-family="sans-serif" font-size="10">storage_bytes: 1062349 to 1089876; y: 0 to 197.73 ms</text></svg>

libs/langgraph/bench/delta-results/pilot-postgres-mock700-50.svg

@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" width="1100" height="420" viewBox="0 0 1100 420"><rect width="1100" height="420" fill="white"/><text x="20" y="25" font-family="sans-serif" font-size="16">postgres / mock700 / 50 updates (n=3 frequencies)</text><path d="M 60 60 V 340 H 510" fill="none" stroke="black"/><text x="60" y="50" font-family="sans-serif" font-size="12">Frequency vs independently normalized metrics</text><polyline points="60.00,80.00 320.53,82.82 510.00,82.87" fill="none" stroke="#176baf" stroke-width="2"/><text x="60" y="365" fill="#176baf" font-family="sans-serif" font-size="11">storage_bytes</text><polyline points="60.00,153.18 320.53,134.27 510.00,80.00" fill="none" stroke="#c14b14" stroke-width="2"/><text x="60" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="60.00,80.00 320.53,325.79 510.00,329.51" fill="none" stroke="#6a378b" stroke-width="2"/><text x="60" y="395" fill="#6a378b" font-family="sans-serif" font-size="11">read_max_ms</text><text x="60" y="355" font-family="sans-serif" font-size="10">frequency: 25 to 500; y: 0 to 1.00 normalized</text><path d="M 610 60 V 340 H 1060" fill="none" stroke="black"/><text x="610" y="50" font-family="sans-serif" font-size="12">Storage vs latency (Pareto tradeoff)</text><polyline points="1060.00,335.24 617.01,334.76 610.00,333.38" fill="none" stroke="#176baf" stroke-width="2"/><circle cx="617.005076142132" cy="334.7610466522907" r="5" fill="#176baf"/><text x="623.005076142132" y="326.7610466522907" font-family="sans-serif" font-size="11">f=300</text><text x="610" y="365" fill="#176baf" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="1060.00,80.00 617.01,325.79 610.00,329.51" fill="none" stroke="#c14b14" stroke-width="2"/><circle cx="610.0" cy="329.50511620062144" r="5" fill="#c14b14"/><text x="616.0" y="321.50511620062144" font-family="sans-serif" font-size="11">f=500</text><text x="610" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_max_ms</text><text x="610" y="355" font-family="sans-serif" font-size="10">storage_bytes: 265018 to 267973; y: 0 to 197.73 ms</text></svg>

libs/langgraph/bench/delta-results/pilot-postgres-tool_calls-200.svg

@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" width="1100" height="420" viewBox="0 0 1100 420"><rect width="1100" height="420" fill="white"/><text x="20" y="25" font-family="sans-serif" font-size="16">postgres / tool_calls / 200 updates (n=3 frequencies)</text><path d="M 60 60 V 340 H 510" fill="none" stroke="black"/><text x="60" y="50" font-family="sans-serif" font-size="12">Frequency vs independently normalized metrics</text><polyline points="60.00,80.00 320.53,176.58 510.00,176.63" fill="none" stroke="#176baf" stroke-width="2"/><text x="60" y="365" fill="#176baf" font-family="sans-serif" font-size="11">storage_bytes</text><polyline points="60.00,153.76 320.53,80.00 510.00,118.12" fill="none" stroke="#c14b14" stroke-width="2"/><text x="60" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="60.00,80.00 320.53,171.19 510.00,179.72" fill="none" stroke="#6a378b" stroke-width="2"/><text x="60" y="395" fill="#6a378b" font-family="sans-serif" font-size="11">read_max_ms</text><text x="60" y="355" font-family="sans-serif" font-size="10">frequency: 25 to 500; y: 0 to 1.00 normalized</text><path d="M 610 60 V 340 H 1060" fill="none" stroke="black"/><text x="610" y="50" font-family="sans-serif" font-size="12">Storage vs latency (Pareto tradeoff)</text><polyline points="1060.00,288.02 610.24,267.43 610.00,278.07" fill="none" stroke="#176baf" stroke-width="2"/><circle cx="610.0" cy="278.0659453883102" r="5" fill="#176baf"/><text x="616.0" y="270.0659453883102" font-family="sans-serif" font-size="11">f=500</text><text x="610" y="365" fill="#176baf" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="1060.00,80.00 610.24,171.19 610.00,179.72" fill="none" stroke="#c14b14" stroke-width="2"/><circle cx="610.0" cy="179.717130949707" r="5" fill="#c14b14"/><text x="616.0" y="171.717130949707" font-family="sans-serif" font-size="11">f=500</text><text x="610" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_max_ms</text><text x="610" y="355" font-family="sans-serif" font-size="10">storage_bytes: 1008416 to 1604911; y: 0 to 77.92 ms</text></svg>

libs/langgraph/bench/delta-results/pilot-postgres-tool_calls-50.svg

@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" width="1100" height="420" viewBox="0 0 1100 420"><rect width="1100" height="420" fill="white"/><text x="20" y="25" font-family="sans-serif" font-size="16">postgres / tool_calls / 50 updates (n=3 frequencies)</text><path d="M 60 60 V 340 H 510" fill="none" stroke="black"/><text x="60" y="50" font-family="sans-serif" font-size="12">Frequency vs independently normalized metrics</text><polyline points="60.00,80.00 320.53,123.32 510.00,123.41" fill="none" stroke="#176baf" stroke-width="2"/><text x="60" y="365" fill="#176baf" font-family="sans-serif" font-size="11">storage_bytes</text><polyline points="60.00,80.00 320.53,92.09 510.00,96.96" fill="none" stroke="#c14b14" stroke-width="2"/><text x="60" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="60.00,80.00 320.53,124.55 510.00,172.29" fill="none" stroke="#6a378b" stroke-width="2"/><text x="60" y="395" fill="#6a378b" font-family="sans-serif" font-size="11">read_max_ms</text><text x="60" y="355" font-family="sans-serif" font-size="10">frequency: 25 to 500; y: 0 to 1.00 normalized</text><path d="M 610 60 V 340 H 1060" fill="none" stroke="black"/><text x="610" y="50" font-family="sans-serif" font-size="12">Storage vs latency (Pareto tradeoff)</text><polyline points="1060.00,240.59 610.96,245.21 610.00,247.07" fill="none" stroke="#176baf" stroke-width="2"/><circle cx="610.0" cy="247.07101749792133" r="5" fill="#176baf"/><text x="616.0" y="239.07101749792133" font-family="sans-serif" font-size="11">f=500</text><text x="610" y="365" fill="#176baf" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="1060.00,80.00 610.96,124.55 610.00,172.29" fill="none" stroke="#c14b14" stroke-width="2"/><circle cx="610.0" cy="172.28887735058402" r="5" fill="#c14b14"/><text x="616.0" y="164.28887735058402" font-family="sans-serif" font-size="11">f=500</text><text x="610" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_max_ms</text><text x="610" y="355" font-family="sans-serif" font-size="10">storage_bytes: 251530 to 301946; y: 0 to 17.90 ms</text></svg>

libs/langgraph/bench/delta-results/pilot.csv

@@ -0,0 +1,13 @@
+backend,profile,frequency,updates,read_samples,read_p50_ms,read_p95_ms,read_max_ms,slowest_update,slowest_checkpoint_id,write_p50_ms,write_p95_ms,write_max_ms,write_total_s,storage_bytes,slowest_read_peak_heap_bytes
+postgres,mock700,25,50,50,3.6180599999511287,6.390288000034161,197.7286589999494,1,1f1cf166-9dd2-6d00-8001-fc71abcfb90a,10.653302000037002,17.0728529999451,30.98109600000498,0.5811482070006377,267973,39327
+postgres,mock700,25,200,200,9.459105500070564,18.783542000051057,197.7286589999494,1,1f1cf166-9dd2-6d00-8001-fc71abcfb90a,17.477463499972146,30.98109600000498,168.95307799995862,3.8713906139998926,1089876,35220
+postgres,mock700,300,50,50,3.9841970000225047,6.104986000082135,10.80296499992528,44,1f1cf166-e019-6c30-8082-21934933f621,11.811886500026958,15.252276999945025,28.05887800002438,0.6044334389998767,265064,519513
+postgres,mock700,300,200,200,11.47131300001547,20.832925000036084,35.41703900009452,185,1f1cf167-0ea3-6a4c-8229-79b49eb7c9fe,19.905765000032716,28.92650700005106,37.699279999969804,3.95925508099981,1062772,2113173
+postgres,mock700,500,50,50,5.035277500041957,7.819561000019348,7.981305000043903,50,1f1cf167-1fc9-6f84-8094-7f8f188214b6,12.75971849997859,15.276644999971722,16.033585999934985,0.6406766549997656,265018,584506
+postgres,mock700,500,200,200,13.788552499988782,24.058938000052876,38.209428999948614,185,1f1cf167-5380-6b9e-8229-c84900f5a6db,21.95349900000565,32.21689399993011,39.65822100008154,4.354335682999022,1062349,2110890
+postgres,tool_calls,25,50,50,6.844815500016921,10.651568999946903,17.9017149999936,50,1f1cf167-6761-6981-8094-f53aeefe50f8,14.597708000053444,19.39097000001766,25.841881000019384,0.7417658820006636,301946,1922124
+postgres,tool_calls,25,200,200,15.580024999962916,28.78641699999207,77.9230529999495,180,1f1cf167-a207-6b8c-821a-1a5a322fa361,24.694384999975227,42.27366899999652,59.241608999968776,5.190335378001123,1604911,6679384
+postgres,tool_calls,300,50,50,6.526570999994874,13.39240200002223,14.834163999921657,50,1f1cf167-bca7-6808-8094-f7b7aea71f3d,14.862395000022843,21.3141539999242,43.96138599997812,0.808315155999594,251638,1910246
+postgres,tool_calls,300,200,200,21.750415999974848,33.799963999967986,50.59318900009657,86,1f1cf167-ce64-6010-8100-78d863dad859,31.301175500004774,48.37651699995149,165.45825800005787,6.2137166139996225,1008731,3276689
+postgres,tool_calls,500,50,50,6.39841599996771,10.550998999974581,11.547371999995448,49,1f1cf168-2162-68cd-8091-9494a54b4fdd,14.285579000045345,19.514858999968965,23.103371000047446,0.704547747000106,251530,1872876
+postgres,tool_calls,500,200,200,18.561887000032584,33.225987000037094,48.03742499996133,200,1f1cf168-70e7-6f69-8256-b1395c31fdf1,26.90513599998212,43.38257900008102,53.06229500001791,5.458913677999817,1008416,7571456

libs/langgraph/bench/delta-results/pilot.metadata.json

@@ -0,0 +1,25 @@
+{
+  "schema": 1,
+  "python": "3.14.3",
+  "platform": "Linux-6.1.166-x86_64-with-glibc2.43",
+  "revision": "eb066bd31a2bf01fc0cd58256df4800a3695c233",
+  "backend": "postgres",
+  "profiles": [
+    "mock700",
+    "tool_calls"
+  ],
+  "frequencies": [
+    25,
+    300,
+    500
+  ],
+  "lengths": [
+    50,
+    200
+  ],
+  "script": "deterministic, no provider; IDs follow update index",
+  "read_sampling": "one get_state after every invocation, prefix distribution",
+  "write_sampling": "end-to-end graph.invoke including hydrate, scripted model and checkpoint I/O",
+  "storage": "postgres pg_column_size row estimate (not relation disk bytes); memory serialized payloads across all stores",
+  "max_supersteps_since_snapshot": "5000"
+}

libs/langgraph/bench/delta-results/postgres-postgres-mock700-1000.svg

@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" width="1100" height="420" viewBox="0 0 1100 420"><rect width="1100" height="420" fill="white"/><text x="20" y="25" font-family="sans-serif" font-size="16">postgres / mock700 / 1000 updates (n=1 frequencies)</text><path d="M 60 60 V 340 H 510" fill="none" stroke="black"/><text x="60" y="50" font-family="sans-serif" font-size="12">Frequency vs independently normalized metrics</text><polyline points="60.00,80.00" fill="none" stroke="#176baf" stroke-width="2"/><text x="60" y="365" fill="#176baf" font-family="sans-serif" font-size="11">storage_bytes</text><polyline points="60.00,80.00" fill="none" stroke="#c14b14" stroke-width="2"/><text x="60" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="60.00,80.00" fill="none" stroke="#6a378b" stroke-width="2"/><text x="60" y="395" fill="#6a378b" font-family="sans-serif" font-size="11">read_max_ms</text><text x="60" y="355" font-family="sans-serif" font-size="10">frequency: 25 to 25; y: 0 to 1.00 normalized</text><path d="M 610 60 V 340 H 1060" fill="none" stroke="black"/><text x="610" y="50" font-family="sans-serif" font-size="12">Storage vs latency (Pareto tradeoff)</text><polyline points="610.00,270.42" fill="none" stroke="#176baf" stroke-width="2"/><circle cx="610.0" cy="270.41705212026505" r="5" fill="#176baf"/><text x="616.0" y="262.41705212026505" font-family="sans-serif" font-size="11">f=25</text><text x="610" y="365" fill="#176baf" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="610.00,80.00" fill="none" stroke="#c14b14" stroke-width="2"/><circle cx="610.0" cy="80.0" r="5" fill="#c14b14"/><text x="616.0" y="72.0" font-family="sans-serif" font-size="11">f=25</text><text x="610" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_max_ms</text><text x="610" y="355" font-family="sans-serif" font-size="10">storage_bytes: 5864394 to 5864394; y: 0 to 88.08 ms</text></svg>

libs/langgraph/bench/delta-results/postgres-postgres-mock700-200.svg

@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" width="1100" height="420" viewBox="0 0 1100 420"><rect width="1100" height="420" fill="white"/><text x="20" y="25" font-family="sans-serif" font-size="16">postgres / mock700 / 200 updates (n=1 frequencies)</text><path d="M 60 60 V 340 H 510" fill="none" stroke="black"/><text x="60" y="50" font-family="sans-serif" font-size="12">Frequency vs independently normalized metrics</text><polyline points="60.00,80.00" fill="none" stroke="#176baf" stroke-width="2"/><text x="60" y="365" fill="#176baf" font-family="sans-serif" font-size="11">storage_bytes</text><polyline points="60.00,80.00" fill="none" stroke="#c14b14" stroke-width="2"/><text x="60" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="60.00,80.00" fill="none" stroke="#6a378b" stroke-width="2"/><text x="60" y="395" fill="#6a378b" font-family="sans-serif" font-size="11">read_max_ms</text><text x="60" y="355" font-family="sans-serif" font-size="10">frequency: 25 to 25; y: 0 to 1.00 normalized</text><path d="M 610 60 V 340 H 1060" fill="none" stroke="black"/><text x="610" y="50" font-family="sans-serif" font-size="12">Storage vs latency (Pareto tradeoff)</text><polyline points="610.00,225.40" fill="none" stroke="#176baf" stroke-width="2"/><circle cx="610.0" cy="225.39966722149467" r="5" fill="#176baf"/><text x="616.0" y="217.39966722149467" font-family="sans-serif" font-size="11">f=25</text><text x="610" y="365" fill="#176baf" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="610.00,80.00" fill="none" stroke="#c14b14" stroke-width="2"/><circle cx="610.0" cy="80.0" r="5" fill="#c14b14"/><text x="616.0" y="72.0" font-family="sans-serif" font-size="11">f=25</text><text x="610" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_max_ms</text><text x="610" y="355" font-family="sans-serif" font-size="10">storage_bytes: 1089890 to 1089890; y: 0 to 16.54 ms</text></svg>

libs/langgraph/bench/delta-results/postgres-postgres-mock700-50.svg

@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" width="1100" height="420" viewBox="0 0 1100 420"><rect width="1100" height="420" fill="white"/><text x="20" y="25" font-family="sans-serif" font-size="16">postgres / mock700 / 50 updates (n=1 frequencies)</text><path d="M 60 60 V 340 H 510" fill="none" stroke="black"/><text x="60" y="50" font-family="sans-serif" font-size="12">Frequency vs independently normalized metrics</text><polyline points="60.00,80.00" fill="none" stroke="#176baf" stroke-width="2"/><text x="60" y="365" fill="#176baf" font-family="sans-serif" font-size="11">storage_bytes</text><polyline points="60.00,80.00" fill="none" stroke="#c14b14" stroke-width="2"/><text x="60" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="60.00,80.00" fill="none" stroke="#6a378b" stroke-width="2"/><text x="60" y="395" fill="#6a378b" font-family="sans-serif" font-size="11">read_max_ms</text><text x="60" y="355" font-family="sans-serif" font-size="10">frequency: 25 to 25; y: 0 to 1.00 normalized</text><path d="M 610 60 V 340 H 1060" fill="none" stroke="black"/><text x="610" y="50" font-family="sans-serif" font-size="12">Storage vs latency (Pareto tradeoff)</text><polyline points="610.00,184.65" fill="none" stroke="#176baf" stroke-width="2"/><circle cx="610.0" cy="184.6472608451244" r="5" fill="#176baf"/><text x="616.0" y="176.6472608451244" font-family="sans-serif" font-size="11">f=25</text><text x="610" y="365" fill="#176baf" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="610.00,80.00" fill="none" stroke="#c14b14" stroke-width="2"/><circle cx="610.0" cy="80.0" r="5" fill="#c14b14"/><text x="616.0" y="72.0" font-family="sans-serif" font-size="11">f=25</text><text x="610" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_max_ms</text><text x="610" y="355" font-family="sans-serif" font-size="10">storage_bytes: 267967 to 267967; y: 0 to 5.21 ms</text></svg>

libs/langgraph/bench/delta-results/postgres-postgres-mock700-5000.svg

@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" width="1100" height="420" viewBox="0 0 1100 420"><rect width="1100" height="420" fill="white"/><text x="20" y="25" font-family="sans-serif" font-size="16">postgres / mock700 / 5000 updates (n=1 frequencies)</text><path d="M 60 60 V 340 H 510" fill="none" stroke="black"/><text x="60" y="50" font-family="sans-serif" font-size="12">Frequency vs independently normalized metrics</text><polyline points="60.00,80.00" fill="none" stroke="#176baf" stroke-width="2"/><text x="60" y="365" fill="#176baf" font-family="sans-serif" font-size="11">storage_bytes</text><polyline points="60.00,80.00" fill="none" stroke="#c14b14" stroke-width="2"/><text x="60" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="60.00,80.00" fill="none" stroke="#6a378b" stroke-width="2"/><text x="60" y="395" fill="#6a378b" font-family="sans-serif" font-size="11">read_max_ms</text><text x="60" y="355" font-family="sans-serif" font-size="10">frequency: 25 to 25; y: 0 to 1.00 normalized</text><path d="M 610 60 V 340 H 1060" fill="none" stroke="black"/><text x="610" y="50" font-family="sans-serif" font-size="12">Storage vs latency (Pareto tradeoff)</text><polyline points="610.00,274.30" fill="none" stroke="#176baf" stroke-width="2"/><circle cx="610.0" cy="274.2962318424321" r="5" fill="#176baf"/><text x="616.0" y="266.2962318424321" font-family="sans-serif" font-size="11">f=25</text><text x="610" y="365" fill="#176baf" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="610.00,80.00" fill="none" stroke="#c14b14" stroke-width="2"/><circle cx="610.0" cy="80.0" r="5" fill="#c14b14"/><text x="616.0" y="72.0" font-family="sans-serif" font-size="11">f=25</text><text x="610" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_max_ms</text><text x="610" y="355" font-family="sans-serif" font-size="10">storage_bytes: 39599985 to 39599985; y: 0 to 302.30 ms</text></svg>

libs/langgraph/bench/delta-results/postgres.csv

@@ -0,0 +1,5 @@
+backend,profile,frequency,updates,read_samples,read_p50_ms,read_p95_ms,read_max_ms,slowest_update,slowest_checkpoint_id,write_p50_ms,write_p95_ms,write_max_ms,write_total_s,storage_bytes,slowest_read_peak_heap_bytes
+postgres,mock700,25,50,50,3.110554999921078,4.785336999930223,5.2058580001812516,46,1f1cf189-c1d7-643f-8088-2d4ca5a5b740,10.314074000007167,12.664818999837735,27.913941999941017,0.53071240700001,267967,483221
+postgres,mock700,25,200,200,7.288281000001007,13.18430499986789,16.53531900001326,190,1f1cf189-e494-60d8-8238-d44b3e1f2bf0,15.244594500018138,22.50665300016408,37.84130200006075,3.0723227170001337,1089890,1720217
+postgres,mock700,25,1000,1000,23.5726189999923,34.56026000003476,88.08021400000143,785,1f1cf18b-25f4-6d27-8931-fc21b750ad8c,32.47065949994976,51.13759799996842,134.2724809999254,31.385343082000187,5864394,5889532
+postgres,mock700,25,5000,5000,76.39274549990205,140.11053600006562,302.2979410000062,3412,1f1cf199-7e0d-6560-a7fa-9687f00ad7cc,90.85022650003793,170.46306499969432,414.5206420002978,459.44821525399425,39599985,22671874

libs/langgraph/bench/delta-results/postgres.metadata.json

@@ -0,0 +1,46 @@
+{
+  "schema": 1,
+  "python": "3.14.3",
+  "platform": "Linux-6.1.166-x86_64-with-glibc2.43",
+  "script_sha256": "3a3cb80dde575f1f008c9078e285cc2e8f8a3bb8648da4c3817cb33f9e9a4aed",
+  "revision": "eb066bd31a2bf01fc0cd58256df4800a3695c233",
+  "backend": "postgres",
+  "profiles": [
+    "mock700",
+    "tool_calls"
+  ],
+  "frequencies": [
+    25,
+    50,
+    75,
+    100,
+    125,
+    150,
+    175,
+    200,
+    225,
+    250,
+    275,
+    300,
+    325,
+    350,
+    375,
+    400,
+    425,
+    450,
+    475,
+    500
+  ],
+  "lengths": [
+    50,
+    200,
+    1000,
+    5000,
+    10000
+  ],
+  "script": "deterministic, no provider; IDs follow update index",
+  "read_sampling": "one get_state after every invocation, prefix distribution",
+  "write_sampling": "end-to-end graph.invoke including hydrate, scripted model and checkpoint I/O",
+  "storage": "postgres pg_column_size row estimate (not relation disk bytes); memory serialized payloads across all stores",
+  "max_supersteps_since_snapshot": "5000"
+}

libs/langgraph/bench/delta-results/scaling-pilot-postgres-mock700-1000.svg

@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" width="1100" height="420" viewBox="0 0 1100 420"><rect width="1100" height="420" fill="white"/><text x="20" y="25" font-family="sans-serif" font-size="16">postgres / mock700 / 1000 updates (n=1 frequencies)</text><path d="M 60 60 V 340 H 510" fill="none" stroke="black"/><text x="60" y="50" font-family="sans-serif" font-size="12">Frequency vs independently normalized metrics</text><polyline points="60.00,80.00" fill="none" stroke="#176baf" stroke-width="2"/><text x="60" y="365" fill="#176baf" font-family="sans-serif" font-size="11">storage_bytes</text><polyline points="60.00,80.00" fill="none" stroke="#c14b14" stroke-width="2"/><text x="60" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="60.00,80.00" fill="none" stroke="#6a378b" stroke-width="2"/><text x="60" y="395" fill="#6a378b" font-family="sans-serif" font-size="11">read_max_ms</text><text x="60" y="355" font-family="sans-serif" font-size="10">frequency: 300 to 300; y: 0 to 1.00 normalized</text><path d="M 610 60 V 340 H 1060" fill="none" stroke="black"/><text x="610" y="50" font-family="sans-serif" font-size="12">Storage vs latency (Pareto tradeoff)</text><polyline points="610.00,278.82" fill="none" stroke="#176baf" stroke-width="2"/><circle cx="610.0" cy="278.815545790787" r="5" fill="#176baf"/><text x="616.0" y="270.815545790787" font-family="sans-serif" font-size="11">f=300</text><text x="610" y="365" fill="#176baf" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="610.00,80.00" fill="none" stroke="#c14b14" stroke-width="2"/><circle cx="610.0" cy="80.0" r="5" fill="#c14b14"/><text x="616.0" y="72.0" font-family="sans-serif" font-size="11">f=300</text><text x="610" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_max_ms</text><text x="610" y="355" font-family="sans-serif" font-size="10">storage_bytes: 5365127 to 5365127; y: 0 to 123.55 ms</text></svg>

libs/langgraph/bench/delta-results/scaling-pilot-postgres-tool_calls-1000.svg

@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" width="1100" height="420" viewBox="0 0 1100 420"><rect width="1100" height="420" fill="white"/><text x="20" y="25" font-family="sans-serif" font-size="16">postgres / tool_calls / 1000 updates (n=1 frequencies)</text><path d="M 60 60 V 340 H 510" fill="none" stroke="black"/><text x="60" y="50" font-family="sans-serif" font-size="12">Frequency vs independently normalized metrics</text><polyline points="60.00,80.00" fill="none" stroke="#176baf" stroke-width="2"/><text x="60" y="365" fill="#176baf" font-family="sans-serif" font-size="11">storage_bytes</text><polyline points="60.00,80.00" fill="none" stroke="#c14b14" stroke-width="2"/><text x="60" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="60.00,80.00" fill="none" stroke="#6a378b" stroke-width="2"/><text x="60" y="395" fill="#6a378b" font-family="sans-serif" font-size="11">read_max_ms</text><text x="60" y="355" font-family="sans-serif" font-size="10">frequency: 300 to 300; y: 0 to 1.00 normalized</text><path d="M 610 60 V 340 H 1060" fill="none" stroke="black"/><text x="610" y="50" font-family="sans-serif" font-size="12">Storage vs latency (Pareto tradeoff)</text><polyline points="610.00,281.99" fill="none" stroke="#176baf" stroke-width="2"/><circle cx="610.0" cy="281.9919586818079" r="5" fill="#176baf"/><text x="616.0" y="273.9919586818079" font-family="sans-serif" font-size="11">f=300</text><text x="610" y="365" fill="#176baf" font-family="sans-serif" font-size="11">read_p50_ms</text><polyline points="610.00,80.00" fill="none" stroke="#c14b14" stroke-width="2"/><circle cx="610.0" cy="80.0" r="5" fill="#c14b14"/><text x="616.0" y="72.0" font-family="sans-serif" font-size="11">f=300</text><text x="610" y="380" fill="#c14b14" font-family="sans-serif" font-size="11">read_max_ms</text><text x="610" y="355" font-family="sans-serif" font-size="10">storage_bytes: 6230423 to 6230423; y: 0 to 261.54 ms</text></svg>

libs/langgraph/bench/delta-results/scaling-pilot.csv

@@ -0,0 +1,3 @@
+backend,profile,frequency,updates,read_samples,read_p50_ms,read_p95_ms,read_max_ms,slowest_update,slowest_checkpoint_id,write_p50_ms,write_p95_ms,write_max_ms,write_total_s,storage_bytes,slowest_read_peak_heap_bytes
+postgres,mock700,300,1000,1000,29.075356000021202,49.26721499998621,123.55413900002077,339,1f1cf16e-fccf-68b6-83f7-54e1839f7866,38.3128815000191,57.815339000057975,117.20152399993822,37.95603520900272,5365127,3107507
+postgres,tool_calls,300,1000,1000,58.35261599997921,106.53731800005062,261.5444310000612,900,1f1cf175-2376-6bf5-8a8a-ce7f009ad20f,70.02236149997998,122.92945799993049,376.99114200006534,71.72656674900168,6230423,34206365

libs/langgraph/bench/delta-results/scaling-pilot.metadata.json

@@ -0,0 +1,23 @@
+{
+  "schema": 1,
+  "python": "3.14.3",
+  "platform": "Linux-6.1.166-x86_64-with-glibc2.43",
+  "script_sha256": "3a3cb80dde575f1f008c9078e285cc2e8f8a3bb8648da4c3817cb33f9e9a4aed",
+  "revision": "eb066bd31a2bf01fc0cd58256df4800a3695c233",
+  "backend": "postgres",
+  "profiles": [
+    "mock700",
+    "tool_calls"
+  ],
+  "frequencies": [
+    300
+  ],
+  "lengths": [
+    1000
+  ],
+  "script": "deterministic, no provider; IDs follow update index",
+  "read_sampling": "one get_state after every invocation, prefix distribution",
+  "write_sampling": "end-to-end graph.invoke including hydrate, scripted model and checkpoint I/O",
+  "storage": "postgres pg_column_size row estimate (not relation disk bytes); memory serialized payloads across all stores",
+  "max_supersteps_since_snapshot": "5000"
+}

libs/langgraph/bench/delta-snapshot/Dockerfile

@@ -0,0 +1,19 @@
+FROM ghcr.io/astral-sh/uv:0.13.4 AS uv
+FROM postgres:14.7
+COPY --from=uv /uv /usr/local/bin/uv
+ENV UV_PYTHON_INSTALL_DIR=/opt/python UV_PYTHON_PREFERENCE=only-managed
+RUN apt-get update && apt-get install -y --no-install-recommends git ca-certificates \
+    && rm -rf /var/lib/apt/lists/*
+RUN uv python install 3.14.3
+ARG LANGGRAPH_REVISION
+RUN test "${#LANGGRAPH_REVISION}" -eq 40 \
+    && git clone https://github.com/langchain-ai/langgraph.git /workspace/langgraph \
+    && git -C /workspace/langgraph checkout --detach "$LANGGRAPH_REVISION" \
+    && test "$(git -C /workspace/langgraph rev-parse HEAD)" = "$LANGGRAPH_REVISION"
+WORKDIR /workspace/langgraph/libs/langgraph
+RUN uv sync --frozen --group dev --python 3.14.3 \
+    && uv cache clean \
+    && .venv/bin/python -c 'from langsmith.sandbox import SandboxClient; from langgraph.checkpoint.postgres import PostgresSaver' \
+    && command -v initdb && command -v pg_ctl
+ENTRYPOINT []
+CMD ["/bin/bash"]

libs/langgraph/bench/delta_distributed.py

@@ -0,0 +1,264 @@
+"""Plan, execute and merge isolated LangSmith sandbox benchmark shards."""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import hashlib
+import importlib.util
+import io
+import itertools
+import json
+import shlex
+import threading
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from typing import Any
+
+from langsmith.sandbox import SandboxClient
+
+ROOT = Path(__file__).resolve().parents[1]
+BENCHMARK = ROOT / "tests/test_delta_channel_benchmark.py"
+REMOTE = "/workspace/langgraph/libs/langgraph"
+PROFILES = ("mock700", "tool_calls")
+FREQUENCIES = tuple(range(25, 501, 25))
+LENGTHS = (50, 200, 1000, 5000, 10000)
+
+
+def _atomic_json(path: Path, value: Any) -> None:
+    temp = path.with_suffix(".tmp")
+    temp.write_text(json.dumps(value, indent=2) + "\n")
+    temp.replace(path)
+
+
+def make_plan(args: argparse.Namespace) -> dict:
+    """Bind every shard to source, snapshot, resource and runtime settings."""
+    return {
+        "schema": 1,
+        "snapshot_id": args.snapshot_id,
+        "concurrency": args.concurrency,
+        "timeout_seconds": args.timeout,
+        "vcpus": args.vcpus,
+        "mem_bytes": args.memory_gib * 1024**3,
+        "fs_capacity_bytes": args.disk_gib * 1024**3,
+        "source_sha256": hashlib.sha256(BENCHMARK.read_bytes()).hexdigest(),
+        "worker_sha256": hashlib.sha256(
+            Path(__file__).with_name("delta_shard.py").read_bytes()
+        ).hexdigest(),
+        "lengths": sorted(set(args.lengths)),
+        "shards": [
+            {"key": f"{profile}-{freq}", "profile": profile, "frequency": freq}
+            for profile, freq in itertools.product(args.profiles, args.frequencies)
+        ],
+    }
+
+
+def validate_csv(data: bytes, shard: dict, lengths: list[int]) -> list[dict]:
+    """Reject missing, duplicate, foreign or malformed measurement points."""
+    rows = list(csv.DictReader(io.StringIO(data.decode())))
+    actual = []
+    for row in rows:
+        if (
+            row["backend"] != "postgres"
+            or row["profile"] != shard["profile"]
+            or int(row["frequency"]) != shard["frequency"]
+        ):
+            raise ValueError("Foreign shard row")
+        actual.append(int(row["updates"]))
+        for metric in ("read_p50_ms", "read_p95_ms", "read_max_ms", "storage_bytes"):
+            value = float(row[metric])
+            if not 0 <= value < float("inf"):
+                raise ValueError("Nonfinite or negative metric")
+    if sorted(actual) != sorted(lengths):
+        raise ValueError("Incomplete or duplicate shard lengths")
+    return rows
+
+
+def run_shard(
+    plan: dict, shard: dict, directory: Path, client_factory=SandboxClient
+) -> None:
+    """Use a disposable sandbox and save validated outputs before deleting it."""
+    target = directory / shard["key"]
+    target.mkdir(exist_ok=True)
+    success = target / "success.json"
+    if success.exists():
+        validate_csv((target / "results.csv").read_bytes(), shard, plan["lengths"])
+        if json.loads(success.read_text())["source_sha256"] != plan["source_sha256"]:
+            raise ValueError("Completed shard source differs")
+        return
+    ledger = target / "attempt.json"
+    if ledger.exists() and json.loads(ledger.read_text())["state"] != "deleted":
+        raise RuntimeError(
+            f"Inspect and clean up the previous sandbox in {ledger} before retrying"
+        )
+    sandbox_name = f"delta-bench-{uuid.uuid4().hex}"
+    _atomic_json(ledger, {"sandbox_name": sandbox_name, "state": "creating"})
+    with client_factory() as client:
+        sandbox = None
+        completed = False
+        try:
+            sandbox = client.create_sandbox(
+                snapshot_id=plan["snapshot_id"],
+                name=sandbox_name,
+                timeout=300,
+                idle_ttl_seconds=600,
+                delete_after_stop_seconds=600,
+                vcpus=plan["vcpus"],
+                mem_bytes=plan["mem_bytes"],
+                fs_capacity_bytes=plan["fs_capacity_bytes"],
+            )
+            _atomic_json(ledger, {"sandbox_name": sandbox_name, "state": "running"})
+            sandbox.write(
+                f"{REMOTE}/tests/test_delta_channel_benchmark.py",
+                BENCHMARK.read_bytes(),
+            )
+            sandbox.write(
+                "/tmp/delta_shard.py",
+                Path(__file__).with_name("delta_shard.py").read_bytes(),
+            )
+            sandbox.write(
+                "/tmp/delta-job.json",
+                json.dumps(
+                    {
+                        **shard,
+                        "lengths": plan["lengths"],
+                        "timeout_seconds": plan["timeout_seconds"],
+                    }
+                ),
+            )
+            result = sandbox.run(
+                shlex.join([f"{REMOTE}/.venv/bin/python", "/tmp/delta_shard.py"]),
+                timeout=plan["timeout_seconds"] + 120,
+                cwd=REMOTE,
+                kill_on_disconnect=True,
+            )
+            if result.exit_code != 0:
+                raise RuntimeError(
+                    "Shard command failed; sandbox logs withheld from shared output"
+                )
+            data = sandbox.read("/tmp/delta-results.csv")
+            metadata = json.loads(sandbox.read("/tmp/delta-results.metadata.json"))
+            if metadata["script_sha256"] != plan["source_sha256"]:
+                raise ValueError("Remote benchmark source differs")
+            validate_csv(data, shard, plan["lengths"])
+            (target / "results.csv.tmp").write_bytes(data)
+            (target / "results.csv.tmp").replace(target / "results.csv")
+            _atomic_json(target / "metadata.json", metadata)
+            completed = True
+        finally:
+            try:
+                client.delete_sandbox(sandbox_name)
+            except Exception:
+                _atomic_json(
+                    ledger, {"sandbox_name": sandbox_name, "state": "cleanup_required"}
+                )
+                raise RuntimeError(f"Sandbox cleanup required; see {ledger}") from None
+            _atomic_json(ledger, {"sandbox_name": sandbox_name, "state": "deleted"})
+        if completed:
+            _atomic_json(
+                success,
+                {"source_sha256": plan["source_sha256"], "sandbox_name": sandbox_name},
+            )
+
+
+def merge(plan: dict, directory: Path) -> Path:
+    """Require the entire planned grid before publishing a combined CSV."""
+    rows = []
+    for shard in plan["shards"]:
+        target = directory / shard["key"]
+        if not (target / "success.json").exists():
+            raise ValueError(f"Shard incomplete: {shard['key']}")
+        rows.extend(
+            validate_csv((target / "results.csv").read_bytes(), shard, plan["lengths"])
+        )
+    output = directory / "combined.csv"
+    temp = output.with_suffix(".tmp")
+    with temp.open("w", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=list(rows[0]), lineterminator="\n")
+        writer.writeheader()
+        writer.writerows(rows)
+    temp.replace(output)
+    spec = importlib.util.spec_from_file_location("delta_benchmark", BENCHMARK)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    module._plot_sweep(output)
+    return output
+
+
+def execute(plan: dict, directory: Path, client_factory=SandboxClient) -> None:
+    """Bound active sandboxes and stop admitting shards after a failure."""
+    failed = threading.Event()
+
+    def run(shard: dict) -> None:
+        if failed.is_set():
+            return
+        try:
+            run_shard(plan, shard, directory, client_factory)
+        except Exception as exc:
+            failed.set()
+            print(f"{shard['key']}: {type(exc).__name__}; see shard ledger", flush=True)
+            raise
+
+    with ThreadPoolExecutor(max_workers=plan["concurrency"]) as pool:
+        list(pool.map(run, plan["shards"]))
+    merge(plan, directory)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--snapshot-id", required=True)
+    parser.add_argument("--concurrency", type=int, required=True)
+    parser.add_argument(
+        "--timeout", type=int, required=True, help="Maximum benchmark seconds per shard"
+    )
+    parser.add_argument("--vcpus", type=int, required=True)
+    parser.add_argument("--memory-gib", type=int, required=True)
+    parser.add_argument("--disk-gib", type=int, required=True)
+    parser.add_argument("--frequencies", nargs="+", type=int, default=list(FREQUENCIES))
+    parser.add_argument("--lengths", nargs="+", type=int, default=list(LENGTHS))
+    parser.add_argument(
+        "--profiles", nargs="+", choices=PROFILES, default=list(PROFILES)
+    )
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument(
+        "--execute",
+        action="store_true",
+        help="Create billable sandboxes; otherwise only write a plan",
+    )
+    args = parser.parse_args()
+    if any(
+        n <= 0
+        for n in [
+            args.concurrency,
+            args.timeout,
+            args.vcpus,
+            args.memory_gib,
+            args.disk_gib,
+            *args.frequencies,
+            *args.lengths,
+        ]
+    ):
+        parser.error("Resource limits, frequencies and lengths must be positive")
+    if args.concurrency > 40 or args.timeout > 86400:
+        parser.error("Concurrency must be <=40 and shard timeout <=86400 seconds")
+    if len(set(args.frequencies)) != len(args.frequencies) or len(
+        set(args.profiles)
+    ) != len(args.profiles):
+        parser.error("Duplicate shard parameters")
+    directory = args.output.resolve()
+    directory.mkdir(parents=True, exist_ok=True)
+    plan = make_plan(args)
+    plan_path = directory / "plan.json"
+    if plan_path.exists() and json.loads(plan_path.read_text()) != plan:
+        parser.error("Existing plan differs; use a new output directory")
+    _atomic_json(plan_path, plan)
+    print(
+        f"{len(plan['shards'])} shards; concurrency {args.concurrency}; upper bound {len(plan['shards']) * (args.timeout + 420) / 3600:.2f} sandbox-hours excluding deletion; execute={args.execute}"
+    )
+    if args.execute:
+        execute(plan, directory)
+
+
+if __name__ == "__main__":
+    main()

libs/langgraph/bench/delta_shard.py

@@ -0,0 +1,91 @@
+"""Execute one scripted benchmark with a sandbox-local disposable Postgres."""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+
+ROOT = Path("/workspace/langgraph/libs/langgraph")
+PGDATA = Path("/tmp/delta-bench-pg")
+
+
+def checked(command: list[str], **kwargs) -> None:
+    subprocess.run(command, check=True, stdout=subprocess.DEVNULL, **kwargs)
+
+
+def main() -> None:
+    job = json.loads(Path("/tmp/delta-job.json").read_text())
+    prefix = ["runuser", "-u", "postgres", "--"] if os.getuid() == 0 else []
+    for binary in ("initdb", "pg_ctl"):
+        if shutil.which(binary) is None:
+            raise RuntimeError(f"Snapshot must provide {binary} on PATH")
+    PGDATA.mkdir(mode=0o700)
+    if prefix:
+        checked(["chown", "postgres:postgres", str(PGDATA)])
+    checked(
+        [
+            *prefix,
+            "initdb",
+            "-D",
+            str(PGDATA),
+            "--auth-local=trust",
+            "--auth-host=trust",
+            "-U",
+            "postgres",
+        ]
+    )
+    started = False
+    try:
+        checked(
+            [
+                *prefix,
+                "pg_ctl",
+                "-D",
+                str(PGDATA),
+                "-l",
+                "/tmp/delta-postgres.log",
+                "-o",
+                "-h 127.0.0.1 -p 5432 -k /tmp",
+                "-w",
+                "start",
+            ]
+        )
+        started = True
+        command = [
+            sys.executable,
+            str(ROOT / "tests/test_delta_channel_benchmark.py"),
+            "--snapshot-sweep",
+            "--backend",
+            "postgres",
+            "--frequencies",
+            str(job["frequency"]),
+            "--profiles",
+            job["profile"],
+            "--lengths",
+            *map(str, job["lengths"]),
+            "--output",
+            "/tmp/delta-results.csv",
+        ]
+        env = {
+            **os.environ,
+            "LANGGRAPH_BENCH_POSTGRES_URI": "postgresql://postgres@127.0.0.1:5432/postgres?sslmode=disable",
+            "LANGSMITH_TRACING": "false",
+            "LANGCHAIN_TRACING_V2": "false",
+            "NO_COLOR": "1",
+        }
+        subprocess.run(
+            command, cwd=ROOT, env=env, timeout=job["timeout_seconds"], check=True
+        )
+    finally:
+        if started:
+            checked(
+                [*prefix, "pg_ctl", "-D", str(PGDATA), "-m", "immediate", "-w", "stop"]
+            )
+
+
+if __name__ == "__main__":
+    main()

libs/langgraph/bench/delta_snapshot.py

@@ -0,0 +1,113 @@
+"""Build a dedicated LangSmith benchmark snapshot from a minimal Docker context."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import re
+import subprocess
+from pathlib import Path
+
+from langsmith.sandbox import SandboxClient
+
+CONTEXT = Path(__file__).resolve().with_name("delta-snapshot")
+
+
+def snapshot_plan(args: argparse.Namespace) -> dict:
+    """Validate resource bounds and pin the public repository to a full SHA."""
+    if not re.fullmatch(r"[0-9a-f]{40}", args.revision):
+        raise ValueError("Revision must be a full lowercase 40-character commit SHA")
+    if not re.fullmatch(r"[a-z0-9][a-z0-9-]{0,62}", args.name):
+        raise ValueError(
+            "Name must contain only lowercase letters, numbers and hyphens"
+        )
+    if not (
+        1 <= args.vcpus <= 16
+        and 1 <= args.memory_gib <= 64
+        and 10 <= args.disk_gib <= 100
+        and 60 <= args.timeout <= 7200
+    ):
+        raise ValueError(
+            "Build limits: CPU 1–16, memory 1–64 GiB, disk 10–100 GiB, timeout 60–7200 seconds"
+        )
+    if sorted(p.name for p in CONTEXT.iterdir()) != ["Dockerfile"]:
+        raise ValueError(
+            "Build context must contain only Dockerfile; refusing other files"
+        )
+    return {
+        "name": args.name,
+        "revision": args.revision,
+        "dockerfile_sha256": hashlib.sha256(
+            (CONTEXT / "Dockerfile").read_bytes()
+        ).hexdigest(),
+        "vcpus": args.vcpus,
+        "mem_bytes": args.memory_gib * 1024**3,
+        "fs_capacity_bytes": args.disk_gib * 1024**3,
+        "timeout": args.timeout,
+    }
+
+
+def build_snapshot(plan: dict, client_factory=SandboxClient) -> dict:
+    """Create a billable builder, then return the reusable snapshot receipt."""
+    with client_factory() as client:
+        snapshot = client.create_snapshot_from_dockerfile(
+            name=plan["name"],
+            dockerfile="Dockerfile",
+            context=CONTEXT,
+            fs_capacity_bytes=plan["fs_capacity_bytes"],
+            build_args={"LANGGRAPH_REVISION": plan["revision"]},
+            vcpus=plan["vcpus"],
+            mem_bytes=plan["mem_bytes"],
+            timeout=plan["timeout"],
+        )
+    return {**plan, "snapshot_id": snapshot.id, "status": snapshot.status}
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--name", required=True)
+    parser.add_argument(
+        "--revision",
+        default=None,
+        help="Full public langgraph commit SHA; defaults to local HEAD",
+    )
+    parser.add_argument("--vcpus", type=int, default=2)
+    parser.add_argument("--memory-gib", type=int, default=8)
+    parser.add_argument("--disk-gib", type=int, default=30)
+    parser.add_argument("--timeout", type=int, default=1800)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument(
+        "--execute",
+        action="store_true",
+        help="Create a billable builder and persistent snapshot",
+    )
+    args = parser.parse_args()
+    if args.output.exists():
+        parser.error(
+            "Output already exists; use a fresh receipt path to avoid duplicate builds"
+        )
+    if args.revision is None:
+        args.revision = subprocess.check_output(
+            ["git", "-C", str(CONTEXT), "rev-parse", "HEAD"], text=True
+        ).strip()
+    try:
+        plan = snapshot_plan(args)
+    except ValueError as exc:
+        parser.error(str(exc))
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps({**plan, "state": "planned"}, indent=2) + "\n")
+    print(json.dumps(plan, indent=2))
+    if not args.execute:
+        print(
+            "Dry run: no resources created. Use --execute and a fresh receipt path to build."
+        )
+        return
+    args.output.write_text(json.dumps({**plan, "state": "building"}, indent=2) + "\n")
+    receipt = build_snapshot(plan)
+    args.output.write_text(json.dumps(receipt, indent=2) + "\n")
+    print(f"Snapshot ready: {receipt['snapshot_id']}")
+
+
+if __name__ == "__main__":
+    main()

libs/langgraph/tests/test_delta_channel_benchmark.py

@@ -20,16 +20,29 @@ Scenarios cover the dimensions where this branch's optimizations matter:
 
 from __future__ import annotations
 
+import argparse
 import contextlib
+import csv
 import gc
+import hashlib
+import itertools
+import json
+import math
 import os
+import platform
+import statistics
+import subprocess
 import sys
 import time
 import tracemalloc
+import uuid
+from collections.abc import Sequence
+from pathlib import Path
 from typing import Annotated, Any
 
 import pytest
-from langchain_core.messages import HumanMessage
+from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
 from langgraph.checkpoint.memory import MemorySaver
 from typing_extensions import TypedDict
 
@@ -210,10 +223,10 @@ def _pg_saver(thread_id: str):
         with saver._cursor() as cur:
             for tbl in ("checkpoints", "checkpoint_blobs", "checkpoint_writes"):
                 cur.execute(f"DELETE FROM {tbl} WHERE thread_id = %s", (thread_id,))
-        yield saver
-        with saver._cursor() as cur:
-            for tbl in ("checkpoints", "checkpoint_blobs", "checkpoint_writes"):
-                cur.execute(f"DELETE FROM {tbl} WHERE thread_id = %s", (thread_id,))
+        try:
+            yield saver
+        finally:
+            saver.delete_thread(thread_id)
 
 
 def _checkpointers() -> list[tuple[str, Any]]:
@@ -314,8 +327,389 @@ def test_delta_channel_benchmark(capsys: Any) -> None:
 # ---------------------------------------------------------------------------
 
 
+def test_scripted_snapshot_sweep() -> None:
+    saver = MemorySaver()
+    graph = _scripted_graph(2, "tool_calls", saver)
+    config = {"configurable": {"thread_id": "scripted-test"}}
+    for _ in range(3):
+        graph.invoke({}, config)
+    state = graph.get_state(config).values
+    assert len(state["messages"]) == 12
+    assert len(state["files"]) == 3
+    assert state["messages"][1].tool_calls[0]["id"] == "call-0"
+    assert state["messages"][2].tool_call_id == "call-0"
+    assert graph.get_state(config).values == state
+    assert _memory_storage_bytes(saver, "scripted-test") > 0
+
+
+def test_sweep_metrics_and_cleanup() -> None:
+    rows = list(_measure_sweep(2, [2, 4], "mock700", "memory"))
+    assert [row["read_samples"] for row in rows] == [2, 4]
+    assert rows[1]["storage_bytes"] > rows[0]["storage_bytes"]
+    assert rows[1]["read_max_ms"] >= rows[1]["read_p95_ms"] >= rows[1]["read_p50_ms"]
+    assert 1 <= rows[1]["slowest_update"] <= 4
+    assert _percentile([1, 2, 3, 4], 0.95) == 4
+    points = [
+        {"storage_bytes": 100, "read_max_ms": 1, "frequency": 25},
+        {"storage_bytes": 50, "read_max_ms": 2, "frequency": 50},
+        {"storage_bytes": 10, "read_max_ms": 10, "frequency": 100},
+    ]
+    assert _pareto_knee(points, "read_max_ms")["frequency"] == 50
+
+
+FREQUENCIES = list(range(25, 501, 25))
+LENGTHS = [50, 200, 1000, 5000, 10000]
+PROFILES = ["mock700", "tool_calls"]
+
+
+def _files_delta(state: dict, writes: Sequence[dict]) -> dict:
+    return {**state, **{key: value for write in writes for key, value in write.items()}}
+
+
+def _script(i: int, profile: str) -> list:
+    """Return reproducible, ID-stable messages without contacting a provider."""
+    if profile == "mock700":
+        return [
+            HumanMessage(content="h" * 350, id=f"human-{i}"),
+            AIMessage(content="a" * 350, id=f"answer-{i}"),
+        ]
+    call_id = f"call-{i}"
+    return [
+        HumanMessage(content=_human_content(i), id=f"human-{i}"),
+        AIMessage(
+            content="I will inspect the source and write a report.",
+            id=f"call-message-{i}",
+            tool_calls=[
+                {"name": "read_file", "args": {"path": f"/src/{i}.py"}, "id": call_id}
+            ],
+        ),
+        ToolMessage(
+            content=(
+                "def transform(record):\n    return record.get('value', 0)\n" * 64
+            ),
+            tool_call_id=call_id,
+            id=f"tool-{i}",
+        ),
+        AIMessage(
+            content="The implementation is deterministic.\n" * 16, id=f"answer-{i}"
+        ),
+    ]
+
+
+def _scripted_graph(freq: int, profile: str, saver: Any) -> Any:
+    """Append one scripted conversation and file update per graph invocation."""
+    fields = {
+        "messages": Annotated[
+            list, DeltaChannel(_messages_delta_reducer, snapshot_frequency=freq)
+        ],
+        "files": Annotated[dict, DeltaChannel(_files_delta, snapshot_frequency=freq)],
+    }
+    state_cls = TypedDict("ScriptedState", fields)
+    counter = itertools.count()
+
+    def turn(state: dict) -> dict:
+        i = next(counter)
+        script = _script(i, profile)
+        model = GenericFakeChatModel(
+            messages=iter(m for m in script if isinstance(m, AIMessage))
+        )
+        history = list(state.get("messages", []))
+        emitted = []
+        for message in script:
+            actual = (
+                model.invoke(history) if isinstance(message, AIMessage) else message
+            )
+            emitted.append(actual)
+            history.append(actual)
+        content = "report\n" * (16 if profile == "mock700" else 256)
+        return {
+            "messages": emitted,
+            "files": {
+                f"/reports/{i}.txt": {
+                    "content": content.splitlines(),
+                    "created_at": "2026-01-01T00:00:00Z",
+                    "modified_at": "2026-01-01T00:00:00Z",
+                }
+            },
+        }
+
+    builder = StateGraph(state_cls)
+    builder.add_node("scripted_chat", turn)
+    builder.set_entry_point("scripted_chat")
+    builder.add_edge("scripted_chat", END)
+    return builder.compile(checkpointer=saver)
+
+
+def _percentile(values: list[float], percentile: float) -> float:
+    return sorted(values)[max(0, math.ceil(len(values) * percentile) - 1)]
+
+
+def _memory_storage_bytes(saver: MemorySaver, thread_id: str) -> int:
+    """Count serialized payload bytes across all three memory saver stores."""
+    blobs = sum(
+        len(blob)
+        for (tid, *_), (_, blob) in saver.blobs.items()
+        if tid == thread_id and blob is not None
+    )
+    checkpoints = sum(
+        len(checkpoint[1]) + len(metadata[1])
+        for ns in saver.storage[thread_id].values()
+        for checkpoint, metadata, _ in ns.values()
+    )
+    writes = sum(
+        len(value[2][1])
+        for (tid, *_), batch in saver.writes.items()
+        if tid == thread_id
+        for value in batch.values()
+    )
+    return blobs + checkpoints + writes
+
+
+def _measure_sweep(freq: int, lengths: list[int], profile: str, backend: str):
+    """Measure every prefix read, including pre-snapshot phases, without tracing."""
+    thread_id = f"delta-sweep-{uuid.uuid4()}"
+    ctx = (
+        _pg_saver(thread_id)
+        if backend == "postgres"
+        else contextlib.nullcontext(MemorySaver())
+    )
+    with ctx as saver:
+        graph = _scripted_graph(freq, profile, saver)
+        config = {"configurable": {"thread_id": thread_id}}
+        warm_config = {"configurable": {"thread_id": f"{thread_id}-warmup"}}
+        try:
+            warm_graph = _scripted_graph(freq, profile, saver)
+            warm_graph.invoke({}, warm_config)
+            warm_graph.get_state(warm_config)
+        finally:
+            saver.delete_thread(warm_config["configurable"]["thread_id"])
+        reads: list[float] = []
+        writes: list[float] = []
+        slowest_update = 0
+        slowest_config = config
+        peak_read = -1.0
+        try:
+            for update in range(1, max(lengths) + 1):
+                start = time.perf_counter()
+                graph.invoke({}, config)
+                writes.append((time.perf_counter() - start) * 1000)
+                start = time.perf_counter()
+                snapshot = graph.get_state(config)
+                elapsed = (time.perf_counter() - start) * 1000
+                reads.append(elapsed)
+                if elapsed > peak_read:
+                    peak_read = elapsed
+                    slowest_update = update
+                    slowest_config = snapshot.config
+                if update not in lengths:
+                    continue
+                storage = (
+                    _postgres_storage_bytes(saver, thread_id)
+                    if backend == "postgres"
+                    else _memory_storage_bytes(saver, thread_id)
+                )
+                gc.collect()
+                tracemalloc.start()
+                try:
+                    graph.get_state(slowest_config)
+                    _, peak = tracemalloc.get_traced_memory()
+                finally:
+                    tracemalloc.stop()
+                yield {
+                    "backend": backend,
+                    "profile": profile,
+                    "frequency": freq,
+                    "updates": update,
+                    "read_samples": len(reads),
+                    "read_p50_ms": statistics.median(reads),
+                    "read_p95_ms": _percentile(reads, 0.95),
+                    "read_max_ms": max(reads),
+                    "slowest_update": slowest_update,
+                    "slowest_checkpoint_id": slowest_config["configurable"][
+                        "checkpoint_id"
+                    ],
+                    "write_p50_ms": statistics.median(writes),
+                    "write_p95_ms": _percentile(writes, 0.95),
+                    "write_max_ms": max(writes),
+                    "write_total_s": sum(writes) / 1000,
+                    "storage_bytes": storage,
+                    "slowest_read_peak_heap_bytes": peak,
+                }
+        finally:
+            saver.delete_thread(thread_id)
+
+
+def _pareto_knee(rows: list[dict], latency: str) -> dict:
+    """Pick the nondominated point nearest the min-max normalized ideal."""
+    frontier = [
+        r
+        for r in rows
+        if not any(
+            float(o["storage_bytes"]) <= float(r["storage_bytes"])
+            and float(o[latency]) <= float(r[latency])
+            and (
+                float(o["storage_bytes"]) < float(r["storage_bytes"])
+                or float(o[latency]) < float(r[latency])
+            )
+            for o in rows
+        )
+    ]
+    bounds = {
+        k: (min(float(r[k]) for r in rows), max(float(r[k]) for r in rows))
+        for k in ("storage_bytes", latency)
+    }
+
+    def score(row: dict) -> float:
+        return sum(
+            ((float(row[k]) - lo) / (hi - lo or 1)) ** 2
+            for k, (lo, hi) in bounds.items()
+        )
+
+    return min(frontier, key=score)
+
+
+def _plot_sweep(csv_path: Path) -> None:
+    """Render dependency-free SVG frequency and Pareto plots for each group."""
+    with csv_path.open() as handle:
+        rows = list(csv.DictReader(handle))
+    groups = sorted({(r["backend"], r["profile"], r["updates"]) for r in rows})
+    for backend, profile, updates in groups:
+        group = sorted(
+            (
+                r
+                for r in rows
+                if (r["backend"], r["profile"], r["updates"])
+                == (backend, profile, updates)
+            ),
+            key=lambda r: int(r["frequency"]),
+        )
+        svg = [
+            '<svg xmlns="http://www.w3.org/2000/svg" width="1100" height="420" viewBox="0 0 1100 420"><rect width="1100" height="420" fill="white"/>'
+        ]
+        svg.append(
+            f'<text x="20" y="25" font-family="sans-serif" font-size="16">{backend} / {profile} / {updates} updates (n={len(group)} frequencies)</text>'
+        )
+        for panel, (xkey, ykeys, title) in enumerate(
+            [
+                (
+                    "frequency",
+                    ["storage_bytes", "read_p50_ms", "read_max_ms"],
+                    "Frequency vs independently normalized metrics",
+                ),
+                (
+                    "storage_bytes",
+                    ["read_p50_ms", "read_max_ms"],
+                    "Storage vs latency (Pareto tradeoff)",
+                ),
+            ]
+        ):
+            left = 60 + panel * 550
+            svg.append(
+                f'<path d="M {left} 60 V 340 H {left + 450}" fill="none" stroke="black"/>'
+            )
+            svg.append(
+                f'<text x="{left}" y="50" font-family="sans-serif" font-size="12">{title}</text>'
+            )
+            xs = [float(r[xkey]) for r in group]
+            xmin, xmax = min(xs), max(xs)
+            ymax = max(float(r[k]) for r in group for k in ykeys)
+            for color, key in zip(["#176baf", "#c14b14", "#6a378b"], ykeys):
+                scale = max(float(r[key]) for r in group) if panel == 0 else ymax
+
+                def coords(row: dict, scale: float = scale) -> tuple[float, float]:
+                    return left + (float(row[xkey]) - xmin) / (
+                        xmax - xmin or 1
+                    ) * 450, 340 - float(row[key]) / (scale or 1) * 260
+
+                points = " ".join(f"{x:.2f},{y:.2f}" for x, y in map(coords, group))
+                svg.append(
+                    f'<polyline points="{points}" fill="none" stroke="{color}" stroke-width="2"/>'
+                )
+                if panel == 1:
+                    knee = _pareto_knee(group, key)
+                    x, y = coords(knee)
+                    svg.append(
+                        f'<circle cx="{x}" cy="{y}" r="5" fill="{color}"/><text x="{x + 6}" y="{y - 8}" font-family="sans-serif" font-size="11">f={knee["frequency"]}</text>'
+                    )
+                svg.append(
+                    f'<text x="{left}" y="{365 + ykeys.index(key) * 15}" fill="{color}" font-family="sans-serif" font-size="11">{key}</text>'
+                )
+            svg.append(
+                f'<text x="{left}" y="355" font-family="sans-serif" font-size="10">{xkey}: {xmin:.0f} to {xmax:.0f}; y: 0 to {1 if panel == 0 else ymax:.2f}{" normalized" if panel == 0 else " ms"}</text>'
+            )
+        svg.append("</svg>")
+        csv_path.with_name(
+            f"{csv_path.stem}-{backend}-{profile}-{updates}.svg"
+        ).write_text("".join(svg))
+
+
+def run_snapshot_sweep(args: argparse.Namespace) -> None:
+    """Append completed grid points and regenerate plots after every point."""
+    if args.backend == "postgres" and not _POSTGRES_AVAILABLE:
+        raise RuntimeError("Postgres dependencies unavailable; no fallback is allowed")
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    metadata_path = args.output.with_suffix(".metadata.json")
+    metadata = {
+        "schema": 1,
+        "python": platform.python_version(),
+        "platform": platform.platform(),
+        "script_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
+        "revision": subprocess.check_output(
+            ["git", "-C", str(Path(__file__).resolve().parent), "rev-parse", "HEAD"],
+            text=True,
+        ).strip(),
+        "backend": args.backend,
+        "profiles": args.profiles,
+        "frequencies": args.frequencies,
+        "lengths": args.lengths,
+        "script": "deterministic, no provider; IDs follow update index",
+        "read_sampling": "one get_state after every invocation, prefix distribution",
+        "write_sampling": "end-to-end graph.invoke including hydrate, scripted model and checkpoint I/O",
+        "storage": "postgres pg_column_size row estimate (not relation disk bytes); memory serialized payloads across all stores",
+        "max_supersteps_since_snapshot": os.environ.get(
+            "LANGGRAPH_DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT", "5000"
+        ),
+    }
+    if metadata_path.exists() and json.loads(metadata_path.read_text()) != metadata:
+        raise ValueError("Resume metadata differs; use a new output path")
+    metadata_path.write_text(json.dumps(metadata, indent=2) + "\n")
+    existing = []
+    if args.output.exists():
+        with args.output.open() as handle:
+            existing = list(csv.DictReader(handle))
+    done = {(r["profile"], int(r["frequency"]), int(r["updates"])) for r in existing}
+    for profile in args.profiles:
+        for freq in args.frequencies:
+            missing = [n for n in args.lengths if (profile, freq, n) not in done]
+            if not missing:
+                continue
+            for row in _measure_sweep(freq, missing, profile, args.backend):
+                fresh = not args.output.exists() or args.output.stat().st_size == 0
+                with args.output.open("a", newline="") as handle:
+                    writer = csv.DictWriter(
+                        handle, fieldnames=list(row), lineterminator="\n"
+                    )
+                    if fresh:
+                        writer.writeheader()
+                    writer.writerow(row)
+                    handle.flush()
+                print(json.dumps(row), flush=True)
+                _plot_sweep(args.output)
+
+
 if __name__ == "__main__":
-    print("DeltaChannel benchmark — multi-channel reads, mixed frequencies")
-    print("=" * 78)
-    run_benchmark()
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--snapshot-sweep", action="store_true")
+    parser.add_argument("--backend", choices=["memory", "postgres"], default="memory")
+    parser.add_argument("--frequencies", type=int, nargs="+", default=FREQUENCIES)
+    parser.add_argument("--lengths", type=int, nargs="+", default=LENGTHS)
+    parser.add_argument("--profiles", choices=PROFILES, nargs="+", default=PROFILES)
+    parser.add_argument("--output", type=Path, default=Path("delta-sweep.csv"))
+    args = parser.parse_args()
+    if any(value <= 0 for value in args.frequencies + args.lengths):
+        parser.error("frequencies and lengths must be positive")
+    if args.snapshot_sweep:
+        run_snapshot_sweep(args)
+    else:
+        run_benchmark()
     sys.exit(0)

libs/langgraph/tests/test_delta_distributed.py

@@ -0,0 +1,104 @@
+import importlib.util
+import json
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+SPEC = importlib.util.spec_from_file_location(
+    "delta_distributed",
+    Path(__file__).resolve().parents[1] / "bench/delta_distributed.py",
+)
+bench = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(bench)
+
+
+def test_plan_grid():
+    args = SimpleNamespace(
+        snapshot_id="snapshot",
+        concurrency=4,
+        timeout=3600,
+        vcpus=2,
+        memory_gib=8,
+        disk_gib=30,
+        lengths=list(bench.LENGTHS),
+        profiles=list(bench.PROFILES),
+        frequencies=list(bench.FREQUENCIES),
+    )
+    plan = bench.make_plan(args)
+    assert len(plan["shards"]) == 40
+    assert len({s["key"] for s in plan["shards"]}) == 40
+    assert plan["mem_bytes"] == 8 * 1024**3
+
+
+@pytest.mark.parametrize("command_fails", [False, True])
+def test_shard_cleanup_resume_and_validation(tmp_path, command_fails):
+    shard = {"key": "mock700-25", "profile": "mock700", "frequency": 25}
+    plan = {
+        "snapshot_id": "test",
+        "timeout_seconds": 10,
+        "vcpus": 2,
+        "mem_bytes": 1024,
+        "fs_capacity_bytes": 1024,
+        "lengths": [50],
+        "source_sha256": "hash",
+    }
+    data = b"backend,profile,frequency,updates,read_p50_ms,read_p95_ms,read_max_ms,storage_bytes\npostgres,mock700,25,50,1,2,3,400\n"
+    deleted = []
+
+    class FakeSandbox:
+        def write(self, path, content):
+            assert "API_KEY" not in str(content)
+
+        def run(self, *args, **kwargs):
+            return SimpleNamespace(exit_code=1 if command_fails else 0)
+
+        def read(self, path):
+            return (
+                json.dumps({"script_sha256": "hash"}).encode()
+                if path.endswith("json")
+                else data
+            )
+
+    class FakeClient:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            pass
+
+        def create_sandbox(self, **kwargs):
+            return FakeSandbox()
+
+        def delete_sandbox(self, name):
+            deleted.append(name)
+
+    if command_fails:
+        with pytest.raises(RuntimeError, match="Shard command failed"):
+            bench.run_shard(plan, shard, tmp_path, FakeClient)
+        assert len(deleted) == 1
+        assert not (tmp_path / shard["key"] / "success.json").exists()
+        return
+    bench.run_shard(plan, shard, tmp_path, FakeClient)
+    bench.run_shard(plan, shard, tmp_path, FakeClient)
+    assert len(deleted) == 1
+    assert (
+        json.loads((tmp_path / shard["key"] / "attempt.json").read_text())["state"]
+        == "deleted"
+    )
+    with pytest.raises(ValueError, match="Incomplete"):
+        bench.validate_csv(data, shard, [50, 200])
+    with pytest.raises(ValueError, match="Foreign"):
+        bench.validate_csv(data, {**shard, "frequency": 50}, [50])
+
+
+def test_failed_shard_is_deleted(tmp_path):
+    shard = {"key": "mock700-25", "profile": "mock700", "frequency": 25}
+    target = tmp_path / shard["key"]
+    target.mkdir()
+    (target / "attempt.json").write_text(
+        json.dumps({"state": "running", "sandbox_name": "previous"})
+    )
+    with pytest.raises(RuntimeError, match="previous sandbox"):
+        bench.run_shard({}, shard, tmp_path)
+    assert not (target / "success.json").exists()

libs/langgraph/tests/test_delta_snapshot.py

@@ -0,0 +1,48 @@
+import importlib.util
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import create_autospec
+
+import pytest
+from langsmith.sandbox import Sandbox, SandboxClient
+
+SPEC = importlib.util.spec_from_file_location(
+    "delta_snapshot", Path(__file__).resolve().parents[1] / "bench/delta_snapshot.py"
+)
+builder = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(builder)
+
+
+def test_snapshot_build_contract():
+    args = SimpleNamespace(
+        name="delta-bench",
+        revision="a" * 40,
+        vcpus=2,
+        memory_gib=8,
+        disk_gib=30,
+        timeout=1800,
+    )
+    plan = builder.snapshot_plan(args)
+    client = create_autospec(SandboxClient, instance=True)
+    client.__enter__.return_value = client
+    client.create_snapshot_from_dockerfile.return_value = SimpleNamespace(
+        id="snapshot-id", status="ready"
+    )
+    receipt = builder.build_snapshot(plan, lambda: client)
+    assert receipt["snapshot_id"] == "snapshot-id"
+    call = client.create_snapshot_from_dockerfile.call_args.kwargs
+    assert call["build_args"] == {"LANGGRAPH_REVISION": "a" * 40}
+    assert sorted(p.name for p in call["context"].iterdir()) == ["Dockerfile"]
+    assert call["timeout"] == 1800
+    client.__exit__.assert_called_once()
+    args.revision = "main; echo nope"
+    with pytest.raises(ValueError, match="40-character"):
+        builder.snapshot_plan(args)
+
+
+def test_sandbox_sdk_file_api():
+    sandbox = create_autospec(Sandbox, instance=True)
+    sandbox.write("/tmp/test", b"payload")
+    sandbox.read("/tmp/test")
+    with pytest.raises(AttributeError):
+        sandbox.write_file("/tmp/test", b"payload")