From 3c39cd0f653a3460be3dbe27d9a396af9e39bf5b Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Tue, 19 May 2026 13:33:15 -0400 Subject: [PATCH] fix: Address HITL review findings --- .../routes/agents/__tests__/abort.spec.js | 50 +++++ api/server/routes/agents/index.js | 12 +- packages/api/src/agents/hitl/policy.spec.ts | 22 ++- packages/api/src/agents/hitl/policy.ts | 11 +- .../RedisJobStore.stream_integration.spec.ts | 109 +++++++++++ .../stream/implementations/RedisJobStore.ts | 173 +++++++++++++----- 6 files changed, 318 insertions(+), 59 deletions(-) diff --git a/api/server/routes/agents/__tests__/abort.spec.js b/api/server/routes/agents/__tests__/abort.spec.js index 418c5f4254..6d90d332df 100644 --- a/api/server/routes/agents/__tests__/abort.spec.js +++ b/api/server/routes/agents/__tests__/abort.spec.js @@ -323,6 +323,56 @@ describe('Agent Abort Endpoint', () => { }); describe('Job Not Found', () => { + it('should skip paused fallback jobs and abort the running job', async () => { + mockGenerationJobManager.getJob + .mockResolvedValueOnce(null) + .mockResolvedValueOnce({ + status: 'requires_action', + metadata: { userId: 'test-user-123' }, + }) + .mockResolvedValueOnce({ + status: 'running', + metadata: { userId: 'test-user-123' }, + }); + mockGenerationJobManager.getActiveJobIdsForUser.mockResolvedValue([ + 'paused-stream', + 'running-stream', + ]); + mockGenerationJobManager.abortJob.mockResolvedValue({ + success: true, + jobData: null, + content: [], + text: '', + }); + + const response = await request(app) + .post('/api/agents/chat/abort') + .send({ conversationId: 'new' }); + + expect(response.status).toBe(200); + expect(response.body).toEqual({ success: true, aborted: 'running-stream' }); + expect(mockGenerationJobManager.abortJob).toHaveBeenCalledWith('running-stream'); + }); + + it('should not abort paused fallback jobs', async () => { + mockGenerationJobManager.getJob.mockResolvedValueOnce(null).mockResolvedValueOnce({ + status: 'requires_action', + metadata: { userId: 'test-user-123' }, + }); + mockGenerationJobManager.getActiveJobIdsForUser.mockResolvedValue(['paused-stream']); + + const response = await request(app) + .post('/api/agents/chat/abort') + .send({ conversationId: 'new' }); + + expect(response.status).toBe(404); + expect(response.body).toEqual({ + error: 'Job not found', + streamId: null, + }); + expect(mockGenerationJobManager.abortJob).not.toHaveBeenCalled(); + }); + it('should return 404 when job is not found', async () => { mockGenerationJobManager.getJob.mockResolvedValue(null); mockGenerationJobManager.getActiveJobIdsForUser.mockResolvedValue([]); diff --git a/api/server/routes/agents/index.js b/api/server/routes/agents/index.js index 145a6c0316..568afee3fa 100644 --- a/api/server/routes/agents/index.js +++ b/api/server/routes/agents/index.js @@ -242,11 +242,15 @@ router.post('/chat/abort', async (req, res) => { userId, req.user.tenantId, ); - if (activeJobIds.length > 0) { - // Abort the most recent active job for this user - jobStreamId = activeJobIds[0]; - job = await GenerationJobManager.getJob(jobStreamId); + for (const activeJobId of activeJobIds) { + const activeJob = await GenerationJobManager.getJob(activeJobId); + if (activeJob?.status !== 'running') { + continue; + } + jobStreamId = activeJobId; + job = activeJob; logger.debug(`[AgentStream] Found active job for user: ${jobStreamId}`); + break; } } diff --git a/packages/api/src/agents/hitl/policy.spec.ts b/packages/api/src/agents/hitl/policy.spec.ts index 35c80b95a2..369922a5ff 100644 --- a/packages/api/src/agents/hitl/policy.spec.ts +++ b/packages/api/src/agents/hitl/policy.spec.ts @@ -8,16 +8,16 @@ import { } from './policy'; describe('isHITLEnabled', () => { - test('default-on when no policy configured (SDK default)', () => { - expect(isHITLEnabled(undefined)).toBe(true); + test('default-off when no policy configured', () => { + expect(isHITLEnabled(undefined)).toBe(false); }); - test('default-on when policy is configured but `enabled` is omitted', () => { - expect(isHITLEnabled({})).toBe(true); - expect(isHITLEnabled({ mode: 'default', allow: ['read_*'] })).toBe(true); + test('default-off when policy is configured but `enabled` is omitted', () => { + expect(isHITLEnabled({})).toBe(false); + expect(isHITLEnabled({ mode: 'default', allow: ['read_*'] })).toBe(false); }); - test('explicit false is the only off signal', () => { + test('explicit false is off', () => { expect(isHITLEnabled({ enabled: false })).toBe(false); }); @@ -216,4 +216,14 @@ describe('buildPendingAction', () => { expect(withTtl.expiresAt).toBeGreaterThanOrEqual(before + ttl); expect(withTtl.expiresAt).toBeLessThanOrEqual(after + ttl); }); + + test('honours ttlMs 0 as immediate expiry', () => { + const before = Date.now(); + const action = buildPendingAction(toolApprovalPayload, { ...ctx, ttlMs: 0 }); + const after = Date.now(); + + expect(action.expiresAt).toBeDefined(); + expect(action.expiresAt).toBeGreaterThanOrEqual(before); + expect(action.expiresAt).toBeLessThanOrEqual(after); + }); }); diff --git a/packages/api/src/agents/hitl/policy.ts b/packages/api/src/agents/hitl/policy.ts index 0f9c5eebf8..43013a6f5b 100644 --- a/packages/api/src/agents/hitl/policy.ts +++ b/packages/api/src/agents/hitl/policy.ts @@ -14,10 +14,9 @@ const DEFAULT_REVIEW_DECISIONS: Agents.ToolApprovalDecisionType[] = ['approve', /** * Whether the HITL machinery should run for this policy. * - * `false` is the LibreChat-only admin kill switch — it disables the SDK - * checkpointer fallback and skips installing the policy hook entirely. - * Users wanting "stop asking me" should use `mode: 'bypass'` instead, which - * keeps the machinery in place but auto-approves. + * HITL remains default-off for the rollout; `enabled: true` is the explicit + * opt-in. Users wanting "stop asking me" after opting in should use + * `mode: 'bypass'` instead, which keeps the machinery in place but auto-approves. * * **Wiring caveat (Slice B):** when this returns `true` and the host passes * `humanInTheLoop: { enabled: true }` to `Run.create`, the host MUST also @@ -28,7 +27,7 @@ const DEFAULT_REVIEW_DECISIONS: Agents.ToolApprovalDecisionType[] = ['approve', * assignment at the `Run.create` call site. */ export function isHITLEnabled(policy: TToolApprovalPolicy | undefined): boolean { - return policy?.enabled !== false; + return policy?.enabled === true; } /** @@ -140,6 +139,6 @@ export function buildPendingAction( responseMessageId: ctx.responseMessageId, payload, createdAt, - expiresAt: ctx.ttlMs ? createdAt + ctx.ttlMs : undefined, + expiresAt: typeof ctx.ttlMs === 'number' ? createdAt + ctx.ttlMs : undefined, }; } diff --git a/packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts b/packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts index 23f25b1291..c06b6df563 100644 --- a/packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts +++ b/packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts @@ -22,6 +22,19 @@ describe('RedisJobStore Integration Tests', () => { let ioredisClient: Redis | Cluster | null = null; const testPrefix = 'Stream-Integration-Test'; + function buildPendingAction(streamId: string): Agents.PendingAction { + return { + actionId: `action-${streamId}`, + streamId, + conversationId: streamId, + payload: { + type: 'ask_user_question', + question: { question: 'Approve?' }, + }, + createdAt: Date.now(), + }; + } + beforeAll(async () => { originalEnv = { ...process.env }; @@ -157,6 +170,102 @@ describe('RedisJobStore Integration Tests', () => { }); }); + describe('Requires Action Status Tracking', () => { + test('should count requires_action jobs and remove them from the running set', async () => { + if (!ioredisClient) { + return; + } + + const { RedisJobStore } = await import('../implementations/RedisJobStore'); + const store = new RedisJobStore(ioredisClient); + await store.initialize(); + + const userId = `requires-action-user-${Date.now()}`; + const streamId = `requires-action-${Date.now()}`; + const beforeRunning = await store.getJobCountByStatus('running'); + const beforePaused = await store.getJobCountByStatus('requires_action'); + await store.createJob(streamId, userId, streamId); + + expect(await store.getJobCountByStatus('running')).toBe(beforeRunning + 1); + expect(await store.getJobCountByStatus('requires_action')).toBe(beforePaused); + + await store.updateJob(streamId, { + status: 'requires_action', + pendingAction: buildPendingAction(streamId), + }); + + const runningMembers = await ioredisClient.smembers('stream:running'); + const pausedMembers = await ioredisClient.smembers('stream:requires_action'); + expect(runningMembers).not.toContain(streamId); + expect(pausedMembers).toContain(streamId); + expect(await store.getJobCountByStatus('running')).toBe(beforeRunning); + expect(await store.getJobCountByStatus('requires_action')).toBe(beforePaused + 1); + expect(await store.getActiveJobIdsByUser(userId)).toContain(streamId); + + await store.destroy(); + }); + + test('should return resumed requires_action jobs to the running index', async () => { + if (!ioredisClient) { + return; + } + + const { RedisJobStore } = await import('../implementations/RedisJobStore'); + const store = new RedisJobStore(ioredisClient); + await store.initialize(); + + const streamId = `requires-action-resume-${Date.now()}`; + const beforeRunning = await store.getJobCountByStatus('running'); + const beforePaused = await store.getJobCountByStatus('requires_action'); + await store.createJob(streamId, 'user-1', streamId); + await store.updateJob(streamId, { + status: 'requires_action', + pendingAction: buildPendingAction(streamId), + }); + + await store.updateJob(streamId, { status: 'running', pendingAction: undefined }); + + const job = await store.getJob(streamId); + expect(job?.status).toBe('running'); + expect(job?.pendingAction).toBeUndefined(); + expect(await store.getJobCountByStatus('running')).toBe(beforeRunning + 1); + expect(await store.getJobCountByStatus('requires_action')).toBe(beforePaused); + + await store.destroy(); + }); + + test('should not drop paused jobs from user tracking when cleanup sees a stale running index', async () => { + if (!ioredisClient) { + return; + } + + const { RedisJobStore } = await import('../implementations/RedisJobStore'); + const store = new RedisJobStore(ioredisClient); + await store.initialize(); + + const userId = `requires-action-cleanup-user-${Date.now()}`; + const streamId = `requires-action-cleanup-${Date.now()}`; + await store.createJob(streamId, userId, streamId); + await store.updateJob(streamId, { + status: 'requires_action', + pendingAction: buildPendingAction(streamId), + }); + + await ioredisClient.sadd('stream:running', streamId); + + const cleaned = await store.cleanup(); + const runningMembers = await ioredisClient.smembers('stream:running'); + const pausedMembers = await ioredisClient.smembers('stream:requires_action'); + + expect(cleaned).toBeGreaterThanOrEqual(1); + expect(runningMembers).not.toContain(streamId); + expect(pausedMembers).toContain(streamId); + expect(await store.getActiveJobIdsByUser(userId)).toContain(streamId); + + await store.destroy(); + }); + }); + describe('Horizontal Scaling - Multi-Instance Simulation', () => { test('should share job state between two store instances', async () => { if (!ioredisClient) { diff --git a/packages/api/src/stream/implementations/RedisJobStore.ts b/packages/api/src/stream/implementations/RedisJobStore.ts index effaf66ff4..5192a4c615 100644 --- a/packages/api/src/stream/implementations/RedisJobStore.ts +++ b/packages/api/src/stream/implementations/RedisJobStore.ts @@ -29,6 +29,8 @@ const KEYS = { runSteps: (streamId: string) => `stream:{${streamId}}:runsteps`, /** Running jobs set for cleanup (global set - single slot) */ runningJobs: 'stream:running', + /** Jobs paused for human review (global set - single slot) */ + requiresActionJobs: 'stream:requires_action', /** User's active jobs set, tenant-qualified when tenantId is available */ userJobs: (userId: string, tenantId?: string) => tenantId ? `stream:user:{${tenantId}:${userId}}:jobs` : `stream:user:{${userId}}:jobs`, @@ -167,6 +169,7 @@ export class RedisJobStore implements IJobStore { await this.redis.hset(key, this.serializeJob(job)); await this.redis.expire(key, this.ttl.running); await this.redis.sadd(KEYS.runningJobs, streamId); + await this.redis.srem(KEYS.requiresActionJobs, streamId); await this.redis.sadd(userJobsKey, streamId); if (this.ttl.userJobsSet > 0) { await this.redis.expire(userJobsKey, this.ttl.userJobsSet); @@ -176,6 +179,7 @@ export class RedisJobStore implements IJobStore { pipeline.hset(key, this.serializeJob(job)); pipeline.expire(key, this.ttl.running); pipeline.sadd(KEYS.runningJobs, streamId); + pipeline.srem(KEYS.requiresActionJobs, streamId); pipeline.sadd(userJobsKey, streamId); if (this.ttl.userJobsSet > 0) { pipeline.expire(userJobsKey, this.ttl.userJobsSet); @@ -204,49 +208,19 @@ export class RedisJobStore implements IJobStore { } const fields = Object.entries(serialized).flat(); - const updated = await this.redis.eval( - 'if redis.call("EXISTS", KEYS[1]) == 1 then redis.call("HSET", KEYS[1], unpack(ARGV)) return 1 else return 0 end', - 1, - key, - ...fields, - ); - - if (updated === 0) { - return; - } if (updates.status === 'requires_action') { - // Job paused for human review — non-terminal. - // Remove from runningJobs so getJobCountByStatus('running') stays accurate, - // refresh the hash TTL so the user has the full window to respond, and - // leave chunks/runSteps/user-active-set untouched so resume can rebuild state. - if (this.isCluster) { - await this.redis.srem(KEYS.runningJobs, streamId); - await this.redis.expire(key, this.ttl.running); - } else { - const pipeline = this.redis.pipeline(); - pipeline.srem(KEYS.runningJobs, streamId); - pipeline.expire(key, this.ttl.running); - await pipeline.exec(); - } + await this.transitionToRequiresAction(key, streamId, fields); return; } if (updates.status === 'running') { - // Resume from requires_action — re-add to runningJobs (idempotent), refresh TTL, - // and explicitly clear any stale pendingAction (serializeJob skips `undefined`, - // so the only way to remove a hash field is HDEL). - if (this.isCluster) { - await this.redis.sadd(KEYS.runningJobs, streamId); - await this.redis.expire(key, this.ttl.running); - await this.redis.hdel(key, 'pendingAction'); - } else { - const pipeline = this.redis.pipeline(); - pipeline.sadd(KEYS.runningJobs, streamId); - pipeline.expire(key, this.ttl.running); - pipeline.hdel(key, 'pendingAction'); - await pipeline.exec(); - } + await this.transitionToRunning(key, streamId, fields); + return; + } + + const updated = await this.updateExistingJobHash(key, fields); + if (!updated) { return; } @@ -258,6 +232,7 @@ export class RedisJobStore implements IJobStore { if (this.isCluster) { await this.redis.expire(key, this.ttl.completed); await this.redis.srem(KEYS.runningJobs, streamId); + await this.redis.srem(KEYS.requiresActionJobs, streamId); if (this.ttl.chunksAfterComplete === 0) { await this.redis.del(KEYS.chunks(streamId)); @@ -278,6 +253,7 @@ export class RedisJobStore implements IJobStore { const pipeline = this.redis.pipeline(); pipeline.expire(key, this.ttl.completed); pipeline.srem(KEYS.runningJobs, streamId); + pipeline.srem(KEYS.requiresActionJobs, streamId); if (this.ttl.chunksAfterComplete === 0) { pipeline.del(KEYS.chunks(streamId)); @@ -300,6 +276,78 @@ export class RedisJobStore implements IJobStore { } } + private async updateExistingJobHash(key: string, fields: string[]): Promise { + const updated = await this.redis.eval( + 'if redis.call("EXISTS", KEYS[1]) == 1 then redis.call("HSET", KEYS[1], unpack(ARGV)) return 1 else return 0 end', + 1, + key, + ...fields, + ); + return updated === 1; + } + + private async transitionToRequiresAction( + key: string, + streamId: string, + fields: string[], + ): Promise { + // Job paused for human review — non-terminal. Keep the user-active set + // untouched so resume can rebuild state from the persisted job. + if (this.isCluster) { + const exists = await this.redis.exists(key); + if (exists !== 1) { + return; + } + await this.redis.srem(KEYS.runningJobs, streamId); + await this.redis.sadd(KEYS.requiresActionJobs, streamId); + await this.updateExistingJobHash(key, fields); + await this.redis.expire(key, this.ttl.running); + return; + } + + await this.redis.eval( + 'if redis.call("EXISTS", KEYS[1]) == 0 then return 0 end redis.call("SREM", KEYS[2], ARGV[1]) redis.call("SADD", KEYS[3], ARGV[1]) redis.call("HSET", KEYS[1], unpack(ARGV, 3)) redis.call("EXPIRE", KEYS[1], tonumber(ARGV[2])) return 1', + 3, + key, + KEYS.runningJobs, + KEYS.requiresActionJobs, + streamId, + String(this.ttl.running), + ...fields, + ); + } + + private async transitionToRunning( + key: string, + streamId: string, + fields: string[], + ): Promise { + // Resume from requires_action and clear stale pendingAction. serializeJob skips + // `undefined`, so the hash field must be removed explicitly. + if (this.isCluster) { + const updated = await this.updateExistingJobHash(key, fields); + if (!updated) { + return; + } + await this.redis.hdel(key, 'pendingAction'); + await this.redis.expire(key, this.ttl.running); + await this.redis.srem(KEYS.requiresActionJobs, streamId); + await this.redis.sadd(KEYS.runningJobs, streamId); + return; + } + + await this.redis.eval( + 'if redis.call("EXISTS", KEYS[1]) == 0 then return 0 end redis.call("HSET", KEYS[1], unpack(ARGV, 3)) redis.call("HDEL", KEYS[1], "pendingAction") redis.call("EXPIRE", KEYS[1], tonumber(ARGV[2])) redis.call("SREM", KEYS[2], ARGV[1]) redis.call("SADD", KEYS[3], ARGV[1]) return 1', + 3, + key, + KEYS.requiresActionJobs, + KEYS.runningJobs, + streamId, + String(this.ttl.running), + ...fields, + ); + } + async deleteJob(streamId: string): Promise { this.localGraphCache.delete(streamId); this.localCollectedUsageCache.delete(streamId); @@ -319,6 +367,7 @@ export class RedisJobStore implements IJobStore { pipeline.del(KEYS.runSteps(streamId)); await pipeline.exec(); await this.redis.srem(KEYS.runningJobs, streamId); + await this.redis.srem(KEYS.requiresActionJobs, streamId); if (userJobsKey) { await this.redis.srem(userJobsKey, streamId); } @@ -328,6 +377,7 @@ export class RedisJobStore implements IJobStore { pipeline.del(KEYS.chunks(streamId)); pipeline.del(KEYS.runSteps(streamId)); pipeline.srem(KEYS.runningJobs, streamId); + pipeline.srem(KEYS.requiresActionJobs, streamId); if (userJobsKey) { pipeline.srem(userJobsKey, streamId); } @@ -380,6 +430,15 @@ export class RedisJobStore implements IJobStore { // Job no longer exists (TTL expired) - remove from set if (!job) { await this.redis.srem(KEYS.runningJobs, streamId); + await this.redis.srem(KEYS.requiresActionJobs, streamId); + this.localGraphCache.delete(streamId); + this.localCollectedUsageCache.delete(streamId); + return 1; + } + + if (job.status === 'requires_action') { + await this.redis.srem(KEYS.runningJobs, streamId); + await this.redis.sadd(KEYS.requiresActionJobs, streamId); this.localGraphCache.delete(streamId); this.localCollectedUsageCache.delete(streamId); return 1; @@ -390,6 +449,7 @@ export class RedisJobStore implements IJobStore { // its own completedTtl so clients can still poll for final status. if (job.status !== 'running') { await this.redis.srem(KEYS.runningJobs, streamId); + await this.redis.srem(KEYS.requiresActionJobs, streamId); if (job.userId) { await this.redis.srem(KEYS.userJobs(job.userId, job.tenantId), streamId); } @@ -426,10 +486,11 @@ export class RedisJobStore implements IJobStore { } async getJobCount(): Promise { - // This is approximate - counts jobs in running set + scans for job keys - // For exact count, would need to scan all job:* keys - const runningCount = await this.redis.scard(KEYS.runningJobs); - return runningCount; + const [runningCount, requiresActionCount] = await Promise.all([ + this.redis.scard(KEYS.runningJobs), + this.countJobsInStatusSet(KEYS.requiresActionJobs, 'requires_action'), + ]); + return runningCount + requiresActionCount; } async getJobCountByStatus(status: JobStatus): Promise { @@ -437,11 +498,37 @@ export class RedisJobStore implements IJobStore { return this.redis.scard(KEYS.runningJobs); } - // For other statuses, we'd need to scan - return 0 for now - // In production, consider maintaining separate sets per status if needed + if (status === 'requires_action') { + return this.countJobsInStatusSet(KEYS.requiresActionJobs, status); + } + return 0; } + private async countJobsInStatusSet(setKey: string, status: JobStatus): Promise { + const streamIds = await this.redis.smembers(setKey); + if (streamIds.length === 0) { + return 0; + } + + let count = 0; + const staleIds: string[] = []; + for (const streamId of streamIds) { + const job = await this.getJob(streamId); + if (job?.status === status) { + count++; + } else { + staleIds.push(streamId); + } + } + + if (staleIds.length > 0) { + await this.redis.srem(setKey, ...staleIds); + } + + return count; + } + /** * Get active job IDs for a user. * Returns conversation IDs of running jobs belonging to the user.