From 0e5c6e4f60ed868dfb323be244812e48d8c4eda2 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Tue, 28 Jul 2026 16:47:27 -0400 Subject: [PATCH] =?UTF-8?q?=E2=8F=B1=EF=B8=8F=20fix:=20Persist=20the=20Lab?= =?UTF-8?q?el=20Flag=20at=20Run=20Start,=20Not=20on=20the=20Emit=20Path?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-twenty-one review: two fixed; the other three repeat the maintainer-decided packages/api extraction, the edited+reconnect index limitation, and the parallel-lane header limitation — all answered on-thread with their standing decisions. - Flag ordering, corrected (client.js): sequencing label emission behind the flag persist (previous round) delayed the claim-time reservation while the shared index offset had ALREADY shifted subsequent SDK chunks — reopening the cross-instance hole-compaction overwrite the reservation emit exists to prevent. The reservation emits immediately again; instead, run start (processStream and resume alike) awaits the settled-on-failure persist chain, so the flag is durable before any batch can claim a label. Same guarantee, zero latency on the emit path. - Tail-label cursor (ContentParts.tsx): a filled label at the content tail is consumed into the group header rather than listed in `group.parts`, so the `isLast` check missed it and nothing held the streaming cursor until the next delta. The check now includes `labelPart.idx`. --- api/server/controllers/agents/client.js | 44 ++++++++++++------- .../Chat/Messages/Content/ContentParts.tsx | 9 +++- 2 files changed, 37 insertions(+), 16 deletions(-) diff --git a/api/server/controllers/agents/client.js b/api/server/controllers/agents/client.js index e8e3cc2dc9..1efa2f5c9b 100644 --- a/api/server/controllers/agents/client.js +++ b/api/server/controllers/agents/client.js @@ -718,10 +718,14 @@ class AgentClient extends BaseClient { * 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): label emission is ORDERED after this - * persist, below. The chain settles on failure (warned retry), so a - * lost write can never wedge label emission. */ - const activityLabelsMarked = GenerationJobManager.markActivityLabels(streamId).catch(() => + /** 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 = GenerationJobManager.markActivityLabels(streamId).catch(() => GenerationJobManager.markActivityLabels(streamId).catch(() => { logger.warn( `[AgentClient] Could not flag activity labels for ${streamId}; a label resolving during a resume gap may not be reconciled.`, @@ -785,15 +789,15 @@ class AgentClient extends BaseClient { bumpIndexOffset: () => { this.steerOffsetState.offset += 1; }, - emitLabelEvent: async (index, part) => { - /** ORDERED after the flag persist: resume-gap reconciliation is - * gated on the flag, so a label event must never exist before the - * flag does — an immediate cross-replica reconnect could otherwise - * read the job between the two writes, see neither flag nor - * snapshot label, and skip reconciling a label claimed in the - * snapshot→subscribe window. */ - await activityLabelsMarked; - return GenerationJobManager.emitChunk( + /** 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, @@ -810,8 +814,7 @@ class AgentClient extends BaseClient { * 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); @@ -2466,6 +2469,13 @@ class AgentClient extends BaseClient { 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, @@ -2817,6 +2827,10 @@ class AgentClient extends BaseClient { /** @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, diff --git a/client/src/components/Chat/Messages/Content/ContentParts.tsx b/client/src/components/Chat/Messages/Content/ContentParts.tsx index 1f00f13208..116f51f379 100644 --- a/client/src/components/Chat/Messages/Content/ContentParts.tsx +++ b/client/src/components/Chat/Messages/Content/ContentParts.tsx @@ -491,7 +491,14 @@ const ContentParts = memo(function ContentParts({ key={`tool-group-${groupId}`} parts={group.parts} isSubmitting={effectiveIsSubmitting} - isLast={group.parts.some((p) => p.idx === lastContentIdx)} + /** The label part is CONSUMED into the header, not listed in + * `parts` — a filled label at the content tail must still + * mark its group as last or nothing holds the streaming + * cursor until the next delta. */ + isLast={ + group.parts.some((p) => p.idx === lastContentIdx) || + group.labelPart?.idx === lastContentIdx + } renderPart={renderGroupedPart} lastContentIdx={lastContentIdx} groupAttachments={group.groupAttachments}