authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-26 20:27:56-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 20:43:22-08:00
log9d5c5156805a24c01de1bc7147c80b9bcd43827f
tree2102c4de13d0da6fa2719a10bc6bd734d5c58637
parent0e9677965cca02d925244e7814a245a194bd42de

Always flush buffered assistant text as final delta

- Remove spilled-state cache tracking for buffered assistant messages - Emit any remaining buffered text as a finalize delta on completion - Keep assistant completion payload text empty in buffered mode

1 files changed, 7 insertions(+), 35 deletions(-)

apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+7-35
......@@ -28,8 +28,6 @@ const TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY = 10_000;
2828const TURN_MESSAGE_IDS_BY_TURN_TTL = Duration.minutes(120);
2929const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000;
3030const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL = Duration.minutes(120);
31const BUFFERED_ASSISTANT_MESSAGE_SPILLED_CACHE_CAPACITY = 20_000;
32const BUFFERED_ASSISTANT_MESSAGE_SPILLED_TTL = Duration.minutes(120);
3331const MAX_BUFFERED_ASSISTANT_CHARS = 24_000;
3432
3533type TurnStartRequestedDomainEvent = Extract<
......@@ -172,11 +170,6 @@ const make = Effect.gen(function* () {
172170 timeToLive: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL,
173171 lookup: () => Effect.succeed(""),
174172 });
175 const bufferedAssistantMessageSpilledByMessageId = yield* Cache.make<MessageId, true>({
176 capacity: BUFFERED_ASSISTANT_MESSAGE_SPILLED_CACHE_CAPACITY,
177 timeToLive: BUFFERED_ASSISTANT_MESSAGE_SPILLED_TTL,
178 lookup: () => Effect.succeed(true),
179 });
180173
181174 const rememberAssistantMessageId = (
182175 sessionId: ProviderSessionId,
......@@ -263,26 +256,7 @@ const make = Effect.gen(function* () {
263256 const clearBufferedAssistantText = (messageId: MessageId) =>
264257 Cache.invalidate(bufferedAssistantTextByMessageId, messageId);
265258
266 const markBufferedAssistantMessageSpilled = (messageId: MessageId) =>
267 Cache.set(bufferedAssistantMessageSpilledByMessageId, messageId, true);
268
269 const takeBufferedAssistantMessageSpilled = (messageId: MessageId) =>
270 Cache.getOption(bufferedAssistantMessageSpilledByMessageId, messageId).pipe(
271 Effect.flatMap((spilled) =>
272 Cache.invalidate(bufferedAssistantMessageSpilledByMessageId, messageId).pipe(
273 Effect.as(Option.isSome(spilled)),
274 ),
275 ),
276 );
277
278 const clearBufferedAssistantMessageSpilled = (messageId: MessageId) =>
279 Cache.invalidate(bufferedAssistantMessageSpilledByMessageId, messageId);
280
281 const clearAssistantMessageState = (messageId: MessageId) =>
282 Effect.all(
283 [clearBufferedAssistantText(messageId), clearBufferedAssistantMessageSpilled(messageId)],
284 { concurrency: 1 },
285 ).pipe(Effect.asVoid);
259 const clearAssistantMessageState = (messageId: MessageId) => clearBufferedAssistantText(messageId);
286260
287261 const finalizeAssistantMessage = (input: {
288262 event: ProviderRuntimeEvent;
......@@ -291,16 +265,15 @@ const make = Effect.gen(function* () {
291265 turnId?: TurnId;
292266 createdAt: string;
293267 commandTag: string;
294 spillRemainderCommandTag: string;
268 finalDeltaCommandTag: string;
295269 }) =>
296270 Effect.gen(function* () {
297271 const text = yield* takeBufferedAssistantText(input.messageId);
298 const spilled = yield* takeBufferedAssistantMessageSpilled(input.messageId);
299272
300 if (spilled && text.length > 0) {
273 if (text.length > 0) {
301274 yield* orchestrationEngine.dispatch({
302275 type: "thread.message.assistant.delta",
303 commandId: providerCommandId(input.event, input.spillRemainderCommandTag),
276 commandId: providerCommandId(input.event, input.finalDeltaCommandTag),
304277 threadId: input.threadId,
305278 messageId: input.messageId,
306279 delta: text,
......@@ -315,7 +288,7 @@ const make = Effect.gen(function* () {
315288 threadId: input.threadId,
316289 messageId: input.messageId,
317290 ...(input.turnId ? { turnId: input.turnId } : {}),
318 text: spilled ? "" : text,
291 text: "",
319292 createdAt: input.createdAt,
320293 });
321294 yield* clearAssistantMessageState(input.messageId);
......@@ -418,7 +391,6 @@ const make = Effect.gen(function* () {
418391 if (assistantDeliveryMode === "buffered") {
419392 const spillChunk = yield* appendBufferedAssistantText(assistantMessageId, event.delta);
420393 if (spillChunk.length > 0) {
421 yield* markBufferedAssistantMessageSpilled(assistantMessageId);
422394 yield* orchestrationEngine.dispatch({
423395 type: "thread.message.assistant.delta",
424396 commandId: providerCommandId(event, "assistant-delta-buffer-spill"),
......@@ -456,7 +428,7 @@ const make = Effect.gen(function* () {
456428 ...(turnId ? { turnId } : {}),
457429 createdAt: now,
458430 commandTag: "assistant-complete",
459 spillRemainderCommandTag: "assistant-delta-buffer-remainder",
431 finalDeltaCommandTag: "assistant-delta-finalize",
460432 });
461433
462434 if (turnId) {
......@@ -479,7 +451,7 @@ const make = Effect.gen(function* () {
479451 turnId,
480452 createdAt: now,
481453 commandTag: "assistant-complete-finalize",
482 spillRemainderCommandTag: "assistant-delta-buffer-remainder-finalize",
454 finalDeltaCommandTag: "assistant-delta-finalize-fallback",
483455 });
484456 }),
485457 { concurrency: 1 },