mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-08-04 14:57:42 +00:00
🩹 fix: Address Review Findings on Activity-Label Indexing and Pricing
Replay index (P1). Reserving the slot only in server memory left no event for it, so a cross-instance replay rebuilt content as [tool, hole, later], compacted the hole away, and the fill for the reserved index landed on the following part and overwrote it. The claim now publishes the empty, pending part so the index is real for every consumer, and fill publishes even when generation returned nothing so the client cannot stay pending. It stays invisible: an empty label still DELIMITS its batch in groupSequentialToolCalls but is not attached as the header, so grouping does not re-shuffle when the text lands and the block renders exactly as it does with the feature off. Edited-response index (P1). Edit-and-resubmit replays the kept prefix and the server indexes only new content, so run steps offset by that prefix. Labels are claimed in the same space and now take the identical shift; without it a label could land inside the prefix and overwrite it. Redis flag. deserializeJob never read activityLabels back, so every Redis reload left it undefined and resume skipped label gap reconciliation. Executing agent. RunActivityLabelOptions.agentId selects the executing agent tracing metadata AND its tool-output redaction policy; omitting it let a handoff be redacted under the default agent configuration. Label pricing. An undefined endpointTokenConfig is meaningful for a built-in label endpoint (priced from the shared table), so the nullish fallback billed those labels at a custom primary rates. Inherit only when the label runs on the agent own endpoint. HITL usage sequence. runId is the response message id and the counter was instance-local, so a resume restarted at -1 and the client runId:seq deduper discarded the post-approval label usage. Seeded past the labels already on the response. Also distinguishes "cannot serve" (undefined) from "no label" (null) in the SDK bridge, so a missing run falls back to the direct call instead of filling the slot empty. Version gating already happens at wiring time via the sdkCapable prototype probe.
This commit is contained in:
Danny Avila
parent
e62f50794e
commit
0d7d250520
8 changed files with 229 additions and 44 deletions
|
|
@ -368,7 +368,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) {
|
||||
|
|
@ -377,8 +377,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
|
||||
|
|
@ -443,11 +450,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({
|
||||
|
|
@ -464,6 +486,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. */
|
||||
|
|
@ -485,6 +512,7 @@ class AgentClient extends BaseClient {
|
|||
collectedMetadata,
|
||||
clientOptions.model,
|
||||
endpointTokenConfig,
|
||||
sameEndpoint,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
@ -563,6 +591,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) {
|
||||
|
|
@ -609,11 +646,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,
|
||||
);
|
||||
},
|
||||
};
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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 });
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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, <hole>, 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<LooseContentPart | null | undefined> = [
|
||||
{ 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<LooseContentPart | null | undefined> = [
|
||||
{ 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<LooseContentPart | null | undefined> = [
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<string | null>;
|
||||
generateLabel?: (payload: GenerateLabelPayload) => Promise<string | null | undefined>;
|
||||
/**
|
||||
* 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<string | null> => {
|
||||
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. */
|
||||
|
|
|
|||
|
|
@ -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, <hole>, 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<void>((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();
|
||||
|
|
|
|||
|
|
@ -1831,6 +1831,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,
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue