@@ -113,6 +113,7 @@ import {
113113 convertSteeredMessages,
114114 retainStepMessages,
115115 steeringMarkers,
116+ withoutInjectionMarkers,
116117 type SteeringInjection,
117118} from "./steeringContext.js";
118119
@@ -134,6 +135,7 @@ import {
134135 type InferChatClientData,
135136 type InferChatUIMessage,
136137 type InferChatUIMessageFromTools,
138+ BACKGROUND_INJECTION_TYPE,
137139 PENDING_MESSAGE_INJECTED_TYPE,
138140 upsertIncomingMessage,
139141 chatRunTags,
@@ -2901,13 +2903,6 @@ function spliceHandoverPartial(
29012903 * @internal
29022904 */
29032905const chatBackgroundQueueKey = locals.create<ModelMessage[]>("chat.backgroundQueue");
2904- /**
2905- * Background injections a step-boundary drain handed to the model this turn,
2906- * with the transcript message they followed. Reconciled into the model lane
2907- * and the persisted injections once the turn's response is in.
2908- */
2909- const chatPendingBackgroundKey =
2910- locals.create<{ afterId: string; messages: ModelMessage[] }[]>("chat.pendingBackground");
29112906
29122907/**
29132908 * System-role context injected mid-conversation, held for the instructions lane.
@@ -4450,6 +4445,34 @@ function modelFormOf(m: UIMessage, batch: UIMessage[], injected: ModelMessage[])
44504445 return batch[0] === m ? injected : [];
44514446}
44524447
4448+ /**
4449+ * Record a prepared injection and write its marker into the response, so the
4450+ * model lane rebuilds it at the step boundary where the model received it.
4451+ */
4452+ function markInjection(
4453+ steps: CompactionStep[],
4454+ messageIds: string[],
4455+ messages: ModelMessage[],
4456+ markerMessages: Array<{ id: string; text: string }>,
4457+ kind?: "background"
4458+ ): void {
4459+ const id = generateMessageId();
4460+ const injections = locals.get(chatSteeringInjectionsKey) ?? new Map();
4461+ injections.set(id, { id, messageIds, messages, ...(kind ? { kind } : {}) });
4462+ locals.set(chatSteeringInjectionsKey, injections);
4463+ const response = managedResponse();
4464+ // prepareStep can run ahead of the UI consumer. Admit the marker only
4465+ // once the actual preceding finish-step chunk has entered this stream.
4466+ response.afterToolStep(
4467+ steps.at(-1)?.content.flatMap((part) => (part.toolCallId ? [part.toolCallId] : [])) ?? []
4468+ );
4469+ response.writeData({
4470+ type: kind === "background" ? BACKGROUND_INJECTION_TYPE : PENDING_MESSAGE_INJECTED_TYPE,
4471+ id,
4472+ data: { messageIds, messages: markerMessages },
4473+ });
4474+ }
4475+
44534476/**
44544477 * Drain the steering queue as a batch. Calls `shouldInject` once with all
44554478 * pending messages. If it returns true, calls `prepareMessages` once to
@@ -4468,7 +4491,6 @@ async function drainSteeringQueue(
44684491 steps: CompactionStep[],
44694492 queueOverride?: SteeringQueueEntry[]
44704493): Promise<DrainedSteering> {
4471- if (steps.length === 0) managedResponse().beginGeneration();
44724494 const queue = queueOverride ?? locals.get(chatSteeringQueueKey);
44734495 if (!queue || queue.length === 0) return EMPTY_DRAIN;
44744496
@@ -4606,23 +4628,12 @@ async function drainSteeringQueue(
46064628 }
46074629
46084630 if (injected.length > 0) {
4609- const id = generateMessageId();
4610- const messageIds = claimedUIMessages.map((m) => m.id);
4611- const injections = locals.get(chatSteeringInjectionsKey) ?? new Map();
4612- injections.set(id, { id, messageIds, messages: injected });
4613- locals.set(chatSteeringInjectionsKey, injections);
4614- const response = managedResponse();
4615- // prepareStep can run ahead of the UI consumer. Admit the marker only
4616- // once the actual preceding finish-step chunk has entered this stream.
4617- response.afterStep(steps.length);
4618- response.writeData({
4619- type: PENDING_MESSAGE_INJECTED_TYPE,
4620- id,
4621- data: {
4622- messageIds,
4623- messages: claimedUIMessages.map((m) => ({ id: m.id, text: textOfUIMessage(m) })),
4624- },
4625- });
4631+ markInjection(
4632+ steps,
4633+ claimedUIMessages.map((m) => m.id),
4634+ injected,
4635+ claimedUIMessages.map((m) => ({ id: m.id, text: textOfUIMessage(m) }))
4636+ );
46264637 }
46274638
46284639 // Fire onInjected callback
@@ -5439,13 +5450,7 @@ function toStreamTextOptions(
54395450 if (bgQueue && bgQueue.length > 0) {
54405451 const injected = bgQueue.splice(0); // drain
54415452 resultMessages = [...(resultMessages ?? messages), ...injected];
5442- const pendingBackground = locals.get(chatPendingBackgroundKey) ?? [];
5443- pendingBackground.push({
5444- afterId:
5445- (locals.get(chatCurrentUIMessagesKey) as UIMessage[] | undefined)?.at(-1)?.id ?? "",
5446- messages: injected,
5447- });
5448- locals.set(chatPendingBackgroundKey, pendingBackground);
5453+ markInjection(steps, [], injected, [], "background");
54495454 }
54505455
54515456 // 4. Parallel tool results in call order, the order the next turn's history
@@ -9164,23 +9169,6 @@ function chatAgent<
91649169 // durable snapshot + `session.out` replay (or `hydrateMessages` if
91659170 // registered) — the wire is delta-only now, no longer a seed.
91669171 let accumulatedMessages: ModelMessage[] = [];
9167- /**
9168- * Give the model accumulator the background injections a step-boundary
9169- * drain handed to the model this turn, and record them for persistence.
9170- * Returns how many model messages were appended.
9171- */
9172- const reconcilePendingBackground = (): number => {
9173- const pending = locals.get(chatPendingBackgroundKey);
9174- if (!pending || pending.length === 0) return 0;
9175- locals.set(chatPendingBackgroundKey, []);
9176- let appended = 0;
9177- for (const entry of pending) {
9178- accumulatedMessages.push(...entry.messages);
9179- laneInjections.push(entry);
9180- appended += entry.messages.length;
9181- }
9182- return appended;
9183- };
91849172 /**
91859173 * Give the model accumulator the steering messages a drain consumed,
91869174 * in the form the model actually received. Appended, never reconverted
@@ -9211,6 +9199,15 @@ function chatAgent<
92119199 // but in frontend-friendly UIMessage format (with parts, id, etc.).
92129200 let accumulatedUIMessages: TUIMessage[] = [];
92139201
9202+ const cleanIncomingMessage = (msg: TUIMessage): TUIMessage => {
9203+ if (msg.role !== "assistant") return withoutInjectionMarkers(msg);
9204+ const cleaned = cleanupAbortedParts(msg);
9205+ // Markers are the agent's own; only a known assistant message keeps them.
9206+ return accumulatedUIMessages.some((m) => m.id === msg.id)
9207+ ? cleaned
9208+ : withoutInjectionMarkers(cleaned);
9209+ };
9210+
92149211 // ── Snapshot + replay (gated on prior-state signals) ─────────────
92159212 //
92169213 // With `hydrateMessages` registered the customer owns history — the
@@ -10960,9 +10957,8 @@ function chatAgent<
1096010957 // Cleaning happens once here so `extractLastUserMessageText` and
1096110958 // every downstream consumer see the same message shape — and
1096210959 // `cleanupAbortedParts` no longer has to be re-applied below.
10963- const cleanedIncomingMessages: TUIMessage[] = incomingMessages.map((msg) =>
10964- msg.role === "assistant" ? cleanupAbortedParts(msg) : msg
10965- );
10960+ const cleanedIncomingMessages: TUIMessage[] =
10961+ incomingMessages.map(cleanIncomingMessage);
1096610962 const lastUserMessage = extractLastUserMessageText(cleanedIncomingMessages);
1096710963
1096810964 channelTurnRan =
@@ -11036,7 +11032,6 @@ function chatAgent<
1103611032 locals.set(chatOutGateKey, createChatOutGate());
1103711033 locals.set(chatCompactionStateKey, undefined);
1103811034 locals.set(chatSteeringQueueKey, []);
11039- locals.set(chatPendingBackgroundKey, []);
1104011035 await locals.get(chatManagedResponseKey)?.close();
1104111036 locals.set(chatManagedResponseKey, undefined);
1104211037 locals.set(chatManagedResponseActiveKey, true);
@@ -12308,11 +12303,10 @@ function chatAgent<
1230812303 // before the response is appended so the order stays
1230912304 // steer-then-answer. Outside the `capturedResponseMessage`
1231012305 // branches below, so a turn that captured no response is covered.
12311- const steerTailThisTurn =
12312- reconcilePendingSteer({
12313- turnNew: turnNewModelMessages,
12314- response: capturedResponseMessage,
12315- }).reduce((n, e) => n + e.model.length, 0) + reconcilePendingBackground();
12306+ const steerTailThisTurn = reconcilePendingSteer({
12307+ turnNew: turnNewModelMessages,
12308+ response: capturedResponseMessage,
12309+ }).reduce((n, e) => n + e.model.length, 0);
1231612310
1231712311 // Append the assistant's response (partial or complete) to the accumulator.
1231812312 // The onFinish callback fires even on abort/stop, so partial responses
@@ -13346,7 +13340,8 @@ function chatAgent<
1334613340 // yet (a pre-run hook threw), so fold the wire message in for the
1334713341 // error event + snapshot — the cursor has already advanced past it,
1334813342 // so otherwise it survives in neither the snapshot nor the `.in` tail.
13349- const erroredWireMessage = (currentWirePayload as { message?: TUIMessage }).message;
13343+ const rawWireMessage = (currentWirePayload as { message?: TUIMessage }).message;
13344+ const erroredWireMessage = rawWireMessage && cleanIncomingMessage(rawWireMessage);
1335013345 const erroredUIMessages =
1335113346 erroredWireMessage &&
1335213347 !accumulatedUIMessages.some((m) => m.id === erroredWireMessage.id)
@@ -13414,7 +13409,6 @@ function chatAgent<
1341413409 let erroredNewModelMessages: ModelMessage[] = [];
1341513410
1341613411 const reconciledSteer = reconcilePendingSteer({ response: partialResponse });
13417- const backgroundTailThisTurn = reconcilePendingBackground();
1341813412
1341913413 if (!responseCommitted) {
1342013414 try {
@@ -13442,8 +13436,7 @@ function chatAgent<
1344213436 accumulatedMessages,
1344313437 erroredUIMessages[partialIdx]!,
1344413438 partialResponse!,
13445- reconciledSteer.reduce((n, e) => n + e.model.length, 0) +
13446- backgroundTailThisTurn
13439+ reconciledSteer.reduce((n, e) => n + e.model.length, 0)
1344713440 );
1344813441 if (!ok) {
1344913442 logger.warn(
0 commit comments