= [];
- if (input.prompt) {
- turnInput.push({
- type: "text",
- text: input.prompt,
- });
- }
- for (const attachment of input.attachments ?? []) {
- turnInput.push(attachment);
- }
-
const config = runtimeModeToThreadConfig(input.runtimeMode);
const collaborationMode = buildCodexCollaborationMode({
...(input.interactionMode ? { interactionMode: input.interactionMode } : {}),
@@ -394,7 +426,7 @@ export function buildTurnStartParams(input: {
return decodeCodexTurnStartParamsWithCollaborationMode({
threadId: input.threadId,
- input: turnInput,
+ input: buildCodexTurnInput(input),
approvalPolicy: config.approvalPolicy,
approvalsReviewer: config.approvalsReviewer,
sandboxPolicy: runtimeModeToTurnSandboxPolicy(input.runtimeMode),
@@ -1289,9 +1321,44 @@ export const makeCodexSessionRuntime = (
),
);
}
- const normalizedModel = normalizeCodexModelSlug(
- input.model ?? (yield* Ref.get(sessionRef)).model,
- );
+ const session = yield* Ref.get(sessionRef);
+ const steeringTurnId = resolveCodexSteeringTurnId(session);
+ if (steeringTurnId) {
+ const response = yield* client.request(
+ "turn/steer",
+ buildTurnSteerParams({
+ threadId: providerThreadId,
+ activeTurnId: steeringTurnId,
+ ...(input.input ? { prompt: input.input } : {}),
+ ...(input.attachments ? { attachments: input.attachments } : {}),
+ }),
+ );
+ const turnId = TurnId.make(response.turnId);
+ yield* updateSession(sessionRef, {
+ status: "running",
+ activeTurnId: turnId,
+ });
+ // Codex does not emit a second turn/started notification when
+ // turn/steer keeps the existing turn alive. Reaffirm the active
+ // turn so orchestration can reconcile the steer request with the
+ // running turn instead of leaving a stale pending-turn row.
+ yield* emitEvent({
+ kind: "notification",
+ threadId: options.threadId,
+ method: "turn/started",
+ turnId,
+ message: "Codex turn steered.",
+ });
+ const resumedProviderThreadId = currentProviderThreadId(yield* Ref.get(sessionRef));
+ return {
+ threadId: options.threadId,
+ turnId,
+ ...(resumedProviderThreadId
+ ? { resumeCursor: { threadId: resumedProviderThreadId } }
+ : {}),
+ } satisfies ProviderTurnStartResult;
+ }
+ const normalizedModel = normalizeCodexModelSlug(input.model ?? session.model);
const params = yield* buildTurnStartParams({
threadId: providerThreadId,
runtimeMode: options.runtimeMode,
diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx
index ad704d9e1b2..c0c54a6b882 100644
--- a/apps/web/src/components/ChatView.tsx
+++ b/apps/web/src/components/ChatView.tsx
@@ -164,10 +164,19 @@ import {
nextProjectScriptId,
projectScriptIdFromCommand,
} from "~/projectScripts";
-import { newDraftId, newMessageId, newThreadId } from "~/lib/utils";
+import { newCommandId, newDraftId, newMessageId, newThreadId } from "~/lib/utils";
import { getProviderModelCapabilities, resolveSelectableProvider } from "../providerModels";
import { NO_PROVIDER_MODEL_SELECTION } from "../providerInstances";
import { useClientSettings, useEnvironmentSettings } from "../hooks/useSettings";
+import {
+ beginWebThreadOutboxDispatch,
+ EMPTY_WEB_THREAD_OUTBOX_QUEUE,
+ finishWebThreadOutboxDispatch,
+ shouldDrainWebThreadOutbox,
+ shouldQueueWebThreadMessage,
+ useWebThreadOutboxStore,
+ webThreadOutboxKey,
+} from "../webThreadOutbox";
import { useNowMinute } from "../hooks/useNowMinute";
import { useNewThreadHandler } from "../hooks/useHandleNewThread";
import { resolveAppModelSelectionForInstance } from "../modelSelection";
@@ -1226,6 +1235,12 @@ function ChatViewContent(props: ChatViewProps) {
(store) => store.threadLastVisitedAtById[routeThreadKey],
);
const settings = useEnvironmentSettings(environmentId);
+ const activeThreadOutboxQueue = useWebThreadOutboxStore(
+ (state) =>
+ state.queuesByThreadKey[webThreadOutboxKey(environmentId, props.threadId)] ??
+ EMPTY_WEB_THREAD_OUTBOX_QUEUE,
+ );
+ const pausedOutboxMessageIds = useWebThreadOutboxStore((state) => state.pausedMessageIds);
// New-thread defaults live in the primary environment's settings.json (the
// settings UI never writes to remote environments), so read them from the
// primary server rather than the thread's environment.
@@ -4580,9 +4595,16 @@ function ChatViewContent(props: ChatViewProps) {
}),
);
};
+ const shouldQueueCurrentMessage = shouldQueueWebThreadMessage({
+ activeTurnMessageBehavior: settings.activeTurnMessageBehavior,
+ hasQueuedMessages: activeThreadOutboxQueue.length > 0,
+ isSendBusy,
+ isServerThread,
+ phase,
+ });
if (
!activeThread ||
- isSendBusy ||
+ (isSendBusy && !shouldQueueCurrentMessage) ||
isConnecting ||
threadDetailLoading ||
activeEnvironmentUnavailable ||
@@ -4722,6 +4744,98 @@ function ChatViewContent(props: ChatViewProps) {
return;
}
+ if (shouldQueueCurrentMessage) {
+ sendInFlightRef.current = true;
+ const composerImagesSnapshot = [...composerImages];
+ const composerTerminalContextsSnapshot = [...sendableComposerTerminalContexts];
+ const composerElementContextsSnapshot = [...composerElementContexts];
+ const composerPreviewAnnotationsSnapshot = [...composerPreviewAnnotations];
+ const composerReviewCommentsSnapshot: ReviewCommentContext[] = [...composerReviewComments];
+ const messageTextWithContexts = appendElementContextsToPrompt(
+ appendTerminalContextsToPrompt(promptForSend, composerTerminalContextsSnapshot),
+ composerElementContextsSnapshot,
+ );
+ const messageTextWithPreviewAnnotations = composerPreviewAnnotationsSnapshot.reduce(
+ (text, annotation) => appendPreviewAnnotationPrompt(text, annotation),
+ messageTextWithContexts,
+ );
+ const messageTextForSend = appendReviewCommentsToPrompt(
+ messageTextWithPreviewAnnotations,
+ composerReviewCommentsSnapshot,
+ );
+ const outgoingMessageText = formatOutgoingPrompt({
+ provider: ctxSelectedProvider,
+ model: ctxSelectedModel,
+ models: ctxSelectedProviderModels,
+ effort: ctxSelectedPromptEffort,
+ text: messageTextForSend || IMAGE_ONLY_BOOTSTRAP_PROMPT,
+ });
+ const attachmentsResult = await settlePromise(() =>
+ Promise.all(
+ composerImagesSnapshot.map(async (image) => ({
+ type: "image" as const,
+ name: image.name,
+ mimeType: image.mimeType,
+ sizeBytes: image.sizeBytes,
+ dataUrl: await readFileAsDataUrl(image.file),
+ })),
+ ),
+ );
+ if (attachmentsResult._tag === "Failure") {
+ const error = squashAtomCommandFailure(attachmentsResult);
+ setThreadError(
+ threadIdForSend,
+ error instanceof Error ? error.message : "Failed to prepare the queued message.",
+ );
+ sendInFlightRef.current = false;
+ return;
+ }
+
+ const messageId = newMessageId();
+ const createdAt = new Date().toISOString();
+ const { durable } = useWebThreadOutboxStore.getState().enqueue({
+ environmentId,
+ threadId: threadIdForSend,
+ messageId,
+ commandId: newCommandId(),
+ text: outgoingMessageText,
+ attachments: attachmentsResult.value,
+ modelSelection: ctxSelectedModelSelection,
+ runtimeMode,
+ interactionMode,
+ createdAt,
+ });
+ promptRef.current = "";
+ clearComposerDraftContent(composerDraftTarget);
+ composerRef.current?.resetCursorState();
+ setThreadError(threadIdForSend, null);
+ if (expiredTerminalContextCount > 0) {
+ const toastCopy = buildExpiredTerminalContextToastCopy(
+ expiredTerminalContextCount,
+ "omitted",
+ );
+ toastManager.add(
+ stackedThreadToast({
+ type: "warning",
+ title: toastCopy.title,
+ description: toastCopy.description,
+ }),
+ );
+ }
+ if (!durable) {
+ toastManager.add(
+ stackedThreadToast({
+ type: "warning",
+ title: "Message queued for this session",
+ description:
+ "Browser storage could not save the queue, so this message will not survive a reload.",
+ }),
+ );
+ }
+ sendInFlightRef.current = false;
+ return;
+ }
+
sendInFlightRef.current = true;
if (isDraftHeroState && activeThreadKey) {
let resolveDockStarted: (() => void) | undefined;
@@ -5001,6 +5115,112 @@ function ChatViewContent(props: ChatViewProps) {
}
};
+ const nextQueuedMessage = activeThreadOutboxQueue[0] ?? null;
+ const queuedMessagePaused =
+ nextQueuedMessage !== null && Boolean(pausedOutboxMessageIds[nextQueuedMessage.messageId]);
+
+ useEffect(() => {
+ if (
+ !nextQueuedMessage ||
+ !activeThread ||
+ nextQueuedMessage.environmentId !== environmentId ||
+ nextQueuedMessage.threadId !== activeThread.id ||
+ !shouldDrainWebThreadOutbox({
+ phase,
+ isSendBusy,
+ isConnecting,
+ environmentUnavailable: activeEnvironmentUnavailable,
+ paused: queuedMessagePaused,
+ }) ||
+ !beginWebThreadOutboxDispatch(nextQueuedMessage.messageId)
+ ) {
+ return;
+ }
+
+ const deliver = async () => {
+ beginLocalDispatch({ preparingWorktree: false });
+ const settingsResult = await persistThreadSettingsForNextTurn({
+ threadId: nextQueuedMessage.threadId,
+ createdAt: nextQueuedMessage.createdAt,
+ modelSelection: nextQueuedMessage.modelSelection,
+ runtimeMode: nextQueuedMessage.runtimeMode,
+ interactionMode: nextQueuedMessage.interactionMode,
+ });
+ const startResult =
+ settingsResult._tag === "Failure"
+ ? settingsResult
+ : await startThreadTurn({
+ environmentId: nextQueuedMessage.environmentId,
+ input: {
+ commandId: nextQueuedMessage.commandId,
+ threadId: nextQueuedMessage.threadId,
+ message: {
+ messageId: nextQueuedMessage.messageId,
+ role: "user",
+ text: nextQueuedMessage.text,
+ attachments: nextQueuedMessage.attachments,
+ },
+ modelSelection: nextQueuedMessage.modelSelection,
+ titleSeed: activeThread.title,
+ runtimeMode: nextQueuedMessage.runtimeMode,
+ interactionMode: nextQueuedMessage.interactionMode,
+ createdAt: nextQueuedMessage.createdAt,
+ },
+ });
+
+ if (startResult._tag === "Failure") {
+ useWebThreadOutboxStore.getState().pause(nextQueuedMessage.messageId);
+ resetLocalDispatch();
+ if (!isAtomCommandInterrupted(startResult)) {
+ const error = squashAtomCommandFailure(startResult);
+ setThreadError(
+ nextQueuedMessage.threadId,
+ error instanceof Error ? error.message : "Failed to send the queued message.",
+ );
+ }
+ return;
+ }
+
+ const { durable } = useWebThreadOutboxStore.getState().remove(nextQueuedMessage);
+ if (!durable) {
+ toastManager.add(
+ stackedThreadToast({
+ type: "warning",
+ title: "Queued message sent",
+ description:
+ "Browser storage could not save the queue update. Its stable command ID prevents a duplicate turn if it reappears after reload.",
+ }),
+ );
+ }
+ };
+
+ void deliver().finally(() => {
+ finishWebThreadOutboxDispatch(nextQueuedMessage.messageId);
+ });
+ }, [
+ activeEnvironmentUnavailable,
+ activeThread,
+ beginLocalDispatch,
+ environmentId,
+ isConnecting,
+ isSendBusy,
+ nextQueuedMessage,
+ persistThreadSettingsForNextTurn,
+ phase,
+ queuedMessagePaused,
+ resetLocalDispatch,
+ setThreadError,
+ startThreadTurn,
+ ]);
+
+ const retryQueuedMessages = useCallback(() => {
+ if (!nextQueuedMessage) {
+ return;
+ }
+ useWebThreadOutboxStore.getState().retry(nextQueuedMessage.messageId);
+ setThreadError(nextQueuedMessage.threadId, null);
+ }, [nextQueuedMessage, setThreadError]);
+
const onInterrupt = async () => {
if (!activeThread) return;
const result = await interruptThreadTurn({
@@ -5989,6 +6209,8 @@ function ChatViewContent(props: ChatViewProps) {
isSendBusy={isSendBusy}
sendDisabledReason={threadDetailLoading ? "Messages loading" : null}
isPreparingWorktree={isPreparingWorktree}
+ queuedMessageCount={activeThreadOutboxQueue.length}
+ queuedMessagesPaused={queuedMessagePaused}
environmentUnavailable={activeEnvironmentUnavailableState}
activePendingApproval={activePendingApproval}
pendingApprovals={pendingApprovals}
@@ -6024,6 +6246,7 @@ function ChatViewContent(props: ChatViewProps) {
composerTerminalContextsRef={composerTerminalContextsRef}
composerElementContextsRef={composerElementContextsRef}
onSend={onSend}
+ onRetryQueuedMessages={retryQueuedMessages}
onInterrupt={onInterrupt}
onImplementPlanInNewThread={onImplementPlanInNewThread}
onRespondToApproval={onRespondToApproval}
diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx
index 201fb79c566..cc15086a78e 100644
--- a/apps/web/src/components/chat/ChatComposer.tsx
+++ b/apps/web/src/components/chat/ChatComposer.tsx
@@ -445,6 +445,7 @@ const ComposerFooterPrimaryActions = memo(function ComposerFooterPrimaryActions(
isConnecting: boolean;
isEnvironmentUnavailable: boolean;
hasSendableContent: boolean;
+ activeTurnMessageBehavior: UnifiedSettings["activeTurnMessageBehavior"];
preserveComposerFocusOnPointerDown?: boolean;
onPreviousPendingQuestion: () => void;
onInterrupt: () => void;
@@ -473,6 +474,7 @@ const ComposerFooterPrimaryActions = memo(function ComposerFooterPrimaryActions(
isEnvironmentUnavailable={props.isEnvironmentUnavailable}
isPreparingWorktree={props.isPreparingWorktree}
hasSendableContent={props.hasSendableContent}
+ activeTurnMessageBehavior={props.activeTurnMessageBehavior}
preserveComposerFocusOnPointerDown={props.preserveComposerFocusOnPointerDown ?? false}
onPreviousPendingQuestion={props.onPreviousPendingQuestion}
onInterrupt={props.onInterrupt}
@@ -551,6 +553,8 @@ export interface ChatComposerProps {
isSendBusy: boolean;
sendDisabledReason: string | null;
isPreparingWorktree: boolean;
+ queuedMessageCount: number;
+ queuedMessagesPaused: boolean;
environmentUnavailable: {
readonly label: string;
readonly connection: EnvironmentConnectionPresentation;
@@ -610,6 +614,7 @@ export interface ChatComposerProps {
// Callbacks
onSend: (e?: { preventDefault: () => void }) => void;
+ onRetryQueuedMessages: () => void;
onInterrupt: () => void;
onImplementPlanInNewThread: () => void;
onRespondToApproval: (
@@ -663,6 +668,8 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps)
isSendBusy,
sendDisabledReason,
isPreparingWorktree,
+ queuedMessageCount,
+ queuedMessagesPaused,
environmentUnavailable,
activePendingApproval,
pendingApprovals,
@@ -697,6 +704,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps)
composerTerminalContextsRef,
composerElementContextsRef,
onSend,
+ onRetryQueuedMessages,
onInterrupt,
onImplementPlanInNewThread,
onRespondToApproval,
@@ -1180,6 +1188,8 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps)
isComposerCollapsedMobile && !isComposerApprovalState && pendingUserInputs.length === 0;
const composerFooterHasWideActions = showPlanFollowUpPrompt || activePendingProgress !== null;
+ const isPrimarySendBusy =
+ isSendBusy && !(phase === "running" && settings.activeTurnMessageBehavior === "queue");
const showPlanSidebarToggle = Boolean(activePlan || sidebarProposedPlan || planSidebarOpen);
const composerFooterActionLayoutKey = useMemo(() => {
if (activePendingProgress) {
@@ -1270,15 +1280,19 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps)
[activePendingIsResponding, activePendingProgress, activePendingResolvedAnswers],
);
const collapsedComposerPrimaryActionDisabled =
- phase === "running" ||
- isSendBusy ||
+ isPrimarySendBusy ||
isSendDisabled ||
isConnecting ||
noProviderAvailable ||
projectSelectionRequired ||
environmentUnavailable !== null ||
!composerSendState.hasSendableContent;
- const collapsedComposerPrimaryActionLabel = "Send message";
+ const collapsedComposerPrimaryActionLabel =
+ phase === "running"
+ ? settings.activeTurnMessageBehavior === "queue"
+ ? "Queue message"
+ : "Steer active turn"
+ : "Send message";
const showMobilePendingAnswerActions =
isMobileViewport && !isComposerCollapsedMobile && pendingPrimaryAction !== null;
@@ -2832,6 +2846,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps)
}
isPreparingWorktree={false}
hasSendableContent={false}
+ activeTurnMessageBehavior={settings.activeTurnMessageBehavior}
preserveComposerFocusOnPointerDown
onPreviousPendingQuestion={onPreviousActivePendingUserInputQuestion}
onInterrupt={handleInterruptPrimaryAction}
@@ -3115,6 +3130,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps)
}
isPreparingWorktree={false}
hasSendableContent={false}
+ activeTurnMessageBehavior={settings.activeTurnMessageBehavior}
preserveComposerFocusOnPointerDown
onPreviousPendingQuestion={onPreviousActivePendingUserInputQuestion}
onInterrupt={handleInterruptPrimaryAction}
@@ -3237,7 +3253,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps)
isRunning={phase === "running"}
showPlanFollowUpPrompt={pendingUserInputs.length === 0 && showPlanFollowUpPrompt}
promptHasText={prompt.trim().length > 0}
- isSendBusy={isSendBusy}
+ isSendBusy={isPrimarySendBusy}
sendDisabledReason={sendDisabledReason}
isConnecting={isConnecting}
isEnvironmentUnavailable={
@@ -3247,6 +3263,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps)
}
isPreparingWorktree={isPreparingWorktree}
hasSendableContent={composerSendState.hasSendableContent}
+ activeTurnMessageBehavior={settings.activeTurnMessageBehavior}
preserveComposerFocusOnPointerDown={isMobileViewport}
onPreviousPendingQuestion={onPreviousActivePendingUserInputQuestion}
onInterrupt={handleInterruptPrimaryAction}
@@ -3255,6 +3272,28 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps)
)}
+ {queuedMessageCount > 0 ? (
+
+
+ {queuedMessageCount} queued message{queuedMessageCount === 1 ? "" : "s"} will send
+ one at a time.
+
+ {queuedMessagesPaused ? (
+
+ ) : null}
+
+ ) : null}
diff --git a/apps/web/src/components/chat/ComposerPrimaryActions.tsx b/apps/web/src/components/chat/ComposerPrimaryActions.tsx
index 504b7e1cc44..c3949b537c3 100644
--- a/apps/web/src/components/chat/ComposerPrimaryActions.tsx
+++ b/apps/web/src/components/chat/ComposerPrimaryActions.tsx
@@ -6,6 +6,7 @@ import { StageBackdropButtonArt, useSidebarStageBackdropVariant } from "../Sideb
import { Button } from "../ui/button";
import { Menu, MenuItem, MenuPopup, MenuTrigger } from "../ui/menu";
import { Spinner } from "../ui/spinner";
+import type { ActiveTurnMessageBehavior } from "@t3tools/contracts/settings";
interface PendingActionState {
questionIndex: number;
@@ -27,6 +28,7 @@ interface ComposerPrimaryActionsProps {
isEnvironmentUnavailable: boolean;
isPreparingWorktree: boolean;
hasSendableContent: boolean;
+ activeTurnMessageBehavior: ActiveTurnMessageBehavior;
preserveComposerFocusOnPointerDown?: boolean;
onPreviousPendingQuestion: () => void;
onInterrupt: () => void;
@@ -67,6 +69,7 @@ export const ComposerPrimaryActions = memo(function ComposerPrimaryActions({
isEnvironmentUnavailable,
isPreparingWorktree,
hasSendableContent,
+ activeTurnMessageBehavior,
preserveComposerFocusOnPointerDown = false,
onPreviousPendingQuestion,
onInterrupt,
@@ -132,19 +135,78 @@ export const ComposerPrimaryActions = memo(function ComposerPrimaryActions({
);
}
+ const sendButton = (
+
+ );
+
if (isRunning) {
return (
-
+
+
+ {sendButton}
+
);
}
@@ -202,55 +264,5 @@ export const ComposerPrimaryActions = memo(function ComposerPrimaryActions({
);
}
- return (
-
- );
+ return sendButton;
});
diff --git a/apps/web/src/components/settings/SettingsPanels.tsx b/apps/web/src/components/settings/SettingsPanels.tsx
index e3e0a22a0c3..ace47aaa108 100644
--- a/apps/web/src/components/settings/SettingsPanels.tsx
+++ b/apps/web/src/components/settings/SettingsPanels.tsx
@@ -31,6 +31,8 @@ import {
squashAtomCommandFailure,
} from "@t3tools/client-runtime/state/runtime";
import {
+ type ActiveTurnMessageBehavior,
+ DEFAULT_ACTIVE_TURN_MESSAGE_BEHAVIOR,
DEFAULT_ENVIRONMENT_IDENTIFICATION_MODE,
DEFAULT_UNIFIED_SETTINGS,
type EnvironmentIdentificationMode,
@@ -600,6 +602,9 @@ export function useSettingsRestore(onRestored?: () => void) {
...(settings.timestampFormat !== DEFAULT_UNIFIED_SETTINGS.timestampFormat
? ["Time format"]
: []),
+ ...(settings.activeTurnMessageBehavior !== DEFAULT_UNIFIED_SETTINGS.activeTurnMessageBehavior
+ ? ["Messages while working"]
+ : []),
...(settings.sidebarThreadPreviewCount !== DEFAULT_UNIFIED_SETTINGS.sidebarThreadPreviewCount
? ["Visible threads"]
: []),
@@ -654,6 +659,7 @@ export function useSettingsRestore(onRestored?: () => void) {
isTextGenerationModelDirty,
isBackgroundActivityDirty,
settings.autoOpenPlanSidebar,
+ settings.activeTurnMessageBehavior,
settings.confirmThreadArchive,
settings.confirmThreadDelete,
settings.addProjectBaseDirectory,
@@ -693,6 +699,7 @@ export function useSettingsRestore(onRestored?: () => void) {
setTheme("system");
updateSettings({
timestampFormat: DEFAULT_UNIFIED_SETTINGS.timestampFormat,
+ activeTurnMessageBehavior: DEFAULT_UNIFIED_SETTINGS.activeTurnMessageBehavior,
wordWrap: DEFAULT_UNIFIED_SETTINGS.wordWrap,
diffIgnoreWhitespace: DEFAULT_UNIFIED_SETTINGS.diffIgnoreWhitespace,
environmentIdentificationMode: DEFAULT_UNIFIED_SETTINGS.environmentIdentificationMode,
@@ -1647,6 +1654,52 @@ export function GeneralSettingsPanel() {
return (
+
+ updateSettings({
+ activeTurnMessageBehavior: DEFAULT_ACTIVE_TURN_MESSAGE_BEHAVIOR,
+ })
+ }
+ />
+ ) : null
+ }
+ control={
+
+ }
+ />
+
{
it("matches normalized title substrings", () => {
expect(searchSettings(" WORD WRAP ", ITEMS).map((item) => item.id)).toEqual(["word-wrap"]);
- expect(searchSettings("work")).toEqual([]);
+ expect(searchSettings("work").map((item) => item.id)).toEqual(["messages-while-working"]);
});
it("keeps catalog order for multiple title matches", () => {
@@ -65,6 +65,10 @@ describe("searchSettings", () => {
});
it("serves anchor props to panels from the catalog", () => {
+ expect(searchableSetting("messages-while-working")).toEqual({
+ id: "messages-while-working",
+ title: "Messages while working",
+ });
expect(searchableSetting("word-wrap")).toEqual({ id: "word-wrap", title: "Word wrap" });
expect(searchableSetting("archive")).toEqual({ id: "archive", title: "Archived threads" });
});
diff --git a/apps/web/src/components/settings/settingsSearch.ts b/apps/web/src/components/settings/settingsSearch.ts
index 1ba231a5835..6e904096b25 100644
--- a/apps/web/src/components/settings/settingsSearch.ts
+++ b/apps/web/src/components/settings/settingsSearch.ts
@@ -85,6 +85,11 @@ export const SETTINGS_SEARCH_ITEMS = [
title: "Word wrap",
to: "/settings/appearance",
},
+ {
+ id: "messages-while-working",
+ title: "Messages while working",
+ to: "/settings/general",
+ },
{
id: "project-grouping",
title: "Project grouping",
diff --git a/apps/web/src/webThreadOutbox.test.ts b/apps/web/src/webThreadOutbox.test.ts
new file mode 100644
index 00000000000..568d09cf464
--- /dev/null
+++ b/apps/web/src/webThreadOutbox.test.ts
@@ -0,0 +1,207 @@
+import {
+ CommandId,
+ EnvironmentId,
+ MessageId,
+ ProviderInstanceId,
+ ThreadId,
+} from "@t3tools/contracts";
+import { afterEach, describe, expect, it } from "vite-plus/test";
+
+import {
+ beginWebThreadOutboxDispatch,
+ finishWebThreadOutboxDispatch,
+ shouldDrainWebThreadOutbox,
+ shouldQueueWebThreadMessage,
+ useWebThreadOutboxStore,
+ webThreadOutboxKey,
+ writeWebThreadOutboxStorageForTest,
+ type QueuedWebThreadMessage,
+} from "./webThreadOutbox";
+
+const environmentId = EnvironmentId.make("environment-test");
+const threadId = ThreadId.make("thread-test");
+
+function message(index: number): QueuedWebThreadMessage {
+ return {
+ environmentId,
+ threadId,
+ messageId: MessageId.make(`message-${index}`),
+ commandId: CommandId.make(`command-${index}`),
+ text: `Message ${index}`,
+ attachments: [],
+ modelSelection: {
+ instanceId: ProviderInstanceId.make("codex"),
+ model: "gpt-5",
+ },
+ runtimeMode: "full-access",
+ interactionMode: "default",
+ createdAt: new Date(1_700_000_000_000 + index).toISOString(),
+ };
+}
+
+function resetOutbox(): void {
+ writeWebThreadOutboxStorageForTest("");
+}
+
+function persistedOutbox(messages: ReadonlyArray): string {
+ return JSON.stringify({
+ version: 1,
+ state: {
+ queuesByThreadKey: {
+ [webThreadOutboxKey(environmentId, threadId)]: messages,
+ },
+ },
+ });
+}
+
+afterEach(resetOutbox);
+
+describe("web thread outbox", () => {
+ it("keeps an unbounded FIFO per thread", () => {
+ const store = useWebThreadOutboxStore.getState();
+ for (let index = 0; index < 100; index += 1) {
+ store.enqueue(message(index));
+ }
+
+ const queue =
+ useWebThreadOutboxStore.getState().queuesByThreadKey[
+ webThreadOutboxKey(environmentId, threadId)
+ ];
+ expect(queue).toHaveLength(100);
+ expect(queue?.map((entry) => entry.messageId)).toEqual(
+ Array.from({ length: 100 }, (_, index) => MessageId.make(`message-${index}`)),
+ );
+ });
+
+ it("deduplicates stable message ids and removes only the delivered head", () => {
+ const first = message(1);
+ const second = message(2);
+ const store = useWebThreadOutboxStore.getState();
+ store.enqueue(first);
+ store.enqueue(second);
+ store.enqueue({ ...first, text: "Updated" });
+ store.remove(first);
+
+ const queue =
+ useWebThreadOutboxStore.getState().queuesByThreadKey[
+ webThreadOutboxKey(environmentId, threadId)
+ ];
+ expect(queue?.map((entry) => entry.messageId)).toEqual([second.messageId]);
+ });
+
+ it("merges another tab's durable messages before enqueueing", () => {
+ const first = message(1);
+ const second = message(2);
+ const third = message(3);
+ const store = useWebThreadOutboxStore.getState();
+ store.enqueue(first);
+
+ writeWebThreadOutboxStorageForTest(persistedOutbox([first, second]), {
+ syncStore: false,
+ });
+ store.enqueue(third);
+
+ const queue =
+ useWebThreadOutboxStore.getState().queuesByThreadKey[
+ webThreadOutboxKey(environmentId, threadId)
+ ];
+ expect(queue?.map((entry) => entry.messageId)).toEqual([
+ first.messageId,
+ second.messageId,
+ third.messageId,
+ ]);
+ });
+
+ it("preserves another tab's messages when removing a delivered message", () => {
+ const first = message(1);
+ const second = message(2);
+ const store = useWebThreadOutboxStore.getState();
+ store.enqueue(first);
+
+ writeWebThreadOutboxStorageForTest(persistedOutbox([first, second]), {
+ syncStore: false,
+ });
+ store.remove(first);
+
+ const queue =
+ useWebThreadOutboxStore.getState().queuesByThreadKey[
+ webThreadOutboxKey(environmentId, threadId)
+ ];
+ expect(queue?.map((entry) => entry.messageId)).toEqual([second.messageId]);
+ });
+
+ it("permits only one dispatcher for a stable message id", () => {
+ const queued = message(3);
+ expect(beginWebThreadOutboxDispatch(queued.messageId)).toBe(true);
+ expect(beginWebThreadOutboxDispatch(queued.messageId)).toBe(false);
+ finishWebThreadOutboxDispatch(queued.messageId);
+ expect(beginWebThreadOutboxDispatch(queued.messageId)).toBe(true);
+ finishWebThreadOutboxDispatch(queued.messageId);
+ });
+
+ it("drains only from a ready, connected, unpaused thread", () => {
+ expect(
+ shouldDrainWebThreadOutbox({
+ phase: "ready",
+ isSendBusy: false,
+ isConnecting: false,
+ environmentUnavailable: false,
+ paused: false,
+ }),
+ ).toBe(true);
+ expect(
+ shouldDrainWebThreadOutbox({
+ phase: "running",
+ isSendBusy: false,
+ isConnecting: false,
+ environmentUnavailable: false,
+ paused: false,
+ }),
+ ).toBe(false);
+ expect(
+ shouldDrainWebThreadOutbox({
+ phase: "ready",
+ isSendBusy: false,
+ isConnecting: false,
+ environmentUnavailable: false,
+ paused: true,
+ }),
+ ).toBe(false);
+ });
+
+ it("queues active-turn messages only when queue mode or an existing FIFO requires it", () => {
+ const activeThread = {
+ isServerThread: true,
+ phase: "running" as const,
+ isSendBusy: false,
+ hasQueuedMessages: false,
+ };
+
+ expect(
+ shouldQueueWebThreadMessage({
+ ...activeThread,
+ activeTurnMessageBehavior: "queue",
+ }),
+ ).toBe(true);
+ expect(
+ shouldQueueWebThreadMessage({
+ ...activeThread,
+ activeTurnMessageBehavior: "steer",
+ }),
+ ).toBe(false);
+ expect(
+ shouldQueueWebThreadMessage({
+ ...activeThread,
+ activeTurnMessageBehavior: "steer",
+ hasQueuedMessages: true,
+ }),
+ ).toBe(true);
+ expect(
+ shouldQueueWebThreadMessage({
+ ...activeThread,
+ activeTurnMessageBehavior: "queue",
+ isServerThread: false,
+ }),
+ ).toBe(false);
+ });
+});
diff --git a/apps/web/src/webThreadOutbox.ts b/apps/web/src/webThreadOutbox.ts
new file mode 100644
index 00000000000..c0752bc7fee
--- /dev/null
+++ b/apps/web/src/webThreadOutbox.ts
@@ -0,0 +1,291 @@
+import {
+ CommandId,
+ EnvironmentId,
+ MessageId,
+ ModelSelection,
+ ProviderInteractionMode,
+ RuntimeMode,
+ ThreadId,
+ type UploadChatAttachment,
+} from "@t3tools/contracts";
+import type { ActiveTurnMessageBehavior } from "@t3tools/contracts/settings";
+import { scopedThreadKey, scopeThreadRef } from "@t3tools/client-runtime/environment";
+import * as Schema from "effect/Schema";
+import { create } from "zustand";
+
+import { createMemoryStorage, type StateStorage } from "./lib/storage";
+
+export const WEB_THREAD_OUTBOX_STORAGE_KEY = "t3code:thread-outbox:v1";
+const WEB_THREAD_OUTBOX_STORAGE_VERSION = 1;
+
+const QueuedWebImageAttachment = Schema.Struct({
+ type: Schema.Literal("image"),
+ name: Schema.String,
+ mimeType: Schema.String,
+ sizeBytes: Schema.Number,
+ dataUrl: Schema.String,
+});
+
+const QueuedWebThreadMessageSchema = Schema.Struct({
+ environmentId: EnvironmentId,
+ threadId: ThreadId,
+ messageId: MessageId,
+ commandId: CommandId,
+ text: Schema.String,
+ attachments: Schema.Array(QueuedWebImageAttachment),
+ modelSelection: ModelSelection,
+ runtimeMode: RuntimeMode,
+ interactionMode: ProviderInteractionMode,
+ createdAt: Schema.String,
+});
+
+export interface QueuedWebThreadMessage {
+ readonly environmentId: EnvironmentId;
+ readonly threadId: ThreadId;
+ readonly messageId: MessageId;
+ readonly commandId: CommandId;
+ readonly text: string;
+ readonly attachments: ReadonlyArray;
+ readonly modelSelection: ModelSelection;
+ readonly runtimeMode: RuntimeMode;
+ readonly interactionMode: ProviderInteractionMode;
+ readonly createdAt: string;
+}
+
+const PersistedWebThreadOutboxState = Schema.Struct({
+ queuesByThreadKey: Schema.Record(Schema.String, Schema.Array(QueuedWebThreadMessageSchema)),
+});
+const decodePersistedState = Schema.decodeUnknownSync(PersistedWebThreadOutboxState);
+
+export function webThreadOutboxKey(environmentId: EnvironmentId, threadId: ThreadId): string {
+ return scopedThreadKey(scopeThreadRef(environmentId, threadId));
+}
+
+function readQueue(
+ queues: Record>,
+ threadKey: string,
+): ReadonlyArray {
+ return Object.hasOwn(queues, threadKey) ? (queues[threadKey] ?? []) : [];
+}
+
+function mergeQueues(
+ ...sources: ReadonlyArray>>
+): Record> {
+ const merged: Record> = {};
+ const threadKeys = new Set(sources.flatMap((source) => Object.keys(source)));
+ for (const threadKey of threadKeys) {
+ const messagesById = new Map();
+ for (const source of sources) {
+ for (const message of readQueue(source, threadKey)) {
+ messagesById.set(message.messageId, message);
+ }
+ }
+ const queue = [...messagesById.values()].sort(
+ (left, right) =>
+ left.createdAt.localeCompare(right.createdAt) ||
+ String(left.messageId).localeCompare(String(right.messageId)),
+ );
+ if (queue.length > 0) {
+ merged[threadKey] = queue;
+ }
+ }
+ return merged;
+}
+
+function resolveBaseStorage(): { storage: StateStorage; durable: boolean } {
+ try {
+ if (typeof localStorage !== "undefined") {
+ return { storage: localStorage, durable: true };
+ }
+ } catch {
+ // Sandboxed browsers can reject access to the localStorage property itself.
+ }
+ return { storage: createMemoryStorage(), durable: false };
+}
+
+const { storage: baseOutboxStorage, durable: storageIsDurable } = resolveBaseStorage();
+
+function persistQueues(queues: Record>): {
+ written: boolean;
+ durable: boolean;
+} {
+ try {
+ baseOutboxStorage.setItem(
+ WEB_THREAD_OUTBOX_STORAGE_KEY,
+ JSON.stringify({
+ version: WEB_THREAD_OUTBOX_STORAGE_VERSION,
+ state: { queuesByThreadKey: queues },
+ }),
+ );
+ return { written: true, durable: storageIsDurable };
+ } catch (error) {
+ console.error("[THREAD-OUTBOX] Could not persist queued messages.", error);
+ return { written: false, durable: false };
+ }
+}
+
+function readPersistedQueues(): Record> | null {
+ try {
+ const raw = baseOutboxStorage.getItem(WEB_THREAD_OUTBOX_STORAGE_KEY);
+ if (typeof raw !== "string" || raw.length === 0) {
+ return null;
+ }
+ const parsed: unknown = JSON.parse(raw);
+ const state = (parsed as { state?: unknown } | null)?.state;
+ return state ? decodePersistedState(state).queuesByThreadKey : null;
+ } catch {
+ return null;
+ }
+}
+
+function mergeWithPersistedQueues(
+ queues: Record>,
+): Record> {
+ const persisted = readPersistedQueues();
+ return persisted === null ? queues : mergeQueues(queues, persisted);
+}
+
+interface WebThreadOutboxState {
+ readonly queuesByThreadKey: Record>;
+ readonly pausedMessageIds: Readonly>;
+ readonly enqueue: (message: QueuedWebThreadMessage) => { durable: boolean };
+ readonly remove: (message: QueuedWebThreadMessage) => { durable: boolean };
+ readonly pause: (messageId: MessageId) => void;
+ readonly retry: (messageId: MessageId) => void;
+}
+
+export const useWebThreadOutboxStore = create()((set, get) => ({
+ queuesByThreadKey: {},
+ pausedMessageIds: {},
+ enqueue: (message) => {
+ const threadKey = webThreadOutboxKey(message.environmentId, message.threadId);
+ // Another tab may have updated the shared outbox since this store last
+ // rendered. Merge its durable snapshot before applying this mutation so a
+ // full-key localStorage write cannot discard the other tab's messages.
+ const queues = mergeWithPersistedQueues(get().queuesByThreadKey);
+ const queue = readQueue(queues, threadKey);
+ const nextQueue = [
+ ...queue.filter((candidate) => candidate.messageId !== message.messageId),
+ message,
+ ];
+ const next = { ...queues, [threadKey]: nextQueue };
+ const persisted = persistQueues(next);
+ // Even when browser storage is blocked or full, retain the message for the
+ // current session. The caller reports that it is not reload-safe.
+ set({ queuesByThreadKey: next });
+ return { durable: persisted.written && persisted.durable };
+ },
+ remove: (message) => {
+ const threadKey = webThreadOutboxKey(message.environmentId, message.threadId);
+ const queues = mergeWithPersistedQueues(get().queuesByThreadKey);
+ const nextQueue = readQueue(queues, threadKey).filter(
+ (candidate) => candidate.messageId !== message.messageId,
+ );
+ const next = { ...queues };
+ if (nextQueue.length === 0) {
+ delete next[threadKey];
+ } else {
+ next[threadKey] = nextQueue;
+ }
+ const persisted = persistQueues(next);
+ const pausedMessageIds = { ...get().pausedMessageIds };
+ delete pausedMessageIds[message.messageId];
+ // The command id is stable, so a removal that fails to persist can only
+ // cause an idempotent acknowledgement after reload, never a second turn.
+ set({ queuesByThreadKey: next, pausedMessageIds });
+ return { durable: persisted.written && persisted.durable };
+ },
+ pause: (messageId) => {
+ set((state) => ({
+ pausedMessageIds: { ...state.pausedMessageIds, [messageId]: true },
+ }));
+ },
+ retry: (messageId) => {
+ set((state) => {
+ if (!state.pausedMessageIds[messageId]) {
+ return state;
+ }
+ const pausedMessageIds = { ...state.pausedMessageIds };
+ delete pausedMessageIds[messageId];
+ return { pausedMessageIds };
+ });
+ },
+}));
+
+export const EMPTY_WEB_THREAD_OUTBOX_QUEUE: ReadonlyArray = [];
+
+{
+ const persisted = readPersistedQueues();
+ if (persisted) {
+ useWebThreadOutboxStore.setState({ queuesByThreadKey: persisted });
+ }
+}
+
+if (storageIsDurable && typeof window !== "undefined") {
+ window.addEventListener("storage", (event) => {
+ if (event.key !== WEB_THREAD_OUTBOX_STORAGE_KEY) {
+ return;
+ }
+ useWebThreadOutboxStore.setState({ queuesByThreadKey: readPersistedQueues() ?? {} });
+ });
+}
+
+const dispatchingMessageIds = new Set();
+
+export function beginWebThreadOutboxDispatch(messageId: MessageId): boolean {
+ if (dispatchingMessageIds.has(messageId)) {
+ return false;
+ }
+ dispatchingMessageIds.add(messageId);
+ return true;
+}
+
+export function finishWebThreadOutboxDispatch(messageId: MessageId): void {
+ dispatchingMessageIds.delete(messageId);
+}
+
+export function shouldDrainWebThreadOutbox(input: {
+ readonly phase: "disconnected" | "connecting" | "ready" | "running";
+ readonly isSendBusy: boolean;
+ readonly isConnecting: boolean;
+ readonly environmentUnavailable: boolean;
+ readonly paused: boolean;
+}): boolean {
+ return (
+ input.phase === "ready" &&
+ !input.isSendBusy &&
+ !input.isConnecting &&
+ !input.environmentUnavailable &&
+ !input.paused
+ );
+}
+
+export function shouldQueueWebThreadMessage(input: {
+ readonly activeTurnMessageBehavior: ActiveTurnMessageBehavior;
+ readonly hasQueuedMessages: boolean;
+ readonly isSendBusy: boolean;
+ readonly isServerThread: boolean;
+ readonly phase: "disconnected" | "connecting" | "ready" | "running";
+}): boolean {
+ return (
+ input.isServerThread &&
+ (input.hasQueuedMessages ||
+ (input.activeTurnMessageBehavior === "queue" &&
+ (input.phase === "running" || input.isSendBusy)))
+ );
+}
+
+export function writeWebThreadOutboxStorageForTest(
+ raw: string,
+ options?: { readonly syncStore?: boolean },
+): void {
+ baseOutboxStorage.setItem(WEB_THREAD_OUTBOX_STORAGE_KEY, raw);
+ if (options?.syncStore === false) {
+ return;
+ }
+ useWebThreadOutboxStore.setState({
+ queuesByThreadKey: readPersistedQueues() ?? {},
+ pausedMessageIds: {},
+ });
+ dispatchingMessageIds.clear();
+}
diff --git a/packages/contracts/src/settings.test.ts b/packages/contracts/src/settings.test.ts
index 5bd22e95f20..d9335cef4d6 100644
--- a/packages/contracts/src/settings.test.ts
+++ b/packages/contracts/src/settings.test.ts
@@ -67,6 +67,22 @@ describe("ClientSettings environment identification", () => {
});
});
+describe("ClientSettings messages while working", () => {
+ it("defaults to steering and accepts both delivery behaviors", () => {
+ expect(decodeClientSettings({}).activeTurnMessageBehavior).toBe("steer");
+ expect(
+ decodeClientSettingsPatch({ activeTurnMessageBehavior: "steer" }).activeTurnMessageBehavior,
+ ).toBe("steer");
+ expect(
+ decodeClientSettingsPatch({ activeTurnMessageBehavior: "queue" }).activeTurnMessageBehavior,
+ ).toBe("queue");
+ });
+
+ it("rejects unsupported delivery behaviors", () => {
+ expect(() => decodeClientSettingsPatch({ activeTurnMessageBehavior: "send-later" })).toThrow();
+ });
+});
+
describe("ClientSettings sidebar v2", () => {
it("defaults the beta off with a three-day auto-settle threshold", () => {
const settings = decodeClientSettings({});
diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts
index cbb547b95fb..ad6f41ae437 100644
--- a/packages/contracts/src/settings.ts
+++ b/packages/contracts/src/settings.ts
@@ -102,6 +102,9 @@ export const DEFAULT_TERMINAL_FONT_SIZE: TerminalFontSize = 12;
export const EnvironmentIdentificationMode = Schema.Literals(["artwork", "pill", "none"]);
export type EnvironmentIdentificationMode = typeof EnvironmentIdentificationMode.Type;
export const DEFAULT_ENVIRONMENT_IDENTIFICATION_MODE: EnvironmentIdentificationMode = "artwork";
+export const ActiveTurnMessageBehavior = Schema.Literals(["steer", "queue"]);
+export type ActiveTurnMessageBehavior = typeof ActiveTurnMessageBehavior.Type;
+export const DEFAULT_ACTIVE_TURN_MESSAGE_BEHAVIOR: ActiveTurnMessageBehavior = "steer";
/**
* A user-chosen font family (a single name or a comma-separated list). Empty
@@ -111,6 +114,9 @@ export const FontFamilyPreference = Schema.String.check(Schema.isMaxLength(200))
export type FontFamilyPreference = typeof FontFamilyPreference.Type;
export const ClientSettingsSchema = Schema.Struct({
+ activeTurnMessageBehavior: ActiveTurnMessageBehavior.pipe(
+ Schema.withDecodingDefault(Effect.succeed(DEFAULT_ACTIVE_TURN_MESSAGE_BEHAVIOR)),
+ ),
autoOpenPlanSidebar: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))),
confirmThreadArchive: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))),
confirmThreadDelete: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))),
@@ -747,6 +753,7 @@ export const ServerSettingsPatch = Schema.Struct({
export type ServerSettingsPatch = typeof ServerSettingsPatch.Type;
export const ClientSettingsPatch = Schema.Struct({
+ activeTurnMessageBehavior: Schema.optionalKey(ActiveTurnMessageBehavior),
autoOpenPlanSidebar: Schema.optionalKey(Schema.Boolean),
confirmThreadArchive: Schema.optionalKey(Schema.Boolean),
confirmThreadDelete: Schema.optionalKey(Schema.Boolean),