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; }