diff --git a/api/server/middleware/validateMessageReq.js b/api/server/middleware/validateMessageReq.js index 77f5fe620d..ca631c8950 100644 --- a/api/server/middleware/validateMessageReq.js +++ b/api/server/middleware/validateMessageReq.js @@ -1,4 +1,4 @@ -const { GenerationJobManager } = require('@librechat/api'); +const { GenerationJobManager, isPendingActionStale } = require('@librechat/api'); const { logger } = require('@librechat/data-schemas'); const { getConvo } = require('~/models'); @@ -24,11 +24,11 @@ async function canReadActiveJobConversation(req, conversationId) { // and /chat/active), so a new-conversation run that pauses before its final // save can still recover the prompt — but only while it has a live, // resolvable prompt (missing/malformed or past-expiry reads as inactive). - const pendingAction = job?.metadata?.pendingAction; - const pendingLive = - !!pendingAction && (pendingAction.expiresAt == null || pendingAction.expiresAt > Date.now()); const isActive = - !!job && (job.status === 'running' || (job.status === 'requires_action' && pendingLive)); + !!job && + (job.status === 'running' || + (job.status === 'requires_action' && + !isPendingActionStale({ pendingAction: job.metadata?.pendingAction }))); if (!isActive) { return false; } diff --git a/api/server/routes/agents/index.js b/api/server/routes/agents/index.js index 9a6c6bde6e..7fb23f6c66 100644 --- a/api/server/routes/agents/index.js +++ b/api/server/routes/agents/index.js @@ -4,6 +4,7 @@ const { GenerationJobManager, hasPersistableAbortContent, buildAbortedResponseMetadata, + isPendingActionStale, } = require('@librechat/api'); const { createSseStreamTelemetry } = require('@librechat/api/telemetry'); const { logger } = require('@librechat/data-schemas'); @@ -207,9 +208,8 @@ router.get('/chat/status/:conversationId', async (req, res) => { // only while it has a live, resolvable prompt: a missing/malformed or // past-expiry pendingAction reads as inactive (cleanup/expiry will finalize it). const pendingAction = job.metadata.pendingAction; - const pendingLive = - !!pendingAction && (pendingAction.expiresAt == null || pendingAction.expiresAt > Date.now()); - const isActive = job.status === 'running' || (job.status === 'requires_action' && pendingLive); + const pendingLive = job.status === 'requires_action' && !isPendingActionStale({ pendingAction }); + const isActive = job.status === 'running' || pendingLive; res.json({ active: isActive, diff --git a/packages/api/src/stream/ApprovalLifecycle.ts b/packages/api/src/stream/ApprovalLifecycle.ts index 56fe00d859..e117c7bee2 100644 --- a/packages/api/src/stream/ApprovalLifecycle.ts +++ b/packages/api/src/stream/ApprovalLifecycle.ts @@ -1,6 +1,7 @@ import { logger } from '@librechat/data-schemas'; import type { Agents } from 'librechat-data-provider'; import type { IJobStore } from '~/stream/interfaces/IJobStore'; +import { isPendingActionExpired, isPendingActionStale } from '~/stream/interfaces/IJobStore'; /** * The guarded lifecycle of a run paused for human review (`requires_action`). @@ -54,10 +55,11 @@ export class ApprovalLifecycle { */ async peek(streamId: string): Promise { const job = await this.store.getJob(streamId); - if (!job || job.status !== 'requires_action' || !job.pendingAction) { + if (!job || job.status !== 'requires_action') { return null; } - return this.isExpired(job.pendingAction) ? null : job.pendingAction; + // isPendingActionStale covers both a missing record and a past-expiry one. + return isPendingActionStale(job) ? null : (job.pendingAction ?? null); } /** @@ -85,11 +87,7 @@ export class ApprovalLifecycle { await this.expire(streamId); return false; } - if ( - job?.status === 'requires_action' && - job.pendingAction && - this.isExpired(job.pendingAction) - ) { + if (job?.status === 'requires_action' && job.pendingAction && isPendingActionExpired(job)) { // Target the exact record observed as expired. If the caller didn't pin an // actionId, fall back to the one just read — otherwise a concurrent // resume + re-pause for a new action could let this expire abort it. @@ -128,8 +126,4 @@ export class ApprovalLifecycle { } return ok; } - - private isExpired(pendingAction: Agents.PendingAction): boolean { - return pendingAction.expiresAt != null && pendingAction.expiresAt <= Date.now(); - } } 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 ca97566620..55229ebc5e 100644 --- a/packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts +++ b/packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts @@ -189,9 +189,10 @@ describe('RedisJobStore Integration Tests', () => { 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), + await store.transitionStatus(streamId, { + from: 'running', + to: 'requires_action', + patch: { pendingAction: buildPendingAction(streamId) }, }); const runningMembers = await ioredisClient.smembers('stream:running'); @@ -218,12 +219,17 @@ describe('RedisJobStore Integration Tests', () => { 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.transitionStatus(streamId, { + from: 'running', + to: 'requires_action', + patch: { pendingAction: buildPendingAction(streamId) }, }); - await store.updateJob(streamId, { status: 'running', pendingAction: undefined }); + await store.transitionStatus(streamId, { + from: 'requires_action', + to: 'running', + clear: ['pendingAction'], + }); const job = await store.getJob(streamId); expect(job?.status).toBe('running'); @@ -257,9 +263,10 @@ describe('RedisJobStore Integration Tests', () => { await ioredisClient.expire(chunkKey, 30); await ioredisClient.expire(runStepsKey, 30); - await store.updateJob(streamId, { - status: 'requires_action', - pendingAction: buildPendingAction(streamId), + await store.transitionStatus(streamId, { + from: 'running', + to: 'requires_action', + patch: { pendingAction: buildPendingAction(streamId) }, }); expect(await ioredisClient.ttl(chunkKey)).toBeGreaterThan(30); @@ -268,7 +275,11 @@ describe('RedisJobStore Integration Tests', () => { await ioredisClient.expire(chunkKey, 30); await ioredisClient.expire(runStepsKey, 30); - await store.updateJob(streamId, { status: 'running', pendingAction: undefined }); + await store.transitionStatus(streamId, { + from: 'requires_action', + to: 'running', + clear: ['pendingAction'], + }); expect(await ioredisClient.ttl(chunkKey)).toBeGreaterThan(30); expect(await ioredisClient.ttl(runStepsKey)).toBeGreaterThan(30); @@ -288,9 +299,10 @@ describe('RedisJobStore Integration Tests', () => { 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 store.transitionStatus(streamId, { + from: 'running', + to: 'requires_action', + patch: { pendingAction: buildPendingAction(streamId) }, }); await ioredisClient.sadd('stream:running', streamId); @@ -319,9 +331,10 @@ describe('RedisJobStore Integration Tests', () => { const streamId = `requires-action-expired-${Date.now()}`; const jobKey = `stream:{${streamId}}:job`; await store.createJob(streamId, 'user-1', streamId); - await store.updateJob(streamId, { - status: 'requires_action', - pendingAction: buildPendingAction(streamId), + await store.transitionStatus(streamId, { + from: 'running', + to: 'requires_action', + patch: { pendingAction: buildPendingAction(streamId) }, }); expect(await ioredisClient.smembers('stream:requires_action')).toContain(streamId); diff --git a/packages/api/src/stream/implementations/InMemoryJobStore.ts b/packages/api/src/stream/implementations/InMemoryJobStore.ts index 702b58ca40..f8caf2b9ed 100644 --- a/packages/api/src/stream/implementations/InMemoryJobStore.ts +++ b/packages/api/src/stream/implementations/InMemoryJobStore.ts @@ -136,18 +136,9 @@ export class InMemoryJobStore implements IJobStore { if (!job) { return; } + // Plain field writer. Membership-aware status transitions + // (running ⇄ requires_action) go solely through transitionStatus. Object.assign(job, updates); - // Mirror the guarded transitionStatus path so a pause/resume via this - // generic update behaves identically (parity with RedisJobStore): - // - mirror the flat pendingActionId the stale-decision guard compares; - // - on resume to running, refresh lastActiveAt and drop the flat id. - if (updates.pendingAction) { - job.pendingActionId = updates.pendingAction.actionId; - } - if (updates.status === 'running') { - job.lastActiveAt = updates.lastActiveAt ?? Date.now(); - delete job.pendingActionId; - } } /** diff --git a/packages/api/src/stream/implementations/RedisJobStore.ts b/packages/api/src/stream/implementations/RedisJobStore.ts index 8c12203fe6..32d78ee487 100644 --- a/packages/api/src/stream/implementations/RedisJobStore.ts +++ b/packages/api/src/stream/implementations/RedisJobStore.ts @@ -257,40 +257,16 @@ export class RedisJobStore implements IJobStore { async updateJob(streamId: string, updates: Partial): Promise { const key = KEYS.job(streamId); - // Keep this generic path consistent with the guarded transitionStatus path: - // - mirror `pendingActionId` on any `pendingAction` write so a pause here - // still carries the flat field the stale-decision guard compares; - // - refresh `lastActiveAt` when resuming to `running` so a long-paused job - // isn't reaped by `createdAt` on the next cleanup tick. - let effective: Partial = updates; - if (updates.status === 'running') { - effective = { ...updates, lastActiveAt: updates.lastActiveAt ?? Date.now() }; - } else if (updates.pendingAction) { - effective = { ...updates, pendingActionId: updates.pendingAction.actionId }; - } - - const serialized = this.serializeJob(effective as SerializableJobData); + // Plain field writer. The membership-aware status transitions + // (running ⇄ requires_action — sets, TTLs, the actionId guard) go solely + // through transitionStatus, the single race-safe path. updateJob still + // handles terminal status writes (complete/error/aborted) + their cleanup. + const serialized = this.serializeJob(updates as SerializableJobData); if (Object.keys(serialized).length === 0) { return; } const fields = Object.entries(serialized).flat(); - - if (updates.status === 'requires_action') { - await this.transitionToRequiresAction( - key, - streamId, - fields, - this.pauseTtlSeconds(updates.pendingAction), - ); - return; - } - - if (updates.status === 'running') { - await this.transitionToRunning(key, streamId, fields); - return; - } - const updated = await this.updateExistingJobHash(key, fields); if (!updated) { return; @@ -472,86 +448,6 @@ export class RedisJobStore implements IJobStore { return updated === 1; } - private async transitionToRequiresAction( - key: string, - streamId: string, - fields: string[], - ttlSeconds: number = this.ttl.running, - ): Promise { - // Job paused for human review — non-terminal. Keep the user-active set - // untouched so resume can rebuild state from the persisted job. The live - // TTL covers the approval window (see pauseTtlSeconds) so a long-pending - // job isn't evicted before a decision arrives. - 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, ttlSeconds); - await this.redis.expire(KEYS.chunks(streamId), ttlSeconds); - await this.redis.expire(KEYS.runSteps(streamId), ttlSeconds); - 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])) redis.call("EXPIRE", KEYS[4], tonumber(ARGV[2])) redis.call("EXPIRE", KEYS[5], tonumber(ARGV[2])) return 1', - 5, - key, - KEYS.runningJobs, - KEYS.requiresActionJobs, - KEYS.chunks(streamId), - KEYS.runSteps(streamId), - streamId, - String(ttlSeconds), - ...fields, - ); - } - - private async transitionToRunning( - key: string, - streamId: string, - fields: string[], - ): Promise { - // Resume from requires_action and clear stale pendingAction + its flat - // pendingActionId mirror. serializeJob skips `undefined`, so the hash - // fields must be removed explicitly. - if (this.isCluster) { - const updated = await this.updateExistingJobHash(key, fields); - if (!updated) { - return; - } - await this.redis.hdel(key, 'pendingAction', 'pendingActionId'); - await this.refreshLiveJobTtls(key, streamId); - 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", "pendingActionId") redis.call("EXPIRE", KEYS[1], tonumber(ARGV[2])) redis.call("EXPIRE", KEYS[4], tonumber(ARGV[2])) redis.call("EXPIRE", KEYS[5], tonumber(ARGV[2])) redis.call("SREM", KEYS[2], ARGV[1]) redis.call("SADD", KEYS[3], ARGV[1]) return 1', - 5, - key, - KEYS.requiresActionJobs, - KEYS.runningJobs, - KEYS.chunks(streamId), - KEYS.runSteps(streamId), - streamId, - String(this.ttl.running), - ...fields, - ); - } - - private async refreshLiveJobTtls(key: string, streamId: string): Promise { - const pipeline = this.redis.pipeline(); - pipeline.expire(key, this.ttl.running); - pipeline.expire(KEYS.chunks(streamId), this.ttl.running); - pipeline.expire(KEYS.runSteps(streamId), this.ttl.running); - await pipeline.exec(); - } - async deleteJob(streamId: string): Promise { this.localGraphCache.delete(streamId); this.localCollectedUsageCache.delete(streamId); diff --git a/packages/api/src/stream/index.ts b/packages/api/src/stream/index.ts index 708a94c3d4..ef63f20d82 100644 --- a/packages/api/src/stream/index.ts +++ b/packages/api/src/stream/index.ts @@ -12,6 +12,9 @@ export type { JobStatus, IJobStore, } from './interfaces/IJobStore'; +// Canonical "is this approval live?" predicate — one definition shared by the +// stores, the approval lifecycle, and the status route / message middleware. +export { isPendingActionExpired, isPendingActionStale } from './interfaces/IJobStore'; export { createStreamServices } from './createStreamServices'; export type { StreamServicesConfig, StreamServices } from './createStreamServices';