From a95d9684d5d062b2c3fadcd33f5f935167aa2954 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Mon, 29 Jun 2026 21:52:30 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=9B=A1=EF=B8=8F=20fix:=20Anchor=20any=20p?= =?UTF-8?q?ending-write=20checkpoint;=20stale-only=20eviction=20(Codex)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Broaden the lazy saver's keep-rule from "interrupt-only" to "persist any checkpoint that carries pending writes" (renamed InterruptOnlyMongoSaver → LazyMongoSaver). This makes it robust to delta-channel graphs without changing behavior for LibreChat's graph: - K1 (P1): a delta-channel graph can write a synthetic PARENT/anchor checkpoint (no __interrupt__ mark) that the interrupt checkpoint then points at, with the delta writes stored under the parent id. The old rule discarded that parent, breaking delta-state resume. Now any checkpoint that received putWrites is persisted, so the anchor parent and its writes survive and resume can walk the chain. - K3 (P2): for the same reason, clean delta-write rows are no longer orphaned — their checkpoint is persisted alongside them. (For LibreChat's standard Annotation/messages graph a clean run makes no putWrites at all — verified empirically — so the common path still writes nothing and the optimization is unchanged.) - K2 (P2): the 1024 FIFO cap could evict a valid in-flight id whose put() was just behind Mongo I/O, mis-classifying its interrupt checkpoint as a clean exit. Replaced with time-based eviction: only ids older than 5 min (a put always follows its putWrites within ms) are swept; a recent in-flight id is never dropped, and the map grows rather than evict a valid id if nothing is stale. New integration test: a checkpoint anchored by a NON-interrupt write is persisted. Full agents/HITL suites green (108). --- .../agents/checkpointer.integration.spec.ts | 23 +++- packages/api/src/agents/checkpointer.ts | 111 +++++++++++------- 2 files changed, 90 insertions(+), 44 deletions(-) diff --git a/packages/api/src/agents/checkpointer.integration.spec.ts b/packages/api/src/agents/checkpointer.integration.spec.ts index 3f4a78fef1..7c010f2c35 100644 --- a/packages/api/src/agents/checkpointer.integration.spec.ts +++ b/packages/api/src/agents/checkpointer.integration.spec.ts @@ -122,13 +122,13 @@ describe('checkpointer (mongodb-memory-server integration)', () => { }); }); -describe('InterruptOnlyMongoSaver (lazy persistence — mongodb-memory-server)', () => { - it('does NOT persist a clean-exit checkpoint (a bare put with no interrupt write)', async () => { +describe('LazyMongoSaver (lazy persistence — mongodb-memory-server)', () => { + it('does NOT persist a clean-exit checkpoint (a bare put with no pending writes)', async () => { const saver = await getAgentCheckpointer(MONGO_CFG); const threadId = `convo-${new mongoose.Types.ObjectId().toString()}`; const { config, checkpoint, metadata } = putArgs(threadId); - // A non-paused run's exit put — no preceding interrupt putWrites. + // A non-paused run's exit put — no preceding putWrites. await saver!.put(config, checkpoint, metadata); expect(await saver!.getTuple(readConfig(threadId))).toBeUndefined(); @@ -138,6 +138,23 @@ describe('InterruptOnlyMongoSaver (lazy persistence — mongodb-memory-server)', expect(count).toBe(0); }); + it('persists a checkpoint anchored by a NON-interrupt write (delta-channel safety)', async () => { + // K1/K3: a delta-channel graph can putWrites on a checkpoint that an interrupt + // checkpoint then depends on — even without the __interrupt__ marker. Any pending + // write must anchor its checkpoint so resume can walk the chain. + const saver = await getAgentCheckpointer(MONGO_CFG); + const threadId = `convo-${new mongoose.Types.ObjectId().toString()}`; + const { config, checkpoint, metadata } = putArgs(threadId); + await saver!.putWrites( + { configurable: { thread_id: threadId, checkpoint_ns: '', checkpoint_id: checkpoint.id } }, + [['some_delta_channel', { msgs: ['delta'] }]], + 'task-1', + ); + await saver!.put(config, checkpoint, metadata); + + expect(await saver!.getTuple(readConfig(threadId))).toBeDefined(); + }); + it('persists an interrupt checkpoint and carries its __interrupt__ pending write', async () => { const saver = await getAgentCheckpointer(MONGO_CFG); const threadId = `convo-${new mongoose.Types.ObjectId().toString()}`; diff --git a/packages/api/src/agents/checkpointer.ts b/packages/api/src/agents/checkpointer.ts index 5cb1b41294..1455cc1078 100644 --- a/packages/api/src/agents/checkpointer.ts +++ b/packages/api/src/agents/checkpointer.ts @@ -1,7 +1,6 @@ import mongoose from 'mongoose'; import { logger } from '@librechat/data-schemas'; import { MongoDBSaver } from '@langchain/langgraph-checkpoint-mongodb'; -import { INTERRUPT } from '@langchain/langgraph-checkpoint'; import type { TCheckpointerConfig } from 'librechat-data-provider'; import type { RunnableConfig } from '@langchain/core/runnables'; import type { Checkpoint, CheckpointMetadata, PendingWrite } from '@langchain/langgraph-checkpoint'; @@ -26,40 +25,57 @@ import type { Checkpoint, CheckpointMetadata, PendingWrite } from '@langchain/la */ /** - * Defensive cap on the number of in-flight interrupt checkpoint ids tracked between an - * interrupt `putWrites` and its matching `put`. Under normal flow each id is consumed by - * the immediately-following `put`, so the set holds at most a handful; the cap only guards - * against a process that dies in that microsecond window leaking ids forever. + * Soft size threshold that triggers a sweep of STALE write-anchor ids. The map normally + * holds a handful (each id is consumed by the `put` that immediately follows its + * `putWrites`); this only bounds a slow leak from a process that dies in that window. */ -const MAX_TRACKED_INTERRUPTS = 1024; +const WRITE_ANCHOR_SWEEP_THRESHOLD = 1024; /** - * A `MongoDBSaver` that persists ONLY the checkpoints created at an interrupt (a HITL pause), - * discarding the one LangGraph writes on a CLEAN exit. + * A write-anchor id is considered stale once this much wall-clock has passed without its + * matching `put` — a `put` always follows its `putWrites` within the same exit sequence + * (milliseconds), so anything this old is from a crashed run, never a valid in-flight id. + * Generous on purpose: we would rather keep a tracked id slightly too long than evict a + * valid one and mis-classify its (possibly slow-I/O) interrupt `put` as a clean exit. + */ +const WRITE_ANCHOR_STALE_MS = 5 * 60 * 1000; + +/** + * A `MongoDBSaver` that persists ONLY checkpoints carrying pending writes — an interrupt + * (a HITL pause) or a delta-channel anchor — and discards the no-write checkpoint LangGraph + * writes on a CLEAN exit. * * **Why.** With `durability: 'exit'` (set by the SDK whenever a checkpointer is active) the * graph persists exactly one checkpoint at the exit boundary on EVERY run — paused or not. * A non-paused turn therefore writes a dead checkpoint whose only fate is to be pruned by - * {@link deleteAgentCheckpoint}. HITL only ever resumes an *interrupt* checkpoint, so the - * clean-exit one is pure write+delete churn on the common path. This saver skips it. + * {@link deleteAgentCheckpoint}. HITL only ever resumes a checkpoint that has pending writes, + * so the clean (write-less) exit checkpoint is pure write+delete churn on the common path. + * This saver skips it. * - * **How it tells them apart** (verified empirically against `@langchain/langgraph`): when a - * run interrupts, the runner calls `putWrites` with the `INTERRUPT` (`"__interrupt__"`) - * channel for the checkpoint it is about to create, and that write's `config.checkpoint_id` - * equals the `checkpoint.id` of the `put` that immediately follows. A clean exit calls `put` - * with no preceding interrupt `putWrites`. So we record the checkpoint id of any interrupt - * `putWrites` and persist a `put` only when its `checkpoint.id` was so marked. Keying on the - * globally-unique checkpoint id (NOT thread_id) keeps this correct even when two runs race - * on the same conversation (`thread_id`) — the job-replacement scenario. + * **How it tells them apart** (verified empirically against `@langchain/langgraph`): LangGraph + * calls `putWrites` for a checkpoint BEFORE the `put` that creates it, with `config.checkpoint_id` + * equal to that `put`'s `checkpoint.id`. An interrupt records `INTERRUPT` ("__interrupt__") + * writes; a delta-channel graph records its delta writes (and may anchor them on a synthetic + * parent checkpoint that the interrupt checkpoint then points at). A CLEAN exit produces a + * checkpoint with NO pending writes. So we record the checkpoint id of ANY `putWrites` and + * persist a `put` only when its `checkpoint.id` was so marked — which keeps interrupt + * checkpoints AND any delta-anchor parents (resume can walk the chain), while still dropping + * the write-less clean-exit checkpoint. Keying on the globally-unique checkpoint id (NOT + * thread_id) stays correct even when two runs race on the same conversation (`thread_id`). * - * **Correctness.** Interrupt checkpoints and their pending writes are persisted exactly as - * before, so resume is unchanged. Clean checkpoints were only ever written-then-pruned, so - * not writing them is observationally equivalent; the eager prune stays as the backstop for - * any lingering interrupt checkpoint. `getTuple`/`list`/`deleteThread`/`setup` are inherited. + * For LibreChat's agent graph (standard `Annotation` channels, no `DeltaChannel`) a clean run + * makes no `putWrites` at all, so this is effectively interrupt-only and the common path + * writes nothing; the broader "has pending writes" rule just makes it robust to delta graphs. + * + * **Correctness.** Checkpoints with pending writes (interrupt + delta-anchor) and the writes + * themselves persist exactly as before, so resume is unchanged. The no-write clean checkpoint + * was only ever written-then-pruned, so not writing it is observationally equivalent; the + * pre-run prune + Mongo TTL remain the backstops. `getTuple`/`list`/`deleteThread`/`setup` + * are inherited. */ -export class InterruptOnlyMongoSaver extends MongoDBSaver { - /** checkpoint ids an interrupt `putWrites` flagged; each consumed by its matching `put`. */ - private readonly interruptedCheckpointIds = new Set(); +export class LazyMongoSaver extends MongoDBSaver { + /** checkpoint id → time the `putWrites` flagging it arrived; each consumed by its `put`. */ + private readonly writeAnchorIds = new Map(); override async putWrites( config: RunnableConfig, @@ -67,18 +83,11 @@ export class InterruptOnlyMongoSaver extends MongoDBSaver { taskId: string, ): Promise { const checkpointId = config.configurable?.checkpoint_id as string | undefined; - if (checkpointId && writes.some((write) => write[0] === INTERRUPT)) { - if (this.interruptedCheckpointIds.size >= MAX_TRACKED_INTERRUPTS) { - // Evict the oldest unconsumed id (a leak from a crash between putWrites and put). - const oldest = this.interruptedCheckpointIds.values().next().value; - if (oldest !== undefined) { - this.interruptedCheckpointIds.delete(oldest); - } - } - this.interruptedCheckpointIds.add(checkpointId); + if (checkpointId) { + // A checkpoint that receives ANY pending writes must be persisted by its `put`: an + // interrupt, or a delta-channel anchor whose writes a later checkpoint depends on. + this.recordWriteAnchor(checkpointId); } - // Always persist the writes: an interrupt's pending writes are required for resume, and - // under `durability: 'exit'` a clean run never calls putWrites at all. return super.putWrites(config, writes, taskId); } @@ -87,12 +96,13 @@ export class InterruptOnlyMongoSaver extends MongoDBSaver { checkpoint: Checkpoint, metadata: CheckpointMetadata, ): Promise { - if (this.interruptedCheckpointIds.delete(checkpoint.id)) { - // Produced by an interrupt (pause) — persist it so the run can be resumed. + if (this.writeAnchorIds.delete(checkpoint.id)) { + // Has pending writes (interrupt / delta anchor) — persist so resume can read it. return super.put(config, checkpoint, metadata); } - // Clean exit: discard. Return the config LangGraph expects (pointing at the checkpoint - // it believes was saved) so the run finishes normally; nothing durable is written. + // No pending writes ⇒ a clean exit: discard. Return the config LangGraph expects + // (pointing at the checkpoint it believes was saved) so the run finishes normally; + // nothing durable is written. return { ...config, configurable: { @@ -101,6 +111,25 @@ export class InterruptOnlyMongoSaver extends MongoDBSaver { }, }; } + + /** + * Track a checkpoint id that received pending writes. Evicts ONLY genuinely-stale ids + * (older than {@link WRITE_ANCHOR_STALE_MS}, i.e. from a crashed run whose `put` never + * landed) — never a recent in-flight id — so a slow-I/O interrupt `put` is never + * mis-classified as a clean exit. If nothing is stale the map is allowed to grow rather + * than drop a valid id; the next sweep reclaims the crashed ones. + */ + private recordWriteAnchor(checkpointId: string): void { + const now = Date.now(); + if (this.writeAnchorIds.size >= WRITE_ANCHOR_SWEEP_THRESHOLD) { + for (const [id, recordedAt] of this.writeAnchorIds) { + if (now - recordedAt > WRITE_ANCHOR_STALE_MS) { + this.writeAnchorIds.delete(id); + } + } + } + this.writeAnchorIds.set(checkpointId, now); + } } /** Default approval window and checkpoint TTL: 24h. */ @@ -185,7 +214,7 @@ async function buildMongoSaver( resolved: ResolvedCheckpointerConfig, ): Promise { try { - const saver = new InterruptOnlyMongoSaver({ + const saver = new LazyMongoSaver({ // mongoose vends the live MongoClient; reuse it instead of opening a second // connection. The driver type is structurally identical but resolves to a // different `mongodb` copy than checkpoint-mongodb's, hence the cast.