diff --git a/packages/api/src/agents/checkpointer.integration.spec.ts b/packages/api/src/agents/checkpointer.integration.spec.ts index 7c010f2c35..e355650e0e 100644 --- a/packages/api/src/agents/checkpointer.integration.spec.ts +++ b/packages/api/src/agents/checkpointer.integration.spec.ts @@ -1,6 +1,6 @@ import mongoose from 'mongoose'; import { MongoMemoryServer } from 'mongodb-memory-server'; -import { emptyCheckpoint, INTERRUPT } from '@langchain/langgraph-checkpoint'; +import { emptyCheckpoint, ERROR, INTERRUPT } from '@langchain/langgraph-checkpoint'; import type { MongoDBSaver } from '@langchain/langgraph-checkpoint-mongodb'; import { getAgentCheckpointer, @@ -140,8 +140,8 @@ describe('LazyMongoSaver (lazy persistence — mongodb-memory-server)', () => { 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. + // checkpoint then depends on — even without the __interrupt__ marker. A write on a + // real (non-`__`-prefixed) channel 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); @@ -155,6 +155,28 @@ describe('LazyMongoSaver (lazy persistence — mongodb-memory-server)', () => { expect(await saver!.getTuple(readConfig(threadId))).toBeDefined(); }); + it('does NOT persist an error-only checkpoint (failed non-paused turn — no leak)', async () => { + // A turn that throws before any pause records a pending write on the `__error__` + // bookkeeping channel, then a `put` (probe-confirmed against @langchain/langgraph). That + // checkpoint is never HITL-resumable, so the lazy saver must discard it rather than leave + // it durable until the next fresh-turn prune / Mongo TTL. + 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 } }, + [[ERROR, 'boom']], // '__error__' — bookkeeping channel, not resumable + 'task-1', + ); + await saver!.put(config, checkpoint, metadata); + + expect(await saver!.getTuple(readConfig(threadId))).toBeUndefined(); + const count = await mongoose.connection + .db!.collection('agent_checkpoints') + .countDocuments({ thread_id: threadId }); + expect(count).toBe(0); + }); + 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()}`; @@ -204,4 +226,29 @@ describe('LazyMongoSaver (lazy persistence — mongodb-memory-server)', () => { const state = await pauseGraph.getState({ configurable: { thread_id: tPause } }); expect(state.next.length).toBeGreaterThan(0); // the interrupted node is still pending → resumable }); + + it('end-to-end: a real graph that THROWS before pausing persists no checkpoint', async () => { + // F2: a failed non-paused turn records an `__error__` pending write + a put. The lazy + // saver must discard it so a conversation that errors (and is never retried) leaves nothing + // durable behind — the clean-path prune that used to catch this was removed. + const { StateGraph, START, END, Annotation } = await import('@langchain/langgraph'); + const saver = await getAgentCheckpointer(MONGO_CFG); + const coll = mongoose.connection.db!.collection('agent_checkpoints'); + + const State = Annotation.Root({ x: Annotation }); + const boomGraph = new StateGraph(State) + .addNode('a', () => { + throw new Error('boom'); + }) + .addEdge(START, 'a') + .addEdge('a', END) + .compile({ checkpointer: saver as never }); + + const tErr = `convo-${new mongoose.Types.ObjectId().toString()}`; + await expect( + boomGraph.invoke({ x: 'start' }, { configurable: { thread_id: tErr }, durability: 'exit' }), + ).rejects.toThrow('boom'); + + expect(await coll.countDocuments({ thread_id: tErr })).toBe(0); + }); }); diff --git a/packages/api/src/agents/checkpointer.ts b/packages/api/src/agents/checkpointer.ts index 3eeffb879a..bdab468802 100644 --- a/packages/api/src/agents/checkpointer.ts +++ b/packages/api/src/agents/checkpointer.ts @@ -1,5 +1,6 @@ import mongoose from 'mongoose'; import { logger } from '@librechat/data-schemas'; +import { INTERRUPT } from '@langchain/langgraph-checkpoint'; import { MongoDBSaver } from '@langchain/langgraph-checkpoint-mongodb'; import type { Checkpoint, CheckpointMetadata, PendingWrite } from '@langchain/langgraph-checkpoint'; import type { TCheckpointerConfig } from 'librechat-data-provider'; @@ -41,9 +42,27 @@ const WRITE_ANCHOR_SWEEP_THRESHOLD = 1024; 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. + * Does a pending-write batch make its checkpoint worth persisting for resume? True if it carries + * an interrupt (the HITL pause that resume targets) or any real state/delta channel write (a value + * a later checkpoint's resume depends on). False for pure bookkeeping batches — `__error__` + * (a failed, non-paused turn), `__scheduled__`, `__resume__` — which are never HITL-resumable, so + * persisting their checkpoint would only leak storage until the next prune / Mongo TTL. + * + * `INTERRUPT` is the one `__`-prefixed channel that IS resume-worthy; every other `__…__` channel + * is langgraph bookkeeping. Constants verified against `@langchain/langgraph-checkpoint`. + */ +function hasResumableWrite(writes: PendingWrite[]): boolean { + return (writes ?? []).some(([channel]) => { + const name = String(channel); + return name === INTERRUPT || !name.startsWith('__'); + }); +} + +/** + * A `MongoDBSaver` that persists ONLY checkpoints carrying a {@link hasResumableWrite resumable} + * pending write — an interrupt (a HITL pause) or a real-channel/delta anchor — and discards both + * the no-write checkpoint LangGraph writes on a CLEAN exit and the bookkeeping-only checkpoint of + * a failed (non-paused) turn. * * **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. @@ -52,26 +71,43 @@ const WRITE_ANCHOR_STALE_MS = 5 * 60 * 1000; * 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`): LangGraph + * **How it tells them apart** (verified empirically with throwaway runnable probes against + * `@langchain/langgraph@1.4`, not source-reading): under `durability: 'exit'` 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`). + * equal to that `put`'s `checkpoint.id`. An interrupt records an `INTERRUPT` ("__interrupt__") + * write; a delta-channel graph records its delta writes on a real (non-`__`-prefixed) channel. + * A CLEAN exit produces a checkpoint with NO pending writes. So we record the checkpoint id of + * each `putWrites` that carries a {@link hasResumableWrite resumable} write and persist a `put` + * only when its `checkpoint.id` was so marked — which keeps interrupt checkpoints AND any + * real-channel/delta anchors (resume can walk the chain), while dropping the write-less clean + * exit. Keying on the globally-unique checkpoint id (NOT thread_id) stays correct even when two + * runs race on the same conversation (`thread_id`). * - * 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. + * **Why "resumable" and not "any" write.** A non-paused turn that ERRORS still records a pending + * write — on the `__error__` bookkeeping channel — followed by a `put` (probe-confirmed). Anchoring + * on *any* write would persist that failed-turn checkpoint, and since the clean-path prune was + * removed it would linger until the next fresh turn or the TTL. An errored turn is never + * HITL-resumable, so {@link hasResumableWrite} excludes bookkeeping-only batches (`__error__`, + * `__scheduled__`, `__resume__`) and the checkpoint is discarded at the source — no leak. * - * **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. + * For LibreChat's agent graph (standard `Annotation`/`MessagesAnnotation` channels, no + * `DeltaChannel` — grep-confirmed in `@librechat/agents`) a clean run makes no `putWrites` at all, + * so this is effectively interrupt-only and the common path writes nothing; the broader + * real-channel rule just keeps it honest for delta graphs. + * + * **Invariant.** Correctness depends on `durability: 'exit'` (which the SDK sets whenever a + * checkpointer is active): exactly one parentless boundary checkpoint per run, with its + * `putWrites` ordered before its `put`. Under per-step durability LangGraph instead emits + * `put`-before-`putWrites` for chained checkpoints — the anchor would arrive too late and a + * checkpoint could be wrongly discarded. The SDK never runs HITL that way; if that ever changes, + * this saver must be revisited (a parent-based guard is NOT viable — a resumed turn's clean + * completion is itself a parented, write-less checkpoint that we correctly discard). + * + * **Correctness.** Checkpoints with resumable writes (interrupt + real-channel/delta anchor) and + * the writes themselves persist exactly as before, so resume is unchanged. The write-less clean + * checkpoint (and the now-discarded error-only 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 LazyMongoSaver extends MongoDBSaver { /** checkpoint id → time the `putWrites` flagging it arrived; each consumed by its `put`. */ @@ -83,9 +119,11 @@ export class LazyMongoSaver extends MongoDBSaver { taskId: string, ): Promise { const checkpointId = config.configurable?.checkpoint_id as string | undefined; - 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. + if (checkpointId && hasResumableWrite(writes)) { + // Anchor only checkpoints whose writes matter for resume: an interrupt (a HITL pause), + // or a real state/delta channel a later checkpoint depends on. Bookkeeping-only batches + // (e.g. `__error__` from a failed, non-paused turn) are NOT anchored, so their checkpoint + // is discarded by `put` rather than leaking until the next prune / Mongo TTL. this.recordWriteAnchor(checkpointId); } return super.putWrites(config, writes, taskId); @@ -97,10 +135,12 @@ export class LazyMongoSaver extends MongoDBSaver { metadata: CheckpointMetadata, ): Promise { if (this.writeAnchorIds.delete(checkpoint.id)) { - // Has pending writes (interrupt / delta anchor) — persist so resume can read it. + // Carries a resumable write (interrupt / real-channel delta anchor) — persist so resume + // can read it. return super.put(config, checkpoint, metadata); } - // No pending writes ⇒ a clean exit: discard. Return the config LangGraph expects + // No resumable writes ⇒ a clean exit (a non-paused completion, a resumed turn's clean + // finish, or an error-only checkpoint): discard. Return the config LangGraph expects // (pointing at the checkpoint it believes was saved) so the run finishes normally; // nothing durable is written. return {