authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-26 18:24:17-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 20:43:22-08:00
log001902c881101f0c4d35cebcf2303ddb75f33534
tree677e272c42b91c502b326b15b0a2ecafa7a749f5
parentc7a029a2bd9d4ece94a19aa614545cbfbb315864

Add configurable assistant delivery mode across server and web

- add `AssistantDeliveryMode` contracts and carry mode through turn-start events - default assistant responses to buffered mode and preserve buffered text until completion - support streaming mode end-to-end with ingestion/decider updates and test coverage - add a Settings toggle to enable assistant streaming in the chat UI

9 files changed, 507 insertions(+), 33 deletions(-)

apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+144
......@@ -2,6 +2,7 @@ import type { ProviderRuntimeEvent } from "@t3tools/contracts";
22import {
33 CommandId,
44 EventId,
5 MessageId,
56 ProjectId,
67 ProviderItemId,
78 ProviderSessionId,
......@@ -33,6 +34,7 @@ const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.make
3334const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value);
3435const asItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(value);
3536const asEventId = (value: string): EventId => EventId.makeUnsafe(value);
37const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value);
3638
3739function createProviderServiceHarness() {
3840 const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());
......@@ -258,6 +260,148 @@ describe("ProviderRuntimeIngestion", () => {
258260 expect(message?.streaming).toBe(false);
259261 });
260262
263 it("buffers assistant deltas by default until completion", async () => {
264 const harness = await createHarness();
265 const now = new Date().toISOString();
266
267 harness.emit({
268 type: "turn.started",
269 eventId: asEventId("evt-turn-started-buffered"),
270 provider: "codex",
271 sessionId: asSessionId("sess-1"),
272 createdAt: now,
273 turnId: asProviderTurnId("turn-buffered"),
274 });
275 await waitForThread(
276 harness.engine,
277 (thread) =>
278 thread.session?.status === "running" && thread.session?.activeTurnId === "turn-buffered",
279 );
280
281 harness.emit({
282 type: "message.delta",
283 eventId: asEventId("evt-message-delta-buffered"),
284 provider: "codex",
285 sessionId: asSessionId("sess-1"),
286 createdAt: now,
287 turnId: asProviderTurnId("turn-buffered"),
288 itemId: asItemId("item-buffered"),
289 delta: "buffer me",
290 });
291
292 await Effect.runPromise(Effect.sleep("30 millis"));
293 const midReadModel = await Effect.runPromise(harness.engine.getReadModel());
294 const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
295 expect(midThread?.messages.some((message) => message.id === "assistant:item-buffered")).toBe(
296 false,
297 );
298
299 harness.emit({
300 type: "message.completed",
301 eventId: asEventId("evt-message-completed-buffered"),
302 provider: "codex",
303 sessionId: asSessionId("sess-1"),
304 createdAt: now,
305 turnId: asProviderTurnId("turn-buffered"),
306 itemId: asItemId("item-buffered"),
307 });
308
309 const thread = await waitForThread(
310 harness.engine,
311 (entry) =>
312 entry.messages.some((message) => message.id === "assistant:item-buffered" && !message.streaming),
313 );
314 const message = thread.messages.find((entry) => entry.id === "assistant:item-buffered");
315 expect(message?.text).toBe("buffer me");
316 expect(message?.streaming).toBe(false);
317 });
318
319 it("streams assistant deltas when thread.turn.start requests streaming mode", async () => {
320 const harness = await createHarness();
321 const now = new Date().toISOString();
322
323 await Effect.runPromise(
324 harness.engine.dispatch({
325 type: "thread.turn.start",
326 commandId: CommandId.makeUnsafe("cmd-turn-start-streaming-mode"),
327 threadId: ThreadId.makeUnsafe("thread-1"),
328 message: {
329 messageId: asMessageId("message-streaming-mode"),
330 role: "user",
331 text: "stream please",
332 attachments: [],
333 },
334 assistantDeliveryMode: "streaming",
335 createdAt: now,
336 }),
337 );
338 await Effect.runPromise(Effect.sleep("30 millis"));
339
340 harness.emit({
341 type: "turn.started",
342 eventId: asEventId("evt-turn-started-streaming-mode"),
343 provider: "codex",
344 sessionId: asSessionId("sess-1"),
345 createdAt: now,
346 turnId: asProviderTurnId("turn-streaming-mode"),
347 });
348 await waitForThread(
349 harness.engine,
350 (thread) =>
351 thread.session?.status === "running" &&
352 thread.session?.activeTurnId === "turn-streaming-mode",
353 );
354
355 harness.emit({
356 type: "message.delta",
357 eventId: asEventId("evt-message-delta-streaming-mode"),
358 provider: "codex",
359 sessionId: asSessionId("sess-1"),
360 createdAt: now,
361 turnId: asProviderTurnId("turn-streaming-mode"),
362 itemId: asItemId("item-streaming-mode"),
363 delta: "hello live",
364 });
365
366 const liveThread = await waitForThread(
367 harness.engine,
368 (entry) =>
369 entry.messages.some(
370 (message) =>
371 message.id === "assistant:item-streaming-mode" &&
372 message.streaming &&
373 message.text === "hello live",
374 ),
375 );
376 const liveMessage = liveThread.messages.find(
377 (entry) => entry.id === "assistant:item-streaming-mode",
378 );
379 expect(liveMessage?.streaming).toBe(true);
380
381 harness.emit({
382 type: "message.completed",
383 eventId: asEventId("evt-message-completed-streaming-mode"),
384 provider: "codex",
385 sessionId: asSessionId("sess-1"),
386 createdAt: now,
387 turnId: asProviderTurnId("turn-streaming-mode"),
388 itemId: asItemId("item-streaming-mode"),
389 });
390
391 const finalThread = await waitForThread(
392 harness.engine,
393 (entry) =>
394 entry.messages.some(
395 (message) => message.id === "assistant:item-streaming-mode" && !message.streaming,
396 ),
397 );
398 const finalMessage = finalThread.messages.find(
399 (entry) => entry.id === "assistant:item-streaming-mode",
400 );
401 expect(finalMessage?.text).toBe("hello live");
402 expect(finalMessage?.streaming).toBe(false);
403 });
404
261405 it("does not duplicate assistant completion when message.completed is followed by turn.completed", async () => {
262406 const harness = await createHarness();
263407 const now = new Date().toISOString();
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+302-32
......@@ -1,7 +1,10 @@
11import {
2 type AssistantDeliveryMode,
23 CommandId,
34 MessageId,
5 type OrchestrationEvent,
46 ProviderThreadId,
7 type ThreadId,
58 TurnId,
69 type OrchestrationThreadActivity,
710 type ProviderRuntimeEvent,
......@@ -19,8 +22,32 @@ import {
1922const providerTurnKey = (sessionId: ProviderSessionId, turnId: TurnId) => `${sessionId}:${turnId}`;
2023const providerCommandId = (event: ProviderRuntimeEvent, tag: string): CommandId =>
2124 CommandId.makeUnsafe(`provider:${event.eventId}:${tag}:${crypto.randomUUID()}`);
25const DEFAULT_ASSISTANT_DELIVERY_MODE: AssistantDeliveryMode = "buffered";
2226const TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY = 10_000;
2327const 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);
34const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000;
35const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL = Duration.minutes(120);
36
37type TurnStartRequestedDomainEvent = Extract<
38 OrchestrationEvent,
39 { type: "thread.turn-start-requested" }
40>;
41
42type RuntimeIngestionInput =
43 | {
44 source: "runtime";
45 event: ProviderRuntimeEvent;
46 }
47 | {
48 source: "domain";
49 event: TurnStartRequestedDomainEvent;
50 };
2451
2552function toTurnId(value: string | undefined): TurnId | undefined {
2653 return value === undefined ? undefined : TurnId.makeUnsafe(value);
......@@ -135,6 +162,26 @@ const make = Effect.gen(function* () {
135162 timeToLive: TURN_MESSAGE_IDS_BY_TURN_TTL,
136163 lookup: () => Effect.succeed(new Set<MessageId>()),
137164 });
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 });
180 const bufferedAssistantTextByMessageId = yield* Cache.make<MessageId, string>({
181 capacity: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY,
182 timeToLive: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL,
183 lookup: () => Effect.succeed(""),
184 });
138185
139186 const rememberAssistantMessageId = (
140187 sessionId: ProviderSessionId,
......@@ -189,6 +236,144 @@ const make = Effect.gen(function* () {
189236 const clearAssistantMessageIdsForTurn = (sessionId: ProviderSessionId, turnId: TurnId) =>
190237 Cache.invalidate(turnMessageIdsByTurnKey, providerTurnKey(sessionId, turnId));
191238
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
278 const appendBufferedAssistantText = (messageId: MessageId, delta: string) =>
279 Cache.getOption(bufferedAssistantTextByMessageId, messageId).pipe(
280 Effect.flatMap((existingText) =>
281 Cache.set(
282 bufferedAssistantTextByMessageId,
283 messageId,
284 Option.match(existingText, {
285 onNone: () => delta,
286 onSome: (text) => `${text}${delta}`,
287 }),
288 ),
289 ),
290 );
291
292 const takeBufferedAssistantText = (messageId: MessageId) =>
293 Cache.getOption(bufferedAssistantTextByMessageId, messageId).pipe(
294 Effect.flatMap((existingText) =>
295 Cache.invalidate(bufferedAssistantTextByMessageId, messageId).pipe(
296 Effect.as(Option.getOrElse(existingText, () => "")),
297 ),
298 ),
299 );
300
301 const clearBufferedAssistantText = (messageId: MessageId) =>
302 Cache.invalidate(bufferedAssistantTextByMessageId, messageId);
303
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 });
349
350 const dispatchAssistantCompletion = (input: {
351 event: ProviderRuntimeEvent;
352 threadId: ThreadId;
353 messageId: MessageId;
354 turnId?: TurnId;
355 assistantDeliveryMode: AssistantDeliveryMode;
356 createdAt: string;
357 commandTag: string;
358 }) =>
359 Effect.gen(function* () {
360 const text =
361 input.assistantDeliveryMode === "buffered"
362 ? yield* takeBufferedAssistantText(input.messageId)
363 : undefined;
364
365 yield* orchestrationEngine.dispatch({
366 type: "thread.message.assistant.complete",
367 commandId: providerCommandId(input.event, input.commandTag),
368 threadId: input.threadId,
369 messageId: input.messageId,
370 ...(input.turnId ? { turnId: input.turnId } : {}),
371 ...(text !== undefined ? { text } : {}),
372 createdAt: input.createdAt,
373 });
374 yield* clearAssistantMessageState(input.messageId);
375 });
376
192377 const clearTurnStateForSession = (sessionId: ProviderSessionId) =>
193378 Effect.gen(function* () {
194379 const prefix = `${sessionId}:`;
......@@ -196,12 +381,35 @@ const make = Effect.gen(function* () {
196381 yield* Effect.forEach(
197382 turnKeys,
198383 (key) =>
199 key.startsWith(prefix) ? Cache.invalidate(turnMessageIdsByTurnKey, key) : Effect.void,
384 Effect.gen(function* () {
385 if (!key.startsWith(prefix)) {
386 return;
387 }
388
389 const messageIds = yield* Cache.getOption(turnMessageIdsByTurnKey, key);
390 if (Option.isSome(messageIds)) {
391 yield* Effect.forEach(messageIds.value, clearAssistantMessageState, {
392 concurrency: 1,
393 }).pipe(Effect.asVoid);
394 }
395
396 yield* Cache.invalidate(turnMessageIdsByTurnKey, key);
397 }),
398 { concurrency: 1 },
399 ).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,
200408 { concurrency: 1 },
201409 ).pipe(Effect.asVoid);
202410 });
203411
204 const processEvent = (event: ProviderRuntimeEvent) =>
412 const processRuntimeEvent = (event: ProviderRuntimeEvent) =>
205413 Effect.gen(function* () {
206414 const readModel = yield* orchestrationEngine.getReadModel();
207415 const thread = readModel.threads.find(
......@@ -218,8 +426,19 @@ const make = Effect.gen(function* () {
218426 event.type === "turn.started" ||
219427 event.type === "turn.completed"
220428 ) {
221 const activeTurnId =
222 event.type === "turn.started" ? (toTurnId(event.turnId) ?? null) : null;
429 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 }
223442 const providerThreadIdFromEvent =
224443 event.type === "thread.started"
225444 ? ProviderThreadId.makeUnsafe(event.threadId)
......@@ -269,16 +488,27 @@ const make = Effect.gen(function* () {
269488 if (turnId) {
270489 yield* rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId);
271490 }
272
273 yield* orchestrationEngine.dispatch({
274 type: "thread.message.assistant.delta",
275 commandId: providerCommandId(event, "assistant-delta"),
491 const assistantDeliveryMode = yield* resolveAssistantDeliveryModeForMessage({
276492 threadId: thread.id,
493 sessionId: event.sessionId,
494 turnId,
277495 messageId: assistantMessageId,
278 delta: event.delta,
279 ...(turnId ? { turnId } : {}),
280 createdAt: now,
281496 });
497 yield* rememberAssistantDeliveryModeForMessage(assistantMessageId, assistantDeliveryMode);
498
499 if (assistantDeliveryMode === "buffered") {
500 yield* appendBufferedAssistantText(assistantMessageId, event.delta);
501 } else {
502 yield* orchestrationEngine.dispatch({
503 type: "thread.message.assistant.delta",
504 commandId: providerCommandId(event, "assistant-delta"),
505 threadId: thread.id,
506 messageId: assistantMessageId,
507 delta: event.delta,
508 ...(turnId ? { turnId } : {}),
509 createdAt: now,
510 });
511 }
282512 }
283513
284514 if (event.type === "message.completed") {
......@@ -287,14 +517,22 @@ const make = Effect.gen(function* () {
287517 if (turnId) {
288518 yield* rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId);
289519 }
520 const assistantDeliveryMode = yield* resolveAssistantDeliveryModeForMessage({
521 threadId: thread.id,
522 sessionId: event.sessionId,
523 turnId,
524 messageId: assistantMessageId,
525 });
526 yield* rememberAssistantDeliveryModeForMessage(assistantMessageId, assistantDeliveryMode);
290527
291 yield* orchestrationEngine.dispatch({
292 type: "thread.message.assistant.complete",
293 commandId: providerCommandId(event, "assistant-complete"),
528 yield* dispatchAssistantCompletion({
529 event,
294530 threadId: thread.id,
295531 messageId: assistantMessageId,
296532 ...(turnId ? { turnId } : {}),
533 assistantDeliveryMode,
297534 createdAt: now,
535 commandTag: "assistant-complete",
298536 });
299537
300538 if (turnId) {
......@@ -306,22 +544,37 @@ const make = Effect.gen(function* () {
306544 const turnId = toTurnId(event.turnId);
307545 if (turnId) {
308546 const assistantMessageIds = yield* getAssistantMessageIdsForTurn(event.sessionId, turnId);
309 yield* Effect.forEach(assistantMessageIds, (assistantMessageId) =>
310 orchestrationEngine.dispatch({
311 type: "thread.message.assistant.complete",
312 commandId: providerCommandId(event, "assistant-complete-finalize"),
313 threadId: thread.id,
314 messageId: assistantMessageId,
315 turnId,
316 createdAt: now,
317 }),
547 yield* Effect.forEach(
548 assistantMessageIds,
549 (assistantMessageId) =>
550 Effect.gen(function* () {
551 const assistantDeliveryMode =
552 yield* resolveAssistantDeliveryModeForMessage({
553 threadId: thread.id,
554 sessionId: event.sessionId,
555 turnId,
556 messageId: assistantMessageId,
557 });
558 yield* dispatchAssistantCompletion({
559 event,
560 threadId: thread.id,
561 messageId: assistantMessageId,
562 turnId,
563 assistantDeliveryMode,
564 createdAt: now,
565 commandTag: "assistant-complete-finalize",
566 });
567 }),
568 { concurrency: 1 },
318569 ).pipe(Effect.asVoid);
319570 yield* clearAssistantMessageIdsForTurn(event.sessionId, turnId);
571 yield* clearAssistantDeliveryModeForTurn(event.sessionId, turnId);
320572 }
321573 }
322574
323575 if (event.type === "session.exited") {
324576 yield* clearTurnStateForSession(event.sessionId);
577 yield* clearPendingAssistantDeliveryModeForThread(thread.id);
325578 }
326579
327580 if (event.type === "runtime.error") {
......@@ -360,33 +613,50 @@ const make = Effect.gen(function* () {
360613 ).pipe(Effect.asVoid);
361614 });
362615
363 const processEventSafely = (event: ProviderRuntimeEvent) =>
364 processEvent(event).pipe(
616 const processDomainEvent = (event: TurnStartRequestedDomainEvent) =>
617 rememberPendingAssistantDeliveryModeForThread(
618 event.payload.threadId,
619 event.payload.assistantDeliveryMode ?? DEFAULT_ASSISTANT_DELIVERY_MODE,
620 );
621
622 const processInput = (input: RuntimeIngestionInput) =>
623 input.source === "runtime" ? processRuntimeEvent(input.event) : processDomainEvent(input.event);
624
625 const processInputSafely = (input: RuntimeIngestionInput) =>
626 processInput(input).pipe(
365627 Effect.catchCause((cause) => {
366628 if (Cause.hasInterruptsOnly(cause)) {
367629 return Effect.failCause(cause);
368630 }
369631 return Effect.logWarning("provider runtime ingestion failed to process event", {
370 eventId: event.eventId,
371 eventType: event.type,
632 source: input.source,
633 eventId: input.event.eventId,
634 eventType: input.event.type,
372635 cause: Cause.pretty(cause),
373636 });
374637 }),
375638 );
376639
377640 const start: ProviderRuntimeIngestionShape["start"] = Effect.gen(function* () {
378 const providerEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
379
380 yield* Effect.addFinalizer(() => Queue.shutdown(providerEventQueue).pipe(Effect.asVoid));
641 const inputQueue = yield* Queue.unbounded<RuntimeIngestionInput>();
642 yield* Effect.addFinalizer(() => Queue.shutdown(inputQueue).pipe(Effect.asVoid));
381643
382644 yield* Effect.forkScoped(
383 Effect.forever(Queue.take(providerEventQueue).pipe(Effect.flatMap(processEventSafely))),
645 Effect.forever(Queue.take(inputQueue).pipe(Effect.flatMap(processInputSafely))),
384646 );
385647 yield* Effect.forkScoped(
386648 Stream.runForEach(providerService.streamEvents, (event) =>
387 Queue.offer(providerEventQueue, event).pipe(Effect.asVoid),
649 Queue.offer(inputQueue, { source: "runtime", event }).pipe(Effect.asVoid),
388650 ),
389651 );
652 yield* Effect.forkScoped(
653 Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => {
654 if (event.type !== "thread.turn-start-requested") {
655 return Effect.void;
656 }
657 return Queue.offer(inputQueue, { source: "domain", event }).pipe(Effect.asVoid);
658 }),
659 );
390660 });
391661
392662 return {
apps/server/src/orchestration/decider.projectScripts.test.ts+3
......@@ -164,5 +164,8 @@ describe("decider project scripts", () => {
164164 expect(events[0]?.type).toBe("thread.message-sent");
165165 expect(events[1]?.type).toBe("thread.turn-start-requested");
166166 expect(events[1]?.causationEventId).toBe(events[0]?.eventId ?? null);
167 expect((events[1]?.payload as { assistantDeliveryMode?: string }).assistantDeliveryMode).toBe(
168 "buffered",
169 );
167170 });
168171});
apps/server/src/orchestration/decider.ts+3-1
......@@ -14,6 +14,7 @@ import {
1414} from "./commandInvariants.ts";
1515
1616const nowIso = () => new Date().toISOString();
17const DEFAULT_ASSISTANT_DELIVERY_MODE = "buffered" as const;
1718
1819const defaultMetadata: Omit<OrchestrationEvent, "sequence" | "type" | "payload"> = {
1920 eventId: crypto.randomUUID() as OrchestrationEvent["eventId"],
......@@ -252,6 +253,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
252253 messageId: command.message.messageId,
253254 ...(command.model !== undefined ? { model: command.model } : {}),
254255 ...(command.effort !== undefined ? { effort: command.effort } : {}),
256 assistantDeliveryMode: command.assistantDeliveryMode ?? DEFAULT_ASSISTANT_DELIVERY_MODE,
255257 createdAt: command.createdAt,
256258 },
257259 };
......@@ -423,7 +425,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
423425 threadId: command.threadId,
424426 messageId: command.messageId,
425427 role: "assistant",
426 text: "",
428 text: command.text ?? "",
427429 turnId: command.turnId ?? null,
428430 streaming: false,
429431 createdAt: command.createdAt,
apps/server/src/wsServer.test.ts+1
......@@ -959,6 +959,7 @@ describe("WebSocket Server", () => {
959959 text: "hello",
960960 attachments: [],
961961 },
962 assistantDeliveryMode: "streaming",
962963 createdAt,
963964 });
964965 expect(startTurnResponse.error).toBeUndefined();
apps/web/src/appSettings.ts+3
......@@ -11,6 +11,9 @@ const AppSettingsSchema = Schema.Struct({
1111 Schema.withConstructorDefault(() => Option.some("")),
1212 ),
1313 confirmThreadDelete: Schema.Boolean.pipe(Schema.withConstructorDefault(() => Option.some(true))),
14 enableAssistantStreaming: Schema.Boolean.pipe(
15 Schema.withConstructorDefault(() => Option.some(false)),
16 ),
1417});
1518export type AppSettings = typeof AppSettingsSchema.Type;
1619
apps/web/src/components/ChatView.tsx+3
......@@ -107,6 +107,7 @@ import { Toggle } from "./ui/toggle";
107107import { SidebarTrigger } from "./ui/sidebar";
108108import { newCommandId, newMessageId } from "~/lib/utils";
109109import { readNativeApi } from "~/nativeApi";
110import { useAppSettings } from "../appSettings";
110111
111112function formatMessageMeta(createdAt: string, duration: string | null): string {
112113 if (!duration) return formatTimestamp(createdAt);
......@@ -332,6 +333,7 @@ interface ChatViewProps {
332333
333334export default function ChatView({ threadId }: ChatViewProps) {
334335 const { state, dispatch } = useStore();
336 const { settings } = useAppSettings();
335337 const navigate = useNavigate();
336338 const rawSearch = useSearch({
337339 strict: false,
......@@ -1506,6 +1508,7 @@ export default function ChatView({ threadId }: ChatViewProps) {
15061508 },
15071509 model: selectedModel || undefined,
15081510 effort: selectedEffort || undefined,
1511 assistantDeliveryMode: settings.enableAssistantStreaming ? "streaming" : "buffered",
15091512 createdAt: messageCreatedAt,
15101513 });
15111514 } catch (err) {
apps/web/src/routes/_chat.settings.tsx+43
......@@ -237,6 +237,49 @@ function SettingsRouteView() {
237237 </div>
238238 </section>
239239
240 <section className="rounded-2xl border border-border bg-card p-5">
241 <div className="mb-4">
242 <h2 className="text-sm font-medium text-foreground">Responses</h2>
243 <p className="mt-1 text-xs text-muted-foreground">
244 Control how assistant output is rendered during a turn.
245 </p>
246 </div>
247
248 <div className="flex items-center justify-between rounded-lg border border-border bg-background px-3 py-2">
249 <div>
250 <p className="text-sm font-medium text-foreground">Stream assistant messages</p>
251 <p className="text-xs text-muted-foreground">
252 Show token-by-token output while a response is in progress.
253 </p>
254 </div>
255 <Switch
256 checked={settings.enableAssistantStreaming}
257 onCheckedChange={(checked) =>
258 updateSettings({
259 enableAssistantStreaming: Boolean(checked),
260 })
261 }
262 aria-label="Stream assistant messages"
263 />
264 </div>
265
266 {settings.enableAssistantStreaming !== defaults.enableAssistantStreaming ? (
267 <div className="mt-3 flex justify-end">
268 <Button
269 size="xs"
270 variant="outline"
271 onClick={() =>
272 updateSettings({
273 enableAssistantStreaming: defaults.enableAssistantStreaming,
274 })
275 }
276 >
277 Restore default
278 </Button>
279 </div>
280 ) : null}
281 </section>
282
240283 <section className="rounded-2xl border border-border bg-card p-5">
241284 <div className="mb-4">
242285 <h2 className="text-sm font-medium text-foreground">Keybindings</h2>
packages/contracts/src/orchestration.ts+5
......@@ -46,6 +46,8 @@ export const ProviderSandboxMode = Schema.Literals([
4646export type ProviderSandboxMode = typeof ProviderSandboxMode.Type;
4747export const ProviderRequestKind = Schema.Literals(["command", "file-change"]);
4848export type ProviderRequestKind = typeof ProviderRequestKind.Type;
49export const AssistantDeliveryMode = Schema.Literals(["buffered", "streaming"]);
50export type AssistantDeliveryMode = typeof AssistantDeliveryMode.Type;
4951export const ProviderApprovalDecision = Schema.Literals([
5052 "accept",
5153 "acceptForSession",
......@@ -279,6 +281,7 @@ export const ThreadTurnStartCommand = Schema.Struct({
279281 }),
280282 model: Schema.optional(TrimmedNonEmptyString),
281283 effort: Schema.optional(TrimmedNonEmptyString),
284 assistantDeliveryMode: Schema.optional(AssistantDeliveryMode),
282285 createdAt: IsoDateTime,
283286});
284287
......@@ -353,6 +356,7 @@ export const ThreadMessageAssistantCompleteCommand = Schema.Struct({
353356 threadId: ThreadId,
354357 messageId: MessageId,
355358 turnId: Schema.optional(TurnId),
359 text: Schema.optional(Schema.String),
356360 createdAt: IsoDateTime,
357361});
358362
......@@ -492,6 +496,7 @@ export const ThreadTurnStartRequestedPayload = Schema.Struct({
492496 messageId: MessageId,
493497 model: Schema.optional(TrimmedNonEmptyString),
494498 effort: Schema.optional(TrimmedNonEmptyString),
499 assistantDeliveryMode: Schema.optional(AssistantDeliveryMode),
495500 createdAt: IsoDateTime,
496501});
497502