From 771b93bf10194850b646dd9808e8d3c7ea168858 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Wed, 24 Jun 2026 16:47:16 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=AA=9D=20feat:=20HITL=20Tool=20Approval?= =?UTF-8?q?=20Scaffolding=20(Slice=20A)=20(#12938)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * πŸͺ feat: HITL Tool Approval Scaffolding Adds the foundational types, job-state, config schema, and policy module for human-in-the-loop tool approval. Purely additive β€” no behavior change on existing runs. Lands ahead of the agents-SDK interrupt/checkpointer integration so both tracks can land independently. - LangChain HumanInterrupt-shaped types in `Agents.*` namespace (`HumanInterruptPayload`, `ToolApprovalRequest`, `ToolReviewConfig`, `PendingAction`, `ToolApprovalResolution`); `ToolCall`/`ToolCallDelta` gain an optional `approval` field. - New `requires_action` job status (non-terminal) plus `pendingAction` field on `SerializableJobData` and `GenerationJobMetadata`. Both stores treat the status as paused-but-alive; Redis `updateJob` has explicit `requires_action`/`running` transition branches that refresh the hash TTL, manage the `runningJobs` set, and `HDEL pendingAction` on resume. Both stores include `requires_action` in `getActiveJobIdsByUser`. - `GenerationJobManager` gains `markRequiresAction`, `getPendingAction`, `clearPendingAction`; `getJobCountByStatus` aggregates the new status. - `endpoints.agents.toolApproval` config (`default`/`required`/`excluded`) and a policy module exporting `decideToolApproval`, `requiresApproval`, and `buildPendingAction` (the LangChain-shaped payload builder). - 20 unit tests covering policy resolution and the manager lifecycle. * 🧭 refactor: Align HITL Surface with Agents SDK Permissions Model Reshapes Slice A on top of the agents SDK's now-landed HITL surface (`createToolPolicyHook`, discriminated `HumanInterruptPayload`, `'bypass'` mode naming). Host stops reimplementing evaluation logic and becomes a config mapper + payload wrapper. Schema (data-provider): - `toolApproval` shape now mirrors SDK `ToolPolicyConfig` 1:1: `mode: 'default' | 'dontAsk' | 'bypass'`, plus `allow` / `deny` / `ask` glob lists and an optional `reason` template. `enabled` is the LibreChat-only admin kill switch. - `'bypass'` (not `'bypassPermissions'`) β€” matches the SDK's surface. Types (`Agents.*` namespace): - `HumanInterruptType` extended to `'tool_approval' | 'ask_user_question'`. - `HumanInterruptPayload` is now a discriminated union β€” `tool_approval` carries `action_requests` + `review_configs`; `ask_user_question` carries a free-form question with optional curated options. - New: `AskUserQuestionRequest`, `AskUserQuestionOption`, `AskUserQuestionResolution`. - `ToolApprovalDecision` (string union) renamed to `ToolApprovalDecisionType` to free the `Decision` name for the SDK's discriminated object union later. - `ToolApprovalResolution` gains `reason?` and `scope?: 'once' | 'session' | 'always'` so route signatures stabilize before persistence lands. Policy module (`packages/api/src/agents/hitl/policy.ts`): - Drop `decideToolApproval` / `requiresApproval` / `ToolRef` β€” the SDK's `createToolPolicyHook` handles full evaluation (`deny β†’ bypass β†’ allow β†’ ask β†’ dontAsk β†’ fallthrough(ask)`). - Add `isHITLEnabled(policy)` β€” the kill-switch predicate that gates the SDK's `humanInTheLoop: { enabled: false }` opt-out in Slice B. - Add `mapToolApprovalPolicy(policy)` β€” strips `enabled`, returns a `ToolPolicyConfig` to feed `createToolPolicyHook`. Structural mirror of the SDK type so this compiles before the SDK upgrade ships. - Reshape `buildPendingAction(payload, ctx)` to wrap any `HumanInterruptPayload` with job context β€” accepts SDK output directly. - Add `buildToolApprovalPayload(...)` and `buildAskUserQuestionPayload(...)` helpers for synthesizing payloads in tests / pre-SDK flows. Tests: - 22 new unit tests covering the mapper, predicate, and payload builders; 20 β†’ 27 total pass across policy + manager-lifecycle suites. * πŸͺ’ chore: Import ToolPolicyConfig From `@librechat/agents` The SDK type now ships in 3.1.77 (already pinned on `dev`), so the structural mirror in `policy.ts` is redundant. Drop the local interface and import directly so future SDK changes to `ToolPolicyConfig` propagate without our `mapToolApprovalPolicy` going stale. * πŸ”‘ fix: Carry tool_call_id On ToolReviewConfig (HITL) `ToolReviewConfig` was joining with `ToolApprovalRequest` by position only. That breaks the moment a single batch contains the same tool called twice (e.g. a model fanning out parallel `mcp:server:search` calls): the UI can't tell which review config applies to which action request once it filters or reorders. Mirrors the SDK's `ToolApprovalReviewConfig` shape β€” `tool_call_id` is the join key, `action_name` is retained for display only. Also: drop a JSDoc warning on `isHITLEnabled` so a future contributor doesn't wire `humanInTheLoop: { enabled: true }` without supplying a host checkpointer β€” the SDK's `MemorySaver` fallback is process-local and silently breaks resume across worker hops. - `Agents.ToolReviewConfig` adds `tool_call_id: string` - `buildToolApprovalPayload` populates `tool_call_id` per review config - New test covers the duplicate-tool batch case (two parallel calls to the same tool); 27 β†’ 28 tests * fix: Address HITL review findings * fix: Refresh paused HITL Redis state * test: Stabilize HITL abort fallback specs * 🎨 style: Sort imports to satisfy dev lint gate (HITL) * πŸ›οΈ refactor: Deepen HITL approval lifecycle into one race-safe seam Architecture-review candidate #1 (+ #4). The requires_action lifecycle was three shallow pass-throughs over updateJob with the legal transitions smeared across JSDoc, the JobStatus union, and each store adapter β€” and the resume transition was NOT race-safe: the Redis lua checked existence, not status, so two concurrent approval submits both drove the run (re-executing tools / double-billing). - IJobStore.transitionStatus: atomic compare-and-set status transition that only fires if the job is currently `from`. InMemory: sync compare. Redis: single-node lua with a status guard (cluster best-effort, matching the existing posture); reconciles membership sets + TTLs to `to`. - New ApprovalLifecycle module: pause / peek / resolve / expire β€” guarded, race-safe transitions behind one interface. resolve() returns true to exactly one concurrent caller; the previously-undefined requires_action β†’ aborted expiry edge is now explicit; peek treats past-expiresAt as gone (lazy expiry). - GenerationJobManager exposes `approvals` and delegates; the three shallow methods (mark/get/clearPendingAction) are removed β€” callers cross the deep interface. - #4: typeContract.spec asserts the SDK <-> data-provider HITL types stay compatible (fails the build on drift); RedisJobStore validates the pendingAction shape on deserialize instead of a bare JSON.parse (defends the cold-resume path against malformed/stale records). - Tests rewritten at the deep interface: double-resolve wins once, pause-on-terminal rejected, explicit expiry, lazy-expiry peek. No Slice B wiring β€” this deepens the existing scaffolding so the future resume route and run seam are born crossing one race-safe interface. * πŸ›‘οΈ fix: Address Codex review on the HITL approval lifecycle Seven findings on the lifecycle deepening (089ba09f9), all valid: - F3 actionId guard: resolve/expire take an expectedActionId; pause records a flat `pendingActionId` the atomic CAS guards on, so a stale decision can't resume a job that has since paused for a different action. - F4 cluster single-winner: transitionStatus now decides the winner with an atomic CAS on the single-slot job hash (one Lua, cluster-safe), then reconciles cross-slot membership sets β€” two concurrent resolves can no longer both win on Redis Cluster. - F1 resume reaping: resolve refreshes `lastActiveAt`; both stores' stale- running failsafes key off it, so a long-paused approval isn't reaped right after resuming. - F2 expire completedAt: expire writes completedAt so terminal cleanup reclaims the job (InMemory only cleans terminal jobs with completedAt set). - F5 facade: buildJobFacade copies pendingAction into metadata so status/ resume routes can render the prompt. - F6 resume metadata: PendingAction + buildPendingAction carry the SDK interruptId/threadId needed to rebuild Command({ resume }) cross-process. - F7 mirror: data-provider AskUserQuestionRequest gains optional description. Tests added at the interface: stale-actionId resolve rejected, expire sets completedAt. tsc + lint clean; policy + type-contract specs pass. * πŸ›‘οΈ fix: Address Codex round 2 on the HITL Redis adapter Five P2 findings on abf4b86291, all valid Redis-adapter consequences of round 1: - G1 terminal cleanup on expiry: transitionStatus's terminal path now runs the same chunk/run-step/userJobs cleanup as updateJob (extracted into a shared applyTerminalContentCleanup). Expired approvals no longer leave Redis stream contents around for the full running TTL. - G2 pause via updateJob mirrors pendingActionId, so a pause through the generic path carries the flat field the stale-decision guard compares. - G3 resume via updateJob refreshes lastActiveAt (and clears pendingActionId), matching transitionStatus so a long-paused job isn't reaped post-resume. - G4 getActiveJobIdsByUser excludes a requires_action job whose pendingAction is past expiry (both stores), via shared isPendingActionExpired β€” the client stops polling an expired prompt. - G5 createJob clears stale pendingAction/pendingActionId/lastActiveAt on a reused streamId, so a fresh run never exposes a prior run's approval metadata and cleanup keys off the new createdAt. Tests added: expired pending-approval excluded from the active set. tsc + lint clean; policy + type-contract specs pass. * πŸ›‘οΈ fix: Address Codex round 3 β€” approval expiry lifecycle completeness Three P2 findings on 780833d908, all valid: - H1 status consistency: /chat/status now treats a non-expired requires_action job as active (matching /chat/active), so a client refreshing while an approval is pending resumes/subscribes instead of treating the run as finished and stranding it. - H2 active expiry: cleanup now finalizes past-expiry requires_action jobs (β†’ aborted) in both stores instead of only filtering them from the active list β€” an expired prompt no longer lingers resident until key TTL. Redis routes through transitionStatus (terminal content cleanup); in-memory marks terminal + reclaims. - H3 resumed liveness: in-memory stale-running check uses max(lastActivity, lastActiveAt, createdAt), so a just-resumed job isn't reaped on a stale per-chunk lastActivity entry before the next chunk. Test added: in-memory cleanup finalizes + reclaims a past-expiry approval. tsc + lint clean; policy + type-contract specs pass. * πŸ›‘οΈ fix: Address Codex round 4 β€” paused-job edge cases across the stack Five P2 findings on 4324a4e776, all valid: - I1 message validation: validateMessageReq's active-job read bypass now accepts a live requires_action job, so a new-conversation run that pauses before its final save can recover the prompt instead of 404ing. - I2 expire targets the observed record: resolve()'s expired path passes `expectedActionId ?? job.pendingAction.actionId`, so a concurrent resume+re-pause can't let expire abort a different action. - I3 stale/malformed prompts: new isPendingActionStale (missing OR expired) drives active-listing exclusion + cleanup expiry in both stores, and the status route + middleware require a live pendingAction β€” a requires_action job whose pendingAction was dropped on deserialize no longer reads active. - I4 in-memory parity: InMemory updateJob mirrors pendingActionId on pause and clears it + refreshes lastActiveAt on resume (matching RedisJobStore), so a pause via the generic path is still resolvable by actionId. - I5 long approval windows: paused-job live TTL (job/chunks/run-steps) now covers pendingAction.expiresAt + grace (pauseTtlSeconds), on both the transitionStatus and updateJob pause paths, so Redis can't evict a paused job before its decision window closes. tsc + lint clean; policy + type-contract specs pass. * πŸ›‘οΈ fix: Codex round 5 β€” refuse unresolvable resolves; expose pending action Two of three findings on c8abd826e1 (the third deferred to Slice B): - J3 resolve() refuses a requires_action job that has lost its pendingAction (e.g. a malformed record dropped on deserialize): it expires/finalizes the job instead of driving a resumed run with no reviewed interrupt payload β€” consistent with how active-listing + cleanup already treat a stale prompt. - J2 /chat/status returns the live pendingAction for a paused stream, so a client rebuilding from status (reload / cross-replica) has the action id + payload to render and submit the prompt, not just "paused". Deferred (Slice B): J1 β€” emitting a terminal SSE event on approval expiry so already-subscribed clients close. The store-level lifecycle can't emit transport events, and there are no live SSE subscribers to a paused stream until the Slice B runtime wiring exists; tracked for that work. tsc + lint clean; policy + type-contract specs pass. * πŸ›‘οΈ fix: Codex final round β€” paused-job TTL + pendingAction in resume contract Two of three findings on e7d9cf21b6 (third deferred to Slice B): - K2 paused-job TTL: a paused (requires_action) job no longer inherits the 20-minute running TTL β€” it uses a dedicated requires_action backstop (default 24h, configurable) so a no-expiry approval (the buildPendingAction default), which the API treats as live, isn't evicted by Redis mid-window. A longer pendingAction.expiresAt still extends beyond the backstop. - K3 resume contract: pendingAction is now carried on the typed ResumeState (data-provider) and populated by getResumeState for a live paused job, so a reloading / cross-replica client can rebuild the prompt from resumeState (the contract useResumeOnLoad actually reads), not just a loose status field. Deferred (Slice B): K1 β€” emit a terminal SSE event on expiry so already- subscribed clients close. Requires the manager/eventTransport layer (the store-level lifecycle and cleanup loops have no transport access) and has no live subscriber until the Slice B subscribe/resume path exists; tracked there. tsc + lint clean; policy + type-contract specs pass. * ♻️ 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 | 13 +- .../routes/agents/__tests__/abort.spec.js | 53 +++ api/server/routes/agents/index.js | 30 +- packages/api/src/agents/hitl/index.ts | 1 + packages/api/src/agents/hitl/policy.spec.ts | 229 +++++++++ packages/api/src/agents/hitl/policy.ts | 150 ++++++ .../api/src/agents/hitl/typeContract.spec.ts | 44 ++ packages/api/src/agents/index.ts | 1 + packages/api/src/stream/ApprovalLifecycle.ts | 129 +++++ .../api/src/stream/GenerationJobManager.ts | 35 +- .../RedisJobStore.stream_integration.spec.ts | 196 +++++++- .../stream/__tests__/pendingAction.spec.ts | 251 ++++++++++ .../implementations/InMemoryJobStore.ts | 64 ++- .../stream/implementations/RedisJobStore.ts | 450 +++++++++++++++--- packages/api/src/stream/index.ts | 3 + .../api/src/stream/interfaces/IJobStore.ts | 92 +++- packages/api/src/types/stream.ts | 6 +- packages/data-provider/src/config.ts | 47 ++ packages/data-provider/src/types/agents.ts | 168 +++++++ 19 files changed, 1878 insertions(+), 84 deletions(-) create mode 100644 packages/api/src/agents/hitl/index.ts create mode 100644 packages/api/src/agents/hitl/policy.spec.ts create mode 100644 packages/api/src/agents/hitl/policy.ts create mode 100644 packages/api/src/agents/hitl/typeContract.spec.ts create mode 100644 packages/api/src/stream/ApprovalLifecycle.ts create mode 100644 packages/api/src/stream/__tests__/pendingAction.spec.ts diff --git a/api/server/middleware/validateMessageReq.js b/api/server/middleware/validateMessageReq.js index 15967cdc52..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'); @@ -20,7 +20,16 @@ async function canReadActiveJobConversation(req, conversationId) { return false; } - if (!job || job.status !== 'running') { + // A job paused for human review is still active (consistent with /chat/status + // 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 isActive = + !!job && + (job.status === 'running' || + (job.status === 'requires_action' && + !isPendingActionStale({ pendingAction: job.metadata?.pendingAction }))); + if (!isActive) { return false; } diff --git a/api/server/routes/agents/__tests__/abort.spec.js b/api/server/routes/agents/__tests__/abort.spec.js index 418c5f4254..9fe007596c 100644 --- a/api/server/routes/agents/__tests__/abort.spec.js +++ b/api/server/routes/agents/__tests__/abort.spec.js @@ -74,6 +74,10 @@ describe('Agent Abort Endpoint', () => { beforeEach(() => { jest.clearAllMocks(); + mockGenerationJobManager.getJob.mockReset(); + mockGenerationJobManager.abortJob.mockReset(); + mockGenerationJobManager.getActiveJobIdsForUser.mockReset(); + mockSaveMessage.mockReset(); }); describe('POST /chat/abort', () => { @@ -323,6 +327,55 @@ describe('Agent Abort Endpoint', () => { }); describe('Job Not Found', () => { + it('should skip paused fallback jobs and abort the running job', async () => { + mockGenerationJobManager.getJob + .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({ + 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..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'); @@ -202,7 +203,13 @@ router.get('/chat/status/:conversationId', async (req, res) => { // Get resume state which contains aggregatedContent // Avoid calling both getStreamInfo and getResumeState (both fetch content) const resumeState = await GenerationJobManager.getResumeState(conversationId); - const isActive = job.status === 'running'; + // A job paused for human review is still active (consistent with /chat/active), + // so the client resumes/subscribes rather than treating it as finished β€” but + // 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 = job.status === 'requires_action' && !isPendingActionStale({ pendingAction }); + const isActive = job.status === 'running' || pendingLive; res.json({ active: isActive, @@ -211,6 +218,10 @@ router.get('/chat/status/:conversationId', async (req, res) => { aggregatedContent: resumeState?.aggregatedContent ?? [], createdAt: job.createdAt, resumeState, + // Surface the live pending approval so a client rebuilding from /chat/status + // (reload / cross-replica) has the action id + payload to render and submit + // the prompt, not just the knowledge that the stream is paused. + pendingAction: job.status === 'requires_action' && pendingLive ? pendingAction : undefined, }); }); @@ -231,7 +242,10 @@ router.post('/chat/abort', async (req, res) => { // streamId === conversationId, so try any of the provided IDs // Skip "new" as it's a placeholder for new conversations, not an actual ID let jobStreamId = - streamId || (conversationId !== 'new' ? conversationId : null) || abortKey?.split(':')[0]; + streamId || + (conversationId !== 'new' ? conversationId : null) || + abortKey?.split(':')[0] || + null; let job = jobStreamId ? await GenerationJobManager.getJob(jobStreamId) : null; // Fallback: if job not found and we have a userId, look up active jobs for user @@ -242,11 +256,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/index.ts b/packages/api/src/agents/hitl/index.ts new file mode 100644 index 0000000000..62d5b849e4 --- /dev/null +++ b/packages/api/src/agents/hitl/index.ts @@ -0,0 +1 @@ +export * from './policy'; diff --git a/packages/api/src/agents/hitl/policy.spec.ts b/packages/api/src/agents/hitl/policy.spec.ts new file mode 100644 index 0000000000..369922a5ff --- /dev/null +++ b/packages/api/src/agents/hitl/policy.spec.ts @@ -0,0 +1,229 @@ +import type { Agents, TToolApprovalPolicy } from 'librechat-data-provider'; +import { + isHITLEnabled, + mapToolApprovalPolicy, + buildToolApprovalPayload, + buildAskUserQuestionPayload, + buildPendingAction, +} from './policy'; + +describe('isHITLEnabled', () => { + test('default-off when no policy configured', () => { + expect(isHITLEnabled(undefined)).toBe(false); + }); + + 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 off', () => { + expect(isHITLEnabled({ enabled: false })).toBe(false); + }); + + test('explicit true is on', () => { + expect(isHITLEnabled({ enabled: true })).toBe(true); + }); +}); + +describe('mapToolApprovalPolicy', () => { + test('returns undefined when no policy is configured', () => { + expect(mapToolApprovalPolicy(undefined)).toBeUndefined(); + }); + + test('returns undefined when policy is empty after stripping enabled', () => { + expect(mapToolApprovalPolicy({ enabled: true })).toBeUndefined(); + expect(mapToolApprovalPolicy({ enabled: false })).toBeUndefined(); + }); + + test('returns undefined when only empty arrays are present', () => { + expect(mapToolApprovalPolicy({ allow: [], deny: [], ask: [] })).toBeUndefined(); + }); + + test('passes through mode/allow/deny/ask/reason verbatim', () => { + const policy: TToolApprovalPolicy = { + mode: 'dontAsk', + allow: ['read_*', 'mcp:github:*'], + deny: ['delete_*'], + ask: ['execute_*'], + reason: 'Tool {tool} requires review', + }; + expect(mapToolApprovalPolicy(policy)).toEqual({ + mode: 'dontAsk', + allow: ['read_*', 'mcp:github:*'], + deny: ['delete_*'], + ask: ['execute_*'], + reason: 'Tool {tool} requires review', + }); + }); + + test('strips enabled regardless of value (LibreChat-only field)', () => { + expect(mapToolApprovalPolicy({ enabled: false, mode: 'bypass' })).toEqual({ + mode: 'bypass', + }); + expect(mapToolApprovalPolicy({ enabled: true, allow: ['read_*'] })).toEqual({ + allow: ['read_*'], + }); + }); + + test('omits empty list fields from the output', () => { + expect(mapToolApprovalPolicy({ mode: 'default', allow: [], deny: ['rm'] })).toEqual({ + mode: 'default', + deny: ['rm'], + }); + }); +}); + +describe('buildToolApprovalPayload', () => { + const calls = [ + { + name: 'shell', + arguments: { command: 'ls' }, + tool_call_id: 'call_abc', + description: 'List files', + }, + ]; + + test('produces a tool_approval-discriminated payload', () => { + const payload = buildToolApprovalPayload(calls); + expect(payload.type).toBe('tool_approval'); + expect(payload.action_requests).toEqual([ + { + name: 'shell', + arguments: { command: 'ls' }, + tool_call_id: 'call_abc', + description: 'List files', + }, + ]); + }); + + test("default decisions exclude 'respond' (reserved for AskUserQuestion semantics)", () => { + const payload = buildToolApprovalPayload(calls); + expect(payload.review_configs[0].allowed_decisions).toEqual(['approve', 'reject', 'edit']); + }); + + test('respects per-tool decision overrides', () => { + const payload = buildToolApprovalPayload(calls, { + shell: ['approve', 'reject'], + }); + expect(payload.review_configs[0].allowed_decisions).toEqual(['approve', 'reject']); + }); + + test('produces one review_config per call, in order', () => { + const payload = buildToolApprovalPayload([ + { name: 'a', arguments: {}, tool_call_id: '1' }, + { name: 'b', arguments: {}, tool_call_id: '2' }, + ]); + expect(payload.review_configs.map((r) => r.action_name)).toEqual(['a', 'b']); + }); + + test('carries tool_call_id on each review_config (join key for duplicate-tool batches)', () => { + const payload = buildToolApprovalPayload([ + { name: 'mcp:server:search', arguments: { q: 'a' }, tool_call_id: 'call_1' }, + { name: 'mcp:server:search', arguments: { q: 'b' }, tool_call_id: 'call_2' }, + ]); + expect(payload.review_configs).toEqual([ + { + action_name: 'mcp:server:search', + tool_call_id: 'call_1', + allowed_decisions: ['approve', 'reject', 'edit'], + }, + { + action_name: 'mcp:server:search', + tool_call_id: 'call_2', + allowed_decisions: ['approve', 'reject', 'edit'], + }, + ]); + }); +}); + +describe('buildAskUserQuestionPayload', () => { + test('produces an ask_user_question-discriminated payload', () => { + const payload = buildAskUserQuestionPayload({ + question: 'Which environment?', + options: [ + { label: 'Staging', value: 'staging' }, + { label: 'Production', value: 'production' }, + ], + }); + expect(payload.type).toBe('ask_user_question'); + expect(payload.question.question).toBe('Which environment?'); + expect(payload.question.options).toHaveLength(2); + }); + + test('options are optional', () => { + const payload = buildAskUserQuestionPayload({ question: 'Free-form?' }); + expect(payload.question.options).toBeUndefined(); + }); +}); + +describe('buildPendingAction', () => { + const ctx = { + streamId: 'stream-1', + conversationId: 'conv-1', + runId: 'run-1', + responseMessageId: 'msg-1', + }; + + const toolApprovalPayload: Agents.ToolApprovalInterruptPayload = { + type: 'tool_approval', + action_requests: [{ name: 'shell', arguments: { command: 'ls' }, tool_call_id: 'call_abc' }], + review_configs: [ + { action_name: 'shell', tool_call_id: 'call_abc', allowed_decisions: ['approve', 'reject'] }, + ], + }; + + test('wraps a tool_approval payload with job context', () => { + const action = buildPendingAction(toolApprovalPayload, ctx); + expect(action.streamId).toBe('stream-1'); + expect(action.conversationId).toBe('conv-1'); + expect(action.runId).toBe('run-1'); + expect(action.responseMessageId).toBe('msg-1'); + expect(action.payload).toBe(toolApprovalPayload); + expect(typeof action.createdAt).toBe('number'); + }); + + test('wraps an ask_user_question payload with the same envelope', () => { + const askPayload: Agents.AskUserQuestionInterruptPayload = { + type: 'ask_user_question', + question: { question: 'Which env?' }, + }; + const action = buildPendingAction(askPayload, ctx); + expect(action.payload.type).toBe('ask_user_question'); + }); + + test('generates a uuid actionId by default', () => { + const a = buildPendingAction(toolApprovalPayload, ctx); + const b = buildPendingAction(toolApprovalPayload, ctx); + expect(a.actionId).not.toBe(b.actionId); + expect(a.actionId).toMatch(/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i); + }); + + test('honours an explicit actionId', () => { + const action = buildPendingAction(toolApprovalPayload, { ...ctx, actionId: 'fixed-id' }); + expect(action.actionId).toBe('fixed-id'); + }); + + test('sets expiresAt only when ttlMs is provided', () => { + const without = buildPendingAction(toolApprovalPayload, ctx); + expect(without.expiresAt).toBeUndefined(); + + const ttl = 5_000; + const before = Date.now(); + const withTtl = buildPendingAction(toolApprovalPayload, { ...ctx, ttlMs: ttl }); + const after = Date.now(); + expect(withTtl.expiresAt).toBeDefined(); + 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 new file mode 100644 index 0000000000..bfe0543c7c --- /dev/null +++ b/packages/api/src/agents/hitl/policy.ts @@ -0,0 +1,150 @@ +import { randomUUID } from 'crypto'; +import type { Agents, TToolApprovalPolicy } from 'librechat-data-provider'; +import type { ToolPolicyConfig } from '@librechat/agents'; + +/** + * Default decisions offered to the user for a paused tool call. + * + * `'respond'` is intentionally NOT in the default set: it represents the agent + * substituting a synthetic tool result, which is rarely the right ergonomic for + * a stock approval prompt. Hosts that want it can pass an override. + */ +const DEFAULT_REVIEW_DECISIONS: Agents.ToolApprovalDecisionType[] = ['approve', 'reject', 'edit']; + +/** + * Whether the HITL machinery should run for this policy. + * + * 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 + * supply `compileOptions.checkpointer` with a durable saver + * (`LibreChatCheckpointSaver`). Otherwise the SDK installs a process-local + * `MemorySaver` fallback, which silently breaks resume across worker hops in + * any multi-process deployment. Pair this predicate with the checkpointer + * assignment at the `Run.create` call site. + */ +export function isHITLEnabled(policy: TToolApprovalPolicy | undefined): boolean { + return policy?.enabled === true; +} + +/** + * Map a LibreChat tool-approval policy to the SDK's `ToolPolicyConfig`. + * + * Returns `undefined` when there's nothing to configure (so the SDK's own + * defaults apply). The `enabled` field is LibreChat-only and stripped here β€” + * it's consumed separately via {@link isHITLEnabled} to gate the SDK opt-out. + */ +export function mapToolApprovalPolicy( + policy: TToolApprovalPolicy | undefined, +): ToolPolicyConfig | undefined { + if (!policy) { + return undefined; + } + const config: ToolPolicyConfig = {}; + if (policy.mode) { + config.mode = policy.mode; + } + if (policy.allow && policy.allow.length > 0) { + config.allow = policy.allow; + } + if (policy.deny && policy.deny.length > 0) { + config.deny = policy.deny; + } + if (policy.ask && policy.ask.length > 0) { + config.ask = policy.ask; + } + if (policy.reason) { + config.reason = policy.reason; + } + return Object.keys(config).length > 0 ? config : undefined; +} + +/** Tool-call shape consumed by {@link buildToolApprovalPayload}. */ +export interface ToolApprovalCallInput { + name: string; + arguments: string | Record; + tool_call_id: string; + description?: string; +} + +/** + * Build a tool-approval interrupt payload from one or more paused tool calls. + * + * Mirrors the SDK's `ToolApprovalInterruptPayload` shape so this can be used + * to synthesize payloads in tests, or by the host before the SDK upgrade ships. + */ +export function buildToolApprovalPayload( + toolCalls: ToolApprovalCallInput[], + decisionsByToolName?: Record, +): Agents.ToolApprovalInterruptPayload { + return { + type: 'tool_approval', + action_requests: toolCalls.map((tc) => ({ + name: tc.name, + arguments: tc.arguments, + tool_call_id: tc.tool_call_id, + description: tc.description, + })), + review_configs: toolCalls.map((tc) => ({ + action_name: tc.name, + tool_call_id: tc.tool_call_id, + allowed_decisions: decisionsByToolName?.[tc.name] ?? DEFAULT_REVIEW_DECISIONS, + })), + }; +} + +/** Build an ask-user-question interrupt payload. */ +export function buildAskUserQuestionPayload( + question: Agents.AskUserQuestionRequest, +): Agents.AskUserQuestionInterruptPayload { + return { + type: 'ask_user_question', + question, + }; +} + +/** Job-context fields wrapped around a {@link Agents.HumanInterruptPayload}. */ +export interface PendingActionContext { + streamId: string; + conversationId?: string; + /** Stable per-turn identifier (e.g. responseMessageId or LangGraph checkpoint_ns). */ + runId?: string; + responseMessageId?: string; + /** Optional TTL (ms). When set, `expiresAt = createdAt + ttlMs`. */ + ttlMs?: number; + /** Override actionId; defaults to a fresh uuid. */ + actionId?: string; + /** SDK interrupt id (`RunInterruptResult.interruptId`) for cross-process resume. */ + interruptId?: string; + /** LangGraph `thread_id` (`RunInterruptResult.threadId`) for cross-process resume. */ + threadId?: string; +} + +/** + * Wrap a HumanInterruptPayload (from the SDK or synthesized locally) as a + * {@link Agents.PendingAction} record persisted with the job. + * + * Accepts both interrupt categories (`tool_approval` and `ask_user_question`) + * via the discriminated union β€” the host doesn't need to branch. + */ +export function buildPendingAction( + payload: Agents.HumanInterruptPayload, + ctx: PendingActionContext, +): Agents.PendingAction { + const createdAt = Date.now(); + return { + actionId: ctx.actionId ?? randomUUID(), + streamId: ctx.streamId, + conversationId: ctx.conversationId, + runId: ctx.runId, + responseMessageId: ctx.responseMessageId, + payload, + createdAt, + expiresAt: typeof ctx.ttlMs === 'number' ? createdAt + ctx.ttlMs : undefined, + interruptId: ctx.interruptId, + threadId: ctx.threadId, + }; +} diff --git a/packages/api/src/agents/hitl/typeContract.spec.ts b/packages/api/src/agents/hitl/typeContract.spec.ts new file mode 100644 index 0000000000..419eefb39a --- /dev/null +++ b/packages/api/src/agents/hitl/typeContract.spec.ts @@ -0,0 +1,44 @@ +import type { + HumanInterruptPayload as SdkHumanInterruptPayload, + ToolApprovalRequest as SdkToolApprovalRequest, + ToolApprovalDecisionType as SdkToolApprovalDecisionType, +} from '@librechat/agents'; +import type { Agents } from 'librechat-data-provider'; + +/** + * Compile-time contract between the SDK's HITL wire types and LibreChat's + * `Agents.*` mirror in `librechat-data-provider`. The mirror is hand-maintained + * (data-provider can't depend on `@librechat/agents`), so these assignability + * checks are the seam that fails the build when the two drift. + * + * The assertions live inside the function signatures: each `accept*` function's + * parameter type forces TypeScript to prove assignability at compile time. If + * the SDK adds a field the mirror lacks (or a decision literal changes), this + * file stops compiling β€” caught here instead of silently dropped on the Redis + * round-trip. The runtime `expect`s exist only so Jest sees real tests. + */ +describe('HITL type contract: @librechat/agents ↔ librechat-data-provider', () => { + test('the SDK interrupt payload is persistable as the LC mirror', () => { + // Direction that matters most: `Run.getInterrupt()` returns the SDK payload, + // which `approvals.pause()` persists as `Agents.PendingAction.payload`. + // Losing a field here = silent data loss across the pause/resume boundary. + const acceptLcPayload = (p: Agents.HumanInterruptPayload): Agents.HumanInterruptType => p.type; + const fromSdk = (p: SdkHumanInterruptPayload) => acceptLcPayload(p); + expect(typeof fromSdk).toBe('function'); + }); + + test('the SDK action request is persistable as the LC mirror', () => { + const acceptLcRequest = (r: Agents.ToolApprovalRequest): string => r.tool_call_id; + const fromSdk = (r: SdkToolApprovalRequest) => acceptLcRequest(r); + expect(typeof fromSdk).toBe('function'); + }); + + test('decision-type literals match in both directions (resume input contract)', () => { + // What an approval route sends to `run.resume()` must be a valid SDK + // decision, and the LC mirror must enumerate exactly the SDK's literals. + const lcToSdk = (d: Agents.ToolApprovalDecisionType): SdkToolApprovalDecisionType => d; + const sdkToLc = (d: SdkToolApprovalDecisionType): Agents.ToolApprovalDecisionType => d; + expect(typeof lcToSdk).toBe('function'); + expect(typeof sdkToLc).toBe('function'); + }); +}); diff --git a/packages/api/src/agents/index.ts b/packages/api/src/agents/index.ts index 14e0f480fd..fc14367e3e 100644 --- a/packages/api/src/agents/index.ts +++ b/packages/api/src/agents/index.ts @@ -27,3 +27,4 @@ export * from './tools'; export * from './validation'; export * from './added'; export * from './load'; +export * from './hitl'; diff --git a/packages/api/src/stream/ApprovalLifecycle.ts b/packages/api/src/stream/ApprovalLifecycle.ts new file mode 100644 index 0000000000..e117c7bee2 --- /dev/null +++ b/packages/api/src/stream/ApprovalLifecycle.ts @@ -0,0 +1,129 @@ +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`). + * + * Owns the legal transitions β€” pause, resolve, expire β€” behind one interface, + * on top of the store's atomic {@link IJobStore.transitionStatus}. Callers + * (approval routes, the status endpoint, the run seam) cross this seam instead + * of re-implementing the "is this transition legal from the current state, and + * is it safe under a concurrent second submit" logic at each site. + * + * Race-safety is the point. Two approval clicks racing to resume the same job + * must not both drive the run β€” a double-drive re-executes tools and + * double-bills. {@link resolve} returns `true` to exactly one caller; the loser + * gets `false`. The same guard protects {@link pause} (don't pause a job that + * was aborted between the interrupt firing and the mark) and {@link expire}. + * + * State machine: + * ``` + * running ──pause(pendingAction)──▢ requires_action + * requires_action ──resolve()──────▢ running + * requires_action ──expire()───────▢ aborted (the edge that was undefined) + * ``` + */ +export class ApprovalLifecycle { + constructor(private readonly store: IJobStore) {} + + /** + * `running β†’ requires_action`, attaching the pending review record. + * Returns `false` when the job was not running (aborted mid-flight, gone), + * so a late interrupt is dropped rather than pausing a dead job. + */ + async pause(streamId: string, pendingAction: Agents.PendingAction): Promise { + const ok = await this.store.transitionStatus(streamId, { + from: 'running', + to: 'requires_action', + // pendingActionId is the flat mirror the atomic resolve/expire guard on. + patch: { pendingAction, pendingActionId: pendingAction.actionId }, + }); + if (ok) { + logger.debug( + `[ApprovalLifecycle] paused for review: ${streamId} action=${pendingAction.actionId}`, + ); + } + return ok; + } + + /** + * The pending review record, or `null` when the job isn't awaiting review. + * A past-`expiresAt` record reads as `null` (lazy expiry) so a stale prompt + * is never surfaced to a UI or fed to a resume. + */ + async peek(streamId: string): Promise { + const job = await this.store.getJob(streamId); + if (!job || job.status !== 'requires_action') { + return null; + } + // isPendingActionStale covers both a missing record and a past-expiry one. + return isPendingActionStale(job) ? null : (job.pendingAction ?? null); + } + + /** + * `requires_action β†’ running`, atomically. Returns `true` to the single + * caller that won the transition; `false` if the job was not paused, was + * already resumed by a racing submit, no longer matches `expectedActionId`, + * or had expired β€” in which case it is moved to a terminal state instead of + * resumed. + * + * Pass `expectedActionId` (the id the user actually decided on, from the + * approval route) so a stale decision can't resume a job that has since + * paused for a *different* action. Omit it only for callers with no specific + * action in hand. + * + * The caller MUST treat `false` as "do not drive the run": only the `true` + * winner may re-enter the agent. + */ + async resolve(streamId: string, expectedActionId?: string): Promise { + const job = await this.store.getJob(streamId); + if (job?.status === 'requires_action' && !job.pendingAction) { + // The prompt was lost (e.g. a malformed record dropped on deserialize). + // It can't be reviewed, so finalize the job instead of driving a resumed + // run with no reviewed interrupt payload β€” consistent with how the active + // listing and cleanup treat a stale pending action. + await this.expire(streamId); + return false; + } + 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. + await this.expire(streamId, expectedActionId ?? job.pendingAction.actionId); + return false; + } + return this.store.transitionStatus(streamId, { + from: 'requires_action', + to: 'running', + clear: ['pendingAction', 'pendingActionId'], + // Refresh the liveness basis so a long-paused run isn't reaped as stale + // immediately after resuming (cleanup keys off lastActiveAt). + patch: { lastActiveAt: Date.now() }, + expectActionId: expectedActionId, + }); + } + + /** + * `requires_action β†’ aborted`: the edge that fires when no decision arrives + * in time. Previously undefined; now an explicit, idempotent terminal + * transition. Returns `true` to the single caller that expired it. Honors + * `expectedActionId` for the same stale-decision protection as `resolve`. + */ + async expire(streamId: string, expectedActionId?: string): Promise { + const ok = await this.store.transitionStatus(streamId, { + from: 'requires_action', + to: 'aborted', + clear: ['pendingAction', 'pendingActionId'], + // completedAt lets the stores' terminal-cleanup reclaim the job; without + // it an expired approval lingers in the in-memory map indefinitely. + patch: { error: 'Approval expired before a decision was made', completedAt: Date.now() }, + expectActionId: expectedActionId, + }); + if (ok) { + logger.debug(`[ApprovalLifecycle] expired pending review: ${streamId}`); + } + return ok; + } +} diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index 42b396bd30..1986641189 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -31,6 +31,8 @@ import { import { InMemoryEventTransport } from './implementations/InMemoryEventTransport'; import { InMemoryJobStore } from './implementations/InMemoryJobStore'; import { filterPersistableAbortContent } from './abortContent'; +import { isPendingActionStale } from './interfaces/IJobStore'; +import { ApprovalLifecycle } from './ApprovalLifecycle'; /** Error surfaced to any client still attached when a stale/hung job is reaped. */ const REAPED_JOB_ERROR = 'Generation timed out'; @@ -176,6 +178,8 @@ interface RuntimeJobState { class GenerationJobManagerClass { /** Job metadata + content state storage - swappable for Redis, etc. */ private jobStore: IJobStore; + /** Guarded human-review lifecycle (pause / resolve / expire) over the store. */ + private _approvals: ApprovalLifecycle; /** Event pub/sub transport - swappable for Redis Pub/Sub, etc. */ private eventTransport: IEventTransport; @@ -202,6 +206,7 @@ class GenerationJobManagerClass { constructor(options?: GenerationJobManagerOptions) { this.jobStore = options?.jobStore ?? new InMemoryJobStore({ ttlAfterComplete: 0, maxJobs: 1000 }); + this._approvals = new ApprovalLifecycle(this.jobStore); this.eventTransport = options?.eventTransport ?? new InMemoryEventTransport(); this._cleanupOnComplete = options?.cleanupOnComplete ?? true; } @@ -260,6 +265,7 @@ class GenerationJobManagerClass { setGenerationJobsInFlight(previousStore, 0); this.jobStore = services.jobStore; + this._approvals = new ApprovalLifecycle(this.jobStore); this.eventTransport = services.eventTransport; this._isRedis = services.isRedis ?? false; this._cleanupOnComplete = services.cleanupOnComplete ?? true; @@ -486,6 +492,9 @@ class GenerationJobManagerClass { iconURL: jobData.iconURL, model: jobData.model, promptTokens: jobData.promptTokens, + // Surface the pending review so status/resume routes built on the + // facade can render the prompt for a `requires_action` job. + pendingAction: jobData.pendingAction, }, readyPromise: runtime.readyPromise, resolveReady: runtime.resolveReady, @@ -1429,6 +1438,21 @@ class GenerationJobManagerClass { this.jobStore.setGraph(streamId, graph); } + /** + * The guarded human-review lifecycle for paused runs: + * `approvals.pause()` / `peek()` / `resolve()` / `expire()`. + * + * This is the seam approval routes, the status endpoint, and the run wiring + * cross β€” it owns the legal `requires_action` transitions and is race-safe + * against concurrent resumes (a double-resolve would otherwise drive the run + * twice). The job's chunks, run steps, and user-active-set membership are + * preserved across a pause so the resume path can rebuild context; the store + * refreshes the job-hash TTL to give the user the full window to respond. + */ + get approvals(): ApprovalLifecycle { + return this._approvals; + } + /** * Get resume state for reconnecting clients. */ @@ -1499,6 +1523,12 @@ class GenerationJobManagerClass { replayEvents, collectedUsage, contextUsage, + // Carry the live pending approval in the resume contract so a reloading / + // cross-replica client can rebuild the prompt from resumeState. + pendingAction: + jobData.status === 'requires_action' && !isPendingActionStale(jobData) + ? jobData.pendingAction + : undefined, }; } @@ -1678,13 +1708,14 @@ class GenerationJobManagerClass { * Get job count by status. */ async getJobCountByStatus(): Promise> { - const [running, complete, error, aborted] = await Promise.all([ + const [running, complete, error, aborted, requires_action] = await Promise.all([ this.jobStore.getJobCountByStatus('running'), this.jobStore.getJobCountByStatus('complete'), this.jobStore.getJobCountByStatus('error'), this.jobStore.getJobCountByStatus('aborted'), + this.jobStore.getJobCountByStatus('requires_action'), ]); - return { running, complete, error, aborted }; + return { running, complete, error, aborted, requires_action }; } /** 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..55229ebc5e 100644 --- a/packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts +++ b/packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts @@ -1,7 +1,7 @@ +import { StandardGraph } from '@librechat/agents'; import { StepTypes } from 'librechat-data-provider'; import type { Agents } from 'librechat-data-provider'; import type { Redis, Cluster } from 'ioredis'; -import { StandardGraph } from '@librechat/agents'; /** Suppress winston Console transport output (survives jest.resetModules) */ jest.spyOn(console, 'log').mockImplementation(); @@ -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,187 @@ 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.transitionStatus(streamId, { + from: 'running', + to: 'requires_action', + patch: { 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.transitionStatus(streamId, { + from: 'running', + to: 'requires_action', + patch: { pendingAction: buildPendingAction(streamId) }, + }); + + await store.transitionStatus(streamId, { + from: 'requires_action', + to: 'running', + clear: ['pendingAction'], + }); + + 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 refresh resume state TTLs when pausing and resuming a job', async () => { + if (!ioredisClient) { + return; + } + + const { RedisJobStore } = await import('../implementations/RedisJobStore'); + const store = new RedisJobStore(ioredisClient, { runningTtl: 120 }); + await store.initialize(); + + const streamId = `requires-action-ttl-${Date.now()}`; + const chunkKey = `stream:{${streamId}}:chunks`; + const runStepsKey = `stream:{${streamId}}:runsteps`; + + await store.createJob(streamId, 'user-1', streamId); + await store.appendChunk(streamId, { event: 'on_message_delta', data: { text: 'hello' } }); + const runSteps: Partial[] = [ + { id: 'step-1', runId: 'run-1', type: StepTypes.MESSAGE_CREATION, index: 0 }, + ]; + await store.saveRunSteps(streamId, runSteps as Agents.RunStep[]); + + await ioredisClient.expire(chunkKey, 30); + await ioredisClient.expire(runStepsKey, 30); + + await store.transitionStatus(streamId, { + from: 'running', + to: 'requires_action', + patch: { pendingAction: buildPendingAction(streamId) }, + }); + + expect(await ioredisClient.ttl(chunkKey)).toBeGreaterThan(30); + expect(await ioredisClient.ttl(runStepsKey)).toBeGreaterThan(30); + + await ioredisClient.expire(chunkKey, 30); + await ioredisClient.expire(runStepsKey, 30); + + 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); + + 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.transitionStatus(streamId, { + from: 'running', + to: 'requires_action', + patch: { 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(); + }); + + test('should prune expired requires_action IDs during cleanup', async () => { + if (!ioredisClient) { + return; + } + + const { RedisJobStore } = await import('../implementations/RedisJobStore'); + const store = new RedisJobStore(ioredisClient); + await store.initialize(); + + const streamId = `requires-action-expired-${Date.now()}`; + const jobKey = `stream:{${streamId}}:job`; + await store.createJob(streamId, 'user-1', streamId); + await store.transitionStatus(streamId, { + from: 'running', + to: 'requires_action', + patch: { pendingAction: buildPendingAction(streamId) }, + }); + + expect(await ioredisClient.smembers('stream:requires_action')).toContain(streamId); + + await ioredisClient.del(jobKey); + + const cleaned = await store.cleanup(); + const pausedMembers = await ioredisClient.smembers('stream:requires_action'); + + expect(cleaned).toBeGreaterThanOrEqual(1); + expect(pausedMembers).not.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/__tests__/pendingAction.spec.ts b/packages/api/src/stream/__tests__/pendingAction.spec.ts new file mode 100644 index 0000000000..de4136a513 --- /dev/null +++ b/packages/api/src/stream/__tests__/pendingAction.spec.ts @@ -0,0 +1,251 @@ +import type { Agents } from 'librechat-data-provider'; +import { InMemoryEventTransport } from '~/stream/implementations/InMemoryEventTransport'; +import { buildPendingAction, buildToolApprovalPayload } from '~/agents/hitl/policy'; +import { InMemoryJobStore } from '~/stream/implementations/InMemoryJobStore'; +import { GenerationJobManagerClass } from '~/stream/GenerationJobManager'; + +jest.spyOn(console, 'log').mockImplementation(); + +describe('ApprovalLifecycle via GenerationJobManager.approvals (in-memory)', () => { + let manager: GenerationJobManagerClass; + + beforeEach(() => { + manager = new GenerationJobManagerClass(); + manager.configure({ + jobStore: new InMemoryJobStore({ ttlAfterComplete: 60000 }), + eventTransport: new InMemoryEventTransport(), + isRedis: false, + cleanupOnComplete: false, + }); + manager.initialize(); + }); + + afterEach(async () => { + await manager.destroy(); + }); + + function buildAction(streamId: string, overrides: Partial = {}) { + const payload = buildToolApprovalPayload([ + { name: 'shell', arguments: { command: 'ls' }, tool_call_id: 'call_abc' }, + ]); + const action = buildPendingAction(payload, { + streamId, + conversationId: streamId, + runId: 'run-1', + responseMessageId: 'msg-1', + }); + return { ...action, ...overrides }; + } + + describe('pause', () => { + test('running β†’ requires_action, persisting the pending record', async () => { + const streamId = 'stream-pause'; + await manager.createJob(streamId, 'user-1'); + + const action = buildAction(streamId); + expect(await manager.approvals.pause(streamId, action)).toBe(true); + + expect(await manager.getJobStatus(streamId)).toBe('requires_action'); + const pending = await manager.approvals.peek(streamId); + expect(pending?.actionId).toBe(action.actionId); + expect(pending?.payload.type).toBe('tool_approval'); + if (pending?.payload.type === 'tool_approval') { + expect(pending.payload.action_requests[0].name).toBe('shell'); + } + }); + + test('returns false when the job is already terminal', async () => { + const streamId = 'stream-pause-dead'; + await manager.createJob(streamId, 'user-1'); + await manager.completeJob(streamId, 'terminated mid-flight'); + + expect(await manager.approvals.pause(streamId, buildAction(streamId))).toBe(false); + // a late interrupt must NOT resurrect a terminal job into requires_action + expect(await manager.getJobStatus(streamId)).not.toBe('requires_action'); + }); + + test('returns false when the job does not exist', async () => { + expect(await manager.approvals.pause('nonexistent', buildAction('nonexistent'))).toBe(false); + }); + }); + + describe('peek', () => { + test('returns null for jobs not in requires_action', async () => { + const streamId = 'stream-running'; + await manager.createJob(streamId, 'user-1'); + expect(await manager.approvals.peek(streamId)).toBeNull(); + }); + + test('returns null when the job does not exist', async () => { + expect(await manager.approvals.peek('nonexistent')).toBeNull(); + }); + + test('treats a past-expiresAt record as gone (lazy expiry)', async () => { + const streamId = 'stream-expired-peek'; + await manager.createJob(streamId, 'user-1'); + await manager.approvals.pause( + streamId, + buildAction(streamId, { expiresAt: Date.now() - 1000 }), + ); + + expect(await manager.approvals.peek(streamId)).toBeNull(); + }); + }); + + describe('resolve', () => { + test('requires_action β†’ running, clearing the record, returns true once', async () => { + const streamId = 'stream-resolve'; + await manager.createJob(streamId, 'user-1'); + await manager.approvals.pause(streamId, buildAction(streamId)); + + expect(await manager.approvals.resolve(streamId)).toBe(true); + expect(await manager.getJobStatus(streamId)).toBe('running'); + expect(await manager.approvals.peek(streamId)).toBeNull(); + }); + + test('a concurrent double-resolve wins exactly once (race-safe)', async () => { + const streamId = 'stream-double-resolve'; + await manager.createJob(streamId, 'user-1'); + await manager.approvals.pause(streamId, buildAction(streamId)); + + const results = await Promise.all([ + manager.approvals.resolve(streamId), + manager.approvals.resolve(streamId), + ]); + + // Exactly one caller may drive the run β€” the other must be rejected. + expect(results.filter(Boolean)).toHaveLength(1); + expect(await manager.getJobStatus(streamId)).toBe('running'); + }); + + test('returns false when the job is not paused', async () => { + const streamId = 'stream-resolve-running'; + await manager.createJob(streamId, 'user-1'); + expect(await manager.approvals.resolve(streamId)).toBe(false); + }); + + test('rejects a resolve whose actionId no longer matches (stale-decision guard)', async () => { + const streamId = 'stream-stale-action'; + await manager.createJob(streamId, 'user-1'); + const action = buildAction(streamId); + await manager.approvals.pause(streamId, action); + + // A decision targeting a different action must not resume this one. + expect(await manager.approvals.resolve(streamId, 'some-other-action-id')).toBe(false); + expect(await manager.getJobStatus(streamId)).toBe('requires_action'); + + // The matching actionId resolves it. + expect(await manager.approvals.resolve(streamId, action.actionId)).toBe(true); + expect(await manager.getJobStatus(streamId)).toBe('running'); + }); + + test('an expired pending action expires instead of resuming', async () => { + const streamId = 'stream-resolve-expired'; + await manager.createJob(streamId, 'user-1'); + await manager.approvals.pause( + streamId, + buildAction(streamId, { expiresAt: Date.now() - 1000 }), + ); + + expect(await manager.approvals.resolve(streamId)).toBe(false); + expect(await manager.getJobStatus(streamId)).toBe('aborted'); + }); + }); + + describe('expire', () => { + test('requires_action β†’ aborted, clearing the record, returns true once', async () => { + const streamId = 'stream-expire'; + await manager.createJob(streamId, 'user-1'); + await manager.approvals.pause(streamId, buildAction(streamId)); + + expect(await manager.approvals.expire(streamId)).toBe(true); + expect(await manager.getJobStatus(streamId)).toBe('aborted'); + expect(await manager.approvals.peek(streamId)).toBeNull(); + + // idempotent β€” a second expire does not fire again + expect(await manager.approvals.expire(streamId)).toBe(false); + }); + + test('returns false when the job is not paused', async () => { + const streamId = 'stream-expire-running'; + await manager.createJob(streamId, 'user-1'); + expect(await manager.approvals.expire(streamId)).toBe(false); + }); + + test('sets completedAt so terminal cleanup can reclaim the job', async () => { + const streamId = 'stream-expire-completed'; + await manager.createJob(streamId, 'user-1'); + await manager.approvals.pause(streamId, buildAction(streamId)); + + expect(await manager.approvals.expire(streamId)).toBe(true); + const job = await manager.getJob(streamId); + expect(job?.status).toBe('aborted'); + expect(job?.completedAt).toBeGreaterThan(0); + }); + }); + + describe('facade integration', () => { + test('requires_action drops the running count but keeps the user-active set', async () => { + const streamId = 'stream-counts'; + await manager.createJob(streamId, 'user-counts'); + + const before = await manager.getJobCountByStatus(); + expect(before.running).toBe(1); + expect(before.requires_action).toBe(0); + + await manager.approvals.pause(streamId, buildAction(streamId)); + + const after = await manager.getJobCountByStatus(); + expect(after.running).toBe(0); + expect(after.requires_action).toBe(1); + + // Pending-approval jobs still occupy the user's conversation slot. + expect(await manager.getActiveJobIdsForUser('user-counts')).toContain(streamId); + }); + + test('getActiveJobIdsForUser excludes terminal jobs but includes requires_action', async () => { + await manager.createJob('s-running', 'user-mix'); + await manager.createJob('s-paused', 'user-mix'); + await manager.createJob('s-done', 'user-mix'); + + await manager.approvals.pause('s-paused', buildAction('s-paused')); + await manager.completeJob('s-done'); + + const active = await manager.getActiveJobIdsForUser('user-mix'); + expect(active.sort()).toEqual(['s-paused', 's-running']); + }); + + test('excludes a pending-approval job whose prompt has expired', async () => { + const streamId = 'stream-expired-active'; + await manager.createJob(streamId, 'user-exp'); + await manager.approvals.pause( + streamId, + buildAction(streamId, { expiresAt: Date.now() - 1000 }), + ); + + // Still requires_action, but the prompt is past expiry β†’ no longer active. + expect(await manager.getActiveJobIdsForUser('user-exp')).not.toContain(streamId); + }); + }); +}); + +describe('InMemoryJobStore β€” approval expiry cleanup', () => { + test('cleanup() finalizes and reclaims a past-expiry pending-approval job', async () => { + const store = new InMemoryJobStore({ ttlAfterComplete: 0 }); + await store.createJob('s1', 'u1'); + + const action = buildPendingAction( + buildToolApprovalPayload([{ name: 'shell', arguments: {}, tool_call_id: 'c1' }]), + { streamId: 's1', ttlMs: -1000 }, + ); + await store.transitionStatus('s1', { + from: 'running', + to: 'requires_action', + patch: { pendingAction: action, pendingActionId: action.actionId }, + }); + + // A past-expiry approval must be finalized + reclaimed, not left resident. + await store.cleanup(); + expect(await store.getJob('s1')).toBeNull(); + }); +}); diff --git a/packages/api/src/stream/implementations/InMemoryJobStore.ts b/packages/api/src/stream/implementations/InMemoryJobStore.ts index f318f3baf5..f8caf2b9ed 100644 --- a/packages/api/src/stream/implementations/InMemoryJobStore.ts +++ b/packages/api/src/stream/implementations/InMemoryJobStore.ts @@ -6,7 +6,9 @@ import type { UsageMetadata, IJobStore, JobStatus, + JobStatusTransition, } from '~/stream/interfaces/IJobStore'; +import { isPendingActionStale } from '~/stream/interfaces/IJobStore'; /** * Content state for a job - volatile, in-memory only. @@ -134,9 +136,35 @@ 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); } + /** + * Atomic in-memory: the single-threaded event loop makes the + * read-check-write sequence indivisible, so the status guard is exact. + * Membership/counts derive from `job.status` directly, so there are no + * sets to reconcile here. + */ + async transitionStatus(streamId: string, args: JobStatusTransition): Promise { + const job = this.jobs.get(streamId); + if (!job || job.status !== args.from) { + return false; + } + if (args.expectActionId != null && job.pendingActionId !== args.expectActionId) { + return false; + } + job.status = args.to; + if (args.patch) { + Object.assign(job, args.patch); + } + for (const field of args.clear ?? []) { + delete job[field]; + } + return true; + } + async deleteJob(streamId: string): Promise { this.jobs.delete(streamId); this.contentState.delete(streamId); @@ -181,14 +209,32 @@ export class InMemoryJobStore implements IJobStore { if (this.ttlAfterComplete === 0 || now - job.completedAt > this.ttlAfterComplete) { toDelete.push(streamId); } + } else if (job.status === 'requires_action' && isPendingActionStale(job)) { + // Stale approval (expired, or missing/malformed pendingAction): + // finalize it (aborted) so it stops occupying the user slot and its + // content state is reclaimed, mirroring ApprovalLifecycle.expire(). + // Skipping it (active-list filter) alone would leave it resident. + job.status = 'aborted'; + job.completedAt = now; + job.error = 'Approval expired before a decision was made'; + delete job.pendingAction; + delete job.pendingActionId; + if (this.ttlAfterComplete === 0) { + toDelete.push(streamId); + } } else if (this.staleJobTimeout > 0 && job.status === 'running') { // Failsafe: reap jobs stuck in "running" with no generation activity for // longer than the stale timeout. These are crashed/hung generations that // never reached a terminal state; without this they accumulate their - // content state in memory until the process OOMs. Reaping keys off last - // activity (not creation time) so a long but live stream is never reaped, - // mirroring RedisJobStore refreshing the running TTL on each chunk. - const lastActive = this.lastActivity.get(streamId) ?? job.createdAt; + // content state in memory until the process OOMs. Reaping keys off the + // most recent liveness signal (not creation time) so a long but live + // stream is never reaped, and a just-resumed approval (fresh + // `lastActiveAt`) wins over a stale per-chunk `lastActivity` entry. + const lastActive = Math.max( + this.lastActivity.get(streamId) ?? 0, + job.lastActiveAt ?? 0, + job.createdAt, + ); if (now - lastActive > this.staleJobTimeout) { toDelete.push(streamId); staleRunning++; @@ -295,8 +341,14 @@ export class InMemoryJobStore implements IJobStore { for (const streamId of trackedIds) { const job = this.jobs.get(streamId); - // Only include if job exists AND is still running - if (job && job.status === 'running') { + // Include running jobs and jobs paused for human review (e.g. tool approval). + // A pending-approval job still occupies the user's conversation slot β€” but + // only while its prompt is live: a past-`expiresAt` approval no longer + // counts as active (cleanup/expiry will finalize it). + if (job && (job.status === 'running' || job.status === 'requires_action')) { + if (job.status === 'requires_action' && isPendingActionStale(job)) { + continue; + } activeIds.push(streamId); } else { // Self-healing: job completed/deleted but mapping wasn't cleaned - fix it now diff --git a/packages/api/src/stream/implementations/RedisJobStore.ts b/packages/api/src/stream/implementations/RedisJobStore.ts index a00eeba6b4..32d78ee487 100644 --- a/packages/api/src/stream/implementations/RedisJobStore.ts +++ b/packages/api/src/stream/implementations/RedisJobStore.ts @@ -8,7 +8,39 @@ import type { UsageMetadata, IJobStore, JobStatus, + JobStatusTransition, } from '~/stream/interfaces/IJobStore'; +import { isPendingActionStale } from '~/stream/interfaces/IJobStore'; + +/** + * Atomic compare-and-set on the job hash β€” the single-winner decision for a + * status transition. Touches ONLY the job key, which lives on one hash slot, so + * it is atomic on both single-node and Redis Cluster (cross-slot membership + * sets are reconciled by the caller AFTER this decides the winner). + * + * Guards on the current `status` and, when ARGV[2] is non-empty, on the flat + * `pendingActionId` field β€” so a stale decision targeting a different action + * loses. On success: removes `clear` fields, writes `status`+patch pairs, + * refreshes the job-hash TTL. Returns 1 if it fired, 0 otherwise. + * + * KEYS: [job] + * ARGV: [from, expectActionId | "", ttl, hdelCount, ...hdelFields, ...hsetPairs] + */ +const JOB_CAS_LUA = + 'if redis.call("HGET", KEYS[1], "status") ~= ARGV[1] then return 0 end ' + + 'if ARGV[2] ~= "" and redis.call("HGET", KEYS[1], "pendingActionId") ~= ARGV[2] then return 0 end ' + + 'local ttl = tonumber(ARGV[3]) ' + + 'local hdelCount = tonumber(ARGV[4]) ' + + 'local idx = 5 ' + + 'for i = 1, hdelCount do redis.call("HDEL", KEYS[1], ARGV[idx]) idx = idx + 1 end ' + + 'local hset = {} ' + + 'for i = idx, #ARGV do hset[#hset + 1] = ARGV[i] end ' + + 'if #hset > 0 then redis.call("HSET", KEYS[1], unpack(hset)) end ' + + 'redis.call("EXPIRE", KEYS[1], ttl) ' + + 'return 1'; + +/** Decision kinds the SDK can emit, used to sanity-check persisted records. */ +const KNOWN_INTERRUPT_TYPES = new Set(['tool_approval', 'ask_user_question']); /** * Key prefixes for Redis storage. @@ -29,6 +61,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`, @@ -49,6 +83,14 @@ const DEFAULT_TTL = { runStepsAfterComplete: 0, /** Safety-net TTL for per-user job tracking sets (24 hours). Refreshed on each createJob. */ userJobsSet: 86400, + /** + * Backstop TTL for a job paused for human review (24 hours). A paused job is + * NOT a hung generation, so it must not inherit the 20-minute running TTL β€” + * an approval with no explicit `expiresAt` is "live" per the API contract and + * would otherwise be evicted mid-window. A pendingAction with a longer + * `expiresAt` extends beyond this (see pauseTtlSeconds). + */ + requiresAction: 86400, }; /** @@ -83,6 +125,8 @@ export interface RedisJobStoreOptions { runStepsAfterCompleteTtl?: number; /** TTL for per-user job tracking sets in seconds (default: 86400 = 24 hours). 0 = no TTL. */ userJobsSetTtl?: number; + /** Backstop TTL for a paused (requires_action) job in seconds (default: 86400 = 24 hours). */ + requiresActionTtl?: number; } export class RedisJobStore implements IJobStore { @@ -118,6 +162,7 @@ export class RedisJobStore implements IJobStore { chunksAfterComplete: options?.chunksAfterCompleteTtl ?? DEFAULT_TTL.chunksAfterComplete, runStepsAfterComplete: options?.runStepsAfterCompleteTtl ?? DEFAULT_TTL.runStepsAfterComplete, userJobsSet: options?.userJobsSetTtl ?? DEFAULT_TTL.userJobsSet, + requiresAction: options?.requiresActionTtl ?? DEFAULT_TTL.requiresAction, }; // Detect cluster mode using ioredis's isCluster property this.isCluster = (redis as Cluster).isCluster === true; @@ -161,12 +206,24 @@ export class RedisJobStore implements IJobStore { const key = KEYS.job(streamId); const userJobsKey = KEYS.userJobs(userId, tenantId); + // A reused streamId overlays onto any existing hash, so paused-run fields + // from a prior generation could survive. Drop the HITL fields so the fresh + // running job never exposes stale approval metadata and cleanup keys off the + // new createdAt rather than a leftover lastActiveAt. + const staleHitlFields: Array = [ + 'pendingAction', + 'pendingActionId', + 'lastActiveAt', + ]; + // For cluster mode, we can't pipeline keys on different slots // The job key uses hash tag {streamId}, runningJobs and userJobs are on different slots if (this.isCluster) { await this.redis.hset(key, this.serializeJob(job)); + await this.redis.hdel(key, ...staleHitlFields); 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); @@ -174,8 +231,10 @@ export class RedisJobStore implements IJobStore { } else { const pipeline = this.redis.pipeline(); pipeline.hset(key, this.serializeJob(job)); + pipeline.hdel(key, ...staleHitlFields); 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); @@ -198,71 +257,195 @@ export class RedisJobStore implements IJobStore { async updateJob(streamId: string, updates: Partial): Promise { const key = KEYS.job(streamId); + // 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(); + const updated = await this.updateExistingJobHash(key, fields); + if (!updated) { + return; + } + + if (updates.status && ['complete', 'error', 'aborted'].includes(updates.status)) { + await this.applyTerminalContentCleanup(streamId); + } + } + + /** + * Terminal cleanup shared by `updateJob` (complete/error/aborted) and the + * terminal path of `transitionStatus` (approval expiry β†’ aborted): drop the + * job from both membership sets and the user-active set, shorten the job-hash + * TTL to the completed window, and del/shorten the chunk + run-step keys per + * the configured after-complete TTLs. Without sharing this, an expired + * approval left Redis stream contents around for the full running TTL. + */ + private async applyTerminalContentCleanup(streamId: string): Promise { + const key = KEYS.job(streamId); + // Proactively remove from user's job set (requires reading userId from the job hash) + const job = await this.getJob(streamId); + const userJobsKey = job?.userId ? KEYS.userJobs(job.userId, job.tenantId) : null; + + 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)); + } else { + await this.redis.expire(KEYS.chunks(streamId), this.ttl.chunksAfterComplete); + } + + if (this.ttl.runStepsAfterComplete === 0) { + await this.redis.del(KEYS.runSteps(streamId)); + } else { + await this.redis.expire(KEYS.runSteps(streamId), this.ttl.runStepsAfterComplete); + } + + if (userJobsKey) { + await this.redis.srem(userJobsKey, streamId); + } + } else { + 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)); + } else { + pipeline.expire(KEYS.chunks(streamId), this.ttl.chunksAfterComplete); + } + + if (this.ttl.runStepsAfterComplete === 0) { + pipeline.del(KEYS.runSteps(streamId)); + } else { + pipeline.expire(KEYS.runSteps(streamId), this.ttl.runStepsAfterComplete); + } + + if (userJobsKey) { + pipeline.srem(userJobsKey, streamId); + } + + await pipeline.exec(); + } + } + + /** + * Live-key TTL (seconds) for a paused job. A paused job isn't a hung + * generation, so it uses the longer requires_action backstop rather than the + * running TTL β€” otherwise a no-expiry approval (the buildPendingAction + * default), which the API treats as "live", would be evicted after the 20m + * running window. A pendingAction with an `expiresAt` farther out than the + * backstop extends to cover it, plus a grace margin so a decision arriving + * right at the deadline can still resume. + */ + private pauseTtlSeconds(pendingAction?: Agents.PendingAction): number { + const exp = pendingAction?.expiresAt; + if (exp == null) { + return this.ttl.requiresAction; + } + const secondsUntilExpiry = Math.ceil((exp - Date.now()) / 1000) + 60; + return Math.max(this.ttl.requiresAction, secondsUntilExpiry); + } + + /** The membership set a status belongs to; terminal statuses have none. */ + private statusSetKey(status: JobStatus): string | null { + if (status === 'running') { + return KEYS.runningJobs; + } + if (status === 'requires_action') { + return KEYS.requiresActionJobs; + } + return null; + } + + async transitionStatus(streamId: string, args: JobStatusTransition): Promise { + const { from, to, patch, clear, expectActionId } = args; + const key = KEYS.job(streamId); + + // status + patch become HSET pairs; serializeJob skips undefined, so + // cleared fields go through HDEL (`clear`) instead. + const fields = Object.entries( + this.serializeJob({ status: to, ...(patch ?? {}) } as SerializableJobData), + ).flat(); + const clearFields = (clear ?? []).map(String); + + const remSet = this.statusSetKey(from); + const addSet = this.statusSetKey(to); + const terminal = addSet === null; + let ttl = terminal ? this.ttl.completed : this.ttl.running; + if (to === 'requires_action') { + // A paused job must outlive its approval window, even when that window is + // longer than the running TTL β€” otherwise Redis evicts it before a + // decision can resume it. + ttl = this.pauseTtlSeconds(patch?.pendingAction); + } + + // 1) Single-winner decision: an atomic CAS on the single-slot job hash. + // Works identically on cluster and single-node, so two concurrent + // resolves can never both win (and drive the run twice). + const won = await this.redis.eval( + JOB_CAS_LUA, + 1, + key, + from, + expectActionId ?? '', + String(ttl), + String(clearFields.length), + ...clearFields, + ...fields, + ); + if (won !== 1) { + return false; + } + + // 2) Reconcile derived state. Only the winner reaches here; membership is + // self-healed by periodic cleanup, so this non-atomic cross-slot step is + // safe. A terminal target (e.g. approval expiry β†’ aborted) gets the same + // content cleanup as updateJob's terminal path. + if (terminal) { + await this.applyTerminalContentCleanup(streamId); + return true; + } + if (this.isCluster) { + if (remSet) { + await this.redis.srem(remSet, streamId); + } + if (addSet) { + await this.redis.sadd(addSet, streamId); + } + await this.redis.expire(KEYS.chunks(streamId), ttl); + await this.redis.expire(KEYS.runSteps(streamId), ttl); + } else { + const pipeline = this.redis.pipeline(); + if (remSet) { + pipeline.srem(remSet, streamId); + } + if (addSet) { + pipeline.sadd(addSet, streamId); + } + pipeline.expire(KEYS.chunks(streamId), ttl); + pipeline.expire(KEYS.runSteps(streamId), ttl); + await pipeline.exec(); + } + return true; + } + + 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, ); - - if (updated === 0) { - return; - } - - if (updates.status && ['complete', 'error', 'aborted'].includes(updates.status)) { - // Proactively remove from user's job set (requires reading userId from the job hash) - const job = await this.getJob(streamId); - const userJobsKey = job?.userId ? KEYS.userJobs(job.userId, job.tenantId) : null; - - if (this.isCluster) { - await this.redis.expire(key, this.ttl.completed); - await this.redis.srem(KEYS.runningJobs, streamId); - - if (this.ttl.chunksAfterComplete === 0) { - await this.redis.del(KEYS.chunks(streamId)); - } else { - await this.redis.expire(KEYS.chunks(streamId), this.ttl.chunksAfterComplete); - } - - if (this.ttl.runStepsAfterComplete === 0) { - await this.redis.del(KEYS.runSteps(streamId)); - } else { - await this.redis.expire(KEYS.runSteps(streamId), this.ttl.runStepsAfterComplete); - } - - if (userJobsKey) { - await this.redis.srem(userJobsKey, streamId); - } - } else { - const pipeline = this.redis.pipeline(); - pipeline.expire(key, this.ttl.completed); - pipeline.srem(KEYS.runningJobs, streamId); - - if (this.ttl.chunksAfterComplete === 0) { - pipeline.del(KEYS.chunks(streamId)); - } else { - pipeline.expire(KEYS.chunks(streamId), this.ttl.chunksAfterComplete); - } - - if (this.ttl.runStepsAfterComplete === 0) { - pipeline.del(KEYS.runSteps(streamId)); - } else { - pipeline.expire(KEYS.runSteps(streamId), this.ttl.runStepsAfterComplete); - } - - if (userJobsKey) { - pipeline.srem(userJobsKey, streamId); - } - - await pipeline.exec(); - } - } + return updated === 1; } async deleteJob(streamId: string): Promise { @@ -284,6 +467,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); } @@ -293,6 +477,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); } @@ -345,6 +530,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; @@ -355,6 +549,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); } @@ -363,8 +558,11 @@ export class RedisJobStore implements IJobStore { return 1; } - // Stale running job (failsafe - running for > configured TTL) - if (now - job.createdAt > this.ttl.running * 1000) { + // Stale running job (failsafe - running for > configured TTL). + // Keys off `lastActiveAt` when present so a just-resumed approval + // isn't reaped on the basis of its original creation time. + const liveSince = job.lastActiveAt ?? job.createdAt; + if (now - liveSince > this.ttl.running * 1000) { logger.warn(`[RedisJobStore] Cleaning up stale job: ${streamId}`); const userJobsKey = job.userId ? KEYS.userJobs(job.userId, job.tenantId) : null; await this.deleteJobInternal(streamId, userJobsKey); @@ -383,6 +581,8 @@ export class RedisJobStore implements IJobStore { } } + cleaned += await this.cleanupRequiresActionIndex(); + if (cleaned > 0) { logger.debug(`[RedisJobStore] Cleaned up ${cleaned} jobs`); } @@ -390,11 +590,72 @@ export class RedisJobStore implements IJobStore { return cleaned; } + private async cleanupRequiresActionIndex(): Promise { + const streamIds = await this.redis.smembers(KEYS.requiresActionJobs); + let cleaned = 0; + + const BATCH_SIZE = 50; + for (let i = 0; i < streamIds.length; i += BATCH_SIZE) { + const batch = streamIds.slice(i, i + BATCH_SIZE); + const results = await Promise.allSettled( + batch.map(async (streamId) => { + const job = await this.getJob(streamId); + + if (!job) { + 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.requiresActionJobs, streamId); + if (job.status === 'running') { + await this.redis.sadd(KEYS.runningJobs, streamId); + } + return 1; + } + + // Stale approval (expired, or missing/malformed pendingAction): + // finalize it (aborted) so it stops occupying the slot and its stream + // contents are reclaimed, mirroring ApprovalLifecycle.expire(). + // transitionStatus runs the terminal content cleanup (sets, chunks, + // run-steps, userJobs, completed TTL). + if (isPendingActionStale(job)) { + await this.transitionStatus(streamId, { + from: 'requires_action', + to: 'aborted', + clear: ['pendingAction', 'pendingActionId'], + patch: { + error: 'Approval expired before a decision was made', + completedAt: Date.now(), + }, + }); + return 1; + } + + return 0; + }), + ); + + for (const result of results) { + if (result.status === 'fulfilled') { + cleaned += result.value; + } else { + logger.warn(`[RedisJobStore] requires_action cleanup failed for a job:`, result.reason); + } + } + } + + return cleaned; + } + 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 { @@ -402,11 +663,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. @@ -428,8 +715,15 @@ export class RedisJobStore implements IJobStore { for (const streamId of trackedIds) { const job = await this.getJob(streamId); - // Only include if job exists AND is still running - if (job && job.status === 'running') { + // Include running jobs and jobs paused for human review (e.g. tool approval). + // A pending-approval job still occupies the user's conversation slot β€” but + // only while its prompt is live: a past-`expiresAt` approval no longer + // counts as active (cleanup/expiry will finalize it), so the client stops + // polling and can complete. + if (job && (job.status === 'running' || job.status === 'requires_action')) { + if (job.status === 'requires_action' && isPendingActionStale(job)) { + continue; + } activeIds.push(streamId); } else { // Self-healing: job completed/deleted but mapping wasn't cleaned - mark for removal @@ -925,6 +1219,36 @@ export class RedisJobStore implements IJobStore { replayEvents: data.replayEvents || undefined, contextUsage: data.contextUsage || undefined, tokenUsage: data.tokenUsage || undefined, + pendingAction: this.parsePendingAction(data.pendingAction), + pendingActionId: data.pendingActionId || undefined, + lastActiveAt: data.lastActiveAt ? parseInt(data.lastActiveAt, 10) : undefined, }; } + + /** + * Parse a persisted `pendingAction`, defending the cold-resume path against + * malformed or stale records: a corrupt JSON blob or a payload whose shape + * predates the current SDK contract is dropped (logged) rather than crashing + * the resume or feeding a bad record to an approval route. Returns undefined + * when absent/invalid. + */ + private parsePendingAction(raw: string | undefined): Agents.PendingAction | undefined { + if (!raw) { + return undefined; + } + try { + const parsed = JSON.parse(raw) as Agents.PendingAction; + const typeOk = + typeof parsed?.actionId === 'string' && + KNOWN_INTERRUPT_TYPES.has(parsed?.payload?.type as string); + if (!typeOk) { + logger.warn('[RedisJobStore] Dropping malformed pendingAction record'); + return undefined; + } + return parsed; + } catch { + logger.warn('[RedisJobStore] Dropping unparseable pendingAction record'); + return undefined; + } + } } 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'; diff --git a/packages/api/src/stream/interfaces/IJobStore.ts b/packages/api/src/stream/interfaces/IJobStore.ts index f2e905e4fa..f3dcb752bf 100644 --- a/packages/api/src/stream/interfaces/IJobStore.ts +++ b/packages/api/src/stream/interfaces/IJobStore.ts @@ -2,9 +2,13 @@ import type { StandardGraph } from '@librechat/agents'; import type { Agents } from 'librechat-data-provider'; /** - * Job status enum + * Job status enum. + * + * `requires_action` is non-terminal: the run has paused for human review + * (e.g. tool approval) and is expected to be resumed by an approval route. + * Stores must NOT cleanup `requires_action` jobs as if they were complete. */ -export type JobStatus = 'running' | 'complete' | 'error' | 'aborted'; +export type JobStatus = 'running' | 'complete' | 'error' | 'aborted' | 'requires_action'; /** * Serializable job data - no object references, suitable for Redis/external storage @@ -62,6 +66,68 @@ export interface SerializableJobData { iconURL?: string; model?: string; promptTokens?: number; + + /** + * Set when status is `requires_action`. Describes the human review the + * run is waiting on. Cleared by the resume path before the job returns to `running`. + */ + pendingAction?: Agents.PendingAction; + + /** + * Flat mirror of `pendingAction.actionId`, kept as a top-level field so an + * atomic status transition can guard on it (a nested JSON field can't be + * compared inside a Redis Lua CAS). Lets `resolve`/`expire` reject a stale + * decision that targets a different action than the one currently pending. + */ + pendingActionId?: string; + + /** + * Liveness basis for the stale-running failsafe, refreshed when a paused job + * is resumed. Without it, cleanup keys off `createdAt`, so an approval that + * sat in `requires_action` past the running window would be reaped on the + * next tick right after resuming. Falls back to `createdAt` when unset. + */ + lastActiveAt?: number; +} + +/** + * Whether a job's pending review has passed its `expiresAt`. Shared by the + * stores so an expired approval is kept out of active-job listings (the client + * stops polling; cleanup/expiry finalizes it). + */ +export function isPendingActionExpired(job: Pick): boolean { + const exp = job.pendingAction?.expiresAt; + return exp != null && exp <= Date.now(); +} + +/** + * Whether a `requires_action` job has no live, resolvable prompt β€” either the + * pendingAction is missing/malformed (e.g. dropped on deserialize) or past its + * `expiresAt`. Such a job can't be rendered or resolved, so it must be kept out + * of active listings and finalized by cleanup rather than left stuck active. + */ +export function isPendingActionStale(job: Pick): boolean { + return !job.pendingAction || isPendingActionExpired(job); +} + +/** + * Arguments for an atomic {@link IJobStore.transitionStatus} compare-and-set. + */ +export interface JobStatusTransition { + /** Only fire the transition if the job is currently in this status. */ + from: JobStatus; + /** Status to move to when the `from` guard holds. */ + to: JobStatus; + /** Fields written in the same atomic step as the status change. */ + patch?: Partial; + /** Field names removed in the same atomic step (e.g. `pendingAction`). */ + clear?: Array; + /** + * Additional guard: only fire if the job's `pendingActionId` equals this. + * Checked atomically alongside the `from` status so a stale decision can't + * resolve a job that has since paused for a different action. + */ + expectActionId?: string; } /** @@ -204,6 +270,28 @@ export interface IJobStore { /** Update job data */ updateJob(streamId: string, updates: Partial): Promise; + /** + * Atomically transition a job's status, **only if** it is currently `from`. + * Returns `true` when the transition fired, `false` when the job was missing + * or no longer in `from` (lost a race / illegal transition). + * + * `patch` fields are written and `clear` fields removed in the same atomic + * step, and the running / requires_action membership sets plus live-key TTLs + * are reconciled to match `to`. This is the race-safe primitive behind the + * approval lifecycle β€” it prevents two concurrent resumes from both driving a + * paused run (a double-drive would re-execute tools / double-bill). + * + * Distinct from {@link updateJob}, which writes status unconditionally for + * callers that don't know the prior state. Reach for `transitionStatus` + * whenever the legal prior state is known. + * + * Atomicity: fully atomic on in-memory and single-node / sentinel Redis + * (Lua). On Redis Cluster the status guard is best-effort β€” the membership + * sets live on a different hash slot from the job hash β€” matching the store's + * existing cluster posture for status writes. + */ + transitionStatus(streamId: string, args: JobStatusTransition): Promise; + /** Delete a job */ deleteJob(streamId: string): Promise; diff --git a/packages/api/src/types/stream.ts b/packages/api/src/types/stream.ts index dd125a1aab..d43a1112f1 100644 --- a/packages/api/src/types/stream.ts +++ b/packages/api/src/types/stream.ts @@ -1,5 +1,5 @@ -import type { EventEmitter } from 'events'; import type { Agents } from 'librechat-data-provider'; +import type { EventEmitter } from 'events'; import type { ServerSentEvent } from '~/types'; export interface GenerationJobMetadata { @@ -20,9 +20,11 @@ export interface GenerationJobMetadata { model?: string; /** Prompt token count for abort token spending */ promptTokens?: number; + /** Set when the job is paused for human review (status === 'requires_action') */ + pendingAction?: Agents.PendingAction; } -export type GenerationJobStatus = 'running' | 'complete' | 'error' | 'aborted'; +export type GenerationJobStatus = 'running' | 'complete' | 'error' | 'aborted' | 'requires_action'; export interface GenerationJob { streamId: string; diff --git a/packages/data-provider/src/config.ts b/packages/data-provider/src/config.ts index c0034d8a33..959267ea66 100644 --- a/packages/data-provider/src/config.ts +++ b/packages/data-provider/src/config.ts @@ -754,6 +754,51 @@ const remoteApiSchema = z.object({ auth: remoteApiAuthSchema.optional(), }); +/** + * Permission mode applied to a tool call. Mirrors `@librechat/agents`'s + * `ToolPolicyMode` 1:1. + * + * - `default`: ask the user about anything not explicitly allowed (default-on). + * - `dontAsk`: deny anything not explicitly allowed (headless / API-key flows). + * - `bypass`: auto-approve everything that isn't explicitly denied + * (the user-facing "stop asking me" toggle). + * + * Subagents inherit the parent's mode; this is enforced by the SDK and not + * overridable per-subagent. + */ +export const toolApprovalModeSchema = z.enum(['default', 'dontAsk', 'bypass']); +export type ToolApprovalMode = z.infer; + +/** + * Per-endpoint tool-approval policy. + * + * Shape mirrors `@librechat/agents`'s `ToolPolicyConfig` so the host can map it + * directly into `createToolPolicyHook(config)`. The SDK does the evaluation + * (`deny β†’ bypass β†’ allow β†’ ask β†’ dontAsk β†’ fallthrough(ask)`); this config + * just describes the surface. + * + * Conventions: + * - All list entries are matched as globs (`*`). Use `mcp:server:*` to scope + * a rule to every tool from a single MCP server. + * - `deny` always wins, including under `bypass`. + * - `enabled: false` is a LibreChat-only kill switch that disables the entire + * HITL machinery for this endpoint (no checkpointer, no hooks, no prompts). + * This is admin-level; users toggle prompting via `mode: 'bypass'` instead. + */ +export const toolApprovalPolicySchema = z + .object({ + enabled: z.boolean().optional(), + mode: toolApprovalModeSchema.optional(), + allow: z.array(z.string()).optional(), + deny: z.array(z.string()).optional(), + ask: z.array(z.string()).optional(), + /** Optional reason template surfaced in the prompt; `{tool}` is interpolated. */ + reason: z.string().optional(), + }) + .optional(); + +export type TToolApprovalPolicy = z.infer; + export const agentsEndpointSchema = baseEndpointSchema .omit({ baseURL: true }) .merge( @@ -776,6 +821,8 @@ export const agentsEndpointSchema = baseEndpointSchema }) .optional(), remoteApi: remoteApiSchema.optional(), + /** Human-in-the-loop tool approval policy. Off by default. */ + toolApproval: toolApprovalPolicySchema, }), ) .default({ diff --git a/packages/data-provider/src/types/agents.ts b/packages/data-provider/src/types/agents.ts index 0dc96fb6cc..6318dd651f 100644 --- a/packages/data-provider/src/types/agents.ts +++ b/packages/data-provider/src/types/agents.ts @@ -84,6 +84,16 @@ export namespace Agents { auth?: string; /** Expiration time */ expires_at?: number; + /** + * When set, this tool call is paused for human review. + * The presence of this field signals the UI to render approval controls + * instead of the in-flight tool execution state. + */ + approval?: { + actionId: string; + allowed_decisions: ToolApprovalDecisionType[]; + description?: string; + }; }; export type ToolEndEvent = { @@ -236,6 +246,12 @@ export namespace Agents { collectedUsage?: TTokenUsageEvent[]; /** Latest context window snapshot; restores the usage gauge on resume */ contextUsage?: TContextUsageEvent; + /** + * Live pending approval when the run is paused for human review. Carried in + * the resume contract (not just /chat/status) so a reloading or + * cross-replica client can rebuild and render the prompt from `resumeState`. + */ + pendingAction?: PendingAction; } /** * Represents a run step delta i.e. any changed fields on a run step during @@ -267,8 +283,160 @@ export namespace Agents { tool_calls?: ToolCallChunk[]; auth?: string; expires_at?: number; + /** Approval metadata, set when a tool call is paused for human review. */ + approval?: { + actionId: string; + allowed_decisions: ToolApprovalDecisionType[]; + description?: string; + }; }; export type AgentToolCall = FunctionToolCall | ToolCall; + + /** + * Human-in-the-loop interrupt categories. The discriminator on + * {@link HumanInterruptPayload}. + * + * - `tool_approval`: agent paused before executing one or more tools; user + * approves / rejects / edits each call. + * - `ask_user_question`: agent invoked the `AskUserQuestion` tool to gather + * clarification; user replies with free-form text (or selects an option). + * + * `tool_approval` is a permission gate; `ask_user_question` is a clarification + * channel β€” they share the {@link PendingAction} envelope but have different + * UI affordances and resume payloads. + */ + export type HumanInterruptType = 'tool_approval' | 'ask_user_question'; + + /** String enum of decision kinds the user can make on a paused tool call. */ + export type ToolApprovalDecisionType = 'approve' | 'reject' | 'edit' | 'respond'; + + /** + * One pending tool execution awaiting user review. + * Field naming mirrors LangChain HumanInterrupt's `ActionRequest`. + */ + export interface ToolApprovalRequest { + /** Tool name as registered with the agent */ + name: string; + /** Sanitized arguments (no auth tokens / file blobs). May be string or parsed object. */ + arguments: string | Record; + /** Provider tool_call_id linking this request to the model's tool_use block */ + tool_call_id: string; + /** Optional human-readable description shown alongside the prompt */ + description?: string; + } + + /** + * Per-call review configuration: which decisions the user is allowed to make. + * + * `tool_call_id` (NOT `action_name`) is the join key against + * {@link ToolApprovalRequest.tool_call_id}. By-position mapping breaks the + * moment a single batch contains the same tool called twice β€” e.g. a model + * fanning out two `mcp:server:search` calls in parallel β€” so always join + * by `tool_call_id`. `action_name` is retained for display only. + */ + export interface ToolReviewConfig { + action_name: string; + tool_call_id: string; + allowed_decisions: ToolApprovalDecisionType[]; + } + + /** Interrupt payload for a tool-approval pause. */ + export interface ToolApprovalInterruptPayload { + type: 'tool_approval'; + action_requests: ToolApprovalRequest[]; + review_configs: ToolReviewConfig[]; + } + + /** A selectable answer for an ask-user-question prompt. */ + export interface AskUserQuestionOption { + label: string; + value: string; + } + + /** The question itself: free-form prompt with optional curated answers. */ + export interface AskUserQuestionRequest { + question: string; + /** Optional descriptive context for the prompt; mirrors the SDK field. */ + description?: string; + options?: AskUserQuestionOption[]; + } + + /** Interrupt payload for an ask-user-question pause. */ + export interface AskUserQuestionInterruptPayload { + type: 'ask_user_question'; + question: AskUserQuestionRequest; + } + + /** + * Discriminated by `type`. Mirrors `@librechat/agents`'s `HumanInterruptPayload` + * so the SDK's `Run.getInterrupt()` output can be embedded directly. + */ + export type HumanInterruptPayload = + | ToolApprovalInterruptPayload + | AskUserQuestionInterruptPayload; + + /** + * Server-side record of a job that is waiting for user input. + * Persisted with the job; consumed by approval routes and the status endpoint. + */ + export interface PendingAction { + /** Stable identifier used in approval URLs */ + actionId: string; + streamId: string; + conversationId?: string; + /** Stable per-turn identifier (LangGraph checkpoint_ns) when available */ + runId?: string; + responseMessageId?: string; + payload: HumanInterruptPayload; + createdAt: number; + /** Optional expiry; clients should treat past `expiresAt` as stale */ + expiresAt?: number; + /** + * SDK interrupt id (`RunInterruptResult.interruptId`). Persisted so a + * cross-process resume can correlate the decision with the LangGraph + * interrupt after the original `Run` object is gone. + */ + interruptId?: string; + /** + * LangGraph `thread_id` the run was bound to (`RunInterruptResult.threadId`). + * Required, with the checkpointer, to rebuild `Command({ resume })` on a + * worker that didn't originate the run. + */ + threadId?: string; + } + + /** + * Scope of a tool-approval decision β€” drives the "remember this" persistence + * envelope. Storage of session/always decisions is a Slice B+ concern; the + * field is on the wire today so route signatures don't break later. + */ + export type DecisionScope = 'once' | 'session' | 'always'; + + /** + * Per-tool decision returned from the approval UI. + * Wire format. The host adapts each entry to the SDK's discriminated + * `ToolApprovalDecision` (e.g. `{ type: 'edit', updatedInput }`) at the resume route. + * + * Constraints: + * - `editedArguments` is required when `decision === 'edit'`. + * - `responseText` is required when `decision === 'respond'`. + * - `reason` is optional metadata; useful for reject/edit audit trails. + * - `scope` defaults to `'once'`. + */ + export interface ToolApprovalResolution { + tool_call_id: string; + decision: ToolApprovalDecisionType; + editedArguments?: Record; + responseText?: string; + reason?: string; + scope?: DecisionScope; + } + + /** Wire format for an ask-user-question response. */ + export interface AskUserQuestionResolution { + answer: string; + } + export interface ExtendedMessageContent { type?: string; text?: string;