diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 245b6a5cbb2195376ba9e4e7b9e92e73dc4fa6a5..42247506410b5b3fac0ac2a38930e9fa4862b7ca 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -402,6 +402,59 @@ describe("ProviderRuntimeIngestion", () => { expect(finalMessage?.streaming).toBe(false); }); + it("spills oversized buffered deltas and still finalizes full assistant text", async () => { + const harness = await createHarness(); + const now = new Date().toISOString(); + const oversizedText = "x".repeat(40_000); + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-turn-started-buffer-spill"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: now, + turnId: asProviderTurnId("turn-buffer-spill"), + }); + await waitForThread( + harness.engine, + (thread) => + thread.session?.status === "running" && + thread.session?.activeTurnId === "turn-buffer-spill", + ); + + harness.emit({ + type: "message.delta", + eventId: asEventId("evt-message-delta-buffer-spill"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: now, + turnId: asProviderTurnId("turn-buffer-spill"), + itemId: asItemId("item-buffer-spill"), + delta: oversizedText, + }); + harness.emit({ + type: "message.completed", + eventId: asEventId("evt-message-completed-buffer-spill"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: now, + turnId: asProviderTurnId("turn-buffer-spill"), + itemId: asItemId("item-buffer-spill"), + }); + + const thread = await waitForThread( + harness.engine, + (entry) => + entry.messages.some( + (message) => message.id === "assistant:item-buffer-spill" && !message.streaming, + ), + ); + const message = thread.messages.find((entry) => entry.id === "assistant:item-buffer-spill"); + expect(message?.text.length).toBe(oversizedText.length); + expect(message?.text).toBe(oversizedText); + expect(message?.streaming).toBe(false); + }); + it("does not duplicate assistant completion when message.completed is followed by turn.completed", async () => { const harness = await createHarness(); const now = new Date().toISOString(); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 12cf228e39b146f4c04204021fd8a98ba29b4871..2a6b4aa5b615cba74cdbc20ff32935d8b91ad657 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -28,6 +28,10 @@ const TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY = 10_000; const TURN_MESSAGE_IDS_BY_TURN_TTL = Duration.minutes(120); const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000; const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL = Duration.minutes(120); +const BUFFERED_ASSISTANT_MESSAGE_SPILLED_CACHE_CAPACITY = 20_000; +const BUFFERED_ASSISTANT_MESSAGE_SPILLED_TTL = Duration.minutes(120); +const MAX_BUFFERED_ASSISTANT_CHARS = 24_000; +const BUFFERED_ASSISTANT_KEEP_CHARS = 8_000; type TurnStartRequestedDomainEvent = Extract< OrchestrationEvent, @@ -169,6 +173,11 @@ const make = Effect.gen(function* () { timeToLive: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL, lookup: () => Effect.succeed(""), }); + const bufferedAssistantMessageSpilledByMessageId = yield* Cache.make({ + capacity: BUFFERED_ASSISTANT_MESSAGE_SPILLED_CACHE_CAPACITY, + timeToLive: BUFFERED_ASSISTANT_MESSAGE_SPILLED_TTL, + lookup: () => Effect.succeed(true), + }); const rememberAssistantMessageId = ( sessionId: ProviderSessionId, @@ -226,14 +235,23 @@ const make = Effect.gen(function* () { const appendBufferedAssistantText = (messageId: MessageId, delta: string) => Cache.getOption(bufferedAssistantTextByMessageId, messageId).pipe( Effect.flatMap((existingText) => - Cache.set( - bufferedAssistantTextByMessageId, - messageId, - Option.match(existingText, { + Effect.gen(function* () { + const nextText = Option.match(existingText, { onNone: () => delta, onSome: (text) => `${text}${delta}`, - }), - ), + }); + if (nextText.length <= MAX_BUFFERED_ASSISTANT_CHARS) { + yield* Cache.set(bufferedAssistantTextByMessageId, messageId, nextText); + return ""; + } + + // Safety valve: spill oldest text as a persisted assistant delta to cap memory. + const spillChars = Math.max(1, nextText.length - BUFFERED_ASSISTANT_KEEP_CHARS); + const spillChunk = nextText.slice(0, spillChars); + const retainedTail = nextText.slice(spillChars); + yield* Cache.set(bufferedAssistantTextByMessageId, messageId, retainedTail); + return spillChunk; + }), ), ); @@ -249,18 +267,51 @@ const make = Effect.gen(function* () { const clearBufferedAssistantText = (messageId: MessageId) => Cache.invalidate(bufferedAssistantTextByMessageId, messageId); - const clearAssistantMessageState = (messageId: MessageId) => clearBufferedAssistantText(messageId); + const markBufferedAssistantMessageSpilled = (messageId: MessageId) => + Cache.set(bufferedAssistantMessageSpilledByMessageId, messageId, true); - const dispatchAssistantCompletion = (input: { + const takeBufferedAssistantMessageSpilled = (messageId: MessageId) => + Cache.getOption(bufferedAssistantMessageSpilledByMessageId, messageId).pipe( + Effect.flatMap((spilled) => + Cache.invalidate(bufferedAssistantMessageSpilledByMessageId, messageId).pipe( + Effect.as(Option.isSome(spilled)), + ), + ), + ); + + const clearBufferedAssistantMessageSpilled = (messageId: MessageId) => + Cache.invalidate(bufferedAssistantMessageSpilledByMessageId, messageId); + + const clearAssistantMessageState = (messageId: MessageId) => + Effect.all( + [clearBufferedAssistantText(messageId), clearBufferedAssistantMessageSpilled(messageId)], + { concurrency: 1 }, + ).pipe(Effect.asVoid); + + const finalizeAssistantMessage = (input: { event: ProviderRuntimeEvent; threadId: ThreadId; messageId: MessageId; turnId?: TurnId; createdAt: string; commandTag: string; + spillRemainderCommandTag: string; }) => Effect.gen(function* () { const text = yield* takeBufferedAssistantText(input.messageId); + const spilled = yield* takeBufferedAssistantMessageSpilled(input.messageId); + + if (spilled && text.length > 0) { + yield* orchestrationEngine.dispatch({ + type: "thread.message.assistant.delta", + commandId: providerCommandId(input.event, input.spillRemainderCommandTag), + threadId: input.threadId, + messageId: input.messageId, + delta: text, + ...(input.turnId ? { turnId: input.turnId } : {}), + createdAt: input.createdAt, + }); + } yield* orchestrationEngine.dispatch({ type: "thread.message.assistant.complete", @@ -268,7 +319,7 @@ const make = Effect.gen(function* () { threadId: input.threadId, messageId: input.messageId, ...(input.turnId ? { turnId: input.turnId } : {}), - text, + text: spilled ? "" : text, createdAt: input.createdAt, }); yield* clearAssistantMessageState(input.messageId); @@ -369,7 +420,19 @@ const make = Effect.gen(function* () { const assistantDeliveryMode = yield* Ref.get(assistantDeliveryModeRef); if (assistantDeliveryMode === "buffered") { - yield* appendBufferedAssistantText(assistantMessageId, event.delta); + const spillChunk = yield* appendBufferedAssistantText(assistantMessageId, event.delta); + if (spillChunk.length > 0) { + yield* markBufferedAssistantMessageSpilled(assistantMessageId); + yield* orchestrationEngine.dispatch({ + type: "thread.message.assistant.delta", + commandId: providerCommandId(event, "assistant-delta-buffer-spill"), + threadId: thread.id, + messageId: assistantMessageId, + delta: spillChunk, + ...(turnId ? { turnId } : {}), + createdAt: now, + }); + } } else { yield* orchestrationEngine.dispatch({ type: "thread.message.assistant.delta", @@ -390,13 +453,14 @@ const make = Effect.gen(function* () { yield* rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId); } - yield* dispatchAssistantCompletion({ + yield* finalizeAssistantMessage({ event, threadId: thread.id, messageId: assistantMessageId, ...(turnId ? { turnId } : {}), createdAt: now, commandTag: "assistant-complete", + spillRemainderCommandTag: "assistant-delta-buffer-remainder", }); if (turnId) { @@ -412,13 +476,14 @@ const make = Effect.gen(function* () { assistantMessageIds, (assistantMessageId) => Effect.gen(function* () { - yield* dispatchAssistantCompletion({ + yield* finalizeAssistantMessage({ event, threadId: thread.id, messageId: assistantMessageId, turnId, createdAt: now, commandTag: "assistant-complete-finalize", + spillRemainderCommandTag: "assistant-delta-buffer-remainder-finalize", }); }), { concurrency: 1 }, diff --git a/apps/server/src/orchestration/decider.projectScripts.test.ts b/apps/server/src/orchestration/decider.projectScripts.test.ts index 7661352962f0758c91d4f463b0b7d05f23cca960..9973219f629f6494fe54569746e279b5f255ac09 100644 --- a/apps/server/src/orchestration/decider.projectScripts.test.ts +++ b/apps/server/src/orchestration/decider.projectScripts.test.ts @@ -162,10 +162,12 @@ describe("decider project scripts", () => { const events = Array.isArray(result) ? result : [result]; expect(events).toHaveLength(2); expect(events[0]?.type).toBe("thread.message-sent"); - expect(events[1]?.type).toBe("thread.turn-start-requested"); - expect(events[1]?.causationEventId).toBe(events[0]?.eventId ?? null); - expect((events[1]?.payload as { assistantDeliveryMode?: string }).assistantDeliveryMode).toBe( - "buffered", - ); + const turnStartEvent = events[1]; + expect(turnStartEvent?.type).toBe("thread.turn-start-requested"); + expect(turnStartEvent?.causationEventId).toBe(events[0]?.eventId ?? null); + if (turnStartEvent?.type !== "thread.turn-start-requested") { + return; + } + expect(turnStartEvent.payload.assistantDeliveryMode).toBe("buffered"); }); });