From 7083cf8935a40febdb4e5d8e44f58098ee8fba42 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Tue, 14 Jul 2026 11:58:04 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=91=BB=20fix:=20Guard=20Redis=20Stream=20?= =?UTF-8?q?Resume=20Against=20Disposed=20HITL=20Graph=20(#14258)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a run pauses for `ask_user_question` (or tool approval), the paused turn's client is disposed and `disposeClient` (api/server/cleanup.js `graphPropsToClean`) runs `clearHeavyState()` and then NULLS the graph's internal arrays (`messages`, `contentData`). `RedisJobStore` still holds a `WeakRef` to that graph in `localGraphCache`, so on resume/reconnect `getContentParts()` / `getRunSteps()` deref the disposed graph and call `graph.getContentParts()` (`this.messages.slice()`) / `graph.getRunSteps()` (`[...this.contentData]`), throwing "Cannot read properties of null (reading 'slice')" / "this.contentData is not iterable" and aborting the resume. Redis-only: InMemoryJobStore reads `state.contentParts` / `graph?.contentData ?? []` directly and never calls the throwing SDK getters, which is why `USE_REDIS_STREAMS=false` works. RedisJobStore now reads the cached graph through `readCachedGraph()`, which tolerates a disposed graph: it swallows the deref error, drops the stale cache entry, and lets the caller fall back to durable chunk reconstruction instead of crashing. The true root cause (the unguarded null deref) is in `@librechat/agents` `StandardGraph.getContentParts()` / `getRunSteps()` and is addressed there separately; this is the host-side defensive guard. Fixes #14247. Addresses Bug 3 of #14253. --- ...hitlResumeRedis.stream_integration.spec.ts | 95 +++++++++++++++++++ .../stream/implementations/RedisJobStore.ts | 33 ++++++- 2 files changed, 126 insertions(+), 2 deletions(-) create mode 100644 packages/api/src/stream/__tests__/hitlResumeRedis.stream_integration.spec.ts diff --git a/packages/api/src/stream/__tests__/hitlResumeRedis.stream_integration.spec.ts b/packages/api/src/stream/__tests__/hitlResumeRedis.stream_integration.spec.ts new file mode 100644 index 0000000000..f99e0f5bd5 --- /dev/null +++ b/packages/api/src/stream/__tests__/hitlResumeRedis.stream_integration.spec.ts @@ -0,0 +1,95 @@ +import type { Redis, Cluster } from 'ioredis'; + +jest.spyOn(console, 'log').mockImplementation(); + +/** + * Regression coverage for #14247 / #14253 (Bug 3): resuming an `ask_user_question` + * pause with Redis Streams enabled must not crash on a graph disposed at pause time. + * + * Run with real Redis: + * USE_REDIS=true REDIS_URI=redis://127.0.0.1:6379 \ + * npx jest hitlResumeRedis.stream_integration + */ +describe('HITL ask_user_question resume (Redis Streams)', () => { + let ioredisClient: Redis | Cluster | null = null; + let originalEnv: NodeJS.ProcessEnv; + + beforeAll(async () => { + originalEnv = { ...process.env }; + process.env.USE_REDIS = process.env.USE_REDIS ?? 'true'; + process.env.USE_REDIS_CLUSTER = process.env.USE_REDIS_CLUSTER ?? 'false'; + process.env.REDIS_URI = process.env.REDIS_URI ?? 'redis://127.0.0.1:6379'; + process.env.REDIS_KEY_PREFIX = 'HitlResumeRedisTest'; + process.env.REDIS_PING_INTERVAL = '0'; + jest.resetModules(); + const { ioredisClient: client } = await import('../../cache/redisClients'); + ioredisClient = client; + }); + + afterAll(async () => { + if (ioredisClient) { + const keys = await ioredisClient.keys(`HitlResumeRedisTest*`); + const streamKeys = await ioredisClient.keys(`stream:*`); + await Promise.all([...keys, ...streamKeys].map((k) => ioredisClient!.del(k))); + await ioredisClient.quit(); + } + process.env = originalEnv; + }); + + /** + * A `StandardGraph` disposed after a HITL pause: `disposeClient` + * (api/server/cleanup.js `graphPropsToClean`) nulls `messages`/`contentData`, so the + * SDK getters throw exactly as they would on the real object. This is the #14247 crash. + */ + function disposedGraph() { + return { + messages: null, + contentData: null, + startIndex: null, + getContentParts() { + return (this.messages as unknown as unknown[]).slice(0); + }, + getRunSteps() { + return [...(this.contentData as unknown as unknown[])]; + }, + }; + } + + test('getContentParts / getRunSteps tolerate a disposed cached graph (no null.slice crash)', async () => { + if (!ioredisClient) { + console.warn('no redis'); + return; + } + const { RedisJobStore } = await import('../implementations/RedisJobStore'); + const store = new RedisJobStore(ioredisClient); + await store.initialize(); + + const streamId = `disposed-${Date.now()}`; + await store.createJob(streamId, 'user-1', streamId); + // Persist a chunk so reconstruction has durable content to fall back to. + await store.appendChunk(streamId, { + event: 'on_run_step', + data: { + id: 'step-ask', + runId: 'resp-1', + index: 0, + stepDetails: { + type: 'tool_calls', + tool_calls: [{ id: 'call-ask', name: 'ask_user_question', args: '' }], + }, + }, + }); + + // Cache the graph, then dispose it (mirrors AgentClient pause + disposeClient). + store.setGraph(streamId, disposedGraph() as never); + + // Before the fix these threw "Cannot read properties of null (reading 'slice')" + // and "this.contentData is not iterable" respectively — the awaits would reject. + const content = await store.getContentParts(streamId); + expect(content?.content?.[0]).toMatchObject({ tool_call: { name: 'ask_user_question' } }); + const runSteps = await store.getRunSteps(streamId); + expect(Array.isArray(runSteps)).toBe(true); + + await store.destroy(); + }); +}); diff --git a/packages/api/src/stream/implementations/RedisJobStore.ts b/packages/api/src/stream/implementations/RedisJobStore.ts index 2d84f207ba..3000ac843c 100644 --- a/packages/api/src/stream/implementations/RedisJobStore.ts +++ b/packages/api/src/stream/implementations/RedisJobStore.ts @@ -1121,6 +1121,35 @@ export class RedisJobStore implements IJobStore { * @param streamId - The stream identifier * @returns Content parts array or null if not found */ + /** + * Read from a cached {@link StandardGraph}, tolerating one disposed after a HITL + * pause. When a paused turn's client is disposed, `disposeClient` + * (api/server/cleanup.js `graphPropsToClean`) NULLS the graph's internal arrays + * (`messages`, `contentData`) for GC — but this store still holds a WeakRef to + * that object. Calling `getContentParts()` (`this.messages.slice()`) or + * `getRunSteps()` (`[...this.contentData]`) on it then throws + * ("Cannot read properties of null (reading 'slice')" / "not iterable"), which + * aborts the resume (#14247). Swallow it, drop the stale entry, and let the + * caller fall back to durable chunk reconstruction. (The SDK-side null guard in + * `StandardGraph` is a separate agents fix.) + */ + private readCachedGraph( + streamId: string, + graph: StandardGraph, + read: (graph: StandardGraph) => T, + ): T | null { + try { + return read(graph); + } catch (err) { + logger.debug( + `[RedisJobStore] Cached graph for ${streamId} is unusable (likely disposed); falling back to reconstruction:`, + err instanceof Error ? err.message : err, + ); + this.localGraphCache.delete(streamId); + return null; + } + } + async getContentParts(streamId: string): Promise<{ content: Agents.MessageContentComplex[]; } | null> { @@ -1145,7 +1174,7 @@ export class RedisJobStore implements IJobStore { if (graphRef) { const graph = graphRef.deref(); if (graph) { - const localParts = graph.getContentParts(); + const localParts = this.readCachedGraph(streamId, graph, (g) => g.getContentParts()); if (localParts && localParts.length > 0) { return { content: await this.overlayHostSteerParts(streamId, localParts), @@ -1231,7 +1260,7 @@ export class RedisJobStore implements IJobStore { if (graphRef) { const graph = graphRef.deref(); if (graph) { - const localSteps = graph.getRunSteps(); + const localSteps = this.readCachedGraph(streamId, graph, (g) => g.getRunSteps()); if (localSteps && localSteps.length > 0) { return localSteps; }