diff --git a/api/server/controllers/agents/client.js b/api/server/controllers/agents/client.js index 1849cb5c3d..3d9c2152b5 100644 --- a/api/server/controllers/agents/client.js +++ b/api/server/controllers/agents/client.js @@ -370,7 +370,7 @@ class AgentClient extends BaseClient { * 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) { + async recordActivityLabelUsage(collectedMetadata, model, endpointTokenConfig, sameEndpoint) { const appConfig = this.options.req?.config; const collectedUsage = mapCollectedMetadataToUsage(collectedMetadata); if (collectedUsage.length === 0) { @@ -379,8 +379,15 @@ class AgentClient extends BaseClient { 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; fall back to the agent's only when the label runs there. */ - const labelTokenConfig = endpointTokenConfig ?? this.options.endpointTokenConfig; + * 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 @@ -445,11 +452,26 @@ class AgentClient extends BaseClient { * (thread_id) with its own tags — never as an orphan trace. Returns null * when the label could not be generated. */ - async generateActivityLabelViaRun({ entries, context, traceSeed, signal, charLimit, prompt }) { + async generateActivityLabelViaRun({ + entries, + context, + traceSeed, + signal, + charLimit, + prompt, + executingAgentId, + }) { + /** 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 null; + return undefined; } - const { provider, clientOptions, endpointTokenConfig } = await this.resolveActivityLabelLLM(); + const { provider, clientOptions, endpointTokenConfig, sameEndpoint } = + await this.resolveActivityLabelLLM(); const { handleLLMEnd, collected: collectedMetadata } = createMetadataAggregator(); try { const { label } = await this.run.generateActivityLabel({ @@ -466,6 +488,11 @@ class AgentClient extends BaseClient { lastAssistantText: context.lastAssistantText, 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. */ @@ -487,6 +514,7 @@ class AgentClient extends BaseClient { collectedMetadata, clientOptions.model, endpointTokenConfig, + sameEndpoint, ); } } @@ -565,6 +593,15 @@ class AgentClient extends BaseClient { 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; if (abortSignal != null) { if (abortSignal.aborted) { @@ -611,11 +648,13 @@ class AgentClient extends BaseClient { return { callbacks: [{ handleLLMEnd }], collect: async () => { - const { clientOptions, endpointTokenConfig } = await this.resolveActivityLabelLLM(); + const { clientOptions, endpointTokenConfig, sameEndpoint } = + await this.resolveActivityLabelLLM(); await this.recordActivityLabelUsage( collected, clientOptions.model, endpointTokenConfig, + sameEndpoint, ); }, }; diff --git a/client/src/hooks/SSE/useResumableSSE.ts b/client/src/hooks/SSE/useResumableSSE.ts index da39184183..eb816fda12 100644 --- a/client/src/hooks/SSE/useResumableSSE.ts +++ b/client/src/hooks/SSE/useResumableSSE.ts @@ -781,10 +781,11 @@ export default function useResumableSSE( /** * Places an activity-label part at its claimed content index on the - * in-flight response message. Fires twice per block (counts placeholder - * at batch end, fast-model label on resolve); `applyActivityLabelPart` - * is referentially stable on duplicate replays. Same bounded - * next-frame retry as steers for the inject-before-render race. + * in-flight response message. Fires twice per block (the empty + * reservation at batch end, the generated label on resolve); + * `applyActivityLabelPart` is referentially stable on duplicate replays + * and refuses to overwrite filled text with a stale placeholder. Same + * bounded next-frame retry as steers for the inject-before-render race. */ const applyActivityLabelToMessages = (event: TActivityLabelEvent, attempt = 0) => { const retryNextFrame = () => { @@ -805,7 +806,20 @@ export default function useResumableSSE( retryNextFrame(); return; } - const updated = applyActivityLabelPart(messages[index], event); + /** Edit-and-resubmit replays the kept prefix into the response before + * the run starts, and the server indexes only the NEW content — so + * run steps offset by that prefix (`useStepHandler`). The label index + * is claimed in the same server-side space and needs the identical + * shift, or it lands inside the prefix and overwrites kept content. */ + const initialContent = + currentSubmission.editedContent != null + ? ((currentSubmission.initialResponse as TMessage | undefined)?.content ?? []) + : []; + const offsetEvent = + initialContent.length > 0 + ? { ...event, index: event.index + initialContent.length } + : event; + const updated = applyActivityLabelPart(messages[index], offsetEvent); if (updated !== messages[index]) { const nextMessages = [...messages]; nextMessages[index] = updated; diff --git a/client/src/utils/groupToolCalls.ts b/client/src/utils/groupToolCalls.ts index 824a0f3191..9c1b51dd06 100644 --- a/client/src/utils/groupToolCalls.ts +++ b/client/src/utils/groupToolCalls.ts @@ -1,6 +1,7 @@ import { Constants, ContentTypes, ToolCallTypes } from 'librechat-data-provider'; import type { TMessageContentParts, Agents } from 'librechat-data-provider'; import type { PartWithIndex } from '~/components/Chat/Messages/Content/ParallelContent'; +import { getActivityLabelPart, getActivityLabelText } from '~/utils/activityLabels'; export type GroupedPart = | { type: 'single'; part: PartWithIndex } @@ -69,10 +70,19 @@ export function groupSequentialToolCalls(parts: PartWithIndex[]): GroupedPart[] continue; } if (item.part.type === ContentTypes.ACTIVITY_LABEL) { + /** A reserved-but-unfilled slot (and a failed/blank fill) still + * DELIMITS its batch, so grouping does not re-shuffle when the text + * lands — but it is not attached as a labelPart, leaving the group to + * render its generic verb exactly as it would without the feature. */ + const hasText = getActivityLabelText(getActivityLabelPart(item.part)).length > 0; if (currentBlock.length > 0) { - result.push({ type: 'tool-group', parts: [...currentBlock], labelPart: item }); + result.push({ + type: 'tool-group', + parts: [...currentBlock], + ...(hasText && { labelPart: item }), + }); currentBlock = []; - } else { + } else if (hasText) { /** Orphan label (block parts hidden/filtered): renders standalone. */ result.push({ type: 'single', part: item }); } diff --git a/packages/api/src/agents/activityLabels/__tests__/wiring.spec.ts b/packages/api/src/agents/activityLabels/__tests__/wiring.spec.ts index db62b3f39c..5372e2b3d5 100644 --- a/packages/api/src/agents/activityLabels/__tests__/wiring.spec.ts +++ b/packages/api/src/agents/activityLabels/__tests__/wiring.spec.ts @@ -174,22 +174,94 @@ describe('createActivityLabelWiring close gate', () => { await hook(batchInput(), new AbortController().signal); await flushDetached(); - /** The slot is reserved but silent: claiming one emits nothing, so the - * UI is untouched until a real label exists. */ - expect(emitLabelEvent).not.toHaveBeenCalled(); + /** Claiming publishes the reservation so replay cannot compact the index + * away — empty and pending, which renders nothing. */ + expect(emitLabelEvent).toHaveBeenCalledTimes(1); + expect(emitLabelEvent).toHaveBeenCalledWith( + 1, + expect.objectContaining({ activity_label: '', pending: true }), + ); /** Settle timed out: the scope closes, then the straggler resolves. */ closed = true; releaseLabel('Late label that must not land'); await flushDetached(); - expect(emitLabelEvent).not.toHaveBeenCalled(); + /** No SECOND emit: the late fill neither mutates nor publishes. */ + expect(emitLabelEvent).toHaveBeenCalledTimes(1); const labelPart = parts[1] as LooseContentPart; expect(labelPart.activity_label).toBe(''); expect(labelPart.pending).toBe(true); }); }); +describe('createActivityLabelWiring reservation', () => { + /** + * Without a claim-time event the slot exists only in server memory, so a + * cross-instance replay rebuilds [tool, , laterText], compacts the + * hole, and the fill for the reserved index then overwrites `laterText`. + * Publishing the empty part keeps the index real for every consumer. + */ + it('publishes the reserved index before any label exists', async () => { + const parts: Array = [ + { type: 'tool_call', tool_call: { id: 'tool-1' } }, + ]; + const emitted: Array<{ index: number; label: unknown; pending?: boolean }> = []; + const emitLabelEvent = jest.fn(async (index: number, part: LooseContentPart) => { + emitted.push({ index, label: part.activity_label, pending: part.pending }); + return undefined; + }); + const { hook } = createActivityLabelWiring({ + getContentParts: () => parts, + bumpIndexOffset: jest.fn(), + emitLabelEvent, + trackPendingFill: jest.fn(), + resolveLLM: jest.fn(async () => ({ + provider: Providers.OPENAI, + clientOptions: { model: 'm' }, + })), + generateLabel: jest.fn(async () => 'Stored the release notes'), + }); + + await hook(batchInput(), new AbortController().signal); + await flushDetached(); + + /** Reservation first (empty, pending), then the fill at the SAME index. */ + expect(emitted).toEqual([ + { index: 1, label: '', pending: true }, + { index: 1, label: 'Stored the release notes', pending: false }, + ]); + }); + + /** A blank result must still settle the slot, or the client stays pending. */ + it('publishes a settled empty part when generation yields nothing', async () => { + const parts: Array = [ + { type: 'tool_call', tool_call: { id: 'tool-1' } }, + ]; + const emitLabelEvent = jest.fn(async () => undefined); + const { hook } = createActivityLabelWiring({ + getContentParts: () => parts, + bumpIndexOffset: jest.fn(), + emitLabelEvent, + trackPendingFill: jest.fn(), + resolveLLM: jest.fn(async () => ({ + provider: Providers.OPENAI, + clientOptions: { model: 'm' }, + })), + generateLabel: jest.fn(async () => null), + }); + + await hook(batchInput(), new AbortController().signal); + await flushDetached(); + + expect(emitLabelEvent).toHaveBeenCalledTimes(2); + expect(emitLabelEvent).toHaveBeenLastCalledWith( + 1, + expect.objectContaining({ activity_label: '', pending: false }), + ); + }); +}); + describe('createActivityLabelWiring instruction', () => { const runWith = async (prompt?: string) => { const parts: Array = [ diff --git a/packages/api/src/agents/activityLabels/host.ts b/packages/api/src/agents/activityLabels/host.ts index 851c35662c..578abe5af4 100644 --- a/packages/api/src/agents/activityLabels/host.ts +++ b/packages/api/src/agents/activityLabels/host.ts @@ -196,6 +196,7 @@ export async function resolveActivityLabelModel({ clientOptions: clientOptions as ClientOptions, /** Priced with the LABEL endpoint's rates, not the agent's. */ endpointTokenConfig: options.endpointTokenConfig, + sameEndpoint: endpoint === agentEndpoint, }; } diff --git a/packages/api/src/agents/activityLabels/runtime.ts b/packages/api/src/agents/activityLabels/runtime.ts index 1317d26871..becc061285 100644 --- a/packages/api/src/agents/activityLabels/runtime.ts +++ b/packages/api/src/agents/activityLabels/runtime.ts @@ -15,6 +15,13 @@ export interface ActivityLabelLLM { * cross-endpoint label is costed at the wrong rates. */ endpointTokenConfig?: unknown; + /** + * True when the label resolved to the agent's OWN endpoint. Callers need + * this to read an undefined `endpointTokenConfig` correctly: for a built-in + * label endpoint undefined means "price from the shared table", so + * inheriting the agent's custom rates there would misprice the label. + */ + sameEndpoint?: boolean; } /** @@ -69,6 +76,12 @@ export interface GenerateLabelPayload { * never reaches the preferred path. */ prompt?: string; + /** + * Owning agent of the batch. Selects that agent's tracing metadata AND its + * tool-output redaction policy on the SDK path, so a handoff is not traced + * or redacted under the default agent's configuration. + */ + executingAgentId?: string; } /** Per-generation LLM callbacks for usage accounting on the fallback path. */ @@ -88,11 +101,16 @@ export interface ActivityLabelHookOptions { claimSlot: (meta: ActivityLabelBatchMeta) => ActivityLabelSlot; /** * Preferred generation path: host bridges to the SDK's - * `run.generateActivityLabel()` (session-grouped Langfuse tracing). When - * absent — SDK too old — the hook falls back to a direct, untraced model - * call via `resolveLLM`. + * `run.generateActivityLabel()` (session-grouped Langfuse tracing). + * + * Resolve `undefined` to decline — the SDK lacks the API — and the hook + * falls back to a direct, untraced model call via `resolveLLM`. `null` + * means the opposite: this path ran and produced no label, so the slot + * fills empty. Hosts wire this bridge unconditionally (the run does not + * exist yet at construction time), which is why declining has to be + * expressible at call time rather than by omitting the option. */ - generateLabel?: (payload: GenerateLabelPayload) => Promise; + generateLabel?: (payload: GenerateLabelPayload) => Promise; /** * Fallback model resolution for the direct-call path. Memoized here so * hosts can pass a fresh thunk without caching concerns. @@ -297,19 +315,9 @@ export function createActivityLabelHook( * label call — a user abort must not keep paying for generation * until the timeout. */ const signal = buildSignal(opts.signal, hookSignal); - let text: string | null = null; - if (opts.generateLabel != null) { - /** SDK-backed path: session-grouped Langfuse tracing via - * `run.generateActivityLabel()` (host bridges the call). */ - text = await opts.generateLabel({ - entries: input.entries, - context: slot.context ?? {}, - traceSeed: `${input.runId}-activity-${slot.index}`, - signal, - charLimit, - ...(opts.prompt != null && { prompt: opts.prompt }), - }); - } else { + /** Direct, untraced call: the fallback when no SDK bridge is wired or + * when the bridge declines because the package is too old. */ + const generateDirect = async (): Promise => { const { provider, clientOptions } = await getLLM(); const model = initializeModel({ provider, @@ -322,8 +330,28 @@ export function createActivityLabelHook( signal, ...(invokeCallbacks && { callbacks: invokeCallbacks.callbacks }), }); - text = extractText(response?.content); + const direct = extractText(response?.content); await invokeCallbacks?.collect(); + return direct; + }; + + let text: string | null = null; + if (opts.generateLabel != null) { + /** SDK-backed path: session-grouped Langfuse tracing via + * `run.generateActivityLabel()` (host bridges the call). */ + const bridged = await opts.generateLabel({ + entries: input.entries, + context: slot.context ?? {}, + traceSeed: `${input.runId}-activity-${slot.index}`, + signal, + charLimit, + ...(opts.prompt != null && { prompt: opts.prompt }), + ...(input.executingAgentId != null && { executingAgentId: input.executingAgentId }), + }); + /** Declined (no SDK support) — not the same as "no label". */ + text = bridged === undefined ? await generateDirect() : bridged; + } else { + text = await generateDirect(); } /** Trim centrally: a whitespace-only label from either path must * fill null so the UI keeps the deterministic counts fallback. */ diff --git a/packages/api/src/agents/activityLabels/wiring.ts b/packages/api/src/agents/activityLabels/wiring.ts index c3efee18a7..e4486c5de4 100644 --- a/packages/api/src/agents/activityLabels/wiring.ts +++ b/packages/api/src/agents/activityLabels/wiring.ts @@ -251,11 +251,24 @@ export function createActivityLabelWiring(deps: ActivityLabelHostDeps): { }; parts.push(part); deps.bumpIndexOffset(); - /** No claim-time emit. The slot is reserved server-side so indices - * stay stable, but an empty header has nothing to say — emitting it - * would change the UI before the generation it announces exists. - * Until `fill` lands, the client renders the batch exactly as it - * does today. */ + /** + * Publish the reservation immediately, empty and pending. + * + * Reserving the index server-side is not enough on its own: with no + * event for this slot, a cross-instance replay rebuilds content as + * [tool, , laterText] and compacts the hole away, so the fill + * that later arrives for this index lands on `laterText` and + * overwrites it. Publishing the empty part keeps the slot real + * everywhere the content is reconstructed. + * + * It stays invisible: `groupSequentialToolCalls` lets an empty label + * delimit its batch without becoming the header, so the block renders + * exactly as it does with the feature off until `fill` lands. + */ + void Promise.resolve(deps.emitLabelEvent(index, part)).catch(() => { + /** Best-effort: a dropped reservation degrades to the pre-fix + * behavior, and must never break the batch that triggered it. */ + }); let resolveFill: () => void = () => undefined; const fillDone = new Promise((resolve) => { resolveFill = resolve; @@ -272,10 +285,13 @@ export function createActivityLabelWiring(deps: ActivityLabelHostDeps): { return; } part.pending = false; - if (text == null || text.length === 0) { - return; + if (text != null && text.length > 0) { + part[ContentTypes.ACTIVITY_LABEL] = text; } - part[ContentTypes.ACTIVITY_LABEL] = text; + /** Emitted even when generation produced nothing: the claim + * already published a PENDING part, so staying silent here + * would leave the client pinned at pending forever. An empty + * label still renders nothing. */ await deps.emitLabelEvent(index, part); } finally { resolveFill(); diff --git a/packages/api/src/stream/implementations/RedisJobStore.ts b/packages/api/src/stream/implementations/RedisJobStore.ts index 486a4d3824..7f22304e49 100644 --- a/packages/api/src/stream/implementations/RedisJobStore.ts +++ b/packages/api/src/stream/implementations/RedisJobStore.ts @@ -2244,6 +2244,11 @@ export class RedisJobStore implements IJobStore { pendingAction: this.parsePendingAction(data.pendingAction), pendingActionId: data.pendingActionId || undefined, lastActiveAt: data.lastActiveAt ? parseInt(data.lastActiveAt, 10) : undefined, + /** `markActivityLabels` persists this, so it has to be read back: + * without it every Redis reload leaves the flag undefined and resume + * skips activity-label gap reconciliation, silently dropping a label + * that resolved between the snapshot and subscriber attach. */ + activityLabels: data.activityLabels != null ? data.activityLabels === '1' : undefined, }; }