mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-10-24 11:12:49 +00:00
🪝 feat: HITL Tool Approval Scaffolding (Slice A) (#12938)
* 🪝 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 onabf4b86291, 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 on780833d908, 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 on4324a4e776, 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 onc8abd826e1(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 one7d9cf21b6(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.
This commit is contained in:
Danny Avila
GitHub
parent
61016e328a
commit
771b93bf10
19 changed files with 1878 additions and 84 deletions
|
|
@ -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;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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([]);
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
1
packages/api/src/agents/hitl/index.ts
Normal file
1
packages/api/src/agents/hitl/index.ts
Normal file
|
|
@ -0,0 +1 @@
|
|||
export * from './policy';
|
||||
229
packages/api/src/agents/hitl/policy.spec.ts
Normal file
229
packages/api/src/agents/hitl/policy.spec.ts
Normal file
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
150
packages/api/src/agents/hitl/policy.ts
Normal file
150
packages/api/src/agents/hitl/policy.ts
Normal file
|
|
@ -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<string, unknown>;
|
||||
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<string, Agents.ToolApprovalDecisionType[]>,
|
||||
): 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,
|
||||
};
|
||||
}
|
||||
44
packages/api/src/agents/hitl/typeContract.spec.ts
Normal file
44
packages/api/src/agents/hitl/typeContract.spec.ts
Normal file
|
|
@ -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');
|
||||
});
|
||||
});
|
||||
|
|
@ -27,3 +27,4 @@ export * from './tools';
|
|||
export * from './validation';
|
||||
export * from './added';
|
||||
export * from './load';
|
||||
export * from './hitl';
|
||||
|
|
|
|||
129
packages/api/src/stream/ApprovalLifecycle.ts
Normal file
129
packages/api/src/stream/ApprovalLifecycle.ts
Normal file
|
|
@ -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<boolean> {
|
||||
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<Agents.PendingAction | null> {
|
||||
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<boolean> {
|
||||
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<boolean> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
|
@ -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<Record<t.GenerationJobStatus, number>> {
|
||||
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 };
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -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<Agents.RunStep>[] = [
|
||||
{ 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) {
|
||||
|
|
|
|||
251
packages/api/src/stream/__tests__/pendingAction.spec.ts
Normal file
251
packages/api/src/stream/__tests__/pendingAction.spec.ts
Normal file
|
|
@ -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<Agents.PendingAction> = {}) {
|
||||
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();
|
||||
});
|
||||
});
|
||||
|
|
@ -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<boolean> {
|
||||
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<void> {
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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<keyof SerializableJobData> = [
|
||||
'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<SerializableJobData>): Promise<void> {
|
||||
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<void> {
|
||||
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<boolean> {
|
||||
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<boolean> {
|
||||
const updated = await this.redis.eval(
|
||||
'if redis.call("EXISTS", KEYS[1]) == 1 then redis.call("HSET", KEYS[1], unpack(ARGV)) return 1 else return 0 end',
|
||||
1,
|
||||
key,
|
||||
...fields,
|
||||
);
|
||||
|
||||
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<void> {
|
||||
|
|
@ -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<number> {
|
||||
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<number> {
|
||||
// This is approximate - counts jobs in running set + scans for job keys
|
||||
// For exact count, would need to scan all job:* keys
|
||||
const runningCount = await this.redis.scard(KEYS.runningJobs);
|
||||
return runningCount;
|
||||
const [runningCount, requiresActionCount] = await Promise.all([
|
||||
this.redis.scard(KEYS.runningJobs),
|
||||
this.countJobsInStatusSet(KEYS.requiresActionJobs, 'requires_action'),
|
||||
]);
|
||||
return runningCount + requiresActionCount;
|
||||
}
|
||||
|
||||
async getJobCountByStatus(status: JobStatus): Promise<number> {
|
||||
|
|
@ -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<number> {
|
||||
const streamIds = await this.redis.smembers(setKey);
|
||||
if (streamIds.length === 0) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
let count = 0;
|
||||
const staleIds: string[] = [];
|
||||
for (const streamId of streamIds) {
|
||||
const job = await this.getJob(streamId);
|
||||
if (job?.status === status) {
|
||||
count++;
|
||||
} else {
|
||||
staleIds.push(streamId);
|
||||
}
|
||||
}
|
||||
|
||||
if (staleIds.length > 0) {
|
||||
await this.redis.srem(setKey, ...staleIds);
|
||||
}
|
||||
|
||||
return count;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get active job IDs for a user.
|
||||
* Returns conversation IDs of running jobs belonging to the user.
|
||||
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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';
|
||||
|
|
|
|||
|
|
@ -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<SerializableJobData, 'pendingAction'>): 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<SerializableJobData, 'pendingAction'>): 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<SerializableJobData>;
|
||||
/** Field names removed in the same atomic step (e.g. `pendingAction`). */
|
||||
clear?: Array<keyof SerializableJobData & string>;
|
||||
/**
|
||||
* 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<SerializableJobData>): Promise<void>;
|
||||
|
||||
/**
|
||||
* 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<boolean>;
|
||||
|
||||
/** Delete a job */
|
||||
deleteJob(streamId: string): Promise<void>;
|
||||
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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<typeof toolApprovalModeSchema>;
|
||||
|
||||
/**
|
||||
* 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<typeof toolApprovalPolicySchema>;
|
||||
|
||||
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({
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>;
|
||||
/** 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<string, unknown>;
|
||||
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;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue