From 001902c881101f0c4d35cebcf2303ddb75f33534 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 26 Feb 2026 18:24:17 -0800 Subject: [PATCH] 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 --- .../Layers/ProviderRuntimeIngestion.test.ts | 144 ++++++++ .../Layers/ProviderRuntimeIngestion.ts | 334 ++++++++++++++++-- .../decider.projectScripts.test.ts | 3 + apps/server/src/orchestration/decider.ts | 4 +- apps/server/src/wsServer.test.ts | 1 + apps/web/src/appSettings.ts | 3 + apps/web/src/components/ChatView.tsx | 3 + apps/web/src/routes/_chat.settings.tsx | 43 +++ packages/contracts/src/orchestration.ts | 5 + 9 files changed, 507 insertions(+), 33 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 5a87fa973f0380940b196588bab4e7abfa9c29a7..245b6a5cbb2195376ba9e4e7b9e92e73dc4fa6a5 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -2,6 +2,7 @@ import type { ProviderRuntimeEvent } from "@t3tools/contracts"; import { CommandId, EventId, + MessageId, ProjectId, ProviderItemId, ProviderSessionId, @@ -33,6 +34,7 @@ const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.make const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); const asItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(value); const asEventId = (value: string): EventId => EventId.makeUnsafe(value); +const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); function createProviderServiceHarness() { const runtimeEventPubSub = Effect.runSync(PubSub.unbounded()); @@ -258,6 +260,148 @@ describe("ProviderRuntimeIngestion", () => { expect(message?.streaming).toBe(false); }); + it("buffers assistant deltas by default until completion", async () => { + const harness = await createHarness(); + const now = new Date().toISOString(); + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-turn-started-buffered"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: now, + turnId: asProviderTurnId("turn-buffered"), + }); + await waitForThread( + harness.engine, + (thread) => + thread.session?.status === "running" && thread.session?.activeTurnId === "turn-buffered", + ); + + harness.emit({ + type: "message.delta", + eventId: asEventId("evt-message-delta-buffered"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: now, + turnId: asProviderTurnId("turn-buffered"), + itemId: asItemId("item-buffered"), + delta: "buffer me", + }); + + await Effect.runPromise(Effect.sleep("30 millis")); + const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); + const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); + expect(midThread?.messages.some((message) => message.id === "assistant:item-buffered")).toBe( + false, + ); + + harness.emit({ + type: "message.completed", + eventId: asEventId("evt-message-completed-buffered"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: now, + turnId: asProviderTurnId("turn-buffered"), + itemId: asItemId("item-buffered"), + }); + + const thread = await waitForThread( + harness.engine, + (entry) => + entry.messages.some((message) => message.id === "assistant:item-buffered" && !message.streaming), + ); + const message = thread.messages.find((entry) => entry.id === "assistant:item-buffered"); + expect(message?.text).toBe("buffer me"); + expect(message?.streaming).toBe(false); + }); + + it("streams assistant deltas when thread.turn.start requests streaming mode", async () => { + const harness = await createHarness(); + const now = new Date().toISOString(); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.makeUnsafe("cmd-turn-start-streaming-mode"), + threadId: ThreadId.makeUnsafe("thread-1"), + message: { + messageId: asMessageId("message-streaming-mode"), + role: "user", + text: "stream please", + attachments: [], + }, + assistantDeliveryMode: "streaming", + createdAt: now, + }), + ); + await Effect.runPromise(Effect.sleep("30 millis")); + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-turn-started-streaming-mode"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: now, + turnId: asProviderTurnId("turn-streaming-mode"), + }); + await waitForThread( + harness.engine, + (thread) => + thread.session?.status === "running" && + thread.session?.activeTurnId === "turn-streaming-mode", + ); + + harness.emit({ + type: "message.delta", + eventId: asEventId("evt-message-delta-streaming-mode"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: now, + turnId: asProviderTurnId("turn-streaming-mode"), + itemId: asItemId("item-streaming-mode"), + delta: "hello live", + }); + + const liveThread = await waitForThread( + harness.engine, + (entry) => + entry.messages.some( + (message) => + message.id === "assistant:item-streaming-mode" && + message.streaming && + message.text === "hello live", + ), + ); + const liveMessage = liveThread.messages.find( + (entry) => entry.id === "assistant:item-streaming-mode", + ); + expect(liveMessage?.streaming).toBe(true); + + harness.emit({ + type: "message.completed", + eventId: asEventId("evt-message-completed-streaming-mode"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: now, + turnId: asProviderTurnId("turn-streaming-mode"), + itemId: asItemId("item-streaming-mode"), + }); + + const finalThread = await waitForThread( + harness.engine, + (entry) => + entry.messages.some( + (message) => message.id === "assistant:item-streaming-mode" && !message.streaming, + ), + ); + const finalMessage = finalThread.messages.find( + (entry) => entry.id === "assistant:item-streaming-mode", + ); + expect(finalMessage?.text).toBe("hello live"); + expect(finalMessage?.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 e76a355217d5d054da107e9c66f39673f1bbe5b5..420eee0da8aa82884d2b5b4fc6db23a90a71ae13 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -1,7 +1,10 @@ import { + type AssistantDeliveryMode, CommandId, MessageId, + type OrchestrationEvent, ProviderThreadId, + type ThreadId, TurnId, type OrchestrationThreadActivity, type ProviderRuntimeEvent, @@ -19,8 +22,32 @@ import { const providerTurnKey = (sessionId: ProviderSessionId, turnId: TurnId) => `${sessionId}:${turnId}`; const providerCommandId = (event: ProviderRuntimeEvent, tag: string): CommandId => CommandId.makeUnsafe(`provider:${event.eventId}:${tag}:${crypto.randomUUID()}`); +const DEFAULT_ASSISTANT_DELIVERY_MODE: AssistantDeliveryMode = "buffered"; const TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY = 10_000; const TURN_MESSAGE_IDS_BY_TURN_TTL = Duration.minutes(120); +const TURN_DELIVERY_MODE_BY_TURN_CACHE_CAPACITY = 10_000; +const TURN_DELIVERY_MODE_BY_TURN_TTL = Duration.minutes(120); +const PENDING_DELIVERY_MODE_BY_THREAD_CACHE_CAPACITY = 10_000; +const PENDING_DELIVERY_MODE_BY_THREAD_TTL = Duration.minutes(30); +const MESSAGE_DELIVERY_MODE_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000; +const MESSAGE_DELIVERY_MODE_BY_MESSAGE_ID_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); + +type TurnStartRequestedDomainEvent = Extract< + OrchestrationEvent, + { type: "thread.turn-start-requested" } +>; + +type RuntimeIngestionInput = + | { + source: "runtime"; + event: ProviderRuntimeEvent; + } + | { + source: "domain"; + event: TurnStartRequestedDomainEvent; + }; function toTurnId(value: string | undefined): TurnId | undefined { return value === undefined ? undefined : TurnId.makeUnsafe(value); @@ -135,6 +162,26 @@ const make = Effect.gen(function* () { timeToLive: TURN_MESSAGE_IDS_BY_TURN_TTL, lookup: () => Effect.succeed(new Set()), }); + const assistantDeliveryModeByTurnKey = yield* Cache.make({ + capacity: TURN_DELIVERY_MODE_BY_TURN_CACHE_CAPACITY, + timeToLive: TURN_DELIVERY_MODE_BY_TURN_TTL, + lookup: () => Effect.succeed(DEFAULT_ASSISTANT_DELIVERY_MODE), + }); + const pendingAssistantDeliveryModeByThreadId = yield* Cache.make({ + capacity: PENDING_DELIVERY_MODE_BY_THREAD_CACHE_CAPACITY, + timeToLive: PENDING_DELIVERY_MODE_BY_THREAD_TTL, + lookup: () => Effect.succeed(DEFAULT_ASSISTANT_DELIVERY_MODE), + }); + const assistantDeliveryModeByMessageId = yield* Cache.make({ + capacity: MESSAGE_DELIVERY_MODE_BY_MESSAGE_ID_CACHE_CAPACITY, + timeToLive: MESSAGE_DELIVERY_MODE_BY_MESSAGE_ID_TTL, + lookup: () => Effect.succeed(DEFAULT_ASSISTANT_DELIVERY_MODE), + }); + const bufferedAssistantTextByMessageId = yield* Cache.make({ + capacity: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY, + timeToLive: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL, + lookup: () => Effect.succeed(""), + }); const rememberAssistantMessageId = ( sessionId: ProviderSessionId, @@ -189,6 +236,144 @@ const make = Effect.gen(function* () { const clearAssistantMessageIdsForTurn = (sessionId: ProviderSessionId, turnId: TurnId) => Cache.invalidate(turnMessageIdsByTurnKey, providerTurnKey(sessionId, turnId)); + const rememberAssistantDeliveryModeForTurn = ( + sessionId: ProviderSessionId, + turnId: TurnId, + assistantDeliveryMode: AssistantDeliveryMode, + ) => + Cache.set( + assistantDeliveryModeByTurnKey, + providerTurnKey(sessionId, turnId), + assistantDeliveryMode, + ); + + const clearAssistantDeliveryModeForTurn = (sessionId: ProviderSessionId, turnId: TurnId) => + Cache.invalidate(assistantDeliveryModeByTurnKey, providerTurnKey(sessionId, turnId)); + + const rememberPendingAssistantDeliveryModeForThread = ( + threadId: ThreadId, + assistantDeliveryMode: AssistantDeliveryMode, + ) => Cache.set(pendingAssistantDeliveryModeByThreadId, threadId, assistantDeliveryMode); + + const takePendingAssistantDeliveryModeForThread = (threadId: ThreadId) => + Cache.getOption(pendingAssistantDeliveryModeByThreadId, threadId).pipe( + Effect.flatMap((assistantDeliveryMode) => + Cache.invalidate(pendingAssistantDeliveryModeByThreadId, threadId).pipe( + Effect.as(assistantDeliveryMode), + ), + ), + ); + + const clearPendingAssistantDeliveryModeForThread = (threadId: ThreadId) => + Cache.invalidate(pendingAssistantDeliveryModeByThreadId, threadId); + + const rememberAssistantDeliveryModeForMessage = ( + messageId: MessageId, + assistantDeliveryMode: AssistantDeliveryMode, + ) => Cache.set(assistantDeliveryModeByMessageId, messageId, assistantDeliveryMode); + + const clearAssistantDeliveryModeForMessage = (messageId: MessageId) => + Cache.invalidate(assistantDeliveryModeByMessageId, messageId); + + const appendBufferedAssistantText = (messageId: MessageId, delta: string) => + Cache.getOption(bufferedAssistantTextByMessageId, messageId).pipe( + Effect.flatMap((existingText) => + Cache.set( + bufferedAssistantTextByMessageId, + messageId, + Option.match(existingText, { + onNone: () => delta, + onSome: (text) => `${text}${delta}`, + }), + ), + ), + ); + + const takeBufferedAssistantText = (messageId: MessageId) => + Cache.getOption(bufferedAssistantTextByMessageId, messageId).pipe( + Effect.flatMap((existingText) => + Cache.invalidate(bufferedAssistantTextByMessageId, messageId).pipe( + Effect.as(Option.getOrElse(existingText, () => "")), + ), + ), + ); + + const clearBufferedAssistantText = (messageId: MessageId) => + Cache.invalidate(bufferedAssistantTextByMessageId, messageId); + + const clearAssistantMessageState = (messageId: MessageId) => + Effect.all( + [ + clearAssistantDeliveryModeForMessage(messageId), + clearBufferedAssistantText(messageId), + ], + { concurrency: 1 }, + ).pipe(Effect.asVoid); + + const resolveAssistantDeliveryModeForMessage = (input: { + threadId: ThreadId; + sessionId: ProviderSessionId; + turnId?: TurnId; + messageId: MessageId; + }) => + Effect.gen(function* () { + const messageMode = yield* Cache.getOption(assistantDeliveryModeByMessageId, input.messageId); + if (Option.isSome(messageMode)) { + return messageMode.value; + } + + if (input.turnId) { + const turnMode = yield* Cache.getOption( + assistantDeliveryModeByTurnKey, + providerTurnKey(input.sessionId, input.turnId), + ); + if (Option.isSome(turnMode)) { + return turnMode.value; + } + } + + const pendingMode = yield* Cache.getOption( + pendingAssistantDeliveryModeByThreadId, + input.threadId, + ); + if (Option.isSome(pendingMode)) { + if (input.turnId) { + yield* rememberAssistantDeliveryModeForTurn(input.sessionId, input.turnId, pendingMode.value); + yield* clearPendingAssistantDeliveryModeForThread(input.threadId); + } + return pendingMode.value; + } + + return DEFAULT_ASSISTANT_DELIVERY_MODE; + }); + + const dispatchAssistantCompletion = (input: { + event: ProviderRuntimeEvent; + threadId: ThreadId; + messageId: MessageId; + turnId?: TurnId; + assistantDeliveryMode: AssistantDeliveryMode; + createdAt: string; + commandTag: string; + }) => + Effect.gen(function* () { + const text = + input.assistantDeliveryMode === "buffered" + ? yield* takeBufferedAssistantText(input.messageId) + : undefined; + + yield* orchestrationEngine.dispatch({ + type: "thread.message.assistant.complete", + commandId: providerCommandId(input.event, input.commandTag), + threadId: input.threadId, + messageId: input.messageId, + ...(input.turnId ? { turnId: input.turnId } : {}), + ...(text !== undefined ? { text } : {}), + createdAt: input.createdAt, + }); + yield* clearAssistantMessageState(input.messageId); + }); + const clearTurnStateForSession = (sessionId: ProviderSessionId) => Effect.gen(function* () { const prefix = `${sessionId}:`; @@ -196,12 +381,35 @@ const make = Effect.gen(function* () { yield* Effect.forEach( turnKeys, (key) => - key.startsWith(prefix) ? Cache.invalidate(turnMessageIdsByTurnKey, key) : Effect.void, + Effect.gen(function* () { + if (!key.startsWith(prefix)) { + return; + } + + const messageIds = yield* Cache.getOption(turnMessageIdsByTurnKey, key); + if (Option.isSome(messageIds)) { + yield* Effect.forEach(messageIds.value, clearAssistantMessageState, { + concurrency: 1, + }).pipe(Effect.asVoid); + } + + yield* Cache.invalidate(turnMessageIdsByTurnKey, key); + }), + { concurrency: 1 }, + ).pipe(Effect.asVoid); + + const deliveryModeKeys = Array.from(yield* Cache.keys(assistantDeliveryModeByTurnKey)); + yield* Effect.forEach( + deliveryModeKeys, + (key) => + key.startsWith(prefix) + ? Cache.invalidate(assistantDeliveryModeByTurnKey, key) + : Effect.void, { concurrency: 1 }, ).pipe(Effect.asVoid); }); - const processEvent = (event: ProviderRuntimeEvent) => + const processRuntimeEvent = (event: ProviderRuntimeEvent) => Effect.gen(function* () { const readModel = yield* orchestrationEngine.getReadModel(); const thread = readModel.threads.find( @@ -218,8 +426,19 @@ const make = Effect.gen(function* () { event.type === "turn.started" || event.type === "turn.completed" ) { - const activeTurnId = - event.type === "turn.started" ? (toTurnId(event.turnId) ?? null) : null; + const activeTurnId = event.type === "turn.started" ? (toTurnId(event.turnId) ?? null) : null; + if (event.type === "turn.started" && activeTurnId) { + const pendingAssistantDeliveryMode = yield* takePendingAssistantDeliveryModeForThread( + thread.id, + ); + if (Option.isSome(pendingAssistantDeliveryMode)) { + yield* rememberAssistantDeliveryModeForTurn( + event.sessionId, + activeTurnId, + pendingAssistantDeliveryMode.value, + ); + } + } const providerThreadIdFromEvent = event.type === "thread.started" ? ProviderThreadId.makeUnsafe(event.threadId) @@ -269,16 +488,27 @@ const make = Effect.gen(function* () { if (turnId) { yield* rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId); } - - yield* orchestrationEngine.dispatch({ - type: "thread.message.assistant.delta", - commandId: providerCommandId(event, "assistant-delta"), + const assistantDeliveryMode = yield* resolveAssistantDeliveryModeForMessage({ threadId: thread.id, + sessionId: event.sessionId, + turnId, messageId: assistantMessageId, - delta: event.delta, - ...(turnId ? { turnId } : {}), - createdAt: now, }); + yield* rememberAssistantDeliveryModeForMessage(assistantMessageId, assistantDeliveryMode); + + if (assistantDeliveryMode === "buffered") { + yield* appendBufferedAssistantText(assistantMessageId, event.delta); + } else { + yield* orchestrationEngine.dispatch({ + type: "thread.message.assistant.delta", + commandId: providerCommandId(event, "assistant-delta"), + threadId: thread.id, + messageId: assistantMessageId, + delta: event.delta, + ...(turnId ? { turnId } : {}), + createdAt: now, + }); + } } if (event.type === "message.completed") { @@ -287,14 +517,22 @@ const make = Effect.gen(function* () { if (turnId) { yield* rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId); } + const assistantDeliveryMode = yield* resolveAssistantDeliveryModeForMessage({ + threadId: thread.id, + sessionId: event.sessionId, + turnId, + messageId: assistantMessageId, + }); + yield* rememberAssistantDeliveryModeForMessage(assistantMessageId, assistantDeliveryMode); - yield* orchestrationEngine.dispatch({ - type: "thread.message.assistant.complete", - commandId: providerCommandId(event, "assistant-complete"), + yield* dispatchAssistantCompletion({ + event, threadId: thread.id, messageId: assistantMessageId, ...(turnId ? { turnId } : {}), + assistantDeliveryMode, createdAt: now, + commandTag: "assistant-complete", }); if (turnId) { @@ -306,22 +544,37 @@ const make = Effect.gen(function* () { const turnId = toTurnId(event.turnId); if (turnId) { const assistantMessageIds = yield* getAssistantMessageIdsForTurn(event.sessionId, turnId); - yield* Effect.forEach(assistantMessageIds, (assistantMessageId) => - orchestrationEngine.dispatch({ - type: "thread.message.assistant.complete", - commandId: providerCommandId(event, "assistant-complete-finalize"), - threadId: thread.id, - messageId: assistantMessageId, - turnId, - createdAt: now, - }), + yield* Effect.forEach( + assistantMessageIds, + (assistantMessageId) => + Effect.gen(function* () { + const assistantDeliveryMode = + yield* resolveAssistantDeliveryModeForMessage({ + threadId: thread.id, + sessionId: event.sessionId, + turnId, + messageId: assistantMessageId, + }); + yield* dispatchAssistantCompletion({ + event, + threadId: thread.id, + messageId: assistantMessageId, + turnId, + assistantDeliveryMode, + createdAt: now, + commandTag: "assistant-complete-finalize", + }); + }), + { concurrency: 1 }, ).pipe(Effect.asVoid); yield* clearAssistantMessageIdsForTurn(event.sessionId, turnId); + yield* clearAssistantDeliveryModeForTurn(event.sessionId, turnId); } } if (event.type === "session.exited") { yield* clearTurnStateForSession(event.sessionId); + yield* clearPendingAssistantDeliveryModeForThread(thread.id); } if (event.type === "runtime.error") { @@ -360,33 +613,50 @@ const make = Effect.gen(function* () { ).pipe(Effect.asVoid); }); - const processEventSafely = (event: ProviderRuntimeEvent) => - processEvent(event).pipe( + const processDomainEvent = (event: TurnStartRequestedDomainEvent) => + rememberPendingAssistantDeliveryModeForThread( + event.payload.threadId, + event.payload.assistantDeliveryMode ?? DEFAULT_ASSISTANT_DELIVERY_MODE, + ); + + const processInput = (input: RuntimeIngestionInput) => + input.source === "runtime" ? processRuntimeEvent(input.event) : processDomainEvent(input.event); + + const processInputSafely = (input: RuntimeIngestionInput) => + processInput(input).pipe( Effect.catchCause((cause) => { if (Cause.hasInterruptsOnly(cause)) { return Effect.failCause(cause); } return Effect.logWarning("provider runtime ingestion failed to process event", { - eventId: event.eventId, - eventType: event.type, + source: input.source, + eventId: input.event.eventId, + eventType: input.event.type, cause: Cause.pretty(cause), }); }), ); const start: ProviderRuntimeIngestionShape["start"] = Effect.gen(function* () { - const providerEventQueue = yield* Queue.unbounded(); - - yield* Effect.addFinalizer(() => Queue.shutdown(providerEventQueue).pipe(Effect.asVoid)); + const inputQueue = yield* Queue.unbounded(); + yield* Effect.addFinalizer(() => Queue.shutdown(inputQueue).pipe(Effect.asVoid)); yield* Effect.forkScoped( - Effect.forever(Queue.take(providerEventQueue).pipe(Effect.flatMap(processEventSafely))), + Effect.forever(Queue.take(inputQueue).pipe(Effect.flatMap(processInputSafely))), ); yield* Effect.forkScoped( Stream.runForEach(providerService.streamEvents, (event) => - Queue.offer(providerEventQueue, event).pipe(Effect.asVoid), + Queue.offer(inputQueue, { source: "runtime", event }).pipe(Effect.asVoid), ), ); + yield* Effect.forkScoped( + Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => { + if (event.type !== "thread.turn-start-requested") { + return Effect.void; + } + return Queue.offer(inputQueue, { source: "domain", event }).pipe(Effect.asVoid); + }), + ); }); return { diff --git a/apps/server/src/orchestration/decider.projectScripts.test.ts b/apps/server/src/orchestration/decider.projectScripts.test.ts index 78fbb858a0b236b69e341b9e51055f497a0df2e3..7661352962f0758c91d4f463b0b7d05f23cca960 100644 --- a/apps/server/src/orchestration/decider.projectScripts.test.ts +++ b/apps/server/src/orchestration/decider.projectScripts.test.ts @@ -164,5 +164,8 @@ describe("decider project scripts", () => { 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", + ); }); }); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index b716543d3d51b2b3134a572ba0f0f9649fa3a3f0..98acd091ac051e131b2b5efe09ed9385b3838faf 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -14,6 +14,7 @@ import { } from "./commandInvariants.ts"; const nowIso = () => new Date().toISOString(); +const DEFAULT_ASSISTANT_DELIVERY_MODE = "buffered" as const; const defaultMetadata: Omit = { eventId: crypto.randomUUID() as OrchestrationEvent["eventId"], @@ -252,6 +253,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" messageId: command.message.messageId, ...(command.model !== undefined ? { model: command.model } : {}), ...(command.effort !== undefined ? { effort: command.effort } : {}), + assistantDeliveryMode: command.assistantDeliveryMode ?? DEFAULT_ASSISTANT_DELIVERY_MODE, createdAt: command.createdAt, }, }; @@ -423,7 +425,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" threadId: command.threadId, messageId: command.messageId, role: "assistant", - text: "", + text: command.text ?? "", turnId: command.turnId ?? null, streaming: false, createdAt: command.createdAt, diff --git a/apps/server/src/wsServer.test.ts b/apps/server/src/wsServer.test.ts index 57abed02ac8d088199a99ca7ac49261913891706..78394af58557afd93d258465718e17b5bf74d66a 100644 --- a/apps/server/src/wsServer.test.ts +++ b/apps/server/src/wsServer.test.ts @@ -959,6 +959,7 @@ describe("WebSocket Server", () => { text: "hello", attachments: [], }, + assistantDeliveryMode: "streaming", createdAt, }); expect(startTurnResponse.error).toBeUndefined(); diff --git a/apps/web/src/appSettings.ts b/apps/web/src/appSettings.ts index def3272c953a7aa3719c41d836e35601b27b28bf..2c4aa9a2a7f32d53637ccced77db9b81ac306fa0 100644 --- a/apps/web/src/appSettings.ts +++ b/apps/web/src/appSettings.ts @@ -11,6 +11,9 @@ const AppSettingsSchema = Schema.Struct({ Schema.withConstructorDefault(() => Option.some("")), ), confirmThreadDelete: Schema.Boolean.pipe(Schema.withConstructorDefault(() => Option.some(true))), + enableAssistantStreaming: Schema.Boolean.pipe( + Schema.withConstructorDefault(() => Option.some(false)), + ), }); export type AppSettings = typeof AppSettingsSchema.Type; diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index c1bfa62f2b97abcf2bbf5f70957e1fd91d9466f0..95ab08d6c8f25513d1f64879b033c0a0b50aa934 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -107,6 +107,7 @@ import { Toggle } from "./ui/toggle"; import { SidebarTrigger } from "./ui/sidebar"; import { newCommandId, newMessageId } from "~/lib/utils"; import { readNativeApi } from "~/nativeApi"; +import { useAppSettings } from "../appSettings"; function formatMessageMeta(createdAt: string, duration: string | null): string { if (!duration) return formatTimestamp(createdAt); @@ -332,6 +333,7 @@ interface ChatViewProps { export default function ChatView({ threadId }: ChatViewProps) { const { state, dispatch } = useStore(); + const { settings } = useAppSettings(); const navigate = useNavigate(); const rawSearch = useSearch({ strict: false, @@ -1506,6 +1508,7 @@ export default function ChatView({ threadId }: ChatViewProps) { }, model: selectedModel || undefined, effort: selectedEffort || undefined, + assistantDeliveryMode: settings.enableAssistantStreaming ? "streaming" : "buffered", createdAt: messageCreatedAt, }); } catch (err) { diff --git a/apps/web/src/routes/_chat.settings.tsx b/apps/web/src/routes/_chat.settings.tsx index 1068502dc6f9e909467d151becc005d5e7202e7e..2fcdf64725ce6c4fd046f62b780160146fca2bea 100644 --- a/apps/web/src/routes/_chat.settings.tsx +++ b/apps/web/src/routes/_chat.settings.tsx @@ -237,6 +237,49 @@ function SettingsRouteView() { +
+
+

Responses

+

+ Control how assistant output is rendered during a turn. +

+
+ +
+
+

Stream assistant messages

+

+ Show token-by-token output while a response is in progress. +

+
+ + updateSettings({ + enableAssistantStreaming: Boolean(checked), + }) + } + aria-label="Stream assistant messages" + /> +
+ + {settings.enableAssistantStreaming !== defaults.enableAssistantStreaming ? ( +
+ +
+ ) : null} +
+

Keybindings

diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index e9d57d4f395a01537c73a548246bd7de4c20afbc..2148179d3fa3b0501fed2eb88d4ed527752169a9 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -46,6 +46,8 @@ export const ProviderSandboxMode = Schema.Literals([ export type ProviderSandboxMode = typeof ProviderSandboxMode.Type; export const ProviderRequestKind = Schema.Literals(["command", "file-change"]); export type ProviderRequestKind = typeof ProviderRequestKind.Type; +export const AssistantDeliveryMode = Schema.Literals(["buffered", "streaming"]); +export type AssistantDeliveryMode = typeof AssistantDeliveryMode.Type; export const ProviderApprovalDecision = Schema.Literals([ "accept", "acceptForSession", @@ -279,6 +281,7 @@ export const ThreadTurnStartCommand = Schema.Struct({ }), model: Schema.optional(TrimmedNonEmptyString), effort: Schema.optional(TrimmedNonEmptyString), + assistantDeliveryMode: Schema.optional(AssistantDeliveryMode), createdAt: IsoDateTime, }); @@ -353,6 +356,7 @@ export const ThreadMessageAssistantCompleteCommand = Schema.Struct({ threadId: ThreadId, messageId: MessageId, turnId: Schema.optional(TurnId), + text: Schema.optional(Schema.String), createdAt: IsoDateTime, }); @@ -492,6 +496,7 @@ export const ThreadTurnStartRequestedPayload = Schema.Struct({ messageId: MessageId, model: Schema.optional(TrimmedNonEmptyString), effort: Schema.optional(TrimmedNonEmptyString), + assistantDeliveryMode: Schema.optional(AssistantDeliveryMode), createdAt: IsoDateTime, }); -- 2.54.0