diff --git a/packages/api/src/agents/checkpointer.integration.spec.ts b/packages/api/src/agents/checkpointer.integration.spec.ts index e355650e0e..be198c2c9d 100644 --- a/packages/api/src/agents/checkpointer.integration.spec.ts +++ b/packages/api/src/agents/checkpointer.integration.spec.ts @@ -155,11 +155,13 @@ 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 () => { + it('does NOT persist an error-only checkpoint OR its write row (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. + // checkpoint is never HITL-resumable, so the lazy saver must leave NOTHING durable: not the + // checkpoint (discarded by `put`) and not the write row (`putWrites` drops a bookkeeping-only + // batch instead of forwarding it, which would otherwise orphan a row in the writes collection + // 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); @@ -171,10 +173,13 @@ describe('LazyMongoSaver (lazy persistence — mongodb-memory-server)', () => { 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); + const db = mongoose.connection.db!; + expect(await db.collection('agent_checkpoints').countDocuments({ thread_id: threadId })).toBe( + 0, + ); + expect( + await db.collection('agent_checkpoint_writes').countDocuments({ thread_id: threadId }), + ).toBe(0); }); it('persists an interrupt checkpoint and carries its __interrupt__ pending write', async () => { @@ -249,6 +254,40 @@ describe('LazyMongoSaver (lazy persistence — mongodb-memory-server)', () => { boomGraph.invoke({ x: 'start' }, { configurable: { thread_id: tErr }, durability: 'exit' }), ).rejects.toThrow('boom'); + // Nothing durable: neither the checkpoint nor an orphan row in the writes collection. expect(await coll.countDocuments({ thread_id: tErr })).toBe(0); + const writesColl = mongoose.connection.db!.collection('agent_checkpoint_writes'); + expect(await writesColl.countDocuments({ thread_id: tErr })).toBe(0); + }); + + it('end-to-end: an interrupt persists, then resumes to completion with the approval value', async () => { + // Guards the `putWrites` change: the `__interrupt__` write must still be forwarded (it is + // resumable) so a paused run rehydrates and the resume value flows in. Mirrors the real HITL + // round-trip across a fresh `invoke` on the same thread_id. + const { StateGraph, START, END, interrupt, Annotation, Command } = await import( + '@langchain/langgraph' + ); + const saver = await getAgentCheckpointer(MONGO_CFG); + + const State = Annotation.Root({ approved: Annotation }); + const graph = new StateGraph(State) + .addNode('gate', () => ({ approved: interrupt('approve?') })) + .addNode('done', () => ({})) + .addEdge(START, 'gate') + .addEdge('gate', 'done') + .addEdge('done', END) + .compile({ checkpointer: saver as never }); + + const thread = `convo-${new mongoose.Types.ObjectId().toString()}`; + const cfg = { configurable: { thread_id: thread }, durability: 'exit' as const }; + + // Pause at the interrupt. + await graph.invoke({ approved: null }, cfg); + const paused = await graph.getState(cfg); + expect(paused.next.length).toBeGreaterThan(0); + + // Resume with the approval — the paused run rehydrates from the durable interrupt checkpoint. + const out = await graph.invoke(new Command({ resume: 'YES' }), cfg); + expect(out.approved).toBe('YES'); }); }); diff --git a/packages/api/src/agents/checkpointer.ts b/packages/api/src/agents/checkpointer.ts index bdab468802..e55e81c541 100644 --- a/packages/api/src/agents/checkpointer.ts +++ b/packages/api/src/agents/checkpointer.ts @@ -84,11 +84,14 @@ function hasResumableWrite(writes: PendingWrite[]): boolean { * runs race on the same conversation (`thread_id`). * * **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 + * write — on the `__error__` bookkeeping channel — followed by a `put` (probe-confirmed). Such a + * batch is dropped on BOTH paths: `putWrites` does not forward it to the writes collection (else an + * orphan write row with no surviving parent checkpoint would linger until the TTL / next prune), + * and because it is not anchored its `put` discards the checkpoint too. 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. + * `__scheduled__`, `__resume__`) and the failed turn leaves NOTHING durable — verified against a + * real `MongoDBSaver` (mongodb-memory-server): a throwing graph persists 0 checkpoints AND 0 write + * rows, while interrupt→resume is unaffected (its `__interrupt__` write is forwarded as before). * * For LibreChat's agent graph (standard `Annotation`/`MessagesAnnotation` channels, no * `DeltaChannel` — grep-confirmed in `@librechat/agents`) a clean run makes no `putWrites` at all, @@ -118,12 +121,19 @@ export class LazyMongoSaver extends MongoDBSaver { writes: PendingWrite[], taskId: string, ): Promise { + if (!hasResumableWrite(writes)) { + // A bookkeeping-only batch (e.g. `__error__` from a failed, non-paused turn). Its + // checkpoint is discarded by `put`, so forwarding to `super.putWrites` would leave an + // ORPHAN row in the writes collection — a write whose parent checkpoint never persists — + // until the Mongo TTL or the conversation's next pre-run prune. Drop it entirely: a + // non-resumable batch is never read back on resume, so nothing durable is needed. + return; + } + // Anchor the checkpoint so its `put` persists it: an interrupt (a HITL pause), or a real + // state/delta channel a later checkpoint depends on. Keyed on the globally-unique checkpoint + // id so concurrent runs on the same `thread_id` can't cross-consume anchors. const checkpointId = config.configurable?.checkpoint_id as string | undefined; - 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. + if (checkpointId) { this.recordWriteAnchor(checkpointId); } return super.putWrites(config, writes, taskId);