mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-07-10 16:23:44 +00:00
🛡️ fix: Guard HITL checkpoint size against MongoDB 16MB limit (#14157)
Some checks failed
Docker Dev Branch Images Build / build (Dockerfile, lc-dev, node) (push) Waiting to run
Docker Dev Branch Images Build / build (Dockerfile.multi, lc-dev-api, api-build) (push) Waiting to run
Docker Dev Images Build / build (Dockerfile, librechat-dev, node) (push) Has been cancelled
Docker Dev Images Build / build (Dockerfile.multi, librechat-dev-api, api-build) (push) Has been cancelled
GitNexus Index / index (push) Has been cancelled
Sync Locize Translations & Create Translation PR / Sync Translation Keys with Locize (push) Has been cancelled
Sync Helm Chart Tags / Ignore non-main push (push) Has been cancelled
Sync Helm Chart Tags / Sync chart tags (push) Has been cancelled
GitNexus Index / post-index (push) Has been cancelled
Sync Locize Translations & Create Translation PR / Create Translation PR on Version Published (push) Has been cancelled
Some checks failed
Docker Dev Branch Images Build / build (Dockerfile, lc-dev, node) (push) Waiting to run
Docker Dev Branch Images Build / build (Dockerfile.multi, lc-dev-api, api-build) (push) Waiting to run
Docker Dev Images Build / build (Dockerfile, librechat-dev, node) (push) Has been cancelled
Docker Dev Images Build / build (Dockerfile.multi, librechat-dev-api, api-build) (push) Has been cancelled
GitNexus Index / index (push) Has been cancelled
Sync Locize Translations & Create Translation PR / Sync Translation Keys with Locize (push) Has been cancelled
Sync Helm Chart Tags / Ignore non-main push (push) Has been cancelled
Sync Helm Chart Tags / Sync chart tags (push) Has been cancelled
GitNexus Index / post-index (push) Has been cancelled
Sync Locize Translations & Create Translation PR / Create Translation PR on Version Published (push) Has been cancelled
* 🛡️ fix: Guard HITL checkpoint size against MongoDB 16MB limit A LangGraph HITL checkpoint embeds the whole serialized message history in a single BSON document, so a large conversation (inlined base64 media, big tool outputs, long history) can serialize past MongoDB's 16MB document ceiling. `MongoDBSaver.put` would then throw a raw `BSONObjectTooLarge` at pause time and the pause would be lost with no legible error. `LazyMongoSaver` now measures the serialized checkpoint on the persist path (rare HITL pauses only — the clean-exit common path is untouched): debug-logs the size, warns past a soft 8MB threshold, and throws a typed `CheckpointTooLargeError` before the doomed write past a 15MB hard limit (16MB minus headroom for the document's other fields). Thresholds are overridable via the constructor for testing. Adds integration coverage (real serde + mongodb-memory-server) for the under-threshold, soft-warn, and hard-reject cases. * 🧱 fix: Add explicit types for isolatedDeclarations build The production build (tsdown + rolldown-plugin-dts) compiles with --isolatedDeclarations, which requires explicit type annotations on exported bindings whose initializers it can't infer syntactically. `CHECKPOINT_HARD_LIMIT_BYTES` (arithmetic over two consts) tripped TS9010; annotate it and `CHECKPOINT_WARN_BYTES` as `number`. Verified with `tsc --isolatedDeclarations` over the package. * fix: Codex review — include metadata in the size guard + flush parked bookkeeping P1 (lost bookkeeping): `put` consumes the write anchor, then AWAITS assertCheckpointFitsDocument (checkpoint serialization). A bookkeeping-only putWrites dispatched in that window sees neither the anchor nor persistedIds, so it parks — and `put` never flushed it, dropping the marker (e.g. a completed Send-sibling's __no_writes__) so a resume re-executed the sibling. Extract flushBufferedBookkeeping (shared with the anchoring putWrites) and call it after super.put in the persist path. P2 (metadata ignored by guard): MongoDBSaver.put stores the serialized checkpoint AND metadata (plus metadata_search) in the SAME document, but the guard measured only the checkpoint — a just-under-limit checkpoint with large metadata fell through to a raw BSONObjectTooLarge. Measure checkpoint + metadata; the fixed headroom now only covers metadata_search/ids/framing. Two integration regressions added (both fail without the fix, pass with it): metadata-pushes-over-the-ceiling, and flush-during-the-serialization-window. * fix: count metadata_search (raw metadata copy) in the checkpoint size guard Codex follow-up: MongoDBSaver.put stores metadata a SECOND time as `metadata_search: metadata` — the whole raw metadata object as a queryable BSON subdocument in the same agent_checkpoints document. Measuring only checkpoint + serialized metadata undercounted by that raw copy, so a large metadata.writes payload could pass the 15 MB preflight while metadata_search pushed the actual BSON past 16 MB — the raw BSONObjectTooLarge the guard exists to prevent. Add mongoose.mongo.BSON.calculateObjectSize(metadata) for the metadata_search contribution (mongoose already imported; no new dep). Headroom now only covers ids + BSON framing. New integration test sizes a case where checkpoint + serialized metadata is under the limit but the raw metadata_search copy pushes it over — green (24/24). * ci: run packages/api agents integration specs (checkpointer) in CI The checkpointer.integration.spec.ts (durable HITL checkpointer vs a real in-process MongoDB) is a *.integration.spec.ts, which test:ci deliberately excludes — and cache-integration-tests.yml only covers cache/cluster/mcp/ stream, not src/agents/**. So it ran nowhere and its regressions guarded nothing. Add: - test:agents-integration script (jest over src/agents/*.integration.spec.ts, runInBand — mongodb-memory-server is in-process, no external service); - a dedicated agents-integration-tests.yml (mirrors the proven build setup, no Redis) triggered on packages/api/src/agents/** changes; - babel-plugin-replace-ts-export-assignment as a packages/api devDep: the spec imports @langchain/langgraph-checkpoint (whitelisted for babel transform, uses `export =`), whose transform needs this plugin — it was only present under client/node_modules, unresolvable from packages/api, so the suite couldn't load. 24/24 pass locally.
This commit is contained in:
parent
988a14a405
commit
cf9a426d29
5 changed files with 447 additions and 11 deletions
90
.github/workflows/agents-integration-tests.yml
vendored
Normal file
90
.github/workflows/agents-integration-tests.yml
vendored
Normal file
|
|
@ -0,0 +1,90 @@
|
|||
name: Agents Integration Tests
|
||||
|
||||
# Runs the packages/api `src/agents/**` integration specs (e.g. the durable HITL
|
||||
# checkpointer against a real in-process MongoDB via mongodb-memory-server). These
|
||||
# are `*.integration.spec.ts`, which `test:ci` deliberately excludes — without this
|
||||
# job they run nowhere and their regressions guard nothing.
|
||||
on:
|
||||
pull_request:
|
||||
branches:
|
||||
- main
|
||||
- dev
|
||||
- dev-staging
|
||||
- release/*
|
||||
paths:
|
||||
- 'packages/api/src/agents/**'
|
||||
- 'packages/api/package.json'
|
||||
- '.github/workflows/agents-integration-tests.yml'
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
jobs:
|
||||
agents_integration_tests:
|
||||
name: Integration Tests that use in-process MongoDB
|
||||
timeout-minutes: 20
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Use Node.js 24.16.0
|
||||
uses: actions/setup-node@v4
|
||||
with:
|
||||
node-version: '24.16.0'
|
||||
|
||||
- name: Restore node_modules cache
|
||||
id: cache-node-modules
|
||||
uses: actions/cache@v4
|
||||
with:
|
||||
path: |
|
||||
node_modules
|
||||
api/node_modules
|
||||
packages/api/node_modules
|
||||
packages/data-provider/node_modules
|
||||
packages/data-schemas/node_modules
|
||||
key: node-modules-backend-${{ runner.os }}-24.16.0-${{ hashFiles('package-lock.json') }}
|
||||
|
||||
- name: Install dependencies
|
||||
if: steps.cache-node-modules.outputs.cache-hit != 'true'
|
||||
run: npm ci
|
||||
|
||||
- name: Restore data-provider build cache
|
||||
id: cache-data-provider
|
||||
uses: actions/cache@v4
|
||||
with:
|
||||
path: packages/data-provider/dist
|
||||
key: build-data-provider-${{ runner.os }}-${{ hashFiles('packages/data-provider/src/**', 'packages/data-provider/tsconfig*.json', 'packages/data-provider/tsdown.config.mjs', 'packages/data-provider/package.json') }}
|
||||
|
||||
- name: Build data-provider
|
||||
if: steps.cache-data-provider.outputs.cache-hit != 'true'
|
||||
run: npm run build:data-provider
|
||||
|
||||
- name: Restore data-schemas build cache
|
||||
id: cache-data-schemas
|
||||
uses: actions/cache@v4
|
||||
with:
|
||||
path: packages/data-schemas/dist
|
||||
key: build-data-schemas-${{ runner.os }}-${{ hashFiles('packages/data-schemas/src/**', 'packages/data-schemas/tsconfig*.json', 'packages/data-schemas/tsdown.config.mjs', 'packages/data-schemas/package.json', 'packages/data-provider/src/**', 'packages/data-provider/tsconfig*.json', 'packages/data-provider/tsdown.config.mjs', 'packages/data-provider/package.json') }}
|
||||
|
||||
- name: Build data-schemas
|
||||
if: steps.cache-data-schemas.outputs.cache-hit != 'true'
|
||||
run: npm run build:data-schemas
|
||||
|
||||
- name: Restore api build cache
|
||||
id: cache-api
|
||||
uses: actions/cache@v4
|
||||
with:
|
||||
path: packages/api/dist
|
||||
key: build-api-${{ runner.os }}-${{ hashFiles('packages/api/src/**', 'packages/api/tsconfig*.json', 'packages/api/tsdown.config.mjs', 'packages/api/package.json', 'packages/data-provider/src/**', 'packages/data-provider/tsconfig*.json', 'packages/data-provider/tsdown.config.mjs', 'packages/data-provider/package.json', 'packages/data-schemas/src/**', 'packages/data-schemas/tsconfig*.json', 'packages/data-schemas/tsdown.config.mjs', 'packages/data-schemas/package.json') }}
|
||||
|
||||
- name: Build api
|
||||
if: steps.cache-api.outputs.cache-hit != 'true'
|
||||
run: npm run build:api
|
||||
|
||||
- name: Run agents integration tests (in-process MongoDB)
|
||||
working-directory: packages/api
|
||||
env:
|
||||
NODE_ENV: test
|
||||
run: npm run test:agents-integration
|
||||
4
package-lock.json
generated
4
package-lock.json
generated
|
|
@ -20806,7 +20806,8 @@
|
|||
"version": "0.0.2",
|
||||
"resolved": "https://registry.npmjs.org/babel-plugin-replace-ts-export-assignment/-/babel-plugin-replace-ts-export-assignment-0.0.2.tgz",
|
||||
"integrity": "sha512-BiTEG2Ro+O1spuheL5nB289y37FFmz0ISE6GjpNCG2JuA/WNcuEHSYw01+vN8quGf208sID3FnZFDwVyqX18YQ==",
|
||||
"dev": true
|
||||
"dev": true,
|
||||
"license": "ISC"
|
||||
},
|
||||
"node_modules/babel-plugin-root-import": {
|
||||
"version": "6.6.0",
|
||||
|
|
@ -42920,6 +42921,7 @@
|
|||
"@types/winston": "^2.4.4",
|
||||
"@types/yauzl": "^2.10.3",
|
||||
"aws-sdk-client-mock": "^4.1.0",
|
||||
"babel-plugin-replace-ts-export-assignment": "^0.0.2",
|
||||
"dedent": "^1.5.3",
|
||||
"get-stream": "^6.0.1",
|
||||
"jest": "^30.2.0",
|
||||
|
|
|
|||
|
|
@ -33,6 +33,7 @@
|
|||
"test:cache-integration:stream": "jest --testPathPatterns=\"src/stream/.*\\.stream_integration\\.spec\\.ts$\" --coverage=false --runInBand --forceExit",
|
||||
"test:cache-integration": "npm run test:cache-integration:core && npm run test:cache-integration:cluster && npm run test:cache-integration:mcp && npm run test:cache-integration:stream",
|
||||
"test:s3-integration": "jest --testPathPatterns=\"src/storage/s3/.*\\.integration\\.spec\\.ts$\" --coverage=false --runInBand",
|
||||
"test:agents-integration": "jest --testPathPatterns=\"src/agents/.*\\.integration\\.spec\\.ts$\" --coverage=false --runInBand --forceExit",
|
||||
"verify": "npm run test:ci",
|
||||
"b:clean": "bun run rimraf dist",
|
||||
"b:build": "bun run b:clean && bun run tsdown",
|
||||
|
|
@ -77,6 +78,7 @@
|
|||
"@types/winston": "^2.4.4",
|
||||
"@types/yauzl": "^2.10.3",
|
||||
"aws-sdk-client-mock": "^4.1.0",
|
||||
"babel-plugin-replace-ts-export-assignment": "^0.0.2",
|
||||
"dedent": "^1.5.3",
|
||||
"get-stream": "^6.0.1",
|
||||
"jest": "^30.2.0",
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import mongoose from 'mongoose';
|
||||
import { logger } from '@librechat/data-schemas';
|
||||
import { MongoMemoryServer } from 'mongodb-memory-server';
|
||||
import { MongoDBSaver } from '@langchain/langgraph-checkpoint-mongodb';
|
||||
import { emptyCheckpoint, ERROR, INTERRUPT } from '@langchain/langgraph-checkpoint';
|
||||
|
|
@ -6,6 +7,8 @@ import {
|
|||
getAgentCheckpointer,
|
||||
deleteAgentCheckpoint,
|
||||
deleteAgentCheckpoints,
|
||||
LazyMongoSaver,
|
||||
CheckpointTooLargeError,
|
||||
__resetCheckpointerForTests,
|
||||
} from './checkpointer';
|
||||
|
||||
|
|
@ -446,3 +449,197 @@ describe('LazyMongoSaver (lazy persistence — mongodb-memory-server)', () => {
|
|||
expect(effects).toEqual({ a: 1, c: 1 }); // siblings did NOT re-execute
|
||||
});
|
||||
});
|
||||
|
||||
describe('LazyMongoSaver checkpoint size guard (mongodb-memory-server integration)', () => {
|
||||
// Guards the single-document ceiling: a checkpoint embeds the whole message history, so a
|
||||
// large conversation can serialize past MongoDB's 16 MB limit. The guard measures the
|
||||
// serialized state on the persist path, WARNS past a soft threshold, and REJECTS past a hard
|
||||
// limit BEFORE the write — a typed CheckpointTooLargeError instead of a raw BSONObjectTooLarge.
|
||||
// Thresholds are shrunk here so payloads stay tiny while exercising the real serde + Mongo.
|
||||
const clientForSaver = () =>
|
||||
// mongoose vends the live MongoClient; the driver type resolves to a different `mongodb` copy
|
||||
// than checkpoint-mongodb's, so the cast mirrors buildMongoSaver in checkpointer.ts.
|
||||
mongoose.connection.getClient() as unknown as ConstructorParameters<
|
||||
typeof MongoDBSaver
|
||||
>[0]['client'];
|
||||
|
||||
const makeSaver = (overrides?: { warnBytes?: number; hardLimitBytes?: number }) =>
|
||||
new LazyMongoSaver({
|
||||
client: clientForSaver(),
|
||||
checkpointCollectionName: 'agent_checkpoints',
|
||||
checkpointWritesCollectionName: 'agent_checkpoint_writes',
|
||||
ttl: 3600,
|
||||
...overrides,
|
||||
});
|
||||
|
||||
/** Seed an interrupt anchor for a checkpoint whose serialized state is inflated to ~`payloadBytes`. */
|
||||
async function seedSizedInterrupt(saver: LazyMongoSaver, threadId: string, payloadBytes: number) {
|
||||
const checkpoint = emptyCheckpoint();
|
||||
checkpoint.channel_values = { messages: 'x'.repeat(payloadBytes) };
|
||||
await saver.putWrites(
|
||||
{ configurable: { thread_id: threadId, checkpoint_ns: '', checkpoint_id: checkpoint.id } },
|
||||
[[INTERRUPT, 'approve?']],
|
||||
'task-1',
|
||||
);
|
||||
const config = { configurable: { thread_id: threadId, checkpoint_ns: '' } };
|
||||
const metadata = { source: 'input' as const, step: -1, writes: null, parents: {} };
|
||||
return { checkpoint, config, metadata };
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
jest.restoreAllMocks();
|
||||
});
|
||||
|
||||
it('persists a checkpoint under the soft threshold', async () => {
|
||||
const saver = makeSaver({ warnBytes: 5_000, hardLimitBytes: 50_000 });
|
||||
await saver.setup();
|
||||
const threadId = `convo-${new mongoose.Types.ObjectId().toString()}`;
|
||||
const { checkpoint, config, metadata } = await seedSizedInterrupt(saver, threadId, 100);
|
||||
|
||||
await saver.put(config, checkpoint, metadata);
|
||||
|
||||
expect(await saver.getTuple(readConfig(threadId))).toBeDefined();
|
||||
});
|
||||
|
||||
it('persists but WARNS when a checkpoint crosses the soft threshold', async () => {
|
||||
const warnSpy = jest.spyOn(logger, 'warn').mockImplementation(() => logger);
|
||||
const saver = makeSaver({ warnBytes: 500, hardLimitBytes: 50_000 });
|
||||
await saver.setup();
|
||||
const threadId = `convo-${new mongoose.Types.ObjectId().toString()}`;
|
||||
const { checkpoint, config, metadata } = await seedSizedInterrupt(saver, threadId, 2_000);
|
||||
|
||||
await saver.put(config, checkpoint, metadata);
|
||||
|
||||
expect(await saver.getTuple(readConfig(threadId))).toBeDefined();
|
||||
expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining('soft threshold'));
|
||||
});
|
||||
|
||||
it('REFUSES to persist and throws CheckpointTooLargeError over the hard limit', async () => {
|
||||
jest.spyOn(logger, 'error').mockImplementation(() => logger);
|
||||
const saver = makeSaver({ warnBytes: 500, hardLimitBytes: 2_000 });
|
||||
await saver.setup();
|
||||
const threadId = `convo-${new mongoose.Types.ObjectId().toString()}`;
|
||||
const { checkpoint, config, metadata } = await seedSizedInterrupt(saver, threadId, 10_000);
|
||||
|
||||
await expect(saver.put(config, checkpoint, metadata)).rejects.toBeInstanceOf(
|
||||
CheckpointTooLargeError,
|
||||
);
|
||||
|
||||
// Nothing durable was written for that thread — getTuple reads the checkpoint document.
|
||||
expect(await saver.getTuple(readConfig(threadId))).toBeUndefined();
|
||||
expect(
|
||||
await mongoose.connection.db!.collection('agent_checkpoints').countDocuments({
|
||||
thread_id: threadId,
|
||||
}),
|
||||
).toBe(0);
|
||||
});
|
||||
|
||||
it('counts metadata toward the ceiling — refuses when the checkpoint is under but metadata pushes over', async () => {
|
||||
// MongoDBSaver stores the serialized checkpoint AND metadata in one document, so a
|
||||
// just-under-limit checkpoint with large metadata still overflows. The guard must catch it
|
||||
// as a typed CheckpointTooLargeError, not let it fall through to a raw BSONObjectTooLarge.
|
||||
jest.spyOn(logger, 'error').mockImplementation(() => logger);
|
||||
const saver = makeSaver({ warnBytes: 500, hardLimitBytes: 2_000 });
|
||||
await saver.setup();
|
||||
const threadId = `convo-${new mongoose.Types.ObjectId().toString()}`;
|
||||
|
||||
const checkpoint = emptyCheckpoint();
|
||||
checkpoint.channel_values = { messages: 'x'.repeat(400) }; // checkpoint alone well UNDER 2 KB
|
||||
await saver.putWrites(
|
||||
{ configurable: { thread_id: threadId, checkpoint_ns: '', checkpoint_id: checkpoint.id } },
|
||||
[[INTERRUPT, 'approve?']],
|
||||
'task-1',
|
||||
);
|
||||
const config = { configurable: { thread_id: threadId, checkpoint_ns: '' } };
|
||||
// Metadata alone pushes checkpoint + metadata over the 2 KB hard limit.
|
||||
const metadata = {
|
||||
source: 'input' as const,
|
||||
step: -1,
|
||||
writes: { payload: 'y'.repeat(4_000) },
|
||||
parents: {},
|
||||
};
|
||||
|
||||
await expect(saver.put(config, checkpoint, metadata)).rejects.toBeInstanceOf(
|
||||
CheckpointTooLargeError,
|
||||
);
|
||||
expect(await saver.getTuple(readConfig(threadId))).toBeUndefined();
|
||||
});
|
||||
|
||||
it('counts metadata_search (the raw metadata copy) — refuses when serialized-only would pass', async () => {
|
||||
// MongoDBSaver stores `metadata_search: metadata` — the WHOLE raw metadata a
|
||||
// SECOND time in the same document. Sized so checkpoint + serialized metadata is
|
||||
// UNDER the limit but adding the raw metadata_search copy pushes it over: the guard
|
||||
// must count that copy, else the doc slips the preflight and overflows on the wire.
|
||||
jest.spyOn(logger, 'error').mockImplementation(() => logger);
|
||||
const saver = makeSaver({ warnBytes: 500, hardLimitBytes: 6_000 });
|
||||
await saver.setup();
|
||||
const threadId = `convo-${new mongoose.Types.ObjectId().toString()}`;
|
||||
|
||||
const checkpoint = emptyCheckpoint();
|
||||
checkpoint.channel_values = { messages: 'x'.repeat(200) };
|
||||
await saver.putWrites(
|
||||
{ configurable: { thread_id: threadId, checkpoint_ns: '', checkpoint_id: checkpoint.id } },
|
||||
[[INTERRUPT, 'approve?']],
|
||||
'task-1',
|
||||
);
|
||||
const config = { configurable: { thread_id: threadId, checkpoint_ns: '' } };
|
||||
// ~3 KB metadata: checkpoint + serialized metadata (~3.2 KB) is under 6 KB, but the
|
||||
// raw metadata_search copy (~another 3 KB) pushes checkpoint+metadata+metadata_search
|
||||
// over 6 KB. The pre-fix guard (checkpoint + serialized metadata only) would pass here.
|
||||
const metadata = {
|
||||
source: 'loop' as const,
|
||||
step: 1,
|
||||
writes: { payload: 'y'.repeat(3_000) },
|
||||
parents: {},
|
||||
};
|
||||
|
||||
await expect(saver.put(config, checkpoint, metadata)).rejects.toBeInstanceOf(
|
||||
CheckpointTooLargeError,
|
||||
);
|
||||
expect(await saver.getTuple(readConfig(threadId))).toBeUndefined();
|
||||
});
|
||||
|
||||
it('flushes bookkeeping parked during the size-serialization window (not dropped on resume)', async () => {
|
||||
// `put` consumes the write anchor, then AWAITS `assertCheckpointFitsDocument` (serialization).
|
||||
// A bookkeeping-only putWrites dispatched in that window sees no anchor and no persisted
|
||||
// marker, so it parks — and must be flushed once the checkpoint persists, or a resume
|
||||
// re-executes the completed sibling.
|
||||
const NO_WRITES = '__no_writes__';
|
||||
const saver = makeSaver({ warnBytes: 5_000, hardLimitBytes: 50_000 });
|
||||
await saver.setup();
|
||||
const threadId = `convo-${new mongoose.Types.ObjectId().toString()}`;
|
||||
const checkpoint = emptyCheckpoint();
|
||||
const writeCfg = {
|
||||
configurable: { thread_id: threadId, checkpoint_ns: '', checkpoint_id: checkpoint.id },
|
||||
};
|
||||
await saver.putWrites(writeCfg, [[INTERRUPT, 'approve?']], 'task-gate');
|
||||
|
||||
// Pause the checkpoint serialization inside `put` so the bookkeeping batch lands mid-window.
|
||||
const serde = (saver as unknown as { serde: { dumpsTyped: (v: unknown) => unknown } }).serde;
|
||||
const realDumps = serde.dumpsTyped.bind(serde);
|
||||
let releaseGate!: () => void;
|
||||
const gate = new Promise<void>((resolve) => {
|
||||
releaseGate = resolve;
|
||||
});
|
||||
let paused = false;
|
||||
jest.spyOn(serde, 'dumpsTyped').mockImplementation(async (value: unknown) => {
|
||||
if (!paused) {
|
||||
paused = true;
|
||||
await gate; // hold inside assertCheckpointFitsDocument (anchor consumed, not yet persisted)
|
||||
}
|
||||
return realDumps(value);
|
||||
});
|
||||
|
||||
const config = { configurable: { thread_id: threadId, checkpoint_ns: '' } };
|
||||
const metadata = { source: 'input' as const, step: -1, writes: null, parents: {} };
|
||||
const putPromise = saver.put(config, checkpoint, metadata); // suspends at the gate
|
||||
await saver.putWrites(writeCfg, [[NO_WRITES, null]], 'task-sibling'); // parks in the window
|
||||
releaseGate();
|
||||
await putPromise;
|
||||
|
||||
const tuple = await saver.getTuple(readConfig(threadId));
|
||||
const channels = (tuple?.pendingWrites ?? []).map((w) => w[1]);
|
||||
expect(channels).toContain(INTERRUPT);
|
||||
expect(channels).toContain(NO_WRITES); // flushed, not dropped
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -124,6 +124,69 @@ interface BufferedWriteBatch {
|
|||
batches: Array<{ config: RunnableConfig; writes: PendingWrite[]; taskId: string }>;
|
||||
}
|
||||
|
||||
/**
|
||||
* MongoDB's hard per-document ceiling. A checkpoint whose serialized state pushes its
|
||||
* document past this cannot be stored — the driver throws `BSONObjectTooLarge` (code 10334).
|
||||
*/
|
||||
const MAX_BSON_DOCUMENT_BYTES = 16 * 1024 * 1024;
|
||||
|
||||
/**
|
||||
* Headroom reserved below {@link MAX_BSON_DOCUMENT_BYTES} for a checkpoint document's
|
||||
* non-state fields (ids, metadata, `metadata_search`, BSON framing). The serialized
|
||||
* `checkpoint` blob dominates the document; this margin covers everything else so the guard
|
||||
* rejects before Mongo does — with a legible error instead of a raw driver failure.
|
||||
*/
|
||||
const CHECKPOINT_SIZE_HEADROOM_BYTES = 1024 * 1024;
|
||||
|
||||
/**
|
||||
* Reject a checkpoint whose serialized state exceeds this. The pause is unrecoverable either
|
||||
* way (the document can't be written), so failing here as a typed {@link CheckpointTooLargeError}
|
||||
* turns an opaque `BSONObjectTooLarge` crash into an actionable one.
|
||||
*/
|
||||
export const CHECKPOINT_HARD_LIMIT_BYTES: number =
|
||||
MAX_BSON_DOCUMENT_BYTES - CHECKPOINT_SIZE_HEADROOM_BYTES;
|
||||
|
||||
/**
|
||||
* Warn once a persisted checkpoint crosses this soft threshold (~50% of the ceiling), so a
|
||||
* conversation's checkpoint growth is visible in logs well before it reaches the hard limit.
|
||||
*/
|
||||
export const CHECKPOINT_WARN_BYTES: number = 8 * 1024 * 1024;
|
||||
|
||||
/**
|
||||
* A HITL checkpoint whose serialized state exceeds {@link CHECKPOINT_HARD_LIMIT_BYTES} — more
|
||||
* than MongoDB can hold in a single document. Thrown BEFORE the doomed write so the run fails
|
||||
* with a clear, typed message instead of a raw driver `BSONObjectTooLarge`. The pause cannot be
|
||||
* persisted regardless of how it is handled upstream; a durable resume is impossible for this turn.
|
||||
*/
|
||||
export class CheckpointTooLargeError extends Error {
|
||||
readonly code = 'CHECKPOINT_TOO_LARGE';
|
||||
constructor(
|
||||
readonly bytes: number,
|
||||
readonly limit: number,
|
||||
readonly threadId?: string,
|
||||
) {
|
||||
const mb = (n: number): string => (n / 1024 / 1024).toFixed(1);
|
||||
super(
|
||||
`Checkpoint state is ${mb(bytes)} MB, over the ${mb(limit)} MB limit for a durable pause. ` +
|
||||
'This conversation carries too much state to pause for input — large tool outputs or ' +
|
||||
'inlined media are the usual cause. Start a new conversation or reduce context.',
|
||||
);
|
||||
this.name = 'CheckpointTooLargeError';
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Construction options for {@link LazyMongoSaver}: the base saver options plus optional
|
||||
* size-guard overrides. The overrides default to the module thresholds and exist so tests can
|
||||
* exercise the guard at small sizes; production always uses the defaults.
|
||||
*/
|
||||
export type LazyMongoSaverOptions = ConstructorParameters<typeof MongoDBSaver>[0] & {
|
||||
/** Soft warn threshold in bytes. Defaults to {@link CHECKPOINT_WARN_BYTES}. */
|
||||
warnBytes?: number;
|
||||
/** Hard reject limit in bytes. Defaults to {@link CHECKPOINT_HARD_LIMIT_BYTES}. */
|
||||
hardLimitBytes?: number;
|
||||
};
|
||||
|
||||
export class LazyMongoSaver extends MongoDBSaver {
|
||||
/** checkpoint id → time the resumable `putWrites` anchoring it arrived; consumed by `put`. */
|
||||
private readonly writeAnchorIds = new Map<string, number>();
|
||||
|
|
@ -133,6 +196,18 @@ export class LazyMongoSaver extends MongoDBSaver {
|
|||
/** checkpoint id → bookkeeping batches parked until the checkpoint persists or is discarded. */
|
||||
private readonly bufferedBookkeeping = new Map<string, BufferedWriteBatch>();
|
||||
|
||||
/** Soft threshold (bytes) past which a persisted checkpoint is warned about. */
|
||||
private readonly warnBytes: number;
|
||||
/** Hard limit (bytes) past which a checkpoint is refused with {@link CheckpointTooLargeError}. */
|
||||
private readonly hardLimitBytes: number;
|
||||
|
||||
constructor(options: LazyMongoSaverOptions) {
|
||||
const { warnBytes, hardLimitBytes, ...mongoOptions } = options;
|
||||
super(mongoOptions);
|
||||
this.warnBytes = warnBytes ?? CHECKPOINT_WARN_BYTES;
|
||||
this.hardLimitBytes = hardLimitBytes ?? CHECKPOINT_HARD_LIMIT_BYTES;
|
||||
}
|
||||
|
||||
override async putWrites(
|
||||
config: RunnableConfig,
|
||||
writes: PendingWrite[],
|
||||
|
|
@ -174,15 +249,9 @@ export class LazyMongoSaver extends MongoDBSaver {
|
|||
// wrongly discarded.
|
||||
this.recordWriteAnchor(checkpointId);
|
||||
try {
|
||||
const buffered = this.bufferedBookkeeping.get(checkpointId);
|
||||
this.bufferedBookkeeping.delete(checkpointId);
|
||||
if (buffered) {
|
||||
// The checkpoint's fate is now "persist" — flush the bookkeeping batches that
|
||||
// arrived before this anchor so the stored pending writes are complete.
|
||||
await Promise.all(
|
||||
buffered.batches.map((b) => super.putWrites(b.config, b.writes, b.taskId)),
|
||||
);
|
||||
}
|
||||
// The checkpoint's fate is now "persist" — flush the bookkeeping batches that
|
||||
// arrived before this anchor so the stored pending writes are complete.
|
||||
await this.flushBufferedBookkeeping(checkpointId);
|
||||
return await super.putWrites(config, writes, taskId);
|
||||
} catch (err) {
|
||||
// The write batch never landed — best-effort un-anchor so the concurrent `put` doesn't
|
||||
|
|
@ -203,9 +272,17 @@ export class LazyMongoSaver extends MongoDBSaver {
|
|||
// Carries a resumable write (interrupt / real-channel delta anchor) — persist so resume
|
||||
// can read it, and remember the id briefly so any bookkeeping batch dispatched after
|
||||
// this `put` is forwarded rather than parked.
|
||||
await this.assertCheckpointFitsDocument(config, checkpoint, metadata);
|
||||
sweepStale(this.persistedIds, (t) => t);
|
||||
this.persistedIds.set(checkpoint.id, Date.now());
|
||||
return super.put(config, checkpoint, metadata);
|
||||
const persisted = await super.put(config, checkpoint, metadata);
|
||||
// `assertCheckpointFitsDocument` awaits a (potentially slow) serialization AFTER the
|
||||
// anchor was consumed above but BEFORE `persistedIds` was set — a bookkeeping-only
|
||||
// `putWrites` dispatched in that window sees neither marker and parks its batch. Flush
|
||||
// it now that the checkpoint is persisted; without this the marker is dropped and a
|
||||
// resume can re-execute already-completed work.
|
||||
await this.flushBufferedBookkeeping(checkpoint.id);
|
||||
return persisted;
|
||||
}
|
||||
// No resumable writes ⇒ a clean exit (a non-paused completion, a resumed turn's clean
|
||||
// finish, or an error-only turn): discard, and drop the parked bookkeeping batches with
|
||||
|
|
@ -233,6 +310,74 @@ export class LazyMongoSaver extends MongoDBSaver {
|
|||
sweepStale(this.writeAnchorIds, (t) => t);
|
||||
this.writeAnchorIds.set(checkpointId, Date.now());
|
||||
}
|
||||
|
||||
/**
|
||||
* Forward the bookkeeping batches parked for `checkpointId` while its fate was undecided,
|
||||
* now that the checkpoint is being persisted. Snapshot-and-delete before awaiting so a batch
|
||||
* that arrives afterwards can't be double-forwarded — by then the anchor/persisted marker is
|
||||
* set, so it forwards directly instead of parking. Shared by the anchoring `putWrites` and by
|
||||
* `put` (for a batch parked during the size-check serialization window).
|
||||
*/
|
||||
private async flushBufferedBookkeeping(checkpointId: string): Promise<void> {
|
||||
const buffered = this.bufferedBookkeeping.get(checkpointId);
|
||||
if (!buffered) {
|
||||
return;
|
||||
}
|
||||
this.bufferedBookkeeping.delete(checkpointId);
|
||||
await Promise.all(buffered.batches.map((b) => super.putWrites(b.config, b.writes, b.taskId)));
|
||||
}
|
||||
|
||||
/**
|
||||
* Measure the checkpoint's serialized size on the persist path and act on it: `debug`-log it,
|
||||
* `warn` past {@link warnBytes}, and throw {@link CheckpointTooLargeError} past
|
||||
* {@link hardLimitBytes} — BEFORE the write, so an oversize pause fails legibly rather than as a
|
||||
* raw `BSONObjectTooLarge`. Serializes with the same `serde` the base `put` uses, so the measured
|
||||
* bytes match what would be stored. The extra serialization runs only on the (rare) HITL pause
|
||||
* path — never the clean-exit common path, which is discarded before reaching here.
|
||||
*/
|
||||
private async assertCheckpointFitsDocument(
|
||||
config: RunnableConfig,
|
||||
checkpoint: Checkpoint,
|
||||
metadata: CheckpointMetadata,
|
||||
): Promise<void> {
|
||||
// `MongoDBSaver.put` writes THREE size-bearing fields into the same
|
||||
// `agent_checkpoints` document: the serialized `checkpoint`, the serialized
|
||||
// `metadata`, AND `metadata_search` — the WHOLE raw `metadata` object stored a
|
||||
// second time as a queryable BSON subdocument (`metadata_search: metadata`).
|
||||
// So a large `metadata` (e.g. `metadata.writes` holding a big tool result) is
|
||||
// counted twice on the wire. Measuring only checkpoint + serialized metadata
|
||||
// let such a document pass the preflight while `metadata_search` pushed the
|
||||
// actual BSON past 16 MB — the raw `BSONObjectTooLarge` this guard exists to
|
||||
// prevent. Add the raw metadata's BSON size; the headroom now only has to
|
||||
// cover ids and BSON framing.
|
||||
const [, serializedCheckpoint] = await this.serde.dumpsTyped(checkpoint);
|
||||
const [, serializedMetadata] = await this.serde.dumpsTyped(metadata);
|
||||
const metadataSearchBytes = mongoose.mongo.BSON.calculateObjectSize(
|
||||
metadata as unknown as Record<string, unknown>,
|
||||
);
|
||||
const bytes =
|
||||
serializedCheckpoint.byteLength + serializedMetadata.byteLength + metadataSearchBytes;
|
||||
const threadId = config.configurable?.thread_id as string | undefined;
|
||||
const mb = (n: number): string => (n / 1024 / 1024).toFixed(1);
|
||||
if (bytes > this.hardLimitBytes) {
|
||||
// The anchoring write row was already persisted by `putWrites`; the pre-run prune and Mongo
|
||||
// TTL reclaim it. Drop any parked bookkeeping so it doesn't linger in memory.
|
||||
this.bufferedBookkeeping.delete(checkpoint.id);
|
||||
logger.error(
|
||||
`[checkpointer] HITL checkpoint for thread ${threadId ?? 'unknown'} is ${mb(bytes)} MB, over the ${mb(this.hardLimitBytes)} MB durable-pause limit; refusing the write (a document past 16 MB cannot be stored in MongoDB).`,
|
||||
);
|
||||
throw new CheckpointTooLargeError(bytes, this.hardLimitBytes, threadId);
|
||||
}
|
||||
if (bytes >= this.warnBytes) {
|
||||
logger.warn(
|
||||
`[checkpointer] HITL checkpoint for thread ${threadId ?? 'unknown'} is ${mb(bytes)} MB, past the ${mb(this.warnBytes)} MB soft threshold (hard limit ${mb(this.hardLimitBytes)} MB) — approaching MongoDB's single-document ceiling.`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
logger.debug(
|
||||
`[checkpointer] Persisting HITL checkpoint for thread ${threadId ?? 'unknown'}: ${bytes} bytes`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue