LibreChat/api/server/controllers/agents/client.js
Danny Avila 236ee6c1ab
🧭 fix: Re-Anchor Parent Activity Phase Bounds (#14741)
* test: cover parent activity phase finalization

* test(e2e): stabilize parent phase coverage

* fix(agents): reanchor parent activity phase bounds

* fix(agents): preserve delayed tools in activity phases

* test(agents): keep phase slice bounds typed

* fix(agents): preserve sparse activity phase bounds

* test(e2e): read structured phase replies
2026-08-11 10:16:57 -04:00

3697 lines
147 KiB
JavaScript

require('events').EventEmitter.defaultMaxListeners = 100;
const { logger } = require('@librechat/data-schemas');
const { getBufferString, HumanMessage } = require('@librechat/agents/langchain/messages');
const {
createRun,
isEnabled,
checkAccess,
buildToolSet,
logToolError,
sanitizeTitle,
payloadParser,
createSafeUser,
initializeAgent,
resolveConfigHeaders,
countTokens,
getBalanceConfig,
omitTitleOptions,
getProviderConfig,
memoryInstructions,
createTokenCounter,
applyContextToAgent,
isMemoryAgentEnabled,
recordCollectedUsage,
sendEvent,
computeUsageCostUSD,
aggregateEmittedUsage,
resolveAgentTokenConfig,
buildPersistedContextUsage,
computeSummaryUsedTokens,
priorRunOutputTokens,
createSubagentUsageSink,
anyAgentReplaysReasoningContent,
GenerationJobManager,
getTransactionsConfig,
resolveRecursionLimit,
buildPendingAction,
toClientPendingAction,
computeAgentRequestFingerprint,
getRunDiscoveredTools,
captureResumeModelParameters,
pickResumeContext,
getApprovalTtlMs,
isHITLEnabled,
captureAgentCheckpointGeneration,
deleteAgentCheckpoint,
LIBRECHAT_CHECKPOINT_NAMESPACE_KEY,
agentRequestsAskUserQuestion,
attachAskUserQuestionArgs,
hydrateResumeRunSteps,
createContentIndexOffsetHandlers,
createSteerIndexOffsetHandlers,
createSteerDrainHook,
createSteerPreemptBoundaryHook,
createSteerPreemptPoll,
isSteeringSupported,
isSteerPreemptSupported,
buildSteerMedia,
stampSteerPartMedia,
createActivityLabelWiring,
createActivityPhaseWiring,
createAssistantPhaseStampingHandlers,
resolveActivityConfig,
resolveActivityPhaseConfig,
getCustomEndpointConfig,
mapCollectedMetadataToUsage,
resolveActivityLabelModel,
resolveActivityPhaseLabelModel,
traceIdForMessage,
settlePendingLabelFills,
stripActivityLabelParts,
getRequestMemories,
getMemoryAgentId,
createMemoryProcessor,
agentHasInlineMemoryTools,
loadAgent: loadAgentFn,
createMultiAgentMapper,
filterMalformedContentParts,
countFormattedMessageTokens,
prependFileContext,
prependQuotes,
hydrateMissingIndexTokenCounts,
injectSkillPrimes,
collectFreshSkillPrimeNames,
isSkillPrimeMessage,
collectFileIds,
processTextWithTokenLimit,
buildAgentScopedContext,
buildSkillPrimeContentParts,
buildInitialToolSessions,
hasUrlContextTool,
hasYouTubeVideoParts,
appendYouTubeVideoParts,
resolveGoogleVideoError,
resolveYouTubeInjectionConfig,
decrementPendingRequest,
maybePrewarmCodeSandbox,
} = require('@librechat/api');
const {
Run,
Callback,
Providers,
TitleMethod,
formatMessage,
formatAgentMessages,
createMetadataAggregator,
} = require('@librechat/agents');
const {
Constants,
SteerEvents,
ActivityLabelEvents,
UsageEvents,
Permissions,
VisionModes,
ContentTypes,
ApprovalEvents,
EModelEndpoint,
PermissionTypes,
AgentCapabilities,
isAgentsEndpoint,
isEphemeralAgentId,
removeNullishValues,
DEFAULT_MEMORY_MAX_INPUT_TOKENS,
} = require('librechat-data-provider');
const { filterFilesByAgentAccess } = require('~/server/services/Files/permissions');
const { encodeAndFormat } = require('~/server/services/Files/images/encode');
const { createContextHandlers } = require('~/app/clients/prompts');
const { resolveConfigServers, getAccessibleMcpServerNames } = require('~/server/services/MCP');
const { getMCPServerTools } = require('~/server/services/Config');
const BaseClient = require('~/app/clients/BaseClient');
const { getMCPManager } = require('~/config');
const db = require('~/models');
const loadAgent = (params) => loadAgentFn(params, { getAgent: db.getAgent, getMCPServerTools });
const MEMORY_INPUT_CHARS_PER_TOKEN = 8;
class AgentClient extends BaseClient {
constructor(options = {}) {
super(null, options);
/** The current client class
* @type {string} */
this.clientName = EModelEndpoint.agents;
/** @deprecated @type {true} - Is a Chat Completion Request */
this.isChatCompletion = true;
/** @type {number | undefined} */
this.jobCreatedAt = options.jobCreatedAt;
/** Generation-scoped LangGraph checkpoint namespace. Legacy paused jobs
* intentionally use the historical empty namespace. @type {string} */
this.checkpointNamespace = options.checkpointNamespace ?? '';
/** @type {AgentRun} */
this.run;
/** Resolves with the agent run once `chatCompletion` initializes it (or
* `null` if initialization fails), letting immediate-mode title generation
* await the run instead of throwing when fired before the run exists.
* @type {Promise<AgentRun | null> | null} */
this._runReady = null;
/** @type {((run: AgentRun | null) => void) | null} */
this._resolveRun = null;
const {
agentConfigs,
contentParts,
stepMap,
collectedUsage,
collectedThoughtSignatures,
artifactPromises,
maxContextTokens,
subagentAggregatorsByToolCallId,
contextUsageSink,
usageEmitSink,
toolInputValidationErrors,
...clientOptions
} = options;
this.agentConfigs = agentConfigs;
this.maxContextTokens = maxContextTokens;
/** Latest visible context snapshot for this response, captured live by the
* ON_CONTEXT_USAGE handler; persisted on `metadata.contextUsage`.
* @type {{ latest: import('librechat-data-provider').TContextUsageEvent | null } | undefined} */
this.contextUsageSink = contextUsageSink;
/** Every emitted `on_token_usage` payload for this response (primary,
* summarization, sequential, and subagent); aggregated into the rollup
* persisted on `metadata.usage`.
* @type {Array<import('librechat-data-provider').TTokenUsageEvent> | undefined} */
this.usageEmitSink = usageEmitSink;
/** Schema-validation exceptions keyed by tool-call ID. The completion
* handler consumes these to distinguish execution failures from tool
* output that merely contains similar text.
* @type {Map<string, import('@librechat/api').ToolInputValidationError> | undefined} */
this.toolInputValidationErrors = toolInputValidationErrors;
/** @type {MessageContentComplex[]} */
this.contentParts = contentParts;
/** Original run-step identity used by the content aggregator to attach
* completion events to their rendered content indices.
* @type {Map<string, import('@librechat/agents').RunStep | undefined> | undefined} */
this.stepMap = stepMap;
/** @type {Array<UsageMetadata>} */
this.collectedUsage = collectedUsage;
/** Vertex Gemini 3 thought signatures captured during the run, keyed by
* `tool_call_id`. Persisted on `responseMessage.metadata.thoughtSignatures`
* and restored as `additional_kwargs.signatures` on subsequent turns to
* keep tool round-trips valid across DB reconstruction.
* @type {Record<string, string> | undefined} */
this.collectedThoughtSignatures = collectedThoughtSignatures;
/** @type {ArtifactPromises} */
this.artifactPromises = artifactPromises;
/** Per-request map of `createContentAggregator` instances keyed by
* the parent's `tool_call_id`. `ON_SUBAGENT_UPDATE` events stream
* into each aggregator as they arrive; `finalizeSubagentContent`
* harvests `contentParts` onto the matching `subagent` tool_call
* so the child's full activity survives a page refresh. */
this.subagentAggregatorsByToolCallId = subagentAggregatorsByToolCallId ?? new Map();
/** In-flight `on_token_usage` emits from subagent child runs. The sink
* fires the emitter without awaiting, so chatCompletion's finally flushes
* these before returning — otherwise job cleanup can race the persist.
* @type {Promise<void>[]} */
this.pendingSubagentEmits = [];
/** @type {AgentClientOptions} */
this.options = Object.assign({ endpoint: options.endpoint }, clientOptions);
/** @type {string} */
this.model = this.options.agent.model_parameters.model;
/** The key for the usage object's input tokens
* @type {string} */
this.inputTokensKey = 'input_tokens';
/** The key for the usage object's output tokens
* @type {string} */
this.outputTokensKey = 'output_tokens';
/** @type {UsageMetadata} */
this.usage;
/** @type {Record<string, number>} */
this.indexTokenCountMap = {};
/** @type {Array<Record<string, unknown>> | null} */
this.memoryPayload = null;
/** Mutable content-index shift shared with the steer offset handlers.
* Incremented each time a steer part is spliced into `contentParts`, so
* SDK-emitted indices that arrive after an injection land past it.
* @type {import('@librechat/api').SteerOffsetState} */
this.steerOffsetState = { offset: 0 };
/** @type {(messages: BaseMessage[]) => Promise<void>} */
this.processMemory;
}
/**
* Returns the aggregated content parts for the current run.
* @returns {MessageContentComplex[]} */
getContentParts() {
return this.contentParts;
}
/**
* Harvest the `contentParts` from each per-subagent `createContentAggregator`
* instance and attach them onto the matching parent `subagent` tool_call
* as `subagent_content`. Runs once per message save (from
* `sendCompletion`'s `finally`) so the child's full reasoning / tool
* calls / final text survive a page refresh — the client-side Recoil
* atom is session-only. Aggregators keyed by a tool_call_id that never
* appeared in `contentParts` are discarded (no home to attach to).
*/
finalizeSubagentContent() {
const buffer = this.subagentAggregatorsByToolCallId;
if (!buffer || buffer.size === 0 || !Array.isArray(this.contentParts)) {
return;
}
for (const part of this.contentParts) {
if (part?.type !== ContentTypes.TOOL_CALL) continue;
const toolCall = part[ContentTypes.TOOL_CALL];
if (!toolCall || toolCall.name !== Constants.SUBAGENT || !toolCall.id) continue;
const aggregator = buffer.get(toolCall.id);
if (!aggregator) continue;
try {
/** `createContentAggregator` returns a sparse array (undefined
* slots for indices that never received content). Strip those
* so the persisted shape is a clean `TMessageContentParts[]`. */
const parts = Array.isArray(aggregator.contentParts)
? aggregator.contentParts.filter((p) => p != null)
: [];
if (parts.length > 0) {
toolCall.subagent_content = parts;
}
} catch (err) {
logger.warn(
`[AgentClient] Failed to attach subagent content for tool_call ${toolCall.id}: ${err?.message ?? err}`,
);
}
}
buffer.clear();
}
/**
* Apply one drained steer to host state: append the steer content part at
* the live content index, bump the shared index offset so subsequent SDK
* step indices land past it, and emit `on_steer_applied` so the live client
* replaces its pending chip with the inline part (the emitted chunk also
* reaches the Redis chunk log for reconnect reconstruction).
*
* Runs BEFORE the drain hook's media encode so an abort during the encode
* cannot lose the steer. File refs persist from the queue item (sanitized at
* enqueue); replay/token accounting re-fetch owner-scoped and re-encode per
* turn (stampSteerPartMedia), so unauthorized ids drop out there.
*
* @param {string} streamId
* @param {import('@librechat/api').SteerQueueItem} item
*/
async applySteerPart(streamId, item) {
const index = this.contentParts.length;
const part = {
type: ContentTypes.STEER,
[ContentTypes.STEER]: item.text,
steerId: item.steerId,
...(item.clientSteerId && { clientSteerId: item.clientSteerId }),
createdAt: item.createdAt,
...(item.files?.length && { files: item.files }),
};
this.contentParts.push(part);
this.steerOffsetState.offset += 1;
// durable: the chunk-log XADD is this event's recovery record — it must
// commit before the publish or a cross-replica reconnect that missed the
// pub/sub delivery reconstructs content without the steer part.
try {
await GenerationJobManager.emitChunk(
streamId,
{
event: SteerEvents.ON_STEER_APPLIED,
data: {
steerId: item.steerId,
...(item.clientSteerId && { clientSteerId: item.clientSteerId }),
index,
part,
responseMessageId: this.responseMessageId,
conversationId: this.conversationId,
},
},
{
durable: true,
expectedCreatedAt: this.jobCreatedAt,
deliveredSteer: item,
},
);
/** Only a COMMITTED steer is a hard semantic boundary. If its durable
* append failed, the drain restores the queue item and the current
* phase evidence must remain intact for the eventual retry. */
this.activityPhaseWiring?.drop?.();
} catch (error) {
/** The part and its receipt commit as one durable unit. Roll the local
* projection back when that commit fails so the drain can restore the
* claimed item instead of injecting an instruction absent from replay. */
if (this.contentParts[index] === part) {
this.contentParts.splice(index, 1);
this.steerOffsetState.offset -= 1;
}
throw error;
}
}
/**
* The `steering` fragment for `createRun`: the run-scoped PostToolBatch
* drain hook — plus, when the SDK can seal mid-stream, the PreemptBoundary
* twin and the preempt poll built from the SAME drain closures, so both
* boundaries inject byte-identical shapes. `undefined` when there is no
* resumable job surface or the installed SDK cannot inject hook messages
* (draining would drop them).
*
* @param {string | undefined} streamId
*/
buildSteerWiring(streamId) {
if (!streamId || !isSteeringSupported()) {
return undefined;
}
const drainOptions = {
streamId,
jobCreatedAt: this.jobCreatedAt,
applySteer: (item) => this.applySteerPart(streamId, item),
buildMedia: (item) =>
buildSteerMedia({
client: this,
user: this.options.req?.user,
item,
getFiles: db.getFiles,
}),
};
return {
hook: createSteerDrainHook(drainOptions),
...(isSteerPreemptSupported() && {
preemptHook: createSteerPreemptBoundaryHook(drainOptions),
preemption: createSteerPreemptPoll(streamId),
}),
};
}
setOptions(_options) {}
/**
* Resolve provider + client options for the
* tool-batch summary model. Same resolution path as titleConvo minus the
* title-specific branches. Model precedence: the endpoint's
* `activityModel` > its `titleModel` > the agent's own model, on the
* endpoint named by `activityEndpoint` (default: the agent's).
*/
async resolveActivityLabelLLM() {
/** Memoized per response: resolution reads provider config and can hit the
* database for user keys, and nothing it depends on changes between
* batches of the same run — so re-resolving on every batch (twice, with
* usage accounting) is repeated credential work for an identical result.
* The promise is cached rather than the value so concurrent batches share
* one in-flight resolution. */
this.activityLabelLLMPromise =
this.activityLabelLLMPromise ??
resolveActivityLabelModel({
req: this.options.req,
agent: this.options.agent,
/** Same public-endpoint-first field resolution as the wiring gate. */
publicEndpoint: this.options.endpoint,
ids: {
messageId: this.responseMessageId,
conversationId: this.conversationId,
parentMessageId: this.parentMessageId,
},
db: { getUserKey: db.getUserKey, getUserKeyValues: db.getUserKeyValues },
}).catch((error) => {
/** Never cache a rejection: a transient credential read failure would
* otherwise disable labels for the rest of the response. */
this.activityLabelLLMPromise = null;
throw error;
});
return this.activityLabelLLMPromise;
}
/** Phase resolution is independently configurable and memoized per response. */
async resolveActivityPhaseLabelLLM() {
this.activityPhaseLabelLLMPromise =
this.activityPhaseLabelLLMPromise ??
resolveActivityPhaseLabelModel({
req: this.options.req,
agent: this.options.agent,
publicEndpoint: this.options.endpoint,
ids: {
messageId: this.responseMessageId,
conversationId: this.conversationId,
parentMessageId: this.parentMessageId,
},
db: { getUserKey: db.getUserKey, getUserKeyValues: db.getUserKeyValues },
}).catch((error) => {
this.activityPhaseLabelLLMPromise = null;
throw error;
});
return this.activityPhaseLabelLLMPromise;
}
/**
* Bills the label call and folds its usage into the response rollup with
* an `activity-label` tag (subagent precedent) so `metadata.usage` and the
* live cost gauge reflect it. Tagged, so it is not a PRIMARY usage event
* and cannot disturb the context-snapshot pairing in buildResponseMetadata.
*/
async recordActivityLabelUsage(
collectedMetadata,
model,
endpointTokenConfig,
sameEndpoint,
/** Optional suppression gate, defaulting open. The hook-driven paths
* deliberately pass nothing: they invoke accounting ONLY for a
* COMMITTED fill, and a committed (visible) label must bill even when
* its scope closed during the durable emit — the commit flag, not the
* scope, is the billing authority. */
scopeOpen = () => true,
/** The LABEL endpoint's provider — cost math needs it to know whether
* cache tokens are folded into `input_tokens` (additive providers like
* Bedrock keep them separate). */
provider = undefined,
/** Lazy `() => ({ promptText, completionText })` fallback. When the
* provider omits usage metadata entirely, labels bill by ESTIMATE —
* the title convention — from locally counted text rather than going
* unbilled. Invoked only when no entry carries a real token count. */
estimate = undefined,
/** Separate telemetry bucket for parent phase summaries. */
usageType = 'activity-label',
) {
const appConfig = this.options.req?.config;
/** Provider ON EVERY ENTRY, not just the streamed event: `splitUsage`
* keys additive-vs-subset cache math on `usage.provider`, and an
* unknown provider takes the additive branch — for Anthropic/OpenAI
* (cache already inside `input_tokens`) that re-adds cache_read and
* cache_creation on top, double-charging the balance while the
* streamed cost (which carries the provider) disagrees. */
let collectedUsage = mapCollectedMetadataToUsage(collectedMetadata).map((usage) =>
provider != null ? { ...usage, provider } : usage,
);
const hasRealUsage = collectedUsage.some(
(usage) => usage.input_tokens != null || usage.output_tokens != null,
);
if (!hasRealUsage && typeof estimate === 'function') {
try {
const { promptText = '', completionText = '' } = estimate() ?? {};
const [input_tokens, output_tokens] = await Promise.all([
countTokens(promptText),
countTokens(completionText),
]);
collectedUsage = [
provider != null
? { input_tokens, output_tokens, provider }
: { input_tokens, output_tokens },
];
} catch (err) {
logger.warn(
`[AgentClient] Failed to estimate activity-label usage: ${err?.message ?? err}`,
);
}
}
if (
collectedUsage.length === 0 ||
!collectedUsage.some((usage) => usage.input_tokens != null || usage.output_tokens != null)
) {
return;
}
if (!scopeOpen()) {
return;
}
const streamId = this.options.req?._resumableStreamId || null;
const includeCost = this.options.req?.config?.interfaceConfig?.contextCost === true;
/** Cross-endpoint labels (`activityEndpoint`) price with THEIR endpoint's
* rates. `undefined` is a MEANINGFUL result for a built-in label endpoint
* (built-ins price from the shared table, not a per-endpoint map), so it
* must not fall through to the agent's custom rates — a custom primary
* pointing `activityEndpoint` at a built-in would bill the label at its
* own rates. Only inherit when the label actually runs on the agent's
* endpoint. */
const labelTokenConfig = sameEndpoint
? (endpointTokenConfig ?? this.options.endpointTokenConfig)
: endpointTokenConfig;
for (const usage of collectedUsage) {
/** `seq` is normally a position in `collectedUsage` (each emitter
* pushes, then emits with the new length). Label usage is billed
* separately and never appended there, so it has no position: any
* positive value eventually collides with a real one, and the client
* dedupes on `runId:seq`. Labels therefore occupy a NEGATIVE seq
* namespace that positional sequences can never reach. The key is
* only used for Set membership, so the sign is inert. */
this.activityLabelUsageSeq = (this.activityLabelUsageSeq ?? 0) + 1;
const data = {
input_tokens: usage.input_tokens,
output_tokens: usage.output_tokens,
/** Cache tokens ride along (subagent-event shape) so display and
* aggregation price cached label calls at cache rates. */
...(usage.input_token_details != null && {
input_token_details: usage.input_token_details,
}),
...(provider != null && { provider }),
model,
usage_type: usageType,
/**
* Scoped to the GENERATION, not just the response. Editing one
* assistant response reuses its `responseMessageId` while each fresh
* server generation restarts `activityLabelUsageSeq`, so a second
* edit re-emitted `<responseId>:-1` and the client — which dedupes on
* exactly `runId:seq` — discarded the newer usage even though its
* balance transaction was still written. `jobCreatedAt` is the run's
* own epoch: stable across reconnects and HITL resumes of one
* generation, distinct between generations.
*/
runId:
this.jobCreatedAt != null
? `${this.responseMessageId}:${this.jobCreatedAt}`
: this.responseMessageId,
seq: -this.activityLabelUsageSeq,
/** Cost coverage is all-or-nothing in `aggregateEmittedUsage`: an
* event without `cost` suppresses the whole response's cost when
* `interface.contextCost` is on. */
cost: includeCost
? computeUsageCostUSD(
{ ...usage, model, provider },
{ getMultiplier: db.getMultiplier, getCacheMultiplier: db.getCacheMultiplier },
labelTokenConfig,
)
: undefined,
};
/** Fold into the response rollup synchronously, then stream it like
* primary/subagent usage so the live session gauge stays honest.
* Retained and flushed with the subagent emits so job cleanup cannot
* race the persist. */
this.usageEmitSink?.push(data);
if (streamId) {
const emit = GenerationJobManager.emitChunk(
streamId,
{
event: UsageEvents.ON_TOKEN_USAGE,
data,
},
/** Same epoch scoping as the label event: this usage is recorded
* from a detached generation and must not bill against whichever
* generation replaced it. */
{ expectedCreatedAt: this.jobCreatedAt },
).catch((err) => {
logger.warn(`[AgentClient] Failed to emit ${usageType} usage: ${err?.message ?? err}`);
});
this.pendingSubagentEmits.push(emit);
}
}
await this.recordCollectedUsage({
collectedUsage,
context: usageType,
model,
endpointTokenConfig: labelTokenConfig,
/** The label ran elsewhere, so its config governs even when undefined. */
crossEndpoint: sameEndpoint === false,
balance: getBalanceConfig(appConfig),
transactions: getTransactionsConfig(appConfig),
messageId: this.responseMessageId,
/** Billed, but NOT the response's stream usage — see the parameter. */
updateStreamUsage: false,
}).catch((err) => {
logger.error(
'[api/server/controllers/agents/client.js #recordActivityLabelUsage] Error recording usage',
err,
);
});
}
/**
* Bridges label generation to the SDK's `run.generateActivityLabel()` so
* the fast-model call is Langfuse-traced under the conversation's session
* (thread_id) with its own tags — never as an orphan trace. Returns null
* when the label could not be generated.
*/
async generateActivityLabelViaRun({
entries,
context,
previousLabels,
traceSeed,
signal,
charLimit,
prompt,
executingAgentId,
deferUsage,
}) {
/** Version gating happens at wiring time via the `sdkCapable` prototype
* probe, so this only catches a run that is missing or not yet built.
* Resolve `undefined` (not `null`) so the hook reads it as "this path
* cannot serve the request" and falls back to the direct model call;
* `null` would mean "ran, produced no label" and would leave the slot
* permanently empty. */
if (typeof this.run?.generateActivityLabel !== 'function') {
return undefined;
}
const { provider, clientOptions, endpointTokenConfig, sameEndpoint } =
await this.resolveActivityLabelLLM();
const { handleLLMEnd, collected: collectedMetadata } = createMetadataAggregator();
/**
* NO scope gate here: the hook invokes this ONLY for a COMMITTED fill,
* and the commit flag is the single billing authority. A scope that
* closes while the fill's durable emit is in flight does not un-commit
* the label — it is persisted and visible — so gating on the scope here
* turned that race into a completed provider call escaping both the
* label charge and the primary abort accounting. The reverse direction
* (billed but never shown) is enforced by the commit gate itself: a
* dropped fill never reaches this callback.
*/
/**
* The PROMPT THE SDK ACTUALLY SENT, captured at chain start. The hook's
* estimate thunk carries this module's locally built prompt — same
* entries and instruction but different framing — so estimated billing
* on this path would count a prompt that was never sent. When capture
* succeeded, it replaces the thunk's promptText.
*/
let sdkPromptText;
const capturePrompt = {
handleLLMStart: (_llm, prompts) => {
sdkPromptText = Array.isArray(prompts) ? prompts.join('\n') : undefined;
},
handleChatModelStart: (_llm, messages) => {
try {
sdkPromptText = (messages ?? [])
.flat()
.map((message) =>
typeof message?.content === 'string'
? message.content
: JSON.stringify(message?.content ?? ''),
)
.join('\n');
} catch {
/** Estimation falls back to the local approximation. */
}
},
};
const recordUsage = async (estimate) => {
const refined =
typeof estimate === 'function'
? () => {
const base = estimate() ?? {};
return sdkPromptText != null && sdkPromptText.length > 0
? { ...base, promptText: sdkPromptText }
: base;
}
: estimate;
await this.recordActivityLabelUsage(
collectedMetadata,
clientOptions.model,
endpointTokenConfig,
sameEndpoint,
undefined,
provider,
refined,
);
};
/**
* Accounting is DEFERRED to the hook, which runs it only after the slot
* commit settles. Awaiting it here (pre-fill) let the settlement window
* expire during the balance write: the charge landed, then the fill was
* dropped as out-of-scope — billed, never shown. Registered before the
* call so a mid-call throw still bills whatever metadata the provider
* returned, exactly like the old `finally` did.
*/
let usageDeferred = false;
if (typeof deferUsage === 'function') {
usageDeferred = true;
deferUsage(recordUsage);
}
try {
const { label } = await this.run.generateActivityLabel({
provider,
clientOptions,
entries: entries.map(({ toolName, toolInput, toolOutput, error, status }) => ({
toolName,
toolInput,
toolOutput,
error,
status,
})),
thinkingExcerpts: context.thinkingExcerpts,
lastAssistantText: context.lastAssistantText,
lastAssistantPhase: context.lastAssistantPhase,
...(previousLabels != null && { previousLabels }),
traceSeed,
charLimit,
/** Selects the EXECUTING agent's Langfuse metadata and, more
* importantly, its tool-output redaction policy. Omitting it lets a
* handoff's activity be traced and redacted under the default
* agent's configuration, bypassing a stricter per-agent policy. */
...(executingAgentId != null && { agentId: executingAgentId }),
/** The wiring always supplies one (the yaml `activityPrompt` when
* set, else this repo's instruction). Falling through to the SDK's
* built-in prompt would silently use a different register. */
...((prompt ?? this.activityLabelPrompt) != null && {
prompt: prompt ?? this.activityLabelPrompt,
}),
chainOptions: {
signal,
callbacks: [{ handleLLMEnd, ...capturePrompt }],
configurable: {
thread_id: this.conversationId,
user_id: this.user ?? this.options.req?.user?.id,
},
},
});
return label ?? null;
} finally {
/** Safety net for a caller that did not defer (none in-tree): the old
* inline accounting, still scope-gated. */
if (!usageDeferred) {
await recordUsage();
}
}
}
/**
* SDK bridge for parent phase summaries. Usage is returned as a deferred
* collector so the runtime can commit the durable label first and bill
* only summaries that actually became visible.
*/
async generateActivityPhaseViaRun({
activities,
assistantContext,
closingTextPhase,
phaseIndex,
totalActivityCount,
status,
agentIds,
charLimit,
prompt,
signal,
}) {
if (typeof this.run?.generateActivityPhaseLabel !== 'function') {
return {};
}
const { provider, clientOptions, endpointTokenConfig, sameEndpoint } =
await this.resolveActivityPhaseLabelLLM();
const { handleLLMEnd, collected } = createMetadataAggregator();
let sdkPromptText;
const capturePrompt = {
handleLLMStart: (_llm, prompts) => {
sdkPromptText = Array.isArray(prompts) ? prompts.join('\n') : undefined;
},
handleChatModelStart: (_llm, messages) => {
try {
sdkPromptText = (messages ?? [])
.flat()
.map((message) =>
typeof message?.content === 'string'
? message.content
: JSON.stringify(message?.content ?? ''),
)
.join('\n');
} catch {
// Providers with usage metadata do not need the estimate fallback.
}
},
};
let label;
try {
({ label } = await this.run.generateActivityPhaseLabel({
provider,
clientOptions,
activities,
assistantContext,
closingTextPhase,
phaseIndex,
totalActivityCount,
status,
agentIds,
charLimit,
prompt,
sourceRunId: this.responseMessageId,
sourceTraceId: traceIdForMessage(this.responseMessageId),
responseId: this.responseMessageId,
traceSeed: `${this.responseMessageId}-activity-phase-${phaseIndex}`,
chainOptions: {
signal,
callbacks: [{ handleLLMEnd, ...capturePrompt }],
configurable: {
thread_id: this.conversationId,
user_id: this.user ?? this.options.req?.user?.id,
requestBody: { parentMessageId: this.parentMessageId },
},
},
}));
} catch (error) {
if (!signal?.aborted) {
logger.warn('[AgentClient] Activity phase generation failed', error);
}
}
return {
label,
collectUsage: async (completionText) =>
this.recordActivityLabelUsage(
collected,
clientOptions.model,
endpointTokenConfig,
sameEndpoint,
undefined,
provider,
() => ({ promptText: sdkPromptText ?? '', completionText: completionText ?? '' }),
'activity-phase',
),
};
}
/** Bounded settle for in-flight label fills before finalization. On
* timeout the label scope is closed and its abort controller fired, so a
* straggler cannot mutate the saved response or emit into a dead job. */
async settleActivityLabels(timeoutMs = 3000) {
/** Detached even when nothing settled: the wiring attaches its abort
* listener at BUILD time, and a segment can end without a single claim
* (text-only, or handoff batches, which skip labels) — the early
* return below would otherwise leave that listener accumulating across
* HITL approval cycles on the shared job signal. Idempotent. */
const detachScopeListeners = () => {
for (const scope of this.activityLabelScopes ?? []) {
scope.detach?.();
}
};
const pending = this.pendingActivityLabelFills;
if (!pending || pending.length === 0) {
detachScopeListeners();
return;
}
this.pendingActivityLabelFills = [];
await settlePendingLabelFills(pending, timeoutMs, () => {
/** Close EVERY generation's scope: a pre-pause wiring's straggler must
* stay closed even though a resume built a newer one. */
for (const scope of this.activityLabelScopes ?? []) {
scope.closed = true;
scope.abort.abort();
}
});
detachScopeListeners();
}
/**
* Activity-label wiring. At each batch boundary the hook synchronously
* claims a live content slot (steering's index-offset pattern: push
* placeholder with deterministic counts, bump the shared offset so
* subsequent SDK indices land past it) and fills it when the fast-model
* label resolves. Both states reach the live client via the dedicated
* `on_activity_label` event; failures leave the counts-only part.
* @param {string | undefined} streamId
*/
buildActivityLabelWiring(streamId, abortSignal) {
if (!streamId) {
return undefined;
}
/** Per-endpoint opt-in via `activityLabel: true` in librechat.yaml,
* resolved the same way the title options are (endpoints.all > named
* endpoint > custom endpoint config). Custom endpoints live in the
* `endpoints.custom` ARRAY, so their settings are only visible through
* the matched entry — without it every custom endpoint reads as
* disabled. */
const agentEndpoint = this.options.agent?.endpoint ?? '';
const appConfigForActivity = this.options.req?.config;
let customEndpointConfig;
try {
customEndpointConfig = getCustomEndpointConfig({
endpoint: agentEndpoint,
appConfig: appConfigForActivity,
});
} catch {
customEndpointConfig = undefined;
}
const activityConfig = resolveActivityConfig(
appConfigForActivity,
agentEndpoint,
customEndpointConfig,
/** The PUBLIC endpoint (`agents`): `initializeAgent` rewrites
* `agent.endpoint` to the backing provider, so without this an
* admin's `endpoints.agents.activityLabel: true` reads the
* provider's block instead and the feature stays off. */
this.options.endpoint,
);
if (!activityConfig.enabled) {
return undefined;
}
this.activityLabelPrompt = activityConfig.prompt;
/**
* Mark the job so a resume can reconcile label gaps without probing
* content. Retried rather than fire-and-forget: this flag GATES that
* reconciliation, and it is a separate write from the durable label
* append — so a single lost write silently drops a label that the label
* content itself recorded perfectly well. One retry costs nothing at run
* setup and removes the only realistic way the gate goes stale.
*/
/** Retained (not fire-and-forget): the RUN START awaits this persist
* (chatCompletion/resumeCompletion, before processStream/resume), so
* the flag is durable before any batch can claim a label — closing the
* immediate-reconnect race WITHOUT delaying the claim-time reservation
* emit, whose ordering against shifted SDK indices is load-bearing.
* The chain settles on failure (warned retry), so a lost write can
* never wedge run startup. */
this.activityLabelsMarkedPromise =
this.activityLabelsMarkedPromise ??
GenerationJobManager.markActivityLabels(streamId, this.jobCreatedAt).catch(() =>
GenerationJobManager.markActivityLabels(streamId, this.jobCreatedAt).catch(() => {
logger.warn(
`[AgentClient] Could not flag activity labels for ${streamId}; a label resolving during a resume gap may not be reconciled.`,
);
}),
);
/** SDK support probe (steering-style): the Run method and the formatter
* replay skip ship together, so method presence is the capability. */
const sdkCapable = typeof Run?.prototype?.generateActivityLabel === 'function';
/** Label-scoped abort: fired when settle times out so a straggling
* generation stops burning provider time for a finalized response.
* Chained to the run signal so a user abort still cancels labels. */
/** Close state is PER WIRING, not per client: a HITL resume rebuilds the
* wiring, and resetting a shared instance flag would re-open closures
* from the pre-pause segment whose provider call ignored the abort.
* Scopes are retained so settle closes every generation, past included. */
const labelScope = { closed: false, abort: new AbortController() };
this.activityLabelScopes = this.activityLabelScopes ?? [];
this.activityLabelScopes.push(labelScope);
/** Seed the usage sequence past the labels already on this response.
* `runId` is the response message id, so a HITL resume — which builds a
* NEW client for the SAME response — would otherwise restart at -1 and
* the client's `runId:seq` deduper would discard the post-approval
* label's usage as already counted. Each label generation is a single
* non-streaming invoke, so one existing label part == one consumed seq. */
this.activityLabelUsageSeq =
this.activityLabelUsageSeq ??
(this.contentParts ?? []).filter((part) => part?.type === ContentTypes.ACTIVITY_LABEL).length;
this.activityLabelAbort = labelScope.abort;
/** An abort CLOSES the scope, not just cancels the call. The rejected
* generation still runs its catch and calls `fill(null)`; with the scope
* merely aborted that fill would emit — and by then the next generation
* may already own the stream, so the event would land an index from the
* abandoned response onto the new one. */
const closeOnAbort = () => {
labelScope.closed = true;
labelScope.abort.abort();
};
if (abortSignal != null) {
if (abortSignal.aborted) {
closeOnAbort();
} else {
abortSignal.addEventListener('abort', closeOnAbort, { once: true });
/** Detached once this segment settles: HITL runs rebuild a wiring
* per approval cycle on the SAME job signal, and `once` only
* removes the listener if an abort actually fires — long
* multi-approval runs would otherwise accumulate obsolete
* closures toward the listener-limit warning. */
labelScope.detach = () => abortSignal.removeEventListener('abort', closeOnAbort);
}
}
/** Thin wrapper: slot claiming, lane stamping, emit ordering, and settle
* tracking live in `createActivityLabelWiring` (packages/api, TS). */
return createActivityLabelWiring({
maxPerRun: activityConfig.maxPerRun,
charLimit: activityConfig.charLimit,
prompt: activityConfig.prompt,
abortSignal: labelScope.abort.signal,
isClosed: () => labelScope.closed,
getContentParts: () => this.contentParts,
bumpIndexOffset: () => {
this.steerOffsetState.offset += 1;
},
/** Emits IMMEDIATELY — never sequenced behind the flag persist. The
* claim has already bumped the shared index offset, so delaying the
* reservation while shifted SDK chunks persist would let a
* cross-instance reconnect reconstruct a hole, compact it, and have
* the late label event overwrite the part that moved into its index.
* Flag ordering is guaranteed upstream instead: run start awaits the
* persist, so the flag is durable before any batch can claim. */
emitLabelEvent: (index, part) =>
GenerationJobManager.emitChunk(
streamId,
{
event: ActivityLabelEvents.ON_ACTIVITY_LABEL,
data: {
index,
part,
responseMessageId: this.responseMessageId,
conversationId: this.conversationId,
},
},
/** Label generation is detached and can outlive its generation, so
* the emit is scoped to the epoch that claimed the index. Without
* it a straggler from a replaced generation lands its old index on
* the new response — invisibly, since an empty label renders
* nothing — overwriting whatever occupies that slot. */
{ durable: true, expectedCreatedAt: this.jobCreatedAt },
),
trackPendingFill: (fillDone) => {
this.pendingActivityLabelFills = this.pendingActivityLabelFills ?? [];
this.pendingActivityLabelFills.push(fillDone);
},
resolveLLM: () => this.resolveActivityLabelLLM(),
/** Per-generation usage accounting for the direct fallback path;
* the SDK bridge records its own via chainOptions callbacks. */
getInvokeCallbacks: () => {
const { handleLLMEnd, collected } = createMetadataAggregator();
return {
callbacks: [{ handleLLMEnd }],
collect: async (estimate) => {
const { provider, clientOptions, endpointTokenConfig, sameEndpoint } =
await this.resolveActivityLabelLLM();
await this.recordActivityLabelUsage(
collected,
clientOptions.model,
endpointTokenConfig,
sameEndpoint,
/** No scope gate — the hook invokes collect ONLY for a
* COMMITTED fill (the billing authority), and a scope that
* closes during the fill's durable emit must not let a
* visible label escape its charge. Dropped fills never
* reach this callback. */
undefined,
provider,
estimate,
);
},
};
},
...(sdkCapable && {
generateLabel: (payload) => this.generateActivityLabelViaRun(payload),
}),
});
}
/** Builds the independently opt-in parent activity-phase collector. */
buildActivityPhaseWiring(streamId, abortSignal, initialSnapshot) {
if (!streamId || typeof Run?.prototype?.generateActivityPhaseLabel !== 'function') {
return undefined;
}
const agentEndpoint = this.options.agent?.endpoint ?? '';
const appConfig = this.options.req?.config;
let customEndpointConfig;
try {
customEndpointConfig = getCustomEndpointConfig({ endpoint: agentEndpoint, appConfig });
} catch {
customEndpointConfig = undefined;
}
const phaseConfig = resolveActivityPhaseConfig(
appConfig,
agentEndpoint,
customEndpointConfig,
this.options.endpoint,
);
if (!phaseConfig.enabled) {
return undefined;
}
this.activityLabelsMarkedPromise =
this.activityLabelsMarkedPromise ??
GenerationJobManager.markActivityLabels(streamId, this.jobCreatedAt).catch(() =>
GenerationJobManager.markActivityLabels(streamId, this.jobCreatedAt).catch(() => {
logger.warn(
`[AgentClient] Could not flag activity phases for ${streamId}; a phase resolving during a resume gap may not be reconciled.`,
);
}),
);
const scope = { closed: false, abort: new AbortController() };
this.activityLabelScopes = this.activityLabelScopes ?? [];
this.activityLabelScopes.push(scope);
const closeOnAbort = () => {
scope.closed = true;
scope.abort.abort();
};
if (abortSignal != null) {
if (abortSignal.aborted) {
closeOnAbort();
} else {
abortSignal.addEventListener('abort', closeOnAbort, { once: true });
scope.detach = () => abortSignal.removeEventListener('abort', closeOnAbort);
}
}
this.activityLabelUsageSeq =
this.activityLabelUsageSeq ??
(this.contentParts ?? []).filter((part) => part?.type === ContentTypes.ACTIVITY_LABEL).length;
const wiring = createActivityPhaseWiring({
maxPerRun: phaseConfig.maxPerRun,
charLimit: phaseConfig.charLimit,
prompt: phaseConfig.prompt,
initialSnapshot,
abortSignal: scope.abort.signal,
isClosed: () => scope.closed,
getContentParts: () => this.contentParts,
getStepIndex: (stepId) => this.stepMap?.get(stepId)?.index,
bumpIndexOffset: () => {
this.steerOffsetState.offset += 1;
},
emitLabelEvent: (index, part) =>
GenerationJobManager.emitChunk(
streamId,
{
event: ActivityLabelEvents.ON_ACTIVITY_LABEL,
data: {
index,
part,
responseMessageId: this.responseMessageId,
conversationId: this.conversationId,
},
},
{ durable: true, expectedCreatedAt: this.jobCreatedAt },
),
trackPendingFill: (fillDone) => {
this.pendingActivityLabelFills = this.pendingActivityLabelFills ?? [];
this.pendingActivityLabelFills.push(fillDone);
},
generatePhase: (payload) => this.generateActivityPhaseViaRun(payload),
});
this.activityPhaseWiring = wiring;
return wiring;
}
/**
* `AgentClient` is not opinionated about vision requests, so we don't do anything here
* @param {MongoFile[]} attachments
*/
checkVisionRequest() {}
getSaveOptions() {
let runOptions = {};
try {
runOptions = payloadParser(this.options) ?? {};
} catch (error) {
logger.error(
'[api/server/controllers/agents/client.js #getSaveOptions] Error parsing options',
error,
);
}
return removeNullishValues(
Object.assign(
{
spec: this.options.spec,
iconURL: this.options.iconURL,
chatProjectId: this.options.chatProjectId,
endpoint: this.options.endpoint,
agent_id: this.options.agent.id,
modelLabel: this.options.modelLabel,
resendFiles: this.options.resendFiles,
imageDetail: this.options.imageDetail,
maxContextTokens: this.maxContextTokens,
},
// TODO: PARSE OPTIONS BY PROVIDER, MAY CONTAIN SENSITIVE DATA
runOptions,
),
);
}
/**
* Returns build message options. For AgentClient, agent-specific instructions
* are retrieved directly from agent objects in buildMessages, so this returns empty.
* @returns {Object} Empty options object
*/
getBuildMessagesOptions() {
return {};
}
/**
*
* @param {TMessage} message
* @param {Array<MongoFile>} attachments
* @returns {Promise<Array<Partial<MongoFile>>>}
*/
async addImageURLs(message, attachments) {
const { files, image_urls } = await encodeAndFormat(
this.options.req,
attachments,
{
provider: this.options.agent.provider,
endpoint: this.options.endpoint,
},
VisionModes.agents,
);
message.image_urls = image_urls.length ? image_urls : undefined;
return files;
}
async buildMessages(messages, parentMessageId, _buildOptions, opts) {
/** Always pass mapMethod; getMessagesForConversation applies it only to messages with addedConvo flag */
const orderedMessages = this.constructor.getMessagesForConversation({
messages,
parentMessageId,
summary: this.shouldSummarize,
mapMethod: createMultiAgentMapper(this.options.agent, this.agentConfigs),
mapCondition: (message) => message.addedConvo === true,
});
let payload;
/** @type {number | undefined} */
let promptTokens;
/** Normalize instruction fields before applying per-run context. */
const normalizeInstructions = (agent) => {
agent.instructions = agent.instructions?.trim() || undefined;
agent.additional_instructions = agent.additional_instructions?.trim() || undefined;
return agent;
};
/** Collect all agents for unified processing while preserving stable/dynamic instruction fields. */
const allAgents = [
{ agent: normalizeInstructions(this.options.agent), agentId: this.options.agent.id },
...(this.agentConfigs?.size > 0
? Array.from(this.agentConfigs.entries()).map(([agentId, agent]) => ({
agent: normalizeInstructions(agent),
agentId,
}))
: []),
];
/**
* Memory authorization/loading and MCP config resolution do not depend on
* attachment hydration or prompt formatting. Start them before that work,
* but keep the existing context-application barrier below.
*
* Attach a rejection observer immediately because these operations may
* settle while request attachments are still being prepared. Awaiting the
* original promise later still propagates either error.
*/
const earlySharedContextPromise = Promise.all([
this.useMemory(),
resolveConfigServers(this.options.req),
]);
void earlySharedContextPromise.catch(() => {});
const sharedRunAttachmentIds = new Set();
/** @type {ReturnType<typeof buildAgentScopedContext>} */
let agentScopedContextPromise;
const startAgentScopedContext = () => {
const contextPromise = buildAgentScopedContext({
agentIds: allAgents.map(({ agentId }) => agentId),
attachmentsByAgentId: this.options.agentContextAttachmentsByAgentId,
sharedRunAttachmentIds,
req: this.options.req,
tokenCountFn: (text) => countTokens(text),
});
void contextPromise.catch(() => {});
return contextPromise;
};
if (this.options.attachments) {
const attachments = await this.options.attachments;
const latestMessage = orderedMessages[orderedMessages.length - 1];
for (const fileId of collectFileIds(attachments)) {
sharedRunAttachmentIds.add(fileId);
}
/** Agent-scoped extraction only depends on the shared attachment IDs. */
agentScopedContextPromise = startAgentScopedContext();
if (this.message_file_map) {
this.message_file_map[latestMessage.messageId] = attachments;
} else {
this.message_file_map = {
[latestMessage.messageId]: attachments,
};
}
const [, files] = await Promise.all([
this.addFileContextToMessage(latestMessage, attachments),
this.processAttachments(latestMessage, attachments),
]);
this.options.attachments = files;
} else {
agentScopedContextPromise = startAgentScopedContext();
}
/** Note: Bedrock uses legacy RAG API handling */
if (this.message_file_map && !isAgentsEndpoint(this.options.endpoint)) {
this.contextHandlers = createContextHandlers(
this.options.req,
orderedMessages[orderedMessages.length - 1].text,
);
}
/** @type {Record<number, number>} */
const indexTokenCountMap = {};
/** @type {Record<string, number>} */
const tokenCountMap = {};
const memoryPayload = [];
let hasFileContext = false;
let promptTokenTotal = 0;
const encoding = this.getEncoding();
const formattedMessages = orderedMessages.map((message, i) => {
const formattedMessage = formatMessage({
message,
userName: this.options?.name,
assistantName: this.options?.modelLabel,
});
const memoryFormattedMessage = formatMessage({
message,
userName: this.options?.name,
assistantName: this.options?.modelLabel,
});
/**
* Bind file context to the message it belongs to. Historical attachments
* are resent inline, so the current turn's text attachment must be inline
* too instead of living only in the dynamic system tail.
*/
if (message.fileContext) {
hasFileContext = true;
prependFileContext(formattedMessage, message.fileContext);
}
/**
* Durably re-merge quoted excerpts into every user turn that carries them
* (current and historical) so the model receives the referenced context on
* every prompt and the token count matches what was persisted. Applied to
* the memory copy too so the canonical per-message count includes them.
*/
if (Array.isArray(message.quotes) && message.quotes.length > 0) {
prependQuotes(formattedMessage, message.quotes);
prependQuotes(memoryFormattedMessage, message.quotes);
}
memoryPayload.push(memoryFormattedMessage);
const dbTokenCount = Number(orderedMessages[i].tokenCount);
const hasDbTokenCount = Number.isFinite(dbTokenCount) && dbTokenCount > 0;
/**
* Force a recount when the message carries quotes: a plain text-only
* "Save" edit recomputes `tokenCount` from `text` alone while leaving
* `message.quotes` persisted, so the stored count would undercount the
* quote block this turn prepends. Recounting from the quote-merged memory
* copy keeps context accounting accurate (and self-heals stale counts).
*/
const needsCanonicalTokenCount =
!hasDbTokenCount ||
(this.isVisionModel && (message.image_urls || message.files)) ||
(Array.isArray(message.quotes) && message.quotes.length > 0);
let canonicalTokenCount = hasDbTokenCount ? dbTokenCount : 0;
if (needsCanonicalTokenCount) {
canonicalTokenCount = countFormattedMessageTokens(memoryFormattedMessage, encoding);
}
const promptMessageTokenCount = message.fileContext
? countFormattedMessageTokens(formattedMessage, encoding)
: canonicalTokenCount;
/* If message has files, calculate image token cost */
if (this.message_file_map && this.message_file_map[message.messageId]) {
const attachments = this.message_file_map[message.messageId];
for (const file of attachments) {
if (file.embedded) {
this.contextHandlers?.processFile(file);
continue;
}
if (file.metadata?.codeEnvRef) {
continue;
}
}
}
const normalizedCanonicalTokenCount =
Number.isFinite(canonicalTokenCount) && canonicalTokenCount > 0 ? canonicalTokenCount : 0;
const normalizedPromptTokenCount =
Number.isFinite(promptMessageTokenCount) && promptMessageTokenCount > 0
? promptMessageTokenCount
: 0;
orderedMessages[i].tokenCount = normalizedCanonicalTokenCount;
indexTokenCountMap[i] = normalizedPromptTokenCount;
promptTokenTotal += normalizedPromptTokenCount;
if (message.messageId) {
tokenCountMap[message.messageId] = normalizedCanonicalTokenCount;
}
if (isEnabled(process.env.AGENT_DEBUG_LOGGING)) {
const role = message.isCreatedByUser ? 'user' : 'assistant';
const hasSummary =
Array.isArray(message.content) && message.content.some((p) => p && p.type === 'summary');
const suffix = hasSummary ? '[S]' : '';
const id = (message.messageId ?? message.id ?? '').slice(-8);
const recalced = needsCanonicalTokenCount ? normalizedCanonicalTokenCount : null;
const promptRecalced = message.fileContext ? normalizedPromptTokenCount : null;
logger.debug(
`[AgentClient] msg[${i}] ${role}${suffix} id=…${id} db=${dbTokenCount} needsRecount=${needsCanonicalTokenCount} recalced=${recalced} promptRecalced=${promptRecalced} tokens=${normalizedPromptTokenCount}`,
);
}
return formattedMessage;
});
/**
* Native YouTube -> video understanding: when Google `url_context` is enabled
* (resolved to the native `urlContext` provider tool), inject any YouTube URLs
* from the latest user turn as Gemini `fileData` video parts. The URL Context
* tool cannot read YouTube, so this routes those links through the video path
* while other URLs still flow through `urlContext`. Done after token counting
* (video tokens are reported by the provider) and only on the LLM payload, so
* the memory copy and persisted message are untouched.
*/
const latestOrdered = orderedMessages[orderedMessages.length - 1];
const provider = this.options.agent?.provider;
if (
latestOrdered?.isCreatedByUser === true &&
(provider === Providers.GOOGLE || provider === Providers.VERTEXAI) &&
hasUrlContextTool(this.options.agent?.tools)
) {
const latestFormatted = formattedMessages[formattedMessages.length - 1];
/** Use the resolved run model (model_parameters override) rather than the saved base model. */
const resolvedModel =
this.options.agent?.model_parameters?.model ?? this.options.agent?.model;
const { max, mimeType } = resolveYouTubeInjectionConfig({
provider,
model: resolvedModel,
});
latestFormatted.content = appendYouTubeVideoParts({
enabled: true,
text: latestOrdered.text,
content: latestFormatted.content,
max,
mimeType,
});
/** Google rejects an unusable video with a generic `INVALID_ARGUMENT` that names no cause,
* so `#sendCompletion` can only attribute one by knowing this turn carried a video. */
this.injectedYouTubeVideo = hasYouTubeVideoParts(latestFormatted.content);
}
payload = formattedMessages;
if (this.options.resendFiles) {
/** Persisted steer parts of past turns replay with their attachments:
* one batched owner-scoped fetch, re-encoded per turn and stamped as a
* transient `media` array (same resend semantics as message files).
* The stamp lands after the loop above finalized its counts, so the
* re-encoded media (minus the text part the steer part already counted)
* is folded into the budget here — large steered attachments must
* shrink the window like any other resent media. */
const stamped = await stampSteerPartMedia({
client: this,
user: this.options.req?.user,
payload,
// addPreviousAttachments already fetched steer-part refs in its single
// per-turn historical-files query — no second round trip.
docsById: this.authorizedHistoricalFiles,
getFiles: db.getFiles,
});
for (const { index, media, steerText } of stamped) {
/** Count the FULL stamped content and subtract only the steer body
* (already counted inside the assistant message): extracted file
* context prepended into the text part must hit the budget too, or
* large steered documents bypass pruning. */
const fullTokens = countFormattedMessageTokens({ role: 'user', content: media }, encoding);
const bodyTokens = steerText
? countFormattedMessageTokens(
{ role: 'user', content: [{ type: ContentTypes.TEXT, text: steerText }] },
encoding,
)
: 0;
const mediaTokens = Math.max(0, (fullTokens ?? 0) - (bodyTokens ?? 0));
if (Number.isFinite(mediaTokens) && mediaTokens > 0) {
indexTokenCountMap[index] = (indexTokenCountMap[index] ?? 0) + mediaTokens;
promptTokenTotal += mediaTokens;
}
}
}
this.memoryPayload = hasFileContext ? memoryPayload : null;
messages = orderedMessages;
promptTokens = promptTokenTotal;
/**
* Build shared run context - applies to ALL agents in the run.
* Request attachment file context is already bound inline to the latest
* user message above; only side-channel context belongs here.
* Memory context is handled separately and applied per-agent based on config.
*/
const sharedRunContextParts = [];
const [augmentedPrompt, [memories, configServers], agentScopedContext] = await Promise.all([
this.contextHandlers?.createContext(),
earlySharedContextPromise,
agentScopedContextPromise,
]);
/** Augmented prompt from RAG/context handlers */
this.augmentedPrompt = augmentedPrompt;
if (this.augmentedPrompt) {
sharedRunContextParts.push(this.augmentedPrompt);
}
/** Memory context (user preferences/memories). Keyed context (with memory
* keys + token metadata) is reserved for agents that can call
* `delete_memory`; everyone else gets the unkeyed values only. */
/** Partition the loaded memories belong to (the primary agent's). */
const loadedMemoryAgentId = getMemoryAgentId(this.options.agent);
const buildMemoryContext = (text) =>
text ? `${memoryInstructions}\n\n# Existing memory about the user:\n${text}` : undefined;
/** Resolves formatted memories for an agent's own partition. A defined
* `memories` means the run-level gates (permission, opt-out, config)
* passed; agents on other partitions fetch through the request-scoped
* cache so repeated partitions share one query. */
const getAgentPartitionMemories = async (agent) => {
if (!memories) {
return undefined;
}
const agentPartition = getMemoryAgentId(agent);
if (agentPartition === loadedMemoryAgentId) {
return memories;
}
try {
return await getRequestMemories({
req: this.options.req,
userId: this.options.req.user.id + '',
agentId: agentPartition,
getFormattedMemories: db.getFormattedMemories,
});
} catch (error) {
logger.error('[AgentClient] Error loading partition memories', error);
return undefined;
}
};
const sharedRunContext = sharedRunContextParts.join('\n\n');
const memoryAgentEnabled = isMemoryAgentEnabled(this.options.req.config?.memory);
/** Preserve prompt token counts for graph formatting and pruning. */
this.indexTokenCountMap = indexTokenCountMap;
/** Extract contextMeta from the parent response (second-to-last in ordered chain;
* last is the current user message). Seeds the pruner's calibration EMA for this run. */
const parentResponse =
orderedMessages.length >= 2 ? orderedMessages[orderedMessages.length - 2] : undefined;
if (parentResponse?.contextMeta && !parentResponse.isCreatedByUser) {
this.contextMeta = parentResponse.contextMeta;
}
const result = {
prompt: payload,
tokenCountMap,
promptTokens,
messages,
};
if (promptTokens >= 0 && typeof opts?.getReqData === 'function') {
opts.getReqData({ promptTokens });
}
/**
* Apply context to all agents.
* Stable agent/MCP instructions stay on `instructions`; shared runtime context
* is appended to `additional_instructions` as the dynamic system tail.
*
* NOTE: This intentionally mutates agent objects in place. The agentConfigs Map
* holds references to config objects that will be passed to the graph runtime.
*/
const ephemeralAgent = this.options.req.body.ephemeralAgent;
const mcpManager = getMCPManager();
await Promise.all(
allAgents.map(async ({ agent, agentId }) => {
const agentRunContextParts = [sharedRunContext];
const agentHasMemory = agentHasInlineMemoryTools(agent);
if (agentId === this.options.agent.id || memoryAgentEnabled || agentHasMemory) {
const partitionMemories = await getAgentPartitionMemories(agent);
const agentMemoryContext = buildMemoryContext(
agentHasMemory ? partitionMemories?.withKeys : partitionMemories?.withoutKeys,
);
if (agentMemoryContext) {
agentRunContextParts.push(agentMemoryContext);
}
}
const scopedContext = agentScopedContext.get(agentId);
if (scopedContext) {
agentRunContextParts.push(scopedContext);
}
return applyContextToAgent({
agent,
agentId,
logger,
mcpManager,
configServers,
sharedRunContext: agentRunContextParts.filter(Boolean).join('\n\n'),
ephemeralAgent: agentId === this.options.agent.id ? ephemeralAgent : undefined,
});
}),
);
return result;
}
/**
* Creates a promise that resolves with the memory promise result or undefined after a timeout
* @param {Promise<(TAttachment | null)[] | undefined>} memoryPromise - The memory promise to await
* @param {number} timeoutMs - Timeout in milliseconds (default: 3000)
* @returns {Promise<(TAttachment | null)[] | undefined>}
*/
async awaitMemoryWithTimeout(memoryPromise, timeoutMs = 3000) {
if (!memoryPromise) {
return;
}
try {
const timeoutPromise = new Promise((_, reject) =>
setTimeout(() => reject(new Error('Memory processing timeout')), timeoutMs),
);
const attachments = await Promise.race([memoryPromise, timeoutPromise]);
return attachments;
} catch (error) {
if (error.message === 'Memory processing timeout') {
logger.warn('[AgentClient] Memory processing timed out after 3 seconds');
} else {
logger.error('[AgentClient] Error processing memory:', error);
}
return;
}
}
/**
* @returns {Promise<{ withKeys?: string; withoutKeys?: string } | undefined>}
*/
async useMemory() {
const user = this.options.req.user;
if (user.personalization?.memories === false) {
return;
}
const hasAccess = await checkAccess({
user,
permissionType: PermissionTypes.MEMORIES,
permissions: [Permissions.USE],
getRoleByName: db.getRoleByName,
});
if (!hasAccess) {
logger.debug(
`[api/server/controllers/agents/client.js #useMemory] User ${user.id} does not have USE permission for memories`,
);
return;
}
const appConfig = this.options.req.config;
const memoryConfig = appConfig.memory;
if (!memoryConfig || memoryConfig.disabled === true) {
return;
}
const userId = this.options.req.user.id + '';
/** Memory partition of the primary agent; undefined = shared personal pool */
const memoryAgentId = getMemoryAgentId(this.options.agent);
this.processMemory = undefined;
if (!isMemoryAgentEnabled(memoryConfig)) {
try {
const { withKeys, withoutKeys } = await getRequestMemories({
req: this.options.req,
userId,
agentId: memoryAgentId,
getFormattedMemories: db.getFormattedMemories,
});
return { withKeys, withoutKeys };
} catch (error) {
logger.error(
'[api/server/controllers/agents/client.js #useMemory] Error loading memories',
error,
);
return;
}
}
/** @type {Agent} */
let prelimAgent;
const allowedProviders = new Set(
appConfig?.endpoints?.[EModelEndpoint.agents]?.allowedProviders,
);
try {
if (memoryConfig.agent?.id != null && memoryConfig.agent.id !== this.options.agent.id) {
prelimAgent = await loadAgent({
req: this.options.req,
agent_id: memoryConfig.agent.id,
endpoint: EModelEndpoint.agents,
});
} else if (memoryConfig.agent?.id != null) {
prelimAgent = this.options.agent;
} else if (
memoryConfig.agent?.id == null &&
memoryConfig.agent?.model != null &&
memoryConfig.agent?.provider != null
) {
prelimAgent = { id: Constants.EPHEMERAL_AGENT_ID, ...memoryConfig.agent };
}
} catch (error) {
logger.error(
'[api/server/controllers/agents/client.js #useMemory] Error loading agent for memory',
error,
);
}
if (!prelimAgent) {
return;
}
/** Forward the same `execute_code` capability gate the chat flow uses —
* memory agents are unlikely to list `execute_code`, but if one does,
* Phase 8 relies on this flag to expand the string into
* `bash_tool` + `read_file` (pre-Phase 8 the legacy `execute_code`
* tool registered unconditionally; without this passthrough the
* memory path would silently lose code-execution tooling). */
const memoryCapabilities = new Set(appConfig?.endpoints?.[EModelEndpoint.agents]?.capabilities);
const agent = await initializeAgent(
{
req: this.options.req,
res: this.options.res,
agent: prelimAgent,
allowedProviders,
endpointOption: {
endpoint: !isEphemeralAgentId(prelimAgent.id)
? EModelEndpoint.agents
: memoryConfig.agent?.provider,
},
codeEnvAvailable: memoryCapabilities.has(AgentCapabilities.execute_code),
statefulSessionsAvailable: memoryCapabilities.has(AgentCapabilities.stateful_code_sessions),
},
{
getFiles: db.getFiles,
getUserKey: db.getUserKey,
getConvoFiles: db.getConvoFiles,
getAccessibleMcpServerNames,
updateFilesUsage: db.updateFilesUsage,
getUserKeyValues: db.getUserKeyValues,
getToolFilesByIds: db.getToolFilesByIds,
getCodeGeneratedFiles: db.getCodeGeneratedFiles,
filterFilesByAgentAccess,
},
);
if (!agent) {
logger.warn(
'[api/server/controllers/agents/client.js #useMemory] No agent found for memory',
memoryConfig,
);
return;
}
const llmConfig = Object.assign(
{
provider: agent.provider,
model: agent.model,
},
agent.model_parameters,
);
/** @type {import('@librechat/api').MemoryConfig} */
const config = {
validKeys: memoryConfig.validKeys,
instructions: agent.instructions,
llmConfig,
tokenLimit: memoryConfig.tokenLimit,
};
const messageId = this.responseMessageId + '';
const conversationId = this.conversationId + '';
const streamId = this.options.req?._resumableStreamId || null;
const [withoutKeys, processMemory] = await createMemoryProcessor({
userId,
agentId: memoryAgentId,
config,
messageId,
streamId,
jobCreatedAt: this.jobCreatedAt,
conversationId,
memoryMethods: {
setMemory: db.setMemory,
deleteMemory: db.deleteMemory,
getFormattedMemories: db.getFormattedMemories,
},
res: this.options.res,
user: createSafeUser(this.options.req.user),
});
this.processMemory = processMemory;
let withKeys = withoutKeys;
try {
({ withKeys } = await getRequestMemories({
req: this.options.req,
userId,
agentId: memoryAgentId,
getFormattedMemories: db.getFormattedMemories,
}));
} catch (error) {
logger.error(
'[api/server/controllers/agents/client.js #useMemory] Error loading keyed memories',
error,
);
}
return { withKeys, withoutKeys };
}
/**
* Filters out image URLs from message content
* @param {BaseMessage} message - The message to filter
* @returns {BaseMessage} - A new message with image URLs removed
*/
filterImageUrls(message) {
if (!message.content || typeof message.content === 'string') {
return message;
}
if (Array.isArray(message.content)) {
const filteredContent = message.content.filter(
(part) => part.type !== ContentTypes.IMAGE_URL,
);
if (filteredContent.length === 1 && filteredContent[0].type === ContentTypes.TEXT) {
const MessageClass = message.constructor;
return new MessageClass({
content: filteredContent[0].text,
additional_kwargs: message.additional_kwargs,
});
}
const MessageClass = message.constructor;
return new MessageClass({
content: filteredContent,
additional_kwargs: message.additional_kwargs,
});
}
return message;
}
/**
* @param {BaseMessage[]} messages
* @returns {Promise<void | (TAttachment | null)[]>}
*/
async runMemory(messages) {
try {
if (this.processMemory == null) {
return;
}
const appConfig = this.options.req.config;
const memoryConfig = appConfig.memory;
const messageWindowSize = memoryConfig?.messageWindowSize ?? 5;
/**
* Strip skill-primed meta messages before memory extraction. The primes
* sit next to the latest user message and carry large SKILL.md bodies,
* so letting them into the window would crowd out real chat turns and
* pollute extracted memories with synthetic instruction content the
* user never typed.
*/
const chatMessages = messages.filter((m) => !isSkillPrimeMessage(m));
let messagesToProcess = [...chatMessages];
if (chatMessages.length > messageWindowSize) {
for (let i = chatMessages.length - messageWindowSize; i >= 0; i--) {
const potentialWindow = chatMessages.slice(i, i + messageWindowSize);
if (potentialWindow[0]?.role === 'user') {
messagesToProcess = [...potentialWindow];
break;
}
}
if (messagesToProcess.length === chatMessages.length) {
messagesToProcess = [...chatMessages.slice(-messageWindowSize)];
}
}
const filteredMessages = messagesToProcess.map((msg) => this.filterImageUrls(msg));
const bufferString = getBufferString(filteredMessages);
const configuredMaxInputTokens = Number.isFinite(memoryConfig?.maxInputTokens)
? Math.floor(memoryConfig.maxInputTokens)
: undefined;
const maxInputTokens =
configuredMaxInputTokens != null && configuredMaxInputTokens > 0
? configuredMaxInputTokens
: DEFAULT_MEMORY_MAX_INPUT_TOKENS;
const maxInputChars = maxInputTokens * MEMORY_INPUT_CHARS_PER_TOKEN;
const isCharTruncated = bufferString.length > maxInputChars;
const memoryInput = `# Current Chat:\n\n${
isCharTruncated
? `[Earlier chat content omitted due to memory input limit]\n\n${bufferString.slice(
-maxInputChars,
)}`
: bufferString
}`;
const {
text: limitedMemoryInput,
tokenCount,
wasTruncated,
} = await processTextWithTokenLimit({
text: memoryInput,
tokenLimit: maxInputTokens,
tokenCountFn: (text) => countTokens(text),
preserve: 'end',
});
if (isCharTruncated || wasTruncated) {
logger.warn('[MemoryAgent] Memory input truncated before processing', {
tokenCount,
messageId: this.responseMessageId,
conversationId: this.conversationId,
maxInputTokens,
wasTruncated,
maxInputChars,
originalLength: bufferString.length,
});
}
const bufferMessage = new HumanMessage(limitedMemoryInput);
return await this.processMemory([bufferMessage]);
} catch (error) {
logger.error('Memory Agent failed to process memory', error);
}
}
/** @type {sendCompletion} */
async sendCompletion(payload, opts = {}) {
await this.chatCompletion({
payload,
onProgress: opts.onProgress,
userMCPAuthMap: opts.userMCPAuthMap,
abortController: opts.abortController,
});
const completion = filterMalformedContentParts(this.contentParts);
const metadata = this.buildResponseMetadata();
return metadata ? { completion, metadata } : { completion };
}
/**
* Assembles the response message `metadata`: Vertex thought signatures plus
* the persisted context breakdown (Part A) and the usage/cost rollup (Part B),
* which rebuild the gauge breakdown and branch/total cost across reloads.
* Returns undefined when nothing was captured.
* @returns {{
* thoughtSignatures?: Record<string, string>,
* contextUsage?: import('librechat-data-provider').TContextUsageEvent,
* usage?: import('librechat-data-provider').TResponseUsage,
* } | undefined}
*/
buildResponseMetadata() {
/** @type {{
* thoughtSignatures?: Record<string, string>,
* contextUsage?: import('librechat-data-provider').TContextUsageEvent,
* usage?: import('librechat-data-provider').TResponseUsage,
* }} */
const metadata = {};
const signatures = this.collectedThoughtSignatures;
if (signatures && Object.keys(signatures).length > 0) {
metadata.thoughtSignatures = signatures;
}
const usageEvents = this.usageEmitSink ?? [];
/** Persist the breakdown only when the latest snapshot's OWN run completed —
* i.e. a PRIMARY usage event (usage_type == null) from that run's id arrived
* AFTER the snapshot. Matching by run id keeps `completedOutputTokens` a real
* post-snapshot delta even when parallel/direct runs interleave (A snapshot →
* B snapshot → A usage must NOT persist B's snapshot with A's output); an
* interrupted final call that emits no usage falls back to the per-message
* estimate. It still keeps the post-summary snapshot: the summarization detour
* emits an extra snapshot whose following primary usage shares that run's id,
* which the old snapshot-count guard miscounted and wrongly dropped. Events
* without a run id (older lib / resume) match any snapshot for back-compat. */
const latestSnapshot = this.contextUsageSink?.latest;
const latestSnapshotUsageIndex = this.contextUsageSink?.latestUsageIndex ?? 0;
const latestSnapshotRunId = latestSnapshot?.runId;
const hasPrimaryAfterSnapshot = usageEvents
.slice(latestSnapshotUsageIndex)
.some(
(event) =>
event.usage_type == null &&
(latestSnapshotRunId == null ||
event.runId == null ||
event.runId === latestSnapshotRunId),
);
if (latestSnapshot && hasPrimaryAfterSnapshot) {
metadata.contextUsage = buildPersistedContextUsage(latestSnapshot, usageEvents);
}
/** Lightweight summarization marker — persisted whenever this turn compacted
* the context, INDEPENDENT of the snapshot guard above. When the client has
* no usable snapshot on the branch and falls back to the per-message
* estimate, it caps the discarded pre-summary history at this baseline
* instead of re-summing it (the gauge otherwise reads 100% forever). Shared
* with the abort save path via `computeSummaryUsedTokens`. Subtract the
* response's earlier tool-loop outputs (the primaries that preceded the
* latest snapshot, same run): those tokens are inside the snapshot baseline
* AND in the response `tokenCount` the client estimate adds on top, so
* leaving them in the marker double-counts them on a multi-call turn. */
const priorOutputTokens = priorRunOutputTokens(
usageEvents,
latestSnapshotUsageIndex,
latestSnapshotRunId,
);
const summaryUsedTokens = computeSummaryUsedTokens(latestSnapshot, priorOutputTokens);
if (summaryUsedTokens != null) {
metadata.summaryUsedTokens = summaryUsedTokens;
}
const usage = aggregateEmittedUsage(usageEvents);
if (usage) {
metadata.usage = usage;
}
return Object.keys(metadata).length > 0 ? metadata : undefined;
}
/**
* Resolves the endpoint token config for a usage item by its producing agent
* (multi-endpoint graphs: connected agents + subagents). A known agent's
* config is authoritative — including `undefined`, which prices with built-in
* rates (e.g. a non-custom agent in a custom-primary graph). Only an
* untagged/unknown agent falls back to the primary config, so single-endpoint
* graphs are unchanged.
* @param {UsageMetadata} usage
* @returns {import('@librechat/api').EndpointTokenConfig | undefined}
*/
resolveAgentEndpointTokenConfig(usage) {
return resolveAgentTokenConfig({
agentId: usage?.agentId,
byAgentId: this.options.endpointTokenConfigByAgentId,
fallback: this.options.endpointTokenConfig,
});
}
/**
* @param {Object} params
* @param {string} [params.model]
* @param {string} [params.context='message']
* @param {AppConfig['balance']} [params.balance]
* @param {AppConfig['transactions']} [params.transactions]
* @param {UsageMetadata[]} [params.collectedUsage=this.collectedUsage]
*/
async recordCollectedUsage({
model,
balance,
transactions,
context = 'message',
collectedUsage = this.collectedUsage,
/**
* Rates for usage that did NOT run on the agent's endpoint — currently
* activity labels pointed at a different `activityEndpoint`. Without it
* the caller's config was dropped here and the balance transaction was
* written at the primary agent's rates while the UI cost was computed at
* the label's, so the two disagreed. `undefined` keeps the agent default.
*/
endpointTokenConfig,
/**
* True when this usage ran on a DIFFERENT endpoint than the agent, making
* `endpointTokenConfig` authoritative even when it is `undefined` (a
* built-in endpoint prices from the shared table). Presence of the value
* cannot express that, which is why the caller states it outright.
*/
crossEndpoint = false,
/**
* Whether this recording owns `getStreamUsage()`. Only the PRIMARY
* generation does. Secondary usage (activity labels) must still be
* billed, but writing it here would hand `BaseClient` the label's token
* counts as the assistant response's authoritative total — and because
* the primary call returns early when it collected nothing, the wrong
* value would never be replaced, suppressing the text-based token
* fallback and leaving the real generation unbilled.
*/
updateStreamUsage = true,
}) {
/** Per-agent resolution keys off the AGENT's config map, which cannot
* describe a label running on a different endpoint — so an explicit
* config wins outright rather than being second-guessed per usage row.
*
* Keyed on the caller's discriminator, NOT on `endpointTokenConfig !==
* undefined`: a built-in label endpoint prices from the shared table, so
* `undefined` is its meaningful value. Reading that as "no override" is
* what silently restored the primary's custom rates. */
const overrideTokenConfig = crossEndpoint === true;
const result = await recordCollectedUsage(
{
spendTokens: db.spendTokens,
spendStructuredTokens: db.spendStructuredTokens,
pricing: { getMultiplier: db.getMultiplier, getCacheMultiplier: db.getCacheMultiplier },
bulkWriteOps: { insertMany: db.bulkInsertTransactions, updateBalance: db.updateBalance },
},
{
user: this.user ?? this.options.req.user?.id,
conversationId: this.conversationId,
collectedUsage,
model: model ?? this.model ?? this.options.agent.model_parameters.model,
context,
messageId: this.responseMessageId,
balance,
transactions,
endpointTokenConfig: overrideTokenConfig
? endpointTokenConfig
: this.options.endpointTokenConfig,
...(overrideTokenConfig
? {}
: { resolveEndpointTokenConfig: (usage) => this.resolveAgentEndpointTokenConfig(usage) }),
},
);
if (result && updateStreamUsage) {
this.usage = result;
}
}
/**
* Get stream usage as returned by this client's API response.
* @returns {UsageMetadata} The stream usage object.
*/
getStreamUsage() {
return this.usage;
}
/**
* Builds the subagent usage emitter for {@link createSubagentUsageSink}.
* Streams each billed child-run usage to the client as an `on_token_usage`
* event tagged `subagent` (folds into session cost/totals, not the live
* gauge), with the authoritative cost when `interface.contextCost` is on.
* Returns undefined when there's no stream to write to.
* @param {AppConfig} [appConfig]
* @returns {((usage: UsageMetadata) => void) | undefined}
*/
buildSubagentUsageEmitter(appConfig) {
const res = this.options.res;
const streamId = this.options.req?._resumableStreamId || null;
if (!res && !streamId) {
return undefined;
}
const includeCost = appConfig?.interfaceConfig?.contextCost === true;
return (usage) => {
const data = {
input_tokens: usage.input_tokens,
output_tokens: usage.output_tokens,
total_tokens: usage.total_tokens,
input_token_details: this.subagentCacheDetails(usage),
model: usage.model,
provider: usage.provider,
usage_type: 'subagent',
runId: this.responseMessageId,
/** Unique per collected entry (post-push length) for resume dedupe */
seq: this.collectedUsage.length,
/** Price with the SUBAGENT's own endpoint token config (its endpoint may
* differ from the parent's); `usage.agentId` is tagged by the sink. */
cost: includeCost
? computeUsageCostUSD(
usage,
{ getMultiplier: db.getMultiplier, getCacheMultiplier: db.getCacheMultiplier },
this.resolveAgentEndpointTokenConfig(usage),
)
: undefined,
};
/** Fold into the response's usage rollup (synchronously, regardless of
* emit success) so the persisted total matches the live session, which
* also folds subagent usage into its cost/totals. */
if (this.usageEmitSink) {
this.usageEmitSink.push(data);
}
/** The sink fires this without awaiting, so retain the promise and flush
* it in chatCompletion's finally — emitChunk persists (HSET) before
* publishing, and job cleanup must not race that persist or resumed
* clients miss billed subagent usage. */
const emit = (async () => {
try {
if (streamId) {
await GenerationJobManager.emitChunk(
streamId,
{
event: UsageEvents.ON_TOKEN_USAGE,
data,
},
{ expectedCreatedAt: this.jobCreatedAt },
);
} else {
sendEvent(res, { event: UsageEvents.ON_TOKEN_USAGE, data });
}
} catch (err) {
logger.warn('[AgentClient] Failed to emit subagent usage', err);
}
})();
this.pendingSubagentEmits.push(emit);
return emit;
};
}
/** Normalizes a subagent usage event's cache token details for emission. */
subagentCacheDetails(usage) {
const cache_creation =
usage.input_token_details?.cache_creation ?? usage.cache_creation_input_tokens;
const cache_read = usage.input_token_details?.cache_read ?? usage.cache_read_input_tokens;
if (cache_creation == null && cache_read == null) {
return undefined;
}
return { cache_creation, cache_read };
}
/**
* @param {TMessage} responseMessage
* @returns {number}
*/
getTokenCountForResponse({ content }) {
return countFormattedMessageTokens({ role: 'assistant', content }, this.getEncoding());
}
/**
* @param {object} params
* @param {string | ChatCompletionMessageParam[]} params.payload
* @param {Record<string, Record<string, string>>} [params.userMCPAuthMap]
* @param {AbortController} [params.abortController]
*/
/**
* @deprecated Agent Chain — strip hidden intermediate sequential-agent content
* before persistence, keeping only the last part + tool_call parts. Mirrors the
* chat path so a HITL resume doesn't persist/emit intermediate outputs the
* agent's `hide_sequential_outputs` setting is meant to hide.
*/
applyHideSequentialOutputsFilter() {
if (!this.options.agent?.hide_sequential_outputs || !Array.isArray(this.contentParts)) {
return;
}
this.contentParts = this.contentParts.filter(
(part, index) =>
index >= this.contentParts.length - 1 ||
part.type === ContentTypes.TOOL_CALL ||
// Steer parts are user speech, not intermediate agent output — dropping
// one would erase the user's words from the persisted turn.
part.type === ContentTypes.STEER ||
// Activity labels summarize the hidden intermediate outputs — exactly
// the affordance hide_sequential_outputs wants to keep visible.
part.type === ContentTypes.ACTIVITY_LABEL ||
part.tool_call_ids,
);
}
/**
* Rebase parent activity bounds after completion-time content reshaping.
* Object identity links retained parts back to their pre-reshape positions,
* so prepended skill cards cannot enter a phase and a filtered-away leading
* reasoning part advances the bound to the first retained child.
*
* Both arrays may be sparse: the aggregator writes parts at provider-source
* indexes, which can skip slots. Holes must not enter the identity map — a
* hole reads as `undefined`, and one `undefined` key would falsely match
* every hole in `previousParts` as a retained part.
*
* @param {Array<import('librechat-data-provider').ContentPart | null | undefined>} previousParts
*/
rebaseActivityPhaseBounds(previousParts) {
if (!Array.isArray(previousParts) || !Array.isArray(this.contentParts)) {
return;
}
/** Preserve sparse coordinates when completion did not actually reshape
* the content. A phase can reserve a leading hole for a tool part whose
* SDK event lands after the phase closes; scanning retained identities
* in an unchanged array would skip that hole and move the bound past the
* delayed tool before it arrives. */
if (previousParts.length === this.contentParts.length) {
let unchanged = true;
for (let index = 0; index < previousParts.length; index += 1) {
if (previousParts[index] !== this.contentParts[index]) {
unchanged = false;
break;
}
}
if (unchanged) {
return;
}
}
const retainedIndexes = new Map();
for (let index = 0; index < this.contentParts.length; index += 1) {
const part = this.contentParts[index];
if (part != null) {
retainedIndexes.set(part, index);
}
}
for (let markerIndex = 0; markerIndex < this.contentParts.length; markerIndex += 1) {
const marker = this.contentParts[markerIndex];
if (
marker?.type !== ContentTypes.ACTIVITY_LABEL ||
marker.activity_label_type !== 'phase' ||
typeof marker.activity_start_index !== 'number'
) {
continue;
}
const previousMarkerIndex = previousParts.indexOf(marker);
if (previousMarkerIndex < 0) {
continue;
}
const previousStartIndex = Math.min(
previousMarkerIndex,
Math.max(0, marker.activity_start_index),
);
let nextStartIndex = markerIndex;
for (let index = previousStartIndex; index < previousMarkerIndex; index += 1) {
const retainedIndex = retainedIndexes.get(previousParts[index]);
if (retainedIndex != null && retainedIndex < markerIndex) {
nextStartIndex = retainedIndex;
break;
}
}
marker.activity_start_index = nextStartIndex;
}
}
/**
* Surface any human-in-the-loop interrupt the SDK captured during the most
* recent `processStream` / `resume`. When the run paused for tool approval (or
* an ask-user question), mark the job `requires_action`, persist the pending
* review record, and emit it to live clients — then set `this.pendingApproval`
* so the controller leaves the turn unfinalized for the resume route to continue.
*
* No-op when the run completed without an interrupt, or when the job was aborted
* between the interrupt firing and this mark (a late interrupt must not pause a
* dead job — the atomic `pause` transition returns false and we drop it).
*
* @param {AgentRun} run
* @param {string} [streamId]
*/
async handleRunInterrupt(run, streamId) {
if (!streamId || typeof run?.getInterrupt !== 'function') {
return;
}
const interrupt = run.getInterrupt();
if (!interrupt?.payload) {
return;
}
const appConfig = this.options.req?.config;
const checkpointerCfg = appConfig?.endpoints?.[EModelEndpoint.agents]?.checkpointer;
// Persist the generation params (temperature, max tokens, custom endpoint params, …)
// so an ephemeral-agent resume continues with the SAME settings the run paused on.
// The resume payload omits them and they aren't part of the fingerprint, so without
// this the rebuilt ephemeral run falls back to defaults. The paused request body is
// the primary source (UI-form, round-trips the compact-convo schema by construction);
// the resolved llmConfig fills gaps and is sanitized — it carries provider secrets
// (apiKey, credentials) and gateway config — resume re-resolves those server-side.
// (Saved agents source params from the DB record, so this is belt-and-suspenders.)
const resumeContext = pickResumeContext(this.options.req?.body);
const resumeModelParameters = captureResumeModelParameters(
this.options.req?.body,
this.options.agent?.model_parameters,
);
if (resumeModelParameters) {
resumeContext.model_parameters = resumeModelParameters;
}
// Persist the question onto the paused ask tool_call's args NOW: an
// abandoned/expired/stopped pause never reaches the answer-resume stamp,
// and the streamed args were dropped (name-less chunks) — without this the
// unfinished turn saves an empty ask part the record card can't render.
if (interrupt.payload?.type === 'ask_user_question' && Array.isArray(this.contentParts)) {
const stamped = attachAskUserQuestionArgs(
this.contentParts,
Array.isArray(interrupt.payload.questions)
? { questions: interrupt.payload.questions }
: interrupt.payload.question,
interrupt.payload.tool_call_id,
);
if (stamped !== this.contentParts) {
this.contentParts.length = 0;
this.contentParts.push(...stamped);
}
}
const pendingAction = buildPendingAction(interrupt.payload, {
streamId,
conversationId: this.conversationId,
// runId mirrors the LangGraph checkpoint namespace when the SDK provides it
// (its documented meaning), falling back to the response message id.
runId: interrupt.checkpointNs ?? this.responseMessageId,
responseMessageId: this.responseMessageId,
interruptId: interrupt.interruptId,
// thread_id was bound to conversationId at run config (config.configurable);
// fall back to it when the SDK doesn't echo threadId on the interrupt.
threadId: interrupt.threadId ?? this.conversationId,
ttlMs: getApprovalTtlMs(checkpointerCfg),
// Pin the graph-determining request fields so resume can't rebuild this paused
// run on a different agent/tool set (esp. ephemeral agents, whose agent_id is
// undefined so the id guard can't tell two configs apart).
requestFingerprint: computeAgentRequestFingerprint(this.options.req?.body ?? {}),
// Persist those same fields verbatim so the resume route can REPLAY them — a
// reload/cross-replica resume can't reconstruct the ephemeral config client-side,
// so the server restores it and rebuilds the same graph (and the fingerprint matches).
resumeContext,
});
// Job-replacement guard: streamId == conversationId is reused per conversation, so a
// newer request can replace this run's job. If this (older) run hits an interrupt
// after a replacement, pausing would flip the NEWER job to requires_action with this
// stale run's pending action, blocking fresh work behind the wrong approval. Only
// pause when the live job is still the one THIS run created (mirrors request.js).
if (this.jobCreatedAt != null) {
const liveJob = await GenerationJobManager.getJobStore().getJob(streamId);
if (!liveJob || liveJob.createdAt !== this.jobCreatedAt) {
logger.debug(`[AgentClient] Interrupt fired but job ${streamId} was replaced; not pausing`);
return;
}
}
// Snapshot deferred-tool discovery before exposing the pause. Tool-search results
// may live only in the interrupted SDK graph, so they must be committed atomically
// with requires_action for an immediate/cross-replica resume to retain the schemas.
let discoveredTools = [];
try {
discoveredTools = getRunDiscoveredTools(run);
} catch (err) {
logger.warn(
`[AgentClient] Failed to capture discovered tools for resume on ${streamId}`,
err?.message ?? err,
);
}
const paused = await GenerationJobManager.approvals.pause(streamId, pendingAction, {
expectedCreatedAt: this.jobCreatedAt,
...(discoveredTools.length > 0 ? { discoveredTools } : {}),
...(this.activityPhaseWiring?.snapshot != null && {
activityPhaseSnapshot: this.activityPhaseWiring.snapshot(),
}),
persistencePending: true,
});
if (!paused) {
logger.debug(
`[AgentClient] Interrupt fired but job ${streamId} was not running; not pausing`,
);
return;
}
this.pendingApproval = pendingAction;
// Release the concurrency slot this request held the MOMENT the turn is durably
// paused — before the approval card is emitted — so the user's `/resume` can
// re-acquire one immediately. Otherwise a fast Approve races the HTTP-driver
// teardown (request.js pause branch / resume.js finally) that would otherwise
// release it, and `/resume` 429s under LIMIT_CONCURRENT_MESSAGES. Idempotent via
// the flag; if it fails here, the teardown still releases (it checks the flag).
if (!this.pendingRequestReleased) {
try {
await decrementPendingRequest(this.options.req?.user?.id);
this.pendingRequestReleased = true;
} catch (err) {
logger.error(`[AgentClient] Failed to release request slot on pause ${streamId}`, err);
}
}
await GenerationJobManager.emitChunk(
streamId,
{
event: ApprovalEvents.ON_PENDING_ACTION,
data: toClientPendingAction(pendingAction),
},
{ expectedCreatedAt: this.jobCreatedAt },
);
// Steers queued before this pause stay IN the store for the whole approval
// window: `resumeState.pendingSteers` re-seeds the client's chips on
// reload, and the resumed run drains them at its first tool boundary.
// Draining here would leave the only copy in ephemeral client state — a
// reload during the pause would silently lose the user's message. New
// steers are rejected while paused (enqueue is status-guarded), and the
// requires_action TTL extension keeps the queue key alive.
logger.debug(
`[AgentClient] Paused ${streamId} for ${interrupt.payload.type} (action ${pendingAction.actionId})`,
);
}
async chatCompletion({ payload, userMCPAuthMap, abortController = null }) {
/** @type {Partial<GraphRunnableConfig>} */
let config;
/** @type {ReturnType<createRun>} */
let run;
/** @type {Promise<(TAttachment | null)[] | undefined>} */
let memoryPromise;
const appConfig = this.options.req.config;
const balanceConfig = getBalanceConfig(appConfig);
const transactionsConfig = getTransactionsConfig(appConfig);
try {
if (!abortController) {
abortController = new AbortController();
}
/** Fire-and-forget: boot the per-conversation stateful sandbox in
* parallel with generation so the first execute_code/bash call lands
* on a warm VM. No-op unless a reachable agent resolved
* `statefulCodeSessions`. */
maybePrewarmCodeSandbox({
req: this.options.req,
conversationId: this.conversationId,
agents: [this.options.agent, ...(this.agentConfigs?.values() ?? [])],
});
/** @type {AppConfig['endpoints']['agents']} */
const agentsEConfig = appConfig.endpoints?.[EModelEndpoint.agents];
config = {
runName: 'AgentRun',
configurable: {
thread_id: this.conversationId,
// LangGraph owns `checkpoint_ns` and resets it to '' at every root
// invocation. The saver maps this private immutable generation key
// into its physical namespace while tools keep the conversation id.
checkpoint_ns: '',
[LIBRECHAT_CHECKPOINT_NAMESPACE_KEY]: this.checkpointNamespace,
last_agent_index: this.agentConfigs?.size ?? 0,
user_id: this.user ?? this.options.req.user?.id,
hide_sequential_outputs: this.options.agent.hide_sequential_outputs,
requestBody: {
messageId: this.responseMessageId,
conversationId: this.conversationId,
parentMessageId: this.parentMessageId,
},
user: createSafeUser(this.options.req.user),
},
recursionLimit: resolveRecursionLimit(agentsEConfig, this.options.agent),
signal: abortController.signal,
streamMode: 'values',
version: 'v2',
};
const toolSet = buildToolSet(this.options.agent);
const tokenCounter = createTokenCounter(this.getEncoding());
/** Pre-resolve invoked skill bodies + re-prime files before formatting messages */
const skillPrimeResult = this.options.primeInvokedSkills
? await this.options.primeInvokedSkills(payload)
: undefined;
/**
* Seed `Graph.sessions` with code-env files primed across every
* reachable agent (primary, handoff/addedConvo, and nested
* subagents) plus skill-priming output. The merge logic and its
* run-wide semantics live in `buildInitialToolSessions`; see that
* helper's doc for why this is intentionally NOT per-agent.
*/
const initialSessions = buildInitialToolSessions({
skillSessions: skillPrimeResult?.initialSessions,
agents: [this.options.agent, ...(this.agentConfigs ? this.agentConfigs.values() : [])],
});
/**
* Reconstruct `reasoning_content` on prior tool-call turns: DeepSeek
* thinking-mode (#13366) or custom endpoints opting in via
* `customParams.includeReasoningHistory` (e.g. Xiaomi MiMo, Kimi).
* Walks subagents too — the opted-in endpoint may appear only as a
* nested subagent, not the primary or a top-level handoff agent.
*/
const needsReasoningContentFormat = anyAgentReplaysReasoningContent([
this.options.agent,
...(this.agentConfigs ? Array.from(this.agentConfigs.values()) : []),
]);
/**
* Skills primed fresh this turn — manual ($ popover) and always-apply
* (frontmatter). `injectSkillPrimes` (below) splices their SKILL.md
* bodies in, so `formatAgentMessages` must NOT also reconstruct the
* same names from a historical `skill` tool_call — otherwise the body
* lands twice and a prompt-cache marker can pin to the duplicated
* synthetic prefix. Names NOT primed this turn still reconstruct from
* history, preserving sticky manual re-priming across turns.
*/
const manualSkillPrimes = this.options.agent?.manualSkillPrimes;
const alwaysApplySkillPrimes = this.options.agent?.alwaysApplySkillPrimes;
const freshSkillPrimeNames = collectFreshSkillPrimeNames({
manualSkillPrimes,
alwaysApplySkillPrimes,
});
const formatOptions =
needsReasoningContentFormat || freshSkillPrimeNames.size > 0
? {
...(needsReasoningContentFormat ? { preserveReasoningContent: true } : {}),
...(freshSkillPrimeNames.size > 0
? { skipSkillBodyNames: freshSkillPrimeNames }
: {}),
}
: undefined;
let {
messages: initialMessages,
indexTokenCountMap,
summary: initialSummary,
boundaryTokenAdjustment,
} = formatAgentMessages(
stripActivityLabelParts(payload),
this.indexTokenCountMap,
toolSet,
skillPrimeResult?.skills,
formatOptions,
);
if (boundaryTokenAdjustment) {
logger.debug(
`[AgentClient] Boundary token adjustment: ${boundaryTokenAdjustment.original}${boundaryTokenAdjustment.adjusted} (${boundaryTokenAdjustment.remainingChars}/${boundaryTokenAdjustment.totalChars} chars)`,
);
}
/**
* Skill priming — both manual ($ popover) and always-apply (frontmatter).
*
* Splice + index-shift logic lives in `injectSkillPrimes`
* (packages/api/src/agents/skills.ts) so the delicate position math
* can be unit-tested in TS without standing up AgentClient. The
* resolver enforces a combined ceiling (manual-first, always-apply
* truncated first when over cap) before reaching here; the splice
* re-applies the cap as defense-in-depth. Runs for both single-
* agent and multi-agent runs; how primes interact with handoff /
* added-convo agents' per-agent state is an agents-SDK concern,
* not this layer's to gate.
*
* `manualSkillPrimes` / `alwaysApplySkillPrimes` are resolved above
* (used to build `freshSkillPrimeNames` for dedupe against historical
* skill reconstruction).
*/
if (
(manualSkillPrimes && manualSkillPrimes.length > 0) ||
(alwaysApplySkillPrimes && alwaysApplySkillPrimes.length > 0)
) {
const primeResult = injectSkillPrimes({
initialMessages,
indexTokenCountMap,
manualSkillPrimes,
alwaysApplySkillPrimes,
});
indexTokenCountMap = primeResult.indexTokenCountMap;
if (primeResult.inserted > 0) {
const manualNames = (manualSkillPrimes ?? []).map((p) => p.name);
const alwaysApplyNames = (alwaysApplySkillPrimes ?? []).map((p) => p.name);
logger.debug(
`[AgentClient] Primed ${primeResult.inserted} skill(s) at message index ${primeResult.insertIdx} — manual: [${manualNames.join(', ')}], always-apply: [${alwaysApplyNames.join(', ')}]`,
);
}
if (primeResult.alwaysApplyDropped > 0) {
logger.warn(
`[AgentClient] Dropped ${primeResult.alwaysApplyDropped} always-apply prime(s) to stay within MAX_PRIMED_SKILLS_PER_TURN.`,
);
}
}
if (indexTokenCountMap && isEnabled(process.env.AGENT_DEBUG_LOGGING)) {
const entries = Object.entries(indexTokenCountMap);
const perMsg = entries.map(([idx, count]) => {
const msg = initialMessages[Number(idx)];
const type = msg ? msg._getType() : '?';
return `${idx}:${type}=${count}`;
});
logger.debug(
`[AgentClient] Token map after format: [${perMsg.join(', ')}] (payload=${payload.length}, formatted=${initialMessages.length})`,
);
}
indexTokenCountMap = hydrateMissingIndexTokenCounts({
messages: initialMessages,
indexTokenCountMap,
tokenCounter,
});
const memoryMessages =
this.processMemory && this.memoryPayload
? formatAgentMessages(
stripActivityLabelParts(this.memoryPayload),
undefined,
toolSet,
skillPrimeResult?.skills,
formatOptions,
).messages
: initialMessages;
/**
* @param {BaseMessage[]} messages
*/
const runAgents = async (messages) => {
const agents = [this.options.agent];
// Include additional agents when:
// - agentConfigs has agents (from addedConvo parallel execution or agent handoffs)
// - Agents without incoming edges become start nodes and run in parallel automatically
if (this.agentConfigs && this.agentConfigs.size > 0) {
agents.push(...this.agentConfigs.values());
}
// TODO: needs to be added as part of AgentContext initialization
// const noSystemModelRegex = [/\b(o1-preview|o1-mini|amazon\.titan-text)\b/gi];
// const noSystemMessages = noSystemModelRegex.some((regex) =>
// agent.model_parameters.model.match(regex),
// );
// if (noSystemMessages === true && systemContent?.length) {
// const latestMessageContent = _messages.pop().content;
// if (typeof latestMessageContent !== 'string') {
// latestMessageContent[0].text = [systemContent, latestMessageContent[0].text].join('\n');
// _messages.push(new HumanMessage({ content: latestMessageContent }));
// } else {
// const text = [systemContent, latestMessageContent].join('\n');
// _messages.push(new HumanMessage(text));
// }
// }
// let messages = _messages;
// if (agent.useLegacyContent === true) {
// messages = formatContentStrings(messages);
// }
// if (
// agent.model_parameters?.clientOptions?.defaultHeaders?.['anthropic-beta']?.includes(
// 'prompt-caching',
// )
// ) {
// messages = addCacheControl(messages);
// }
if (this.processMemory) {
memoryPromise = this.runMemory(memoryMessages);
}
/** Seed calibration state from previous run if encoding matches */
const currentEncoding = this.getEncoding();
const prevMeta = this.contextMeta;
const encodingMatch = prevMeta?.encoding === currentEncoding;
const calibrationRatio =
encodingMatch && prevMeta?.calibrationRatio > 0 ? prevMeta.calibrationRatio : undefined;
if (prevMeta) {
logger.debug(
`[AgentClient] contextMeta from parent: ratio=${prevMeta.calibrationRatio}, encoding=${prevMeta.encoding}, current=${currentEncoding}, seeded=${calibrationRatio ?? 'none'}`,
);
}
const streamId = this.options.req?._resumableStreamId;
// HITL: establish an empty checkpoint barrier for THIS immutable generation
// before exposing its graph. A retried/recovered initialization may have left
// partial state in the same saver scope; a predecessor uses a different scope,
// so even a late remote write cannot be rehydrated or deleted here. No-op when
// HITL is off or the generation has no remnants. Deliberately unconditional
// per HITL turn: any cheaper Redis flag can go stale across replicas/restarts,
// while these are two indexed, usually-empty deleteMany operations.
// The gate mirrors createRun's checkpointer condition: the approval policy
// OR an ask_user_question-capable agent (which attaches a checkpointer
// WITHOUT the approval policy).
//
// Start the prune alongside graph construction. The all-settled barrier
// below still guarantees it completes before the graph is exposed or run.
const shouldPruneCheckpoint =
streamId &&
(isHITLEnabled(agentsEConfig?.toolApproval) || agents.some(agentRequestsAskUserQuestion));
let checkpointPrunePromise = Promise.resolve();
if (shouldPruneCheckpoint && this.checkpointNamespace !== '') {
checkpointPrunePromise = deleteAgentCheckpoint(
this.conversationId,
agentsEConfig?.checkpointer,
undefined,
{
throwOnError: true,
checkpointNamespace: this.checkpointNamespace,
},
);
} else if (shouldPruneCheckpoint) {
checkpointPrunePromise = captureAgentCheckpointGeneration(
this.conversationId,
agentsEConfig?.checkpointer,
{ throwOnError: true },
).then(async (checkpointGeneration) => {
/** Legacy jobs share LangGraph's root/nested namespaces. Capture
* their immutable ids first, then prove this client still owns
* the exact job epoch before deleting that set. If a replacement
* arrived before/during capture the check fails; if it arrives
* after the check, its newly-written checkpoint ids are outside
* the snapshot and therefore cannot be deleted. */
const liveJob = await GenerationJobManager.getJobStore().getJob(streamId);
if (
!liveJob ||
liveJob.createdAt !== this.jobCreatedAt ||
liveJob.status !== 'running'
) {
throw new Error('Generation replaced before legacy checkpoint cleanup');
}
await deleteAgentCheckpoint(
this.conversationId,
agentsEConfig?.checkpointer,
checkpointGeneration,
{ throwOnError: true },
);
});
}
const activityLabel = this.buildActivityLabelWiring(streamId, abortController.signal);
const activityPhase = this.buildActivityPhaseWiring(streamId, abortController.signal);
const offsetHandlers = createSteerIndexOffsetHandlers(
this.options.eventHandlers,
this.steerOffsetState,
);
const createRunPromise = createRun({
agents,
messages,
// This controller implements the full HITL pause/resume lifecycle (handleRunInterrupt
// persists the pending action; the /resume route rebuilds + continues the run), so it
// opts into the tool-approval wiring. Non-resumable callers (OpenAI-compat, Responses)
// leave this off so an approval-gated tool can't pause where there's no resume path.
hitlCapable: true,
toolInputValidationErrors: this.toolInputValidationErrors,
// Mid-run steering: drain queued user messages at each tool-batch
// boundary and inject them into graph state. The offset wrapper
// shifts SDK content indices past any spliced steer parts.
steering: this.buildSteerWiring(streamId),
activityLabel,
activityPhase,
indexTokenCountMap,
initialSummary,
initialSessions,
calibrationRatio,
runId: this.responseMessageId,
signal: abortController.signal,
/** The phase wrapper stays outermost: it claims and offsets the
* parent slot before the text step reaches the normal handlers. */
customHandlers:
activityPhase?.handlers(offsetHandlers) ??
(activityLabel ? createAssistantPhaseStampingHandlers(offsetHandlers) : offsetHandlers),
requestBody: config.configurable.requestBody,
user: createSafeUser(this.options.req?.user),
tenantId: this.options.req?.user?.tenantId,
summarizationConfig: appConfig?.summarization,
appConfig,
tokenCounter,
/** Bills subagent child-run model calls — child graphs execute
* outside the streamEvents loop, so ModelEndHandler never sees
* them. Entries land in collectedUsage tagged
* `usage_type: 'subagent'` and are spent by recordCollectedUsage.
* The sink also streams each as an `on_token_usage` event so the
* gauge's session cost/totals include billed subagent usage (the
* `subagent` tag keeps it out of the live context meter). */
subagentUsageSink: createSubagentUsageSink(
this.collectedUsage,
this.buildSubagentUsageEmitter(appConfig),
),
}).then((createdRun) => {
if (!createdRun) {
throw new Error('Failed to create run');
}
this.options.startupTelemetry?.mark('run_created');
return createdRun;
});
const [createRunResult, checkpointPruneResult] = await Promise.allSettled([
createRunPromise,
checkpointPrunePromise,
]);
if (createRunResult.status === 'rejected') {
throw createRunResult.reason;
}
if (checkpointPruneResult.status === 'rejected') {
throw checkpointPruneResult.reason;
}
run = createRunResult.value;
this.run = run;
if (this._resolveRun) {
this._resolveRun(run);
this._resolveRun = null;
}
if (streamId && run.Graph) {
GenerationJobManager.setGraph(streamId, run.Graph, this.jobCreatedAt);
}
if (userMCPAuthMap != null) {
config.configurable.userMCPAuthMap = userMCPAuthMap;
}
/** @deprecated Agent Chain */
config.configurable.last_agent_id = agents[agents.length - 1].id;
this.options.startupTelemetry?.mark('stream_processing_started');
/** Flag durable BEFORE the run can claim a label: gap reconciliation
* is gated on it, and ordering it here (one settled-on-failure
* await) keeps the claim-time reservation emit immediate — see
* `emitLabelEvent` in buildActivityLabelWiring. */
if (this.activityLabelsMarkedPromise != null) {
await this.activityLabelsMarkedPromise;
}
await run.processStream({ messages }, config, {
callbacks: {
[Callback.TOOL_ERROR]: logToolError,
},
});
// HITL: if the run paused for tool approval, mark the job
// `requires_action` + emit the prompt and leave the turn unfinalized
// (the resume route continues it). No-op when the run completed.
await this.handleRunInterrupt(run, streamId);
config.signal = null;
};
this.options.startupTelemetry?.mark('run_input_prepared');
await runAgents(initialMessages);
/**
* Surface a completed `skill` tool_call content part per *manually*-
* primed skill so the existing `SkillCall` frontend renderer shows
* a "Skill X loaded" card on the assistant response. Applied after
* the graph finishes to avoid clashing with the aggregator's own
* per-step content indexing. Prepended (not appended) so cards sit
* above the model's output — priming ran before the turn, the
* reply follows.
*
* Always-apply primes intentionally do NOT emit assistant-side
* cards. `extractInvokedSkillsFromPayload` scans history for
* `skill` tool_calls and feeds `primeInvokedSkills`, which is
* Phase 3's sticky-re-prime path — that's the right behavior for
* manual (user picked `$skill` once; re-prime on every subsequent
* turn from history). For always-apply, `resolveAlwaysApplySkills`
* already re-primes every turn from fresh DB state, so persisting
* the card would cause the skill body to get primed twice per
* turn starting on turn 2. The user-facing acknowledgement for
* always-apply lives on the user bubble as the pinned
* `SkillPills` row (`message.alwaysAppliedSkills`), which
* is the durable signal the user wants: "this skill auto-primes".
*
* Live streaming display of manual user-bubble pills is handled
* by `SkillPills` reading `message.manualSkills`. No
* separate SSE emit is needed here; trying to stream a mid-run
* tool_call at index 0 collided with the LLM's first text
* content, while emitting at a sparse offset pushed the card
* below the reply on finalize. Post-run unshift keeps the final
* responseMessage.content in the right order.
*/
/**
* Settle in-flight label fills BEFORE the content is reshaped below.
* A fill emits its claim-time index; the skill-card unshift and the
* hide-sequential filter both shift positions, so a fill landing after
* either would emit a stale index — and a client that already synced
* the reshaped array applies it onto the wrong part. A paused turn
* never gets a final event to repair that. Costs nothing extra: these
* are the same promises the finalization settle would wait on, and
* that later call then sees an empty pending list.
*/
await this.settleActivityLabels();
const contentBeforeReshape = [...this.contentParts];
const manualPrimed = this.options.agent?.manualSkillPrimes ?? [];
if (manualPrimed.length > 0) {
const runId = this.responseMessageId ?? 'skill-prime';
const manualParts = buildSkillPrimeContentParts(manualPrimed, { runId });
this.contentParts.unshift(...manualParts);
}
this.applyHideSequentialOutputsFilter();
this.rebaseActivityPhaseBounds(contentBeforeReshape);
} catch (err) {
if (abortController.signal.aborted) {
logger.debug(
'[api/server/controllers/agents/client.js #sendCompletion] Operation aborted by user',
{ conversationId: this.conversationId, name: err?.name, code: err?.code },
);
} else {
logger.error(
'[api/server/controllers/agents/client.js #sendCompletion] Unhandled error type',
err,
);
const videoError = resolveGoogleVideoError({
error: err,
provider: this.options.agent?.provider,
hasYouTubeVideo: this.injectedYouTubeVideo,
});
this.contentParts.push({
type: ContentTypes.ERROR,
[ContentTypes.ERROR]:
videoError ??
`An error occurred while processing the request${err?.message ? `: ${err.message}` : ''}`,
});
}
} finally {
/** Capture calibration state from the run for persistence on the response message.
* Runs in finally so values are captured even on abort. */
const ratio = this.run?.getCalibrationRatio() ?? 0;
if (ratio > 0 && ratio !== 1) {
this.contextMeta = {
calibrationRatio: Math.round(ratio * 1000) / 1000,
encoding: this.getEncoding(),
};
} else {
this.contextMeta = undefined;
}
this.finalizeSubagentContent();
await this.settleActivityLabels();
/** Flush subagent usage emits the sink fired without awaiting, so their
* persist/publish completes before we return and the job is cleaned up
* (resumed clients read this persisted usage). */
if (this.pendingSubagentEmits.length > 0) {
await Promise.allSettled(this.pendingSubagentEmits);
this.pendingSubagentEmits = [];
}
try {
const attachments = await this.awaitMemoryWithTimeout(memoryPromise);
if (attachments && attachments.length > 0) {
this.artifactPromises.push(...attachments);
}
/** Skip token spending if aborted - the abort handler (abortMiddleware.js) handles it
This prevents double-spending when user aborts via `/api/agents/chat/abort` */
const wasAborted = abortController?.signal?.aborted;
if (!wasAborted) {
await this.recordCollectedUsage({
context: 'message',
balance: balanceConfig,
transactions: transactionsConfig,
});
} else {
logger.debug(
'[api/server/controllers/agents/client.js #chatCompletion] Skipping token spending - handled by abort middleware',
);
}
} catch (err) {
logger.error(
'[api/server/controllers/agents/client.js #chatCompletion] Error in cleanup phase',
err,
);
}
if (this._resolveRun) {
this._resolveRun(this.run ?? null);
this._resolveRun = null;
}
// HITL: a non-paused turn deliberately prunes nothing here. The lazy checkpointer
// (LazyMongoSaver) never persists a clean-exit checkpoint, so there is
// nothing this turn left to delete. Terminal HITL owners eagerly delete their exact
// saver scope, with the Mongo TTL as the backstop for a crashed owner. Dropping a
// generic post-completion prune also removes its
// job-replacement race: an older run's late finally can no longer delete a newer
// paused run's checkpoint, because there is no longer a clean-path prune to race.
run = null;
config = null;
memoryPromise = null;
}
}
/**
* Resume a run that paused for human-in-the-loop review.
*
* The original run lives in a detached background task that exits when the run
* pauses, so resume REBUILDS the run on a fresh graph bound to the same
* `thread_id` (= conversationId), immutable saver scope, and durable checkpointer.
* LangGraph rehydrates the paused graph state from that scoped checkpoint;
* `run.resume(value)` re-enters the interrupted node with the user's decision.
* State comes from the checkpoint, so no message history is rebuilt here —
* `createRun` only needs the agent(s) to reconstruct the graph structure.
*
* `seedContent` is the content streamed before the pause (the assistant message +
* its tool call). In Redis mode the job store's append log already spans the pause,
* so the finalized message is complete regardless; seeding keeps the in-memory store
* complete too. The run drives events through the same `streamId`, so the client's
* open SSE receives the continuation live.
*
* Unlike `chatCompletion`, this does NOT prune the checkpoint in its `finally` — the
* resume controller owns checkpoint lifecycle (it must also clean up on failures that
* happen before this method runs, and keep the checkpoint on a re-pause).
*
* @param {object} params
* @param {Agents.ToolApprovalDecisionMap | { answer: string }} params.resumeValue
* @param {Array} [params.seedContent] - content aggregated before the pause
* @param {Array<import('@librechat/agents').RunStep>} [params.runSteps] - run steps emitted before the pause
* @param {import('@librechat/api').ActivityPhaseSnapshot} [params.activityPhaseSnapshot]
* @param {AbortController} [params.abortController]
* @param {Pick<import('@langchain/langgraph').Command, 'update' | 'goto'>} [params.commandOptions]
*/
async resumeCompletion({
resumeValue,
seedContent = [],
runSteps = [],
abortController = null,
commandOptions,
userMCPAuthMap,
discoveredToolNames,
activityPhaseSnapshot,
}) {
/** @type {Partial<GraphRunnableConfig>} */
let config;
/** @type {ReturnType<createRun>} */
let run;
const appConfig = this.options.req.config;
const balanceConfig = getBalanceConfig(appConfig);
const transactionsConfig = getTransactionsConfig(appConfig);
try {
if (!abortController) {
abortController = new AbortController();
}
/** @type {AppConfig['endpoints']['agents']} */
const agentsEConfig = appConfig.endpoints?.[EModelEndpoint.agents];
config = {
runName: 'AgentRun',
configurable: {
thread_id: this.conversationId,
checkpoint_ns: '',
[LIBRECHAT_CHECKPOINT_NAMESPACE_KEY]: this.checkpointNamespace,
last_agent_index: this.agentConfigs?.size ?? 0,
user_id: this.user ?? this.options.req.user?.id,
hide_sequential_outputs: this.options.agent.hide_sequential_outputs,
requestBody: {
messageId: this.responseMessageId,
conversationId: this.conversationId,
parentMessageId: this.parentMessageId,
},
user: createSafeUser(this.options.req.user),
},
recursionLimit: resolveRecursionLimit(agentsEConfig, this.options.agent),
signal: abortController.signal,
streamMode: 'values',
version: 'v2',
};
// Seed pre-pause content so the in-memory job store reports the complete turn
// (Redis aggregates across the pause via its append log; this covers in-memory).
if (Array.isArray(seedContent) && seedContent.length > 0) {
this.contentParts.push(...seedContent);
}
const tokenCounter = createTokenCounter(this.getEncoding());
const agents = [this.options.agent];
if (this.agentConfigs && this.agentConfigs.size > 0) {
agents.push(...this.agentConfigs.values());
}
// Re-prime skill files invoked in the pre-pause segment (mirrors the normal path's
// `primeInvokedSkills(payload)`), so an approved code/file-backed tool keeps the
// injected skill-file session refs instead of running without them. The pre-pause
// content carries the `skill` tool_calls, so it stands in for the message payload.
let skillSessions;
if (
typeof this.options.primeInvokedSkills === 'function' &&
Array.isArray(seedContent) &&
seedContent.length > 0
) {
try {
const primed = await this.options.primeInvokedSkills([
{ role: 'assistant', content: seedContent },
]);
skillSessions = primed?.initialSessions;
} catch (err) {
logger.warn(
'[api/server/controllers/agents/client.js #resumeCompletion] Failed to re-prime skill sessions',
err?.message ?? err,
);
}
}
// Seed code-env / skill tool sessions so an approved code/file/skill-backed tool
// runs with the same uploaded-file context the pre-pause run had — the rebuilt
// graph otherwise has no `Graph.sessions` entries (especially cross-replica).
const initialSessions = buildInitialToolSessions({ skillSessions, agents });
const streamId = this.options.req?._resumableStreamId;
const activityLabel = this.buildActivityLabelWiring(streamId, abortController.signal);
const activityPhase = this.buildActivityPhaseWiring(
streamId,
abortController.signal,
activityPhaseSnapshot,
);
const offsetHandlers = createSteerIndexOffsetHandlers(
createContentIndexOffsetHandlers(
this.options.eventHandlers,
Array.isArray(seedContent) ? seedContent : [],
),
this.steerOffsetState,
);
run = await createRun({
agents,
// State (messages, tool calls) is rehydrated from the checkpoint by
// run.resume; createRun only needs the agents to rebuild the graph.
messages: [],
// The resumed run can pause AGAIN (another tool, a follow-up question), and this
// controller owns that lifecycle, so it must keep the HITL wiring on the rebuilt run.
hitlCapable: true,
toolInputValidationErrors: this.toolInputValidationErrors,
// Steering stays live across a pause/resume cycle: steers queued while
// the resumed segment runs drain at its tool-batch boundaries.
steering: this.buildSteerWiring(streamId),
// Activity labels likewise survive pause/resume: post-resume tool
// batches keep claiming slots and generating group headers.
activityLabel,
activityPhase,
// Replay deferred tools discovered before the pause. With `messages: []` the
// discovery scan finds nothing, so these names restore the schemas to the
// rebuilt model binding. Undefined/empty for non-deferred turns is a no-op.
discoveredToolNames,
initialSessions,
runId: this.responseMessageId,
signal: abortController.signal,
// The rebuilt graph numbers content indices from 0, but the aggregator was
// just seeded with the pre-pause parts at those same indices — shift every
// resumed step index past the seed, or the new output merges into (or, on a
// type mismatch, is silently dropped against) the pre-pause content. The
// steer wrapper composes on top: resumed indices shift by seed + any
// steer parts spliced in while the resumed segment streams.
customHandlers:
activityPhase?.handlers(offsetHandlers) ??
(activityLabel ? createAssistantPhaseStampingHandlers(offsetHandlers) : offsetHandlers),
requestBody: config.configurable.requestBody,
user: createSafeUser(this.options.req?.user),
tenantId: this.options.req?.user?.tenantId,
summarizationConfig: appConfig?.summarization,
appConfig,
tokenCounter,
subagentUsageSink: createSubagentUsageSink(
this.collectedUsage,
this.buildSubagentUsageEmitter(appConfig),
),
});
if (!run) {
throw new Error('Failed to create run for resume');
}
hydrateResumeRunSteps(runSteps, this.stepMap, run.Graph, seedContent);
this.run = run;
if (this._resolveRun) {
this._resolveRun(run);
this._resolveRun = null;
}
// Do NOT cache the rebuilt graph on resume: it was created with `messages: []`, so
// RedisJobStore.getContentParts() (which prefers a cached graph over reconstructing
// from the chunk log) would return only the resumed segment and drop the pre-pause
// assistant/tool-call content on a same-replica reload/status poll. Skipping it makes
// introspection fall back to the durable chunk reconstruction, which is complete.
// `setContentParts` still points the in-memory store at the seeded client content.
if (streamId && this.contentParts) {
GenerationJobManager.setContentParts(streamId, this.contentParts, this.jobCreatedAt);
}
// Carry the user's MCP auth into the rebuilt run so an approved MCP tool executes
// with the same OAuth/user credentials it had before the pause.
if (userMCPAuthMap != null) {
config.configurable.userMCPAuthMap = userMCPAuthMap;
}
/** @deprecated Agent Chain */
config.configurable.last_agent_id = agents[agents.length - 1].id;
/** Same flag-before-run ordering as chatCompletion's processStream. */
if (this.activityLabelsMarkedPromise != null) {
await this.activityLabelsMarkedPromise;
}
await run.resume(
resumeValue,
config,
{ callbacks: { [Callback.TOOL_ERROR]: logToolError } },
commandOptions,
);
config.signal = null;
// The model may pause AGAIN (another tool needs approval, or a follow-up
// question). Re-arm the same interrupt gate so the cycle can repeat.
await this.handleRunInterrupt(run, streamId);
// Mirror chatCompletion: settle label fills before the filter below can
// shift part positions out from under an in-flight fill's claimed index.
await this.settleActivityLabels();
// Strip hidden intermediate sequential-agent content
// before resume finalize/re-pause persistence reads `this.contentParts`, so a
// resumed sequential chain doesn't persist/emit outputs hide_sequential_outputs
// is meant to hide.
const contentBeforeReshape = [...this.contentParts];
this.applyHideSequentialOutputsFilter();
this.rebaseActivityPhaseBounds(contentBeforeReshape);
} catch (err) {
if (abortController.signal.aborted) {
logger.debug(
'[api/server/controllers/agents/client.js #resumeCompletion] Aborted by user',
{
conversationId: this.conversationId,
name: err?.name,
code: err?.code,
},
);
} else {
logger.error(
'[api/server/controllers/agents/client.js #resumeCompletion] Unhandled error',
err,
);
this.contentParts.push({
type: ContentTypes.ERROR,
[ContentTypes.ERROR]: `An error occurred while resuming the request${err?.message ? `: ${err.message}` : ''}`,
});
}
} finally {
const ratio = this.run?.getCalibrationRatio() ?? 0;
if (ratio > 0 && ratio !== 1) {
this.contextMeta = {
calibrationRatio: Math.round(ratio * 1000) / 1000,
encoding: this.getEncoding(),
};
} else {
this.contextMeta = undefined;
}
this.finalizeSubagentContent();
await this.settleActivityLabels();
if (this.pendingSubagentEmits.length > 0) {
await Promise.allSettled(this.pendingSubagentEmits);
this.pendingSubagentEmits = [];
}
try {
const wasAborted = abortController?.signal?.aborted;
if (!wasAborted) {
await this.recordCollectedUsage({
context: 'message',
balance: balanceConfig,
transactions: transactionsConfig,
});
}
} catch (err) {
logger.error(
'[api/server/controllers/agents/client.js #resumeCompletion] Error in cleanup phase',
err,
);
}
if (this._resolveRun) {
this._resolveRun(this.run ?? null);
this._resolveRun = null;
}
run = null;
config = null;
}
}
/**
* Resolves with the agent run once it is initialized, or `null` if
* initialization fails. Lets immediate-mode title generation await the run
* instead of throwing when fired before `chatCompletion` assigns `this.run`.
* Rejects promptly if the provided signal aborts before the run is ready.
* @param {AbortSignal} [signal]
* @returns {Promise<AgentRun | null>}
*/
_waitForRun(signal) {
if (this.run) {
return Promise.resolve(this.run);
}
if (!this._runReady) {
this._runReady = new Promise((resolve) => {
this._resolveRun = resolve;
});
}
if (!signal) {
return this._runReady;
}
if (signal.aborted) {
return Promise.reject(new Error('Aborted before run initialization'));
}
return new Promise((resolve, reject) => {
const onAbort = () => reject(new Error('Aborted before run initialization'));
signal.addEventListener('abort', onAbort, { once: true });
this._runReady.then((run) => {
signal.removeEventListener('abort', onAbort);
resolve(run);
});
});
}
/**
* @param {Object} params
* @param {string} params.text
* @param {AbortController} params.abortController
* @param {boolean} [params.immediate] When true, the title is generated as soon
* as the request is made — the run is awaited (instead of throwing) and the
* title derives from the user's input only (`contentParts` is empty).
*/
async titleConvo({ text, abortController, immediate = false }) {
if (!this.run) {
if (!immediate) {
throw new Error('Run not initialized');
}
await this._waitForRun(abortController?.signal);
if (!this.run) {
logger.debug(
'[api/server/controllers/agents/client.js #titleConvo] Run unavailable for immediate title generation',
);
return;
}
}
const { handleLLMEnd, collected: collectedMetadata } = createMetadataAggregator();
const { req, agent } = this.options;
if (req?.body?.isTemporary) {
logger.debug(
`[api/server/controllers/agents/client.js #titleConvo] Skipping title generation for temporary conversation`,
);
return;
}
const appConfig = req.config;
let endpoint = agent.endpoint;
/** @type {import('@librechat/agents').ClientOptions} */
let clientOptions = {
model: agent.model || agent.model_parameters.model,
};
let titleProviderConfig = getProviderConfig({ provider: endpoint, appConfig });
/** @type {TEndpoint | undefined} */
const endpointConfig =
appConfig.endpoints?.all ??
appConfig.endpoints?.[endpoint] ??
titleProviderConfig.customEndpointConfig;
if (!endpointConfig) {
logger.debug(
`[api/server/controllers/agents/client.js #titleConvo] No endpoint config for "${endpoint}"`,
);
}
if (endpointConfig?.titleConvo === false) {
logger.debug(
`[api/server/controllers/agents/client.js #titleConvo] Title generation disabled for endpoint "${endpoint}"`,
);
return;
}
if (endpointConfig?.titleEndpoint && endpointConfig.titleEndpoint !== endpoint) {
try {
titleProviderConfig = getProviderConfig({
provider: endpointConfig.titleEndpoint,
appConfig,
});
endpoint = endpointConfig.titleEndpoint;
} catch (error) {
logger.warn(
`[api/server/controllers/agents/client.js #titleConvo] Error getting title endpoint config for "${endpointConfig.titleEndpoint}", falling back to default`,
error,
);
// Fall back to original provider config
endpoint = agent.endpoint;
titleProviderConfig = getProviderConfig({ provider: endpoint, appConfig });
}
}
if (
endpointConfig &&
endpointConfig.titleModel &&
endpointConfig.titleModel !== Constants.CURRENT_MODEL
) {
clientOptions.model = endpointConfig.titleModel;
}
const options = await titleProviderConfig.getOptions({
req,
endpoint,
model_parameters: clientOptions,
db: {
getUserKey: db.getUserKey,
getUserKeyValues: db.getUserKeyValues,
},
});
let provider = options.provider ?? titleProviderConfig.overrideProvider ?? agent.provider;
if (
endpoint === EModelEndpoint.azureOpenAI &&
options.llmConfig?.azureOpenAIApiInstanceName == null
) {
provider = Providers.OPENAI;
} else if (
endpoint === EModelEndpoint.azureOpenAI &&
options.llmConfig?.azureOpenAIApiInstanceName != null &&
provider !== Providers.AZURE
) {
provider = Providers.AZURE;
}
/** @type {import('@librechat/agents').ClientOptions} */
clientOptions = { ...options.llmConfig };
if (options.configOptions) {
clientOptions.configuration = options.configOptions;
}
if (clientOptions.maxTokens != null) {
delete clientOptions.maxTokens;
}
if (clientOptions?.modelKwargs?.max_completion_tokens != null) {
delete clientOptions.modelKwargs.max_completion_tokens;
}
if (clientOptions?.modelKwargs?.max_output_tokens != null) {
delete clientOptions.modelKwargs.max_output_tokens;
}
/** `omitTitleOptions` drops the Anthropic `clientOptions` carrier (thinking,
* streaming, etc.), which would also drop its `defaultHeaders` — preserve the
* original `clientOptions` object so gateway/reverse-proxy metadata still
* reaches title requests (the proxy may require it for auth/routing). Restore
* the SAME object reference, not a copy: the Vertex `createClient` closure from
* `getLLMConfig` closes over this object, so `resolveConfigHeaders` must mutate
* the very object the client is built from. */
const anthropicClientOptions = clientOptions?.clientOptions;
clientOptions = Object.assign(
Object.fromEntries(
Object.entries(clientOptions).filter(([key]) => !omitTitleOptions.has(key)),
),
);
if (anthropicClientOptions?.defaultHeaders != null && clientOptions.clientOptions == null) {
clientOptions.clientOptions = anthropicClientOptions;
}
if (
provider === Providers.GOOGLE &&
(endpointConfig?.titleMethod === TitleMethod.FUNCTIONS ||
endpointConfig?.titleMethod === TitleMethod.STRUCTURED)
) {
clientOptions.json = true;
}
/** Resolve request-based headers across provider-specific header locations:
* OpenAI `configuration.defaultHeaders`, Anthropic `clientOptions.defaultHeaders`
* (preserved above), and Google `customHeaders`. Uses the `req` captured at
* entry — `disposeClient` nulls `this.options.req` and can race this async
* title flow, which would blank the user context mid-generation.
*/
resolveConfigHeaders({
llmConfig: clientOptions,
user: createSafeUser(req?.user),
body: {
messageId: this.responseMessageId,
conversationId: this.conversationId,
parentMessageId: this.parentMessageId,
},
});
try {
const titleResult = await this.run.generateTitle({
provider,
clientOptions,
inputText: text,
contentParts: immediate ? [] : this.contentParts,
titleMethod: endpointConfig?.titleMethod,
titlePrompt: endpointConfig?.titlePrompt,
titlePromptTemplate: endpointConfig?.titlePromptTemplate,
chainOptions: {
runName: 'TitleRun',
signal: abortController.signal,
callbacks: [
{
handleLLMEnd,
},
],
configurable: {
thread_id: this.conversationId,
user_id: this.user ?? this.options.req.user?.id,
},
},
});
const collectedUsage = collectedMetadata.map((item) => {
let input_tokens, output_tokens;
if (item.usage) {
input_tokens =
item.usage.prompt_tokens || item.usage.input_tokens || item.usage.inputTokens;
output_tokens =
item.usage.completion_tokens || item.usage.output_tokens || item.usage.outputTokens;
} else if (item.tokenUsage) {
input_tokens = item.tokenUsage.promptTokens;
output_tokens = item.tokenUsage.completionTokens;
} else if (item.usage_metadata) {
input_tokens = item.usage_metadata.input_tokens;
output_tokens = item.usage_metadata.output_tokens;
}
return {
input_tokens: input_tokens,
output_tokens: output_tokens,
};
});
const balanceConfig = getBalanceConfig(appConfig);
const transactionsConfig = getTransactionsConfig(appConfig);
await this.recordCollectedUsage({
collectedUsage,
context: 'title',
model: clientOptions.model,
balance: balanceConfig,
transactions: transactionsConfig,
messageId: this.responseMessageId,
}).catch((err) => {
logger.error(
'[api/server/controllers/agents/client.js #titleConvo] Error recording collected usage',
err,
);
});
return sanitizeTitle(titleResult.title);
} catch (err) {
logger.error('[api/server/controllers/agents/client.js #titleConvo] Error', err);
return;
}
}
/**
* @param {object} params
* @param {number} params.promptTokens
* @param {number} params.completionTokens
* @param {string} [params.model]
* @param {OpenAIUsageMetadata} [params.usage]
* @param {AppConfig['balance']} [params.balance]
* @param {string} [params.context='message']
* @returns {Promise<void>}
*/
async recordTokenUsage({
model,
usage,
balance,
promptTokens,
completionTokens,
context = 'message',
}) {
try {
await db.spendTokens(
{
model,
context,
balance,
messageId: this.responseMessageId,
conversationId: this.conversationId,
user: this.user ?? this.options.req.user?.id,
endpointTokenConfig: this.options.endpointTokenConfig,
},
{ promptTokens, completionTokens },
);
if (
usage &&
typeof usage === 'object' &&
'reasoning_tokens' in usage &&
typeof usage.reasoning_tokens === 'number'
) {
await db.spendTokens(
{
model,
balance,
context: 'reasoning',
messageId: this.responseMessageId,
conversationId: this.conversationId,
user: this.user ?? this.options.req.user?.id,
endpointTokenConfig: this.options.endpointTokenConfig,
},
{ completionTokens: usage.reasoning_tokens },
);
}
} catch (error) {
logger.error(
'[api/server/controllers/agents/client.js #recordTokenUsage] Error recording token usage',
error,
);
}
}
/** Anthropic Claude models use a distinct BPE tokenizer; all others default to o200k_base. */
getEncoding() {
if (this.model && this.model.toLowerCase().includes('claude')) {
return 'claude';
}
return 'o200k_base';
}
}
module.exports = AgentClient;