♻️ refactor: dedup HITL transition path + liveness predicate (arch review)

Two follow-ups from the post-hardening architecture re-review — both pure
dedup, no behavior change:

A — collapse the dual status-transition path. transitionStatus is now the
   sole membership-aware transition (running ⇄ requires_action). Removed the
   updateJob requires_action/running branches and the now-orphaned
   transitionToRequiresAction / transitionToRunning / refreshLiveJobTtls, plus
   the per-store pause/resume mirror logic that had to be re-synced into parity
   across review rounds (G2/G3/I4/I5). updateJob is back to a plain field
   writer + terminal cleanup. The Redis integration tests that drove
   updateJob({status}) now drive transitionStatus (the real path).

B — one canonical "is this approval live?" predicate. isPendingActionStale /
   isPendingActionExpired are exported from @librechat/api and used by the
   stores, ApprovalLifecycle (dropped its private isExpired), the /chat/status
   route, and validateMessageReq — replacing 3 inlined re-derivations that were
   the drift source behind several review findings.

tsc + lint clean; policy + type-contract specs pass. Redis integration specs
(migrated) are CI-verified.

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

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,

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<Agents.PendingAction | null> {
     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();
-  }
 }

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

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

packages/api/src/stream/implementations/RedisJobStore.ts

@@ -257,40 +257,16 @@ export class RedisJobStore implements IJobStore {
   async updateJob(streamId: string, updates: Partial<SerializableJobData>): Promise<void> {
     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<SerializableJobData> = 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<void> {
-    // 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<void> {
-    // 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<void> {
-    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<void> {
     this.localGraphCache.delete(streamId);
     this.localCollectedUsageCache.delete(streamId);

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