authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-26 19:51:29-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 20:43:22-08:00
log32a17b91098edf78a3fab93f978e8b765fd112fb
tree4cc6d4c47c4c2cfcd28584ec21468c12bf61c59e
parente119e18ed36f8a909036ddb29c56bf5969bd8b77

Simplify assistant delivery mode state in runtime ingestion

- Replace per-thread/turn/message delivery-mode caches with a single `Ref` - Always complete assistant messages with buffered text from message state - Remove obsolete delivery-mode cleanup and resolution paths

1 files changed, 15 insertions(+), 161 deletions(-)

apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+15-161
......@@ -10,7 +10,7 @@ import {
1010 type ProviderRuntimeEvent,
1111 type ProviderSessionId,
1212} from "@t3tools/contracts";
13import { Cache, Cause, Duration, Effect, Layer, Option, Queue, Stream } from "effect";
13import { Cache, Cause, Duration, Effect, Layer, Option, Queue, Ref, Stream } from "effect";
1414
1515import { ProviderService } from "../../provider/Services/ProviderService.ts";
1616import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
......@@ -22,15 +22,10 @@ import {
2222const providerTurnKey = (sessionId: ProviderSessionId, turnId: TurnId) => `${sessionId}:${turnId}`;
2323const providerCommandId = (event: ProviderRuntimeEvent, tag: string): CommandId =>
2424 CommandId.makeUnsafe(`provider:${event.eventId}:${tag}:${crypto.randomUUID()}`);
25
2526const DEFAULT_ASSISTANT_DELIVERY_MODE: AssistantDeliveryMode = "buffered";
2627const TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY = 10_000;
2728const TURN_MESSAGE_IDS_BY_TURN_TTL = Duration.minutes(120);
28const TURN_DELIVERY_MODE_BY_TURN_CACHE_CAPACITY = 10_000;
29const TURN_DELIVERY_MODE_BY_TURN_TTL = Duration.minutes(120);
30const PENDING_DELIVERY_MODE_BY_THREAD_CACHE_CAPACITY = 10_000;
31const PENDING_DELIVERY_MODE_BY_THREAD_TTL = Duration.minutes(30);
32const MESSAGE_DELIVERY_MODE_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000;
33const MESSAGE_DELIVERY_MODE_BY_MESSAGE_ID_TTL = Duration.minutes(120);
3429const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000;
3530const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL = Duration.minutes(120);
3631
......@@ -115,6 +110,7 @@ function runtimeEventToActivities(
115110 },
116111 ];
117112 }
113
118114 case "tool.completed": {
119115 return [
120116 {
......@@ -131,6 +127,7 @@ function runtimeEventToActivities(
131127 },
132128 ];
133129 }
130
134131 case "tool.started": {
135132 return [
136133 {
......@@ -157,26 +154,16 @@ const make = Effect.gen(function* () {
157154 const orchestrationEngine = yield* OrchestrationEngineService;
158155 const providerService = yield* ProviderService;
159156
157 const assistantDeliveryModeRef = yield* Ref.make<AssistantDeliveryMode>(
158 DEFAULT_ASSISTANT_DELIVERY_MODE,
159 );
160
160161 const turnMessageIdsByTurnKey = yield* Cache.make<string, Set<MessageId>>({
161162 capacity: TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY,
162163 timeToLive: TURN_MESSAGE_IDS_BY_TURN_TTL,
163164 lookup: () => Effect.succeed(new Set<MessageId>()),
164165 });
165 const assistantDeliveryModeByTurnKey = yield* Cache.make<string, AssistantDeliveryMode>({
166 capacity: TURN_DELIVERY_MODE_BY_TURN_CACHE_CAPACITY,
167 timeToLive: TURN_DELIVERY_MODE_BY_TURN_TTL,
168 lookup: () => Effect.succeed(DEFAULT_ASSISTANT_DELIVERY_MODE),
169 });
170 const pendingAssistantDeliveryModeByThreadId = yield* Cache.make<ThreadId, AssistantDeliveryMode>({
171 capacity: PENDING_DELIVERY_MODE_BY_THREAD_CACHE_CAPACITY,
172 timeToLive: PENDING_DELIVERY_MODE_BY_THREAD_TTL,
173 lookup: () => Effect.succeed(DEFAULT_ASSISTANT_DELIVERY_MODE),
174 });
175 const assistantDeliveryModeByMessageId = yield* Cache.make<MessageId, AssistantDeliveryMode>({
176 capacity: MESSAGE_DELIVERY_MODE_BY_MESSAGE_ID_CACHE_CAPACITY,
177 timeToLive: MESSAGE_DELIVERY_MODE_BY_MESSAGE_ID_TTL,
178 lookup: () => Effect.succeed(DEFAULT_ASSISTANT_DELIVERY_MODE),
179 });
166
180167 const bufferedAssistantTextByMessageId = yield* Cache.make<MessageId, string>({
181168 capacity: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY,
182169 timeToLive: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL,
......@@ -236,45 +223,6 @@ const make = Effect.gen(function* () {
236223 const clearAssistantMessageIdsForTurn = (sessionId: ProviderSessionId, turnId: TurnId) =>
237224 Cache.invalidate(turnMessageIdsByTurnKey, providerTurnKey(sessionId, turnId));
238225
239 const rememberAssistantDeliveryModeForTurn = (
240 sessionId: ProviderSessionId,
241 turnId: TurnId,
242 assistantDeliveryMode: AssistantDeliveryMode,
243 ) =>
244 Cache.set(
245 assistantDeliveryModeByTurnKey,
246 providerTurnKey(sessionId, turnId),
247 assistantDeliveryMode,
248 );
249
250 const clearAssistantDeliveryModeForTurn = (sessionId: ProviderSessionId, turnId: TurnId) =>
251 Cache.invalidate(assistantDeliveryModeByTurnKey, providerTurnKey(sessionId, turnId));
252
253 const rememberPendingAssistantDeliveryModeForThread = (
254 threadId: ThreadId,
255 assistantDeliveryMode: AssistantDeliveryMode,
256 ) => Cache.set(pendingAssistantDeliveryModeByThreadId, threadId, assistantDeliveryMode);
257
258 const takePendingAssistantDeliveryModeForThread = (threadId: ThreadId) =>
259 Cache.getOption(pendingAssistantDeliveryModeByThreadId, threadId).pipe(
260 Effect.flatMap((assistantDeliveryMode) =>
261 Cache.invalidate(pendingAssistantDeliveryModeByThreadId, threadId).pipe(
262 Effect.as(assistantDeliveryMode),
263 ),
264 ),
265 );
266
267 const clearPendingAssistantDeliveryModeForThread = (threadId: ThreadId) =>
268 Cache.invalidate(pendingAssistantDeliveryModeByThreadId, threadId);
269
270 const rememberAssistantDeliveryModeForMessage = (
271 messageId: MessageId,
272 assistantDeliveryMode: AssistantDeliveryMode,
273 ) => Cache.set(assistantDeliveryModeByMessageId, messageId, assistantDeliveryMode);
274
275 const clearAssistantDeliveryModeForMessage = (messageId: MessageId) =>
276 Cache.invalidate(assistantDeliveryModeByMessageId, messageId);
277
278226 const appendBufferedAssistantText = (messageId: MessageId, delta: string) =>
279227 Cache.getOption(bufferedAssistantTextByMessageId, messageId).pipe(
280228 Effect.flatMap((existingText) =>
......@@ -301,66 +249,18 @@ const make = Effect.gen(function* () {
301249 const clearBufferedAssistantText = (messageId: MessageId) =>
302250 Cache.invalidate(bufferedAssistantTextByMessageId, messageId);
303251
304 const clearAssistantMessageState = (messageId: MessageId) =>
305 Effect.all(
306 [
307 clearAssistantDeliveryModeForMessage(messageId),
308 clearBufferedAssistantText(messageId),
309 ],
310 { concurrency: 1 },
311 ).pipe(Effect.asVoid);
312
313 const resolveAssistantDeliveryModeForMessage = (input: {
314 threadId: ThreadId;
315 sessionId: ProviderSessionId;
316 turnId?: TurnId;
317 messageId: MessageId;
318 }) =>
319 Effect.gen(function* () {
320 const messageMode = yield* Cache.getOption(assistantDeliveryModeByMessageId, input.messageId);
321 if (Option.isSome(messageMode)) {
322 return messageMode.value;
323 }
324
325 if (input.turnId) {
326 const turnMode = yield* Cache.getOption(
327 assistantDeliveryModeByTurnKey,
328 providerTurnKey(input.sessionId, input.turnId),
329 );
330 if (Option.isSome(turnMode)) {
331 return turnMode.value;
332 }
333 }
334
335 const pendingMode = yield* Cache.getOption(
336 pendingAssistantDeliveryModeByThreadId,
337 input.threadId,
338 );
339 if (Option.isSome(pendingMode)) {
340 if (input.turnId) {
341 yield* rememberAssistantDeliveryModeForTurn(input.sessionId, input.turnId, pendingMode.value);
342 yield* clearPendingAssistantDeliveryModeForThread(input.threadId);
343 }
344 return pendingMode.value;
345 }
346
347 return DEFAULT_ASSISTANT_DELIVERY_MODE;
348 });
252 const clearAssistantMessageState = (messageId: MessageId) => clearBufferedAssistantText(messageId);
349253
350254 const dispatchAssistantCompletion = (input: {
351255 event: ProviderRuntimeEvent;
352256 threadId: ThreadId;
353257 messageId: MessageId;
354258 turnId?: TurnId;
355 assistantDeliveryMode: AssistantDeliveryMode;
356259 createdAt: string;
357260 commandTag: string;
358261 }) =>
359262 Effect.gen(function* () {
360 const text =
361 input.assistantDeliveryMode === "buffered"
362 ? yield* takeBufferedAssistantText(input.messageId)
363 : undefined;
263 const text = yield* takeBufferedAssistantText(input.messageId);
364264
365265 yield* orchestrationEngine.dispatch({
366266 type: "thread.message.assistant.complete",
......@@ -368,7 +268,7 @@ const make = Effect.gen(function* () {
368268 threadId: input.threadId,
369269 messageId: input.messageId,
370270 ...(input.turnId ? { turnId: input.turnId } : {}),
371 ...(text !== undefined ? { text } : {}),
271 text,
372272 createdAt: input.createdAt,
373273 });
374274 yield* clearAssistantMessageState(input.messageId);
......@@ -397,16 +297,6 @@ const make = Effect.gen(function* () {
397297 }),
398298 { concurrency: 1 },
399299 ).pipe(Effect.asVoid);
400
401 const deliveryModeKeys = Array.from(yield* Cache.keys(assistantDeliveryModeByTurnKey));
402 yield* Effect.forEach(
403 deliveryModeKeys,
404 (key) =>
405 key.startsWith(prefix)
406 ? Cache.invalidate(assistantDeliveryModeByTurnKey, key)
407 : Effect.void,
408 { concurrency: 1 },
409 ).pipe(Effect.asVoid);
410300 });
411301
412302 const processRuntimeEvent = (event: ProviderRuntimeEvent) =>
......@@ -427,18 +317,6 @@ const make = Effect.gen(function* () {
427317 event.type === "turn.completed"
428318 ) {
429319 const activeTurnId = event.type === "turn.started" ? (toTurnId(event.turnId) ?? null) : null;
430 if (event.type === "turn.started" && activeTurnId) {
431 const pendingAssistantDeliveryMode = yield* takePendingAssistantDeliveryModeForThread(
432 thread.id,
433 );
434 if (Option.isSome(pendingAssistantDeliveryMode)) {
435 yield* rememberAssistantDeliveryModeForTurn(
436 event.sessionId,
437 activeTurnId,
438 pendingAssistantDeliveryMode.value,
439 );
440 }
441 }
442320 const providerThreadIdFromEvent =
443321 event.type === "thread.started"
444322 ? ProviderThreadId.makeUnsafe(event.threadId)
......@@ -488,14 +366,8 @@ const make = Effect.gen(function* () {
488366 if (turnId) {
489367 yield* rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId);
490368 }
491 const assistantDeliveryMode = yield* resolveAssistantDeliveryModeForMessage({
492 threadId: thread.id,
493 sessionId: event.sessionId,
494 ...(turnId ? { turnId } : {}),
495 messageId: assistantMessageId,
496 });
497 yield* rememberAssistantDeliveryModeForMessage(assistantMessageId, assistantDeliveryMode);
498369
370 const assistantDeliveryMode = yield* Ref.get(assistantDeliveryModeRef);
499371 if (assistantDeliveryMode === "buffered") {
500372 yield* appendBufferedAssistantText(assistantMessageId, event.delta);
501373 } else {
......@@ -517,20 +389,12 @@ const make = Effect.gen(function* () {
517389 if (turnId) {
518390 yield* rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId);
519391 }
520 const assistantDeliveryMode = yield* resolveAssistantDeliveryModeForMessage({
521 threadId: thread.id,
522 sessionId: event.sessionId,
523 ...(turnId ? { turnId } : {}),
524 messageId: assistantMessageId,
525 });
526 yield* rememberAssistantDeliveryModeForMessage(assistantMessageId, assistantDeliveryMode);
527392
528393 yield* dispatchAssistantCompletion({
529394 event,
530395 threadId: thread.id,
531396 messageId: assistantMessageId,
532397 ...(turnId ? { turnId } : {}),
533 assistantDeliveryMode,
534398 createdAt: now,
535399 commandTag: "assistant-complete",
536400 });
......@@ -548,19 +412,11 @@ const make = Effect.gen(function* () {
548412 assistantMessageIds,
549413 (assistantMessageId) =>
550414 Effect.gen(function* () {
551 const assistantDeliveryMode =
552 yield* resolveAssistantDeliveryModeForMessage({
553 threadId: thread.id,
554 sessionId: event.sessionId,
555 turnId,
556 messageId: assistantMessageId,
557 });
558415 yield* dispatchAssistantCompletion({
559416 event,
560417 threadId: thread.id,
561418 messageId: assistantMessageId,
562419 turnId,
563 assistantDeliveryMode,
564420 createdAt: now,
565421 commandTag: "assistant-complete-finalize",
566422 });
......@@ -568,13 +424,11 @@ const make = Effect.gen(function* () {
568424 { concurrency: 1 },
569425 ).pipe(Effect.asVoid);
570426 yield* clearAssistantMessageIdsForTurn(event.sessionId, turnId);
571 yield* clearAssistantDeliveryModeForTurn(event.sessionId, turnId);
572427 }
573428 }
574429
575430 if (event.type === "session.exited") {
576431 yield* clearTurnStateForSession(event.sessionId);
577 yield* clearPendingAssistantDeliveryModeForThread(thread.id);
578432 }
579433
580434 if (event.type === "runtime.error") {
......@@ -614,8 +468,8 @@ const make = Effect.gen(function* () {
614468 });
615469
616470 const processDomainEvent = (event: TurnStartRequestedDomainEvent) =>
617 rememberPendingAssistantDeliveryModeForThread(
618 event.payload.threadId,
471 Ref.set(
472 assistantDeliveryModeRef,
619473 event.payload.assistantDeliveryMode ?? DEFAULT_ASSISTANT_DELIVERY_MODE,
620474 );
621475