CERBERUS/LibreChat
1
0
Fork 0
mirror of https://github.com/danny-avila/LibreChat.git synced 2026-07-10 08:13:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

fix: Address HITL review findings

Browse source
This commit is contained in:
Danny Avila 2026-05-19 13:33:15 -04:00
parent c01f27eff6
commit 3c39cd0f65
6 changed files with 318 additions and 59 deletions
Download patch file Download diff file Expand all files Collapse all files

50
api/server/routes/agents/__tests__/abort.spec.js
View file

@ -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([]);

12
api/server/routes/agents/index.js
View file

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

22
packages/api/src/agents/hitl/policy.spec.ts
View file

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

11
packages/api/src/agents/hitl/policy.ts
View file

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

109
packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts
View file

@ -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) {

173
packages/api/src/stream/implementations/RedisJobStore.ts
View file

@ -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<boolean> {
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<void> {
// 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<void> {
// 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<void> {
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<number> {
// 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<number> {
@ -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<number> {
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.