From 2fcbfb50351eb45839de82634234e23781aff1c1 Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Sun, 25 Oct 2026 03:07:05 +0000 Subject: [PATCH 01/11] refactor(web): send composer turns through a turn delivery seam Routes the composer's turn dispatch through a client-runtime turn delivery service and moves the outbox transport classification there. The policy is unchanged: one send, any failure restores the draft. The new delivery tests fail against it; the next commit makes them pass. Co-Authored-By: Claude Opus 5.5 --- apps/mobile/src/state/thread-outbox-model.ts | 31 +- .../src/components/chat/useComposerSend.ts | 201 +++++++------ apps/web/src/state/threads.ts | 1 + apps/web/src/state/turnDeliveries.ts | 68 +++++ .../src/state/use-orchestration-command.ts | 53 +++- packages/client-runtime/package.json | 4 + .../client-runtime/src/errors/transport.ts | 29 ++ .../src/state/threadCommands.test.ts | 87 ++++++ .../src/state/turnDelivery.test.ts | 273 ++++++++++++++++++ .../client-runtime/src/state/turnDelivery.ts | 79 +++++ 10 files changed, 698 insertions(+), 128 deletions(-) create mode 100644 apps/web/src/state/turnDeliveries.ts create mode 100644 packages/client-runtime/src/state/turnDelivery.test.ts create mode 100644 packages/client-runtime/src/state/turnDelivery.ts diff --git a/apps/mobile/src/state/thread-outbox-model.ts b/apps/mobile/src/state/thread-outbox-model.ts index 6c5da0db20991..6bcc3146525fe 100644 --- a/apps/mobile/src/state/thread-outbox-model.ts +++ b/apps/mobile/src/state/thread-outbox-model.ts @@ -1,4 +1,4 @@ -import { isTransportConnectionErrorMessage } from "@t3tools/client-runtime/errors"; +import { isRetryableDeliveryFailure } from "@t3tools/client-runtime/errors"; import { clampFileAttachmentUploadBytes, fileAttachmentTooLargeMessage, @@ -241,34 +241,7 @@ export function isQueuedThreadCreationSendable(message: QueuedThreadMessage): bo return message.creation.workspaceMode !== "worktree" || Boolean(message.creation.branch); } -function errorMessage(error: unknown): string | null { - if (error instanceof Error) { - return error.message; - } - if (typeof error === "object" && error !== null && "message" in error) { - return typeof error.message === "string" ? error.message : null; - } - return typeof error === "string" ? error : null; -} - -/** Dispatch/auth rejections invalidate payloads; other typed failures may be disconnected transports and must retry. */ -export function shouldRetryThreadOutboxDelivery(error: unknown): boolean { - if (typeof error === "object" && error !== null && "_tag" in error) { - switch (error._tag) { - case "OrchestrationDispatchCommandError": - case "EnvironmentAuthorizationError": - return false; - case "ConnectionTransientError": - case "RpcClientError": - case "EnvironmentRpcUnavailableError": - case "EnvironmentNotRegisteredError": - return true; - default: - break; - } - } - return isTransportConnectionErrorMessage(errorMessage(error)); -} +export const shouldRetryThreadOutboxDelivery = isRetryableDeliveryFailure; export type ThreadOutboxCommandStage = "settings-sync" | "start-turn"; export type ThreadOutboxFailureAction = "retry" | "restore"; diff --git a/apps/web/src/components/chat/useComposerSend.ts b/apps/web/src/components/chat/useComposerSend.ts index a5fa30ffe15a9..13a64b220b09d 100644 --- a/apps/web/src/components/chat/useComposerSend.ts +++ b/apps/web/src/components/chat/useComposerSend.ts @@ -17,6 +17,7 @@ import { import { isUsageLimitsCommand } from "@t3tools/shared/usageLimits"; import { AuthOrchestrationOperateScope, + CommandId, DEFAULT_MODEL, type EnvironmentId, type ModelSelection, @@ -59,7 +60,7 @@ import { import { resolvePlanFollowUpSubmission } from "../../proposedPlan"; import { type ChatMessage } from "../../types"; import { stackedThreadToast, toastManager } from "../ui/toast"; -import { newMessageId, newThreadId } from "~/lib/utils"; +import { newMessageId, newThreadId, randomUUID } from "~/lib/utils"; import { useNewThreadHandler } from "../../hooks/useHandleNewThread"; import { buildThreadRouteParams } from "../../threadRoutes"; import { @@ -83,6 +84,8 @@ import { } from "../../lib/composerContextRecords"; import { environmentServerConfigsAtom } from "../../state/server"; import { threadEnvironment } from "../../state/threads"; +import { turnDeliveries } from "../../state/turnDeliveries"; +import type { TurnDeliveryFailure } from "@t3tools/client-runtime/state/turn-delivery"; import { runMobileComposerTransition } from "./draftHeroTransition"; import { useClientSettingsHydrated, useEnvironmentSettings } from "../../hooks/useSettings"; import { environmentPresentations } from "../../state/presentation"; @@ -1189,7 +1192,7 @@ export function useComposerSend(input: ComposerSendInput) { ctxSelectedModelSelection.options, ); - let failure: AtomCommandResult | null = null; + let failure: TurnDeliveryFailure | null = null; if (failure === null && isServerThread) { const settingsResult = await persistThreadSettingsForNextTurn({ @@ -1242,6 +1245,90 @@ export function useComposerSend(input: ComposerSendInput) { let backgroundDraftOpened = false; let turnStartSucceeded = false; + let sendCancelled = false; + const finishStartedTurn = () => { + setKeepFullHistory(routeThreadKey, false); + clearUsageLimitsFor(routeThreadKey); + if (turnUsesAttachmentUploads) { + releaseDraftAttachments(composerAttachmentsSnapshot); + } + // A draft still on screen finalizes once its thread route takes over. + if (sentDraftThreadRef) { + if (backgroundDraftOpened || currentRouteThreadKeyRef.current !== routeThreadKey) { + finalizePromotedDraftThreadByRef(sentDraftThreadRef); + } else { + clearBackgroundDraftSubmissionByRef(sentDraftThreadRef); + } + } + }; + // `failure` is null when the user cancelled a send that was waiting on a reconnect. + const restoreUnsentDraft = (failure: TurnDeliveryFailure | null) => { + if (isLocalDraftThread && draftId && draftThread) { + restoreFailedDraftThread( + draftId, + draftThread, + // A cancelled launch may still have committed, so its draft must not reuse the thread id. + failure === null || wasBootstrapThreadDeleted(squashAtomCommandFailure(failure)) + ? newThreadId() + : threadIdForSend, + ); + clearBackgroundDraftSubmissionByRef(scopeThreadRef(environmentId, threadIdForSend)); + } + if ( + !composerDraftHasUserContent( + useComposerDraftStore.getState().getComposerDraft(composerDraftTarget), + ) + ) { + setOptimisticUserMessages((existing) => { + const removed = existing.filter((message) => message.id === messageIdForSend); + for (const message of removed) { + revokeUserMessagePreviewUrls(message); + } + const next = existing.filter((message) => message.id !== messageIdForSend); + return next.length === existing.length ? existing : next; + }); + const retryComposerImages = restoreComposerDraft(composerDraftTarget, sendSnapshot); + if (currentRouteThreadKeyRef.current === routeThreadKey) { + promptRef.current = messageTextForSend; + composerImagesRef.current = retryComposerImages; + composerFilesRef.current = composerFilesSnapshot; + composerTerminalContextsRef.current = composerTerminalContextsSnapshot; + composerRef.current?.resetCursorState({ + cursor: collapseExpandedComposerCursor(messageTextForSend, messageTextForSend.length), + prompt: messageTextForSend, + detectTrigger: true, + }); + } + } + if (failure !== null && !isAtomCommandInterrupted(failure)) { + const error = squashAtomCommandFailure(failure); + setThreadError( + threadIdForSend, + error instanceof Error ? error.message : "Failed to send message.", + ); + if ( + draftId && + (backgroundDraftOpened || currentRouteThreadKeyRef.current !== routeThreadKey) + ) { + toastManager.add( + stackedThreadToast({ + type: "error", + title: + submissionIntent === "background" + ? "Background task failed" + : "Could not start thread", + description: error instanceof Error ? error.message : "Failed to send message.", + actionProps: { + children: "Open draft", + onClick: () => { + void navigate({ to: "/draft/$draftId", params: { draftId } }); + }, + }, + }), + ); + } + } + }; if (failure === null && turnAttachmentsResult._tag === "Success") { const bootstrap = isLocalDraftThread || baseBranchForWorktree @@ -1277,9 +1364,11 @@ export function useComposerSend(input: ComposerSendInput) { ? scopeThreadRef(environmentId, threadIdForSend) : null; if (backgroundThreadRef) beginBackgroundDraftSubmissionByRef(backgroundThreadRef); - const startPromise = startThreadTurn({ + const startPromise = turnDeliveries.deliver( environmentId, - input: { + { + // Allocated once so a replay after a lost acknowledgement is the same command. + commandId: CommandId.make(randomUUID()), threadId: threadIdForSend, message: { messageId: messageIdForSend, @@ -1300,7 +1389,8 @@ export function useComposerSend(input: ComposerSendInput) { ...(bootstrap ? { bootstrap } : {}), createdAt: messageCreatedAt, }, - }); + optimisticMessage, + ); if (backgroundThreadRef) { try { backgroundDraftOpened = Boolean( @@ -1324,25 +1414,29 @@ export function useComposerSend(input: ComposerSendInput) { ); } } - const startResult = await startPromise; - if (startResult._tag === "Failure") { - failure = startResult; + const outcome = await startPromise; + if (outcome.type === "rejected") { + failure = outcome.failure; + } else if (outcome.type === "cancelled") { + sendCancelled = true; + } else if (outcome.type === "waiting") { + // The server may already hold the message, so it stays on screen as waiting to send. + turnStartSucceeded = true; + if (!sendQueuesBehindSetup) resetLocalDispatch(); + void outcome.settled.then((end) => { + if (end.type === "delivered") { + finishStartedTurn(); + return; + } + restoreUnsentDraft(end.type === "rejected" ? end.failure : null); + setDockedDraftHeroThreadKey((currentThreadKey) => + currentThreadKey === activeThreadKey ? null : currentThreadKey, + ); + }); } else { turnStartSucceeded = true; - setKeepFullHistory(routeThreadKey, false); - clearUsageLimitsFor(routeThreadKey); - if (turnUsesAttachmentUploads) { - releaseDraftAttachments(composerAttachmentsSnapshot); - } + finishStartedTurn(); acknowledgeActiveThreadWoke(); - // A draft still on screen finalizes once its thread route takes over. - if (sentDraftThreadRef) { - if (backgroundDraftOpened || currentRouteThreadKeyRef.current !== routeThreadKey) { - finalizePromotedDraftThreadByRef(sentDraftThreadRef); - } else { - clearBackgroundDraftSubmissionByRef(sentDraftThreadRef); - } - } if (backgroundThreadRef) { if (backgroundDraftOpened) { toastManager.add( @@ -1371,69 +1465,8 @@ export function useComposerSend(input: ComposerSendInput) { } } - if (failure !== null) { - if (isLocalDraftThread && draftId && draftThread) { - restoreFailedDraftThread( - draftId, - draftThread, - wasBootstrapThreadDeleted(squashAtomCommandFailure(failure)) - ? newThreadId() - : threadIdForSend, - ); - clearBackgroundDraftSubmissionByRef(scopeThreadRef(environmentId, threadIdForSend)); - } - if ( - !composerDraftHasUserContent( - useComposerDraftStore.getState().getComposerDraft(composerDraftTarget), - ) - ) { - setOptimisticUserMessages((existing) => { - const next = existing.filter((message) => message.id !== messageIdForSend); - return next.length === existing.length ? existing : next; - }); - // The restored draft clones its images, so the sent copies are no longer used. - optimisticBlobPreviews.release(optimisticMessage); - const retryComposerImages = restoreComposerDraft(composerDraftTarget, sendSnapshot); - if (currentRouteThreadKeyRef.current === routeThreadKey) { - promptRef.current = messageTextForSend; - composerImagesRef.current = retryComposerImages; - composerFilesRef.current = composerFilesSnapshot; - composerTerminalContextsRef.current = composerTerminalContextsSnapshot; - composerRef.current?.resetCursorState({ - cursor: collapseExpandedComposerCursor(messageTextForSend, messageTextForSend.length), - prompt: messageTextForSend, - detectTrigger: true, - }); - } - } - if (!isAtomCommandInterrupted(failure)) { - const error = squashAtomCommandFailure(failure); - setThreadError( - threadIdForSend, - error instanceof Error ? error.message : "Failed to send message.", - ); - if ( - draftId && - (backgroundDraftOpened || currentRouteThreadKeyRef.current !== routeThreadKey) - ) { - toastManager.add( - stackedThreadToast({ - type: "error", - title: - submissionIntent === "background" - ? "Background task failed" - : "Could not start thread", - description: error instanceof Error ? error.message : "Failed to send message.", - actionProps: { - children: "Open draft", - onClick: () => { - void navigate({ to: "/draft/$draftId", params: { draftId } }); - }, - }, - }), - ); - } - } + if (failure !== null || sendCancelled) { + restoreUnsentDraft(failure); } sendInFlightRef.current = false; setSetupQueueSendThreadKey((current) => (current === activeThreadKey ? null : current)); diff --git a/apps/web/src/state/threads.ts b/apps/web/src/state/threads.ts index 07fffa97cfbe0..6b4a28d0e06b3 100644 --- a/apps/web/src/state/threads.ts +++ b/apps/web/src/state/threads.ts @@ -34,6 +34,7 @@ export const threadEnvironment = createThreadEnvironmentAtoms( threadFamilies.snapshotAtom, ); const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime); +export const environmentThreadStateAtom = environmentThreads.stateAtom; export const environmentThreadDetails = createEnvironmentThreadDetailAtoms( environmentThreads.stateAtom, ); diff --git a/apps/web/src/state/turnDeliveries.ts b/apps/web/src/state/turnDeliveries.ts new file mode 100644 index 0000000000000..217741d47c4d7 --- /dev/null +++ b/apps/web/src/state/turnDeliveries.ts @@ -0,0 +1,68 @@ +import { AVAILABLE_CONNECTION_STATE } from "@t3tools/client-runtime/connection"; +import { EMPTY_ENVIRONMENT_THREAD_STATE } from "@t3tools/client-runtime/state/threads"; +import { + createTurnDeliveries, + type PendingTurnDelivery, +} from "@t3tools/client-runtime/state/turn-delivery"; +import type { EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts"; +import * as Option from "effect/Option"; +import { AsyncResult } from "effect/reactivity"; +import { useSyncExternalStore } from "react"; + +import { environmentCatalog } from "../connection/catalog"; +import { appAtomRegistry } from "../rpc/atomRegistry"; +import type { ChatMessage } from "../types"; +import { environmentThreadStateAtom, threadEnvironment } from "./threads"; +import { runOrchestrationCommand } from "./use-orchestration-command"; + +/** Outlives ChatView, so a send waiting on a reconnect survives navigating between threads. */ +export const turnDeliveries = createTurnDeliveries({ + send: (environmentId, input) => + runOrchestrationCommand( + threadEnvironment.startTurn, + { environmentId, input }, + { reportFailure: false }, + ), + observeThread: (environmentId, threadId, listener) => + appAtomRegistry.subscribe( + environmentThreadStateAtom(environmentId, threadId), + (result) => + listener(Option.getOrElse(AsyncResult.value(result), () => EMPTY_ENVIRONMENT_THREAD_STATE)), + { immediate: true }, + ), + observeConnection: (environmentId, listener) => + appAtomRegistry.subscribe( + environmentCatalog.stateAtom(environmentId), + (result) => + listener(Option.getOrElse(AsyncResult.value(result), () => AVAILABLE_CONNECTION_STATE)), + { immediate: true }, + ), +}); + +const EMPTY: ReadonlyArray> = []; + +function usePendingTurnDeliverySnapshot() { + return useSyncExternalStore(turnDeliveries.subscribe, turnDeliveries.getSnapshot); +} + +export function usePendingTurnDeliveries( + environmentId: EnvironmentId, + threadId: ThreadId | null, +): ReadonlyArray> { + const pending = usePendingTurnDeliverySnapshot().filter( + (delivery) => delivery.environmentId === environmentId && delivery.threadId === threadId, + ); + return pending.length === 0 ? EMPTY : pending; +} + +export function usePendingTurnDelivery( + environmentId: EnvironmentId, + messageId: MessageId, +): PendingTurnDelivery | null { + const snapshot = usePendingTurnDeliverySnapshot(); + return ( + snapshot.find( + (delivery) => delivery.environmentId === environmentId && delivery.messageId === messageId, + ) ?? null + ); +} diff --git a/apps/web/src/state/use-orchestration-command.ts b/apps/web/src/state/use-orchestration-command.ts index 98f7efeecbfe6..7731a70fed84f 100644 --- a/apps/web/src/state/use-orchestration-command.ts +++ b/apps/web/src/state/use-orchestration-command.ts @@ -1,7 +1,8 @@ -import type { - AtomCommand, - AtomCommandOptions, - AtomCommandResult, +import { + type AtomCommand, + type AtomCommandOptions, + type AtomCommandResult, + runAtomCommand, } from "@t3tools/client-runtime/state/runtime"; import { AuthOrchestrationOperateScope, @@ -11,19 +12,19 @@ import { import * as Cause from "effect/Cause"; import { AsyncResult } from "effect/reactivity"; +import { appAtomRegistry } from "../rpc/atomRegistry"; + import { readEnvironmentScope } from "./session"; import { useAtomCommand } from "./use-atom-command"; const ORCHESTRATION_OPERATE_DENIED_MESSAGE = "This connection cannot change threads or projects."; -export function useOrchestrationCommand( - command: AtomCommand, - options?: string | AtomCommandOptions, -): (value: W) => Promise> { - const run = useAtomCommand(command, options); - return async (value: W): Promise> => { - if (!readEnvironmentScope(value.environmentId, AuthOrchestrationOperateScope)) { - return AsyncResult.failure( +function operateDenied( + environmentId: EnvironmentId, +): AtomCommandResult | null { + return readEnvironmentScope(environmentId, AuthOrchestrationOperateScope) + ? null + : AsyncResult.failure( Cause.fail( new EnvironmentAuthorizationError({ requiredScope: AuthOrchestrationOperateScope, @@ -31,7 +32,29 @@ export function useOrchestrationCommand( + command: AtomCommand, + options?: string | AtomCommandOptions, +): (value: W) => Promise> { + const run = useAtomCommand(command, options); + return async (value: W): Promise> => + operateDenied(value.environmentId) ?? run(value); +} + +/** For callers that outlive the component that started them. */ +export async function runOrchestrationCommand< + W extends { readonly environmentId: EnvironmentId }, + A, + E, +>( + command: AtomCommand, + value: W, + options?: AtomCommandOptions, +): Promise> { + return ( + operateDenied(value.environmentId) ?? + runAtomCommand(appAtomRegistry, command, value, options) + ); } diff --git a/packages/client-runtime/package.json b/packages/client-runtime/package.json index 9600ad7ad356d..692734401f2eb 100644 --- a/packages/client-runtime/package.json +++ b/packages/client-runtime/package.json @@ -307,6 +307,10 @@ "types": "./src/state/threads.ts", "default": "./src/state/threads.ts" }, + "./state/turn-delivery": { + "types": "./src/state/turnDelivery.ts", + "default": "./src/state/turnDelivery.ts" + }, "./state/subagent-display": { "types": "./src/state/subagentDisplay.ts", "default": "./src/state/subagentDisplay.ts" diff --git a/packages/client-runtime/src/errors/transport.ts b/packages/client-runtime/src/errors/transport.ts index a38f0d7b9f35f..a29748ed7ed07 100644 --- a/packages/client-runtime/src/errors/transport.ts +++ b/packages/client-runtime/src/errors/transport.ts @@ -35,3 +35,32 @@ export function isTransportConnectionErrorMessage(message: string | null | undef export function sanitizeThreadErrorMessage(message: string | null | undefined): string | null { return isTransportConnectionErrorMessage(message) ? null : (message ?? null); } + +function errorMessage(error: unknown): string | null { + if (error instanceof Error) { + return error.message; + } + if (typeof error === "object" && error !== null && "message" in error) { + return typeof error.message === "string" ? error.message : null; + } + return typeof error === "string" ? error : null; +} + +/** Dispatch/auth rejections invalidate payloads; other typed failures may be disconnected transports and must retry. */ +export function isRetryableDeliveryFailure(error: unknown): boolean { + if (typeof error === "object" && error !== null && "_tag" in error) { + switch (error._tag) { + case "OrchestrationDispatchCommandError": + case "EnvironmentAuthorizationError": + return false; + case "ConnectionTransientError": + case "RpcClientError": + case "EnvironmentRpcUnavailableError": + case "EnvironmentNotRegisteredError": + return true; + default: + break; + } + } + return isTransportConnectionErrorMessage(errorMessage(error)); +} diff --git a/packages/client-runtime/src/state/threadCommands.test.ts b/packages/client-runtime/src/state/threadCommands.test.ts index 2b1b1162d6bc9..8ee50e280fc90 100644 --- a/packages/client-runtime/src/state/threadCommands.test.ts +++ b/packages/client-runtime/src/state/threadCommands.test.ts @@ -25,11 +25,14 @@ import * as Scope from "effect/Scope"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { Atom, AtomRegistry } from "effect/reactivity"; +import { AVAILABLE_CONNECTION_STATE } from "../connection/model.ts"; import * as EnvironmentRegistry from "../connection/registry.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import type { RpcSession } from "../rpc/session.ts"; import { createThreadEnvironmentAtoms } from "./threadCommands.ts"; import * as ThreadHistoryController from "./threadHistoryController.ts"; +import { EMPTY_ENVIRONMENT_THREAD_STATE, type EnvironmentThreadState } from "./threadState.ts"; +import { createTurnDeliveries } from "./turnDelivery.ts"; const ENVIRONMENT_ID = EnvironmentId.make("remote"); const THREAD_ID = ThreadId.make("thread"); @@ -507,3 +510,87 @@ describe("remote thread lifecycle commands", () => { }), ); }); + +describe("turn delivery across a lost acknowledgement", () => { + const threadState = (status: EnvironmentThreadState["status"]): EnvironmentThreadState => ({ + ...EMPTY_ENVIRONMENT_THREAD_STATE, + status, + data: Option.some({ messages: [], runs: [], turnItems: [] } as never), + }); + + it.effect("keeps the send, lets Stop through, and replays the same command after resync", () => + Effect.gen(function* () { + const h = yield* makeHarness(); + const threadListeners = new Set<(state: EnvironmentThreadState) => void>(); + const deliveries = createTurnDeliveries({ + send: (environmentId, input) => + h.commands.startTurn.run(h.registry, { environmentId, input }), + observeThread: (_environmentId, _threadId, listener) => { + threadListeners.add(listener); + listener(threadState("live")); + return () => threadListeners.delete(listener); + }, + observeConnection: (_environmentId, listener) => { + listener({ ...AVAILABLE_CONNECTION_STATE, phase: "connected", generation: 1 }); + return () => {}; + }, + }); + const emitThread = (status: EnvironmentThreadState["status"]) => { + for (const listener of threadListeners) listener(threadState(status)); + }; + + const outcome = deliveries.deliver( + ENVIRONMENT_ID, + { + commandId: CommandId.make("send-command"), + threadId: THREAD_ID, + message: { + messageId: MessageId.make("send-message"), + role: "user", + text: "Delivery check. Reply only with RECEIVED.", + attachments: [], + }, + runtimeMode: "full-access", + interactionMode: "default", + dispatchMode: "start", + }, + "optimistic row", + ); + const first = yield* Queue.take(h.requests); + expect(first.command).toMatchObject({ + type: "message.dispatch", + commandId: "send-command", + messageId: "send-message", + }); + // The frame reached the server; only the reply was lost with the socket. + yield* Deferred.fail(first.reply, new Error("SocketCloseError: connection lost")); + const waiting = yield* Effect.promise(() => outcome); + expect(waiting.type).toBe("waiting"); + expect(deliveries.getSnapshot()).toEqual([ + expect.objectContaining({ messageId: "send-message", state: "waiting" }), + ]); + + const stop = h.commands.interruptTurn.run(h.registry, { + environmentId: ENVIRONMENT_ID, + input: { threadId: THREAD_ID, runId: RunId.make("running-run") }, + }); + const interrupt = yield* Queue.take(h.requests); + expect(interrupt.command.type).toBe("run.interrupt"); + yield* Deferred.succeed(interrupt.reply, { sequence: 2 }); + expect((yield* Effect.promise(() => stop))._tag).toBe("Success"); + + emitThread("cached"); + emitThread("synchronizing"); + emitThread("live"); + const replay = yield* Queue.take(h.requests); + expect(replay.command).toEqual(first.command); + yield* Deferred.succeed(replay.reply, { sequence: 3 }); + if (waiting.type !== "waiting") return; + expect(yield* Effect.promise(() => waiting.settled)).toEqual({ + type: "delivered", + value: { sequence: 3 }, + }); + expect(deliveries.getSnapshot()).toEqual([]); + }), + ); +}); diff --git a/packages/client-runtime/src/state/turnDelivery.test.ts b/packages/client-runtime/src/state/turnDelivery.test.ts new file mode 100644 index 0000000000000..d27b7a2739977 --- /dev/null +++ b/packages/client-runtime/src/state/turnDelivery.test.ts @@ -0,0 +1,273 @@ +import { + CommandId, + EnvironmentId, + MessageId, + ProjectId, + ProviderInstanceId, + ThreadId, + type OrchestrationV2ThreadProjection, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Option from "effect/Option"; +import { AsyncResult } from "effect/reactivity"; +import { describe, expect, it } from "@effect/vitest"; + +import { AVAILABLE_CONNECTION_STATE, type SupervisorConnectionState } from "../connection/model.ts"; +import type { AtomCommandResult } from "./runtime.ts"; +import { EMPTY_ENVIRONMENT_THREAD_STATE, type EnvironmentThreadState } from "./threadState.ts"; +import { createTurnDeliveries, type TurnDeliveryInput } from "./turnDelivery.ts"; + +const ENVIRONMENT_ID = EnvironmentId.make("environment"); +const THREAD_ID = ThreadId.make("thread"); +const MESSAGE_ID = MessageId.make("message"); +const MODEL_SELECTION = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }; + +const INPUT: TurnDeliveryInput = { + commandId: CommandId.make("command"), + threadId: THREAD_ID, + message: { + messageId: MESSAGE_ID, + role: "user", + text: "Delivery check. Reply only with RECEIVED.", + attachments: [], + }, + modelSelection: MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", +}; + +const LAUNCH: TurnDeliveryInput = { + ...INPUT, + bootstrap: { + createThread: { + projectId: ProjectId.make("project"), + title: "New thread", + modelSelection: MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: "2026-10-25T00:00:00.000Z", + }, + }, +}; + +type SendResult = AtomCommandResult<{ sequence: number }, unknown>; + +const LOST_ACK: SendResult = AsyncResult.failure( + Cause.fail({ _tag: "RpcClientError", message: "SocketCloseError" }), +); +const REJECTED: SendResult = AsyncResult.failure( + Cause.fail({ _tag: "OrchestrationDispatchCommandError", message: "Thread is archived." }), +); +const ACCEPTED: SendResult = AsyncResult.success({ sequence: 7 }); + +function connection( + phase: SupervisorConnectionState["phase"], + generation: number, +): SupervisorConnectionState { + return { ...AVAILABLE_CONNECTION_STATE, phase, generation }; +} + +function thread( + status: EnvironmentThreadState["status"], + projection: Partial> = {}, +): EnvironmentThreadState { + return { + ...EMPTY_ENVIRONMENT_THREAD_STATE, + status, + data: Option.some({ + messages: [], + runs: [], + turnItems: [], + ...projection, + } as unknown as OrchestrationV2ThreadProjection), + }; +} + +function setup(results: ReadonlyArray>) { + const queue = [...results]; + const sent: Array = []; + let threadListener: ((state: EnvironmentThreadState) => void) | null = null; + let connectionListener: ((state: SupervisorConnectionState) => void) | null = null; + let threadObservations = 0; + const deliveries = createTurnDeliveries<{ sequence: number }, unknown, string>({ + send: async (_environmentId, input) => { + sent.push(input); + return queue.shift() ?? ACCEPTED; + }, + observeThread: (_environmentId, _threadId, listener) => { + threadObservations += 1; + threadListener = listener; + listener(thread("live")); + return () => { + threadListener = null; + }; + }, + observeConnection: (_environmentId, listener) => { + connectionListener = listener; + listener(connection("connected", 1)); + return () => { + connectionListener = null; + }; + }, + }); + return { + deliveries, + sent, + threadObservations: () => threadObservations, + emitThread: (state: EnvironmentThreadState) => threadListener?.(state), + emitConnection: (state: SupervisorConnectionState) => connectionListener?.(state), + }; +} + +async function deliverUntilWaiting(harness: ReturnType, input = INPUT) { + const outcome = await harness.deliveries.deliver(ENVIRONMENT_ID, input, "optimistic row"); + expect(outcome.type).toBe("waiting"); + if (outcome.type !== "waiting") throw new Error(`Expected waiting, got ${outcome.type}.`); + return outcome.settled; +} + +describe("turn deliveries", () => { + it("keeps a send whose acknowledgement was lost instead of rejecting it", async () => { + const harness = setup([LOST_ACK]); + await deliverUntilWaiting(harness); + + expect(harness.deliveries.getSnapshot()).toEqual([ + expect.objectContaining({ + messageId: MESSAGE_ID, + state: "waiting", + display: "optimistic row", + }), + ]); + expect(harness.sent).toHaveLength(1); + }); + + it("does not replay until the thread resyncs after the lost send", async () => { + const harness = setup([LOST_ACK]); + await deliverUntilWaiting(harness); + + harness.emitThread(thread("live")); + harness.emitThread(thread("synchronizing")); + + expect(harness.sent).toHaveLength(1); + }); + + it("replays the same command and payload once the thread resyncs without the message", async () => { + const harness = setup([LOST_ACK, ACCEPTED]); + const settled = await deliverUntilWaiting(harness); + + harness.emitThread(thread("cached")); + harness.emitThread(thread("synchronizing")); + harness.emitThread(thread("live")); + + expect(await settled).toEqual({ type: "delivered", value: { sequence: 7 } }); + expect(harness.sent).toEqual([INPUT, INPUT]); + expect(harness.deliveries.getSnapshot()).toEqual([]); + }); + + it("settles without a resend when the resynced thread already holds the message", async () => { + const harness = setup([LOST_ACK]); + const settled = await deliverUntilWaiting(harness); + + harness.emitThread(thread("synchronizing")); + harness.emitThread( + thread("live", { + messages: [ + { id: MESSAGE_ID, role: "user" } as OrchestrationV2ThreadProjection["messages"][number], + ], + }), + ); + + expect(await settled).toEqual({ type: "delivered", value: null }); + expect(harness.sent).toHaveLength(1); + }); + + it("settles without a resend when the resynced thread queued the message", async () => { + const harness = setup([LOST_ACK]); + const settled = await deliverUntilWaiting(harness); + + harness.emitThread(thread("synchronizing")); + harness.emitThread( + thread("live", { + runs: [ + { + userMessageId: MESSAGE_ID, + status: "queued", + } as OrchestrationV2ThreadProjection["runs"][number], + ], + }), + ); + + expect(await settled).toEqual({ type: "delivered", value: null }); + expect(harness.sent).toHaveLength(1); + }); + + it("ends once with the rejection when the replay is refused", async () => { + const harness = setup([LOST_ACK, REJECTED]); + const settled = await deliverUntilWaiting(harness); + + harness.emitThread(thread("synchronizing")); + harness.emitThread(thread("live")); + + expect(await settled).toEqual({ type: "rejected", failure: REJECTED }); + harness.emitThread(thread("synchronizing")); + harness.emitThread(thread("live")); + expect(harness.sent).toHaveLength(2); + expect(harness.deliveries.getSnapshot()).toEqual([]); + }); + + it("stops replaying after a cancel", async () => { + const harness = setup([LOST_ACK]); + const settled = await deliverUntilWaiting(harness); + const [pending] = harness.deliveries.getSnapshot(); + + harness.deliveries.cancel(pending!.key); + harness.emitThread(thread("synchronizing")); + harness.emitThread(thread("live")); + + expect(await settled).toEqual({ type: "cancelled" }); + expect(harness.sent).toHaveLength(1); + expect(harness.deliveries.getSnapshot()).toEqual([]); + }); + + it("keeps a replay the server accepted even when cancelled while it was in flight", async () => { + let acceptReplay!: (result: SendResult) => void; + const replay = new Promise((resolve) => { + acceptReplay = resolve; + }); + const harness = setup([LOST_ACK, replay]); + const settled = await deliverUntilWaiting(harness); + + harness.emitThread(thread("synchronizing")); + harness.emitThread(thread("live")); + const [pending] = harness.deliveries.getSnapshot(); + expect(pending?.state).toBe("retrying"); + harness.deliveries.cancel(pending!.key); + acceptReplay(ACCEPTED); + + expect(await settled).toEqual({ type: "delivered", value: { sequence: 7 } }); + }); + + it("returns a rejection right away without holding the send", async () => { + const harness = setup([REJECTED]); + const outcome = await harness.deliveries.deliver(ENVIRONMENT_ID, INPUT, "optimistic row"); + + expect(outcome).toEqual({ type: "rejected", failure: REJECTED }); + expect(harness.deliveries.getSnapshot()).toEqual([]); + }); + + it("replays a first message after reconnecting without watching the unborn thread", async () => { + const harness = setup([LOST_ACK, ACCEPTED]); + const settled = await deliverUntilWaiting(harness, LAUNCH); + + harness.emitConnection(connection("connected", 1)); + expect(harness.sent).toHaveLength(1); + harness.emitConnection(connection("connecting", 2)); + harness.emitConnection(connection("connected", 2)); + + expect(await settled).toEqual({ type: "delivered", value: { sequence: 7 } }); + expect(harness.sent).toEqual([LAUNCH, LAUNCH]); + expect(harness.threadObservations()).toBe(0); + }); +}); diff --git a/packages/client-runtime/src/state/turnDelivery.ts b/packages/client-runtime/src/state/turnDelivery.ts new file mode 100644 index 0000000000000..f1becd0c44b2a --- /dev/null +++ b/packages/client-runtime/src/state/turnDelivery.ts @@ -0,0 +1,79 @@ +import type { CommandId, EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts"; + +import type { SupervisorConnectionState } from "../connection/model.ts"; +import type { StartThreadTurnInput } from "../operations/commands.ts"; +import type { AtomCommandResult } from "./runtime.ts"; +import type { EnvironmentThreadState } from "./threadState.ts"; + +export type TurnDeliveryInput = StartThreadTurnInput & { readonly commandId: CommandId }; + +export type TurnDeliveryFailure = Extract, { _tag: "Failure" }>; + +export type TurnDeliveryEnd = + /** `value` is null when a resynced thread already held the message. */ + | { readonly type: "delivered"; readonly value: A | null } + | { readonly type: "rejected"; readonly failure: TurnDeliveryFailure } + | { readonly type: "cancelled" }; + +export type TurnDeliveryOutcome = + | TurnDeliveryEnd + /** The server may have committed it; `settled` ends after a later resync or cancel. */ + | { readonly type: "waiting"; readonly settled: Promise> }; + +export interface PendingTurnDelivery { + readonly key: string; + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; + readonly messageId: MessageId; + readonly state: "waiting" | "retrying"; + readonly display: M; +} + +export interface TurnDeliveryDependencies { + readonly send: ( + environmentId: EnvironmentId, + input: TurnDeliveryInput, + ) => Promise>; + /** Must call the listener with the current state before returning. */ + readonly observeThread: ( + environmentId: EnvironmentId, + threadId: ThreadId, + listener: (state: EnvironmentThreadState) => void, + ) => () => void; + /** Must call the listener with the current state before returning. */ + readonly observeConnection: ( + environmentId: EnvironmentId, + listener: (state: SupervisorConnectionState) => void, + ) => () => void; +} + +export function turnDeliveryKey( + environmentId: EnvironmentId, + threadId: ThreadId, + messageId: MessageId, +): string { + return JSON.stringify([environmentId, threadId, messageId]); +} + +/** Sends a turn once; any failure is final. */ +export function createTurnDeliveries(deps: TurnDeliveryDependencies) { + const snapshot: ReadonlyArray> = []; + const deliver = async ( + environmentId: EnvironmentId, + input: TurnDeliveryInput, + _display: M, + ): Promise> => { + const result = await deps.send(environmentId, input); + return result._tag === "Success" + ? { type: "delivered", value: result.value } + : { type: "rejected", failure: result }; + }; + return { + deliver, + cancel: (_key: string) => {}, + getSnapshot: (): ReadonlyArray> => snapshot, + subscribe: (_listener: () => void) => () => {}, + }; +} + +export type TurnDeliveries = ReturnType>; From 616606f84867e32a9ec6af12739ab73677179067 Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Sun, 25 Oct 2026 03:11:42 +0000 Subject: [PATCH 02/11] fix(web): keep a send whose acknowledgement was lost and deliver it once When the socket dropped after a turn reached the server, the composer treated the missing reply as a rejection: the message vanished and the draft came back, so resending duplicated it once the original appeared. The send now stays on screen as waiting to send, in a store that outlives ChatView. Once the thread resyncs on a new session, it settles if the server already holds the message (sent or queued), and otherwise replays the same command id and payload. A first message replays on the next connection, relying on its command receipt. Rejection or Cancel restores the draft only into an empty composer. Waiting happens outside the thread's serial command lane, so Stop is not held behind it. Co-Authored-By: Claude Opus 5.5 --- apps/web/src/components/ChatView.tsx | 21 +- .../chat/TimelineUserMessageRow.tsx | 21 ++ .../src/components/chat/useComposerSend.ts | 5 + docs/user/composer.md | 5 + .../src/state/turnDelivery.test.ts | 19 +- .../client-runtime/src/state/turnDelivery.ts | 201 ++++++++++++++++-- 6 files changed, 244 insertions(+), 28 deletions(-) diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index cf48ddcb842dd..64728ba26f531 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -305,6 +305,7 @@ import { QueuedRunsControl, type QueuedRunsControlHandle } from "./chat/QueuedRu import { useThreadLinkedPullRequest } from "./pullRequest/useThreadLinkedPullRequest"; import { useProactivePanels } from "../hooks/useProactivePanels"; import { useIncrementalTimelineEntries } from "../hooks/useIncrementalTimelineEntries"; +import { usePendingTurnDeliveries } from "../state/turnDeliveries"; import { useRightPanelSurfaceClose } from "../hooks/useRightPanelSurfaceClose"; import { ComposerBannerStack, type ComposerBannerStackItem } from "./chat/ComposerBannerStack"; import { modelFallbackBannerItem } from "./chat/ComposerUsageLimits"; @@ -653,6 +654,20 @@ export default function ChatView(props: ChatViewProps) { setHeldOptimisticUserMessages((state) => applyOptimisticUserMessagesAction(state, threadId, action), ); + // Sends waiting on a reconnect live outside this view, so they stay visible across thread switches. + const pendingTurnDeliveries = usePendingTurnDeliveries(environmentId, threadId); + const displayedOptimisticMessages = + pendingTurnDeliveries.length === 0 + ? optimisticUserMessages + : [ + ...optimisticUserMessages, + ...pendingTurnDeliveries.flatMap((delivery) => + serverAcknowledgedUserMessageIds.has(delivery.messageId) || + optimisticUserMessages.some((message) => message.id === delivery.messageId) + ? [] + : [delivery.display], + ), + ]; // Draft ids survive promotion and expire on navigation, keeping the pending resend scoped correctly. const [workLocallyResendDraftId, setWorkLocallyResendDraftId] = useState(null); const [feedbackSubmissionsByThreadKey, setFeedbackSubmissionsByThreadKey] = useState< @@ -1498,7 +1513,7 @@ export default function ChatView(props: ChatViewProps) { ], ); const timelineEntriesInput = { - optimisticMessages: optimisticUserMessages, + optimisticMessages: displayedOptimisticMessages, anchoredMessages: anchoredTimelineMessages, attachmentUrlById: timelineAttachmentUrlById, ...(serverProjection === null @@ -1519,7 +1534,7 @@ export default function ChatView(props: ChatViewProps) { ...timelineEntriesInput, visibleTurnItems: serverTimelineRows, }); - const draftTimelineEntries = optimisticUserMessages.map( + const draftTimelineEntries = displayedOptimisticMessages.map( (message) => ({ id: message.id, @@ -2939,7 +2954,7 @@ export default function ChatView(props: ChatViewProps) { })} environmentId={activeThread.environmentId} threadId={activeThread.id} - optimisticMessages={optimisticUserMessages} + optimisticMessages={displayedOptimisticMessages} editingRunId={queuedRunEdit.editingRunId} editBlockedReason={queuedRunEditBlockedReason({ openQuestionCount: pendingUserInputs.length, diff --git a/apps/web/src/components/chat/TimelineUserMessageRow.tsx b/apps/web/src/components/chat/TimelineUserMessageRow.tsx index c4363a1b45983..dcc7522ac3eea 100644 --- a/apps/web/src/components/chat/TimelineUserMessageRow.tsx +++ b/apps/web/src/components/chat/TimelineUserMessageRow.tsx @@ -11,6 +11,7 @@ import { import { resolveUserMessagePresentation } from "@t3tools/client-runtime/user-message"; import { Link } from "@tanstack/react-router"; import { useQuietThreadShell } from "../../state/entities"; +import { turnDeliveries, usePendingTurnDelivery } from "../../state/turnDeliveries"; import { useEnvironmentQuery } from "../../state/query"; import { serverEnvironment } from "../../state/server"; import { replaceComposerContextReferences } from "@t3tools/shared/composerContextReferences"; @@ -292,6 +293,25 @@ function UserMessageProjectedStatus({ row }: { row: Extract + {delivery.state === "waiting" ? "Waiting to send" : "Sending"} + {delivery.state === "waiting" ? ( + turnDeliveries.cancel(delivery.key)}> + Cancel + + ) : null} + + ); +} + function UserTimelineRow({ row }: { row: Extract }) { const ctx = use(TimelineRowCtx); const { onImageExpand, onFileOpen } = ctx; @@ -711,6 +731,7 @@ function UserTimelineRow({ row }: { row: Extract + {!folded || expanded ? (
diff --git a/apps/web/src/components/chat/useComposerSend.ts b/apps/web/src/components/chat/useComposerSend.ts index 13a64b220b09d..24c0231c7ebda 100644 --- a/apps/web/src/components/chat/useComposerSend.ts +++ b/apps/web/src/components/chat/useComposerSend.ts @@ -1299,6 +1299,11 @@ export function useComposerSend(input: ComposerSendInput) { detectTrigger: true, }); } + } else { + setOptimisticUserMessages((existing) => { + const next = existing.filter((message) => message.id !== messageIdForSend); + return next.length === existing.length ? existing : next; + }); } if (failure !== null && !isAtomCommandInterrupted(failure)) { const error = squashAtomCommandFailure(failure); diff --git a/docs/user/composer.md b/docs/user/composer.md index 714572dfd5005..eedee4fba2212 100644 --- a/docs/user/composer.md +++ b/docs/user/composer.md @@ -85,6 +85,11 @@ messages while disconnected. Uploads resume when you reconnect. Drafts and queue messages survive app restarts. Signing out of T3 Connect keeps that work on your device until you sign back into the same account. +On web and desktop, a message sent as the connection drops shows **Waiting to send**. When the +thread reconnects, T3 Code sends it once, or keeps it if the server already received it. Cancel +stops the send and restores the message to an empty composer. It can't unsend a message the +server already received. + ## Custom models On web and desktop, use Settings → Providers → **Models** to add an unlisted model with a custom diff --git a/packages/client-runtime/src/state/turnDelivery.test.ts b/packages/client-runtime/src/state/turnDelivery.test.ts index d27b7a2739977..0c9aea8729fe7 100644 --- a/packages/client-runtime/src/state/turnDelivery.test.ts +++ b/packages/client-runtime/src/state/turnDelivery.test.ts @@ -10,7 +10,7 @@ import { import * as Cause from "effect/Cause"; import * as Option from "effect/Option"; import { AsyncResult } from "effect/reactivity"; -import { describe, expect, it } from "@effect/vitest"; +import { describe, expect, it } from "vite-plus/test"; import { AVAILABLE_CONNECTION_STATE, type SupervisorConnectionState } from "../connection/model.ts"; import type { AtomCommandResult } from "./runtime.ts"; @@ -125,7 +125,8 @@ async function deliverUntilWaiting(harness: ReturnType, input = IN const outcome = await harness.deliveries.deliver(ENVIRONMENT_ID, input, "optimistic row"); expect(outcome.type).toBe("waiting"); if (outcome.type !== "waiting") throw new Error(`Expected waiting, got ${outcome.type}.`); - return outcome.settled; + // Wrapped: an async function returning the promise would wait for it to settle. + return { settled: outcome.settled }; } describe("turn deliveries", () => { @@ -155,7 +156,7 @@ describe("turn deliveries", () => { it("replays the same command and payload once the thread resyncs without the message", async () => { const harness = setup([LOST_ACK, ACCEPTED]); - const settled = await deliverUntilWaiting(harness); + const { settled } = await deliverUntilWaiting(harness); harness.emitThread(thread("cached")); harness.emitThread(thread("synchronizing")); @@ -168,7 +169,7 @@ describe("turn deliveries", () => { it("settles without a resend when the resynced thread already holds the message", async () => { const harness = setup([LOST_ACK]); - const settled = await deliverUntilWaiting(harness); + const { settled } = await deliverUntilWaiting(harness); harness.emitThread(thread("synchronizing")); harness.emitThread( @@ -185,7 +186,7 @@ describe("turn deliveries", () => { it("settles without a resend when the resynced thread queued the message", async () => { const harness = setup([LOST_ACK]); - const settled = await deliverUntilWaiting(harness); + const { settled } = await deliverUntilWaiting(harness); harness.emitThread(thread("synchronizing")); harness.emitThread( @@ -205,7 +206,7 @@ describe("turn deliveries", () => { it("ends once with the rejection when the replay is refused", async () => { const harness = setup([LOST_ACK, REJECTED]); - const settled = await deliverUntilWaiting(harness); + const { settled } = await deliverUntilWaiting(harness); harness.emitThread(thread("synchronizing")); harness.emitThread(thread("live")); @@ -219,7 +220,7 @@ describe("turn deliveries", () => { it("stops replaying after a cancel", async () => { const harness = setup([LOST_ACK]); - const settled = await deliverUntilWaiting(harness); + const { settled } = await deliverUntilWaiting(harness); const [pending] = harness.deliveries.getSnapshot(); harness.deliveries.cancel(pending!.key); @@ -237,7 +238,7 @@ describe("turn deliveries", () => { acceptReplay = resolve; }); const harness = setup([LOST_ACK, replay]); - const settled = await deliverUntilWaiting(harness); + const { settled } = await deliverUntilWaiting(harness); harness.emitThread(thread("synchronizing")); harness.emitThread(thread("live")); @@ -259,7 +260,7 @@ describe("turn deliveries", () => { it("replays a first message after reconnecting without watching the unborn thread", async () => { const harness = setup([LOST_ACK, ACCEPTED]); - const settled = await deliverUntilWaiting(harness, LAUNCH); + const { settled } = await deliverUntilWaiting(harness, LAUNCH); harness.emitConnection(connection("connected", 1)); expect(harness.sent).toHaveLength(1); diff --git a/packages/client-runtime/src/state/turnDelivery.ts b/packages/client-runtime/src/state/turnDelivery.ts index f1becd0c44b2a..e806dbf889355 100644 --- a/packages/client-runtime/src/state/turnDelivery.ts +++ b/packages/client-runtime/src/state/turnDelivery.ts @@ -1,21 +1,29 @@ import type { CommandId, EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Option from "effect/Option"; +import { AsyncResult } from "effect/reactivity"; import type { SupervisorConnectionState } from "../connection/model.ts"; +import { isRetryableDeliveryFailure } from "../errors/transport.ts"; import type { StartThreadTurnInput } from "../operations/commands.ts"; -import type { AtomCommandResult } from "./runtime.ts"; +import { + type AtomCommandResult, + isAtomCommandInterrupted, + squashAtomCommandFailure, +} from "./runtime.ts"; import type { EnvironmentThreadState } from "./threadState.ts"; export type TurnDeliveryInput = StartThreadTurnInput & { readonly commandId: CommandId }; export type TurnDeliveryFailure = Extract, { _tag: "Failure" }>; -export type TurnDeliveryEnd = +type TurnDeliveryEnd = /** `value` is null when a resynced thread already held the message. */ | { readonly type: "delivered"; readonly value: A | null } | { readonly type: "rejected"; readonly failure: TurnDeliveryFailure } | { readonly type: "cancelled" }; -export type TurnDeliveryOutcome = +type TurnDeliveryOutcome = | TurnDeliveryEnd /** The server may have committed it; `settled` ends after a later resync or cancel. */ | { readonly type: "waiting"; readonly settled: Promise> }; @@ -29,7 +37,7 @@ export interface PendingTurnDelivery { readonly display: M; } -export interface TurnDeliveryDependencies { +interface TurnDeliveryDependencies { readonly send: ( environmentId: EnvironmentId, input: TurnDeliveryInput, @@ -47,7 +55,23 @@ export interface TurnDeliveryDependencies { ) => () => void; } -export function turnDeliveryKey( +interface Entry { + readonly pending: Omit, "state">; + readonly input: TurnDeliveryInput; + readonly launch: boolean; + state: "sending" | "waiting"; + failedOnce: boolean; + cancelRequested: boolean; + /** The thread left live sync since the last dispatch. */ + resynced: boolean; + thread: EnvironmentThreadState | null; + connection: SupervisorConnectionState | null; + dispatchGeneration: number; + settle: (end: TurnDeliveryEnd) => void; + stopObserving: Array<() => void>; +} + +function turnDeliveryKey( environmentId: EnvironmentId, threadId: ThreadId, messageId: MessageId, @@ -55,25 +79,170 @@ export function turnDeliveryKey( return JSON.stringify([environmentId, threadId, messageId]); } -/** Sends a turn once; any failure is final. */ +function threadHoldsUserMessage(state: EnvironmentThreadState, messageId: MessageId): boolean { + return Option.match(state.data, { + onNone: () => false, + onSome: (projection) => + projection.messages.some((message) => message.id === messageId) || + projection.runs.some((run) => run.userMessageId === messageId) || + projection.turnItems.some( + (item) => item.type === "user_message" && item.messageId === messageId, + ), + }); +} + +/** + * Keeps a send whose acknowledgement was lost until the thread resyncs on a later session, then + * settles it when the server already holds the message or replays the same command otherwise. + * Waiting happens here, outside the thread's serial command lane, so Stop is never held behind it. + */ export function createTurnDeliveries(deps: TurnDeliveryDependencies) { - const snapshot: ReadonlyArray> = []; + const entries = new Map>(); + const listeners = new Set<() => void>(); + let snapshot: ReadonlyArray> = []; + + const publish = () => { + snapshot = [...entries.values()].flatMap((entry) => + entry.failedOnce + ? [ + { + ...entry.pending, + state: entry.state === "waiting" ? ("waiting" as const) : ("retrying" as const), + }, + ] + : [], + ); + for (const listener of listeners) listener(); + }; + + const finish = (entry: Entry, end: TurnDeliveryEnd) => { + if (entries.get(entry.pending.key) !== entry) return; + entries.delete(entry.pending.key); + for (const stop of entry.stopObserving) stop(); + if (entry.failedOnce) publish(); + entry.settle(end); + }; + + // A launch's thread may not exist yet, and its command receipt makes a replay safe. + const readyToReplay = (entry: Entry) => + entry.launch + ? entry.connection?.phase === "connected" && + entry.connection.generation > entry.dispatchGeneration + : entry.resynced && entry.thread?.status === "live"; + + const evaluate = (entry: Entry) => { + if (entry.state !== "waiting") return; + if (!entry.launch && entry.thread?.status === "deleted") { + finish(entry, { + type: "rejected", + failure: AsyncResult.failure(Cause.fail(new Error("The thread was deleted."))), + }); + return; + } + if (!readyToReplay(entry)) return; + if (!entry.launch && threadHoldsUserMessage(entry.thread!, entry.pending.messageId)) { + finish(entry, { type: "delivered", value: null }); + return; + } + void dispatch(entry).then((end) => { + if (end !== null) finish(entry, end); + }); + }; + + const dispatch = async (entry: Entry): Promise | null> => { + entry.state = "sending"; + entry.resynced = false; + entry.dispatchGeneration = entry.connection?.generation ?? 0; + if (entry.failedOnce) publish(); + const result = await deps.send(entry.pending.environmentId, entry.input); + if (result._tag === "Success") return { type: "delivered", value: result.value }; + const retryable = + isAtomCommandInterrupted(result) || + isRetryableDeliveryFailure(squashAtomCommandFailure(result)); + if (entry.cancelRequested) return { type: "cancelled" }; + if (!retryable) return { type: "rejected", failure: result }; + entry.state = "waiting"; + entry.failedOnce = true; + publish(); + evaluate(entry); + return null; + }; + const deliver = async ( environmentId: EnvironmentId, input: TurnDeliveryInput, - _display: M, + display: M, ): Promise> => { - const result = await deps.send(environmentId, input); - return result._tag === "Success" - ? { type: "delivered", value: result.value } - : { type: "rejected", failure: result }; + const key = turnDeliveryKey(environmentId, input.threadId, input.message.messageId); + let settle!: (end: TurnDeliveryEnd) => void; + const settled = new Promise>((resolve) => { + settle = resolve; + }); + const entry: Entry = { + pending: { + key, + environmentId, + threadId: input.threadId, + messageId: input.message.messageId, + display, + }, + input, + // Observing a thread that does not exist yet would park its state as deleted. + launch: input.bootstrap?.createThread !== undefined, + state: "sending", + failedOnce: false, + cancelRequested: false, + resynced: false, + thread: null, + connection: null, + dispatchGeneration: 0, + settle, + stopObserving: [], + }; + const previous = entries.get(key); + if (previous !== undefined) finish(previous, { type: "cancelled" }); + entries.set(key, entry); + entry.stopObserving.push( + deps.observeConnection(environmentId, (state) => { + entry.connection = state; + evaluate(entry); + }), + ); + if (!entry.launch) { + entry.stopObserving.push( + deps.observeThread(environmentId, input.threadId, (state) => { + entry.thread = state; + if (state.status !== "live") entry.resynced = true; + evaluate(entry); + }), + ); + } + const end = await dispatch(entry); + if (end === null) return { type: "waiting", settled }; + finish(entry, end); + return end; }; + + /** Stops local delivery only; a command the server already committed stays committed. */ + const cancel = (key: string) => { + const entry = entries.get(key); + if (entry === undefined) return; + if (entry.state === "waiting") { + finish(entry, { type: "cancelled" }); + return; + } + entry.cancelRequested = true; + }; + return { deliver, - cancel: (_key: string) => {}, + cancel, getSnapshot: (): ReadonlyArray> => snapshot, - subscribe: (_listener: () => void) => () => {}, + subscribe: (listener: () => void) => { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }, }; } - -export type TurnDeliveries = ReturnType>; From c075684baf6549920c522da8473edf2328a62a60 Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Sun, 25 Oct 2026 03:33:34 +0000 Subject: [PATCH 03/11] fix(web): keep unsent and waiting messages intact across cancels, switches and launches - Cancel or refusal puts the message back after any newer draft instead of dropping it. - A replayed send's row stays until the thread shows the message, so it never vanishes after a thread switch. - A launch checks the resynced thread list first and reconciles a thread that already exists rather than replaying it, so an expired upload can't refuse a committed launch. - The waiting row owns its own image previews, freed when the delivery ends. - Plan follow-ups send through the same delivery path. Co-Authored-By: Claude Opus 5.5 --- apps/web/src/components/ChatView.logic.ts | 38 +--- .../chat/TimelineUserMessageRow.tsx | 2 +- .../src/components/chat/chatViewSend.test.ts | 43 ++++ apps/web/src/components/chat/chatViewSend.ts | 52 +++-- .../src/components/chat/useComposerSend.ts | 133 ++++++------ .../components/chat/useProposedPlanSend.ts | 85 ++++---- apps/web/src/state/turnDeliveries.ts | 16 +- docs/user/composer.md | 4 +- .../src/state/threadCommands.test.ts | 22 +- .../src/state/turnDelivery.test.ts | 110 +++++++--- .../client-runtime/src/state/turnDelivery.ts | 192 +++++++++++------- 11 files changed, 427 insertions(+), 270 deletions(-) create mode 100644 apps/web/src/components/chat/chatViewSend.test.ts diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index bef8f7ab405f1..f2e31cd996894 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -93,7 +93,7 @@ import { } from "../lib/composerContextRecords"; import type { ComposerDispatchMode } from "@t3tools/client-runtime/state/composer-dispatch"; import type { DraftThreadEnvMode } from "../composerDraftStore"; -import { collapseExpandedComposerCursor, type ComposerSubmissionIntent } from "../composer-logic"; +import type { ComposerSubmissionIntent } from "../composer-logic"; import type { ReviewCommentContext } from "../reviewCommentContext"; import { derivePhase, type TimelineEntry } from "../session-logic"; import type { PreviewMiniPlayerSource } from "../previewMiniPlayerStore"; @@ -1166,42 +1166,6 @@ export function shouldRefocusComposerOnWindowFocus( ); } -export interface PlanFollowUpComposerSnapshot { - readonly prompt: string; - readonly terminalContexts: ReadonlyArray; - readonly reviewComments: ReadonlyArray; - readonly previewAnnotations: ReadonlyArray; - readonly threadContexts: ReadonlyArray; - readonly environmentContexts: ReadonlyArray; -} - -export function restorePlanFollowUpComposer(input: { - readonly snapshot: PlanFollowUpComposerSnapshot; - readonly writePrompt: (prompt: string) => void; - readonly writeTerminalContexts: (contexts: ReadonlyArray) => void; - readonly writeReviewComments: (comments: ReadonlyArray) => void; - readonly writePreviewAnnotations: (annotations: ReadonlyArray) => void; - readonly writeThreadContexts: (records: ReadonlyArray) => void; - readonly writeEnvironmentContexts: (records: ReadonlyArray) => void; - readonly resetCursor: (options: { - cursor: number; - prompt: string; - detectTrigger: boolean; - }) => void; -}): void { - input.writePrompt(input.snapshot.prompt); - input.writeTerminalContexts(input.snapshot.terminalContexts); - input.writeReviewComments(input.snapshot.reviewComments); - input.writePreviewAnnotations(input.snapshot.previewAnnotations); - input.writeThreadContexts(input.snapshot.threadContexts); - input.writeEnvironmentContexts(input.snapshot.environmentContexts); - input.resetCursor({ - cursor: collapseExpandedComposerCursor(input.snapshot.prompt, input.snapshot.prompt.length), - prompt: input.snapshot.prompt, - detectTrigger: true, - }); -} - /** Includes the annotation screenshot only within the attachment cap; exceeding it rejects the whole turn. */ export function mergeDirectPreviewAnnotation(input: { readonly images: ReadonlyArray; diff --git a/apps/web/src/components/chat/TimelineUserMessageRow.tsx b/apps/web/src/components/chat/TimelineUserMessageRow.tsx index dcc7522ac3eea..de55ffe064d29 100644 --- a/apps/web/src/components/chat/TimelineUserMessageRow.tsx +++ b/apps/web/src/components/chat/TimelineUserMessageRow.tsx @@ -296,7 +296,7 @@ function UserMessageProjectedStatus({ row }: { row: Extract { + useComposerDraftStore.setState({ draftsByThreadKey: {}, draftThreadsByThreadKey: {} }); + const store = useComposerDraftStore.getState(); + store.setPrompt(target, "Newer draft"); + store.addImages(target, [image("newer")]); + const unsent = snapshotComposerSend({ + prompt: "Unsent message", + images: [image("unsent")], + files: [], + terminalContexts: [], + sendableTerminalContexts: [], + previewAnnotations: [], + reviewComments: [], + threadContexts: [], + }); + + restoreComposerDraft(target, unsent); + + const draft = store.getComposerDraft(target); + expect(draft?.prompt).toBe("Newer draft\n\nUnsent message"); + expect(draft?.images.map((attachment) => attachment.id)).toEqual(["newer", "unsent"]); +}); diff --git a/apps/web/src/components/chat/chatViewSend.ts b/apps/web/src/components/chat/chatViewSend.ts index 81a00823a98be..3df000650cf62 100644 --- a/apps/web/src/components/chat/chatViewSend.ts +++ b/apps/web/src/components/chat/chatViewSend.ts @@ -26,6 +26,7 @@ import { newThreadId } from "~/lib/utils"; import { type ComposerFileAttachment, type ComposerImageAttachment, + composerDraftHasUserContent, useComposerDraftStore, DraftId, } from "../../composerDraftStore"; @@ -252,25 +253,48 @@ export async function rewindToMessage(input: { } /** - * Writes a failed send's composer contents back into its draft. Returns the - * restored images, which are fresh copies so their preview URLs outlive the - * optimistic row that held the originals. + * Writes an unsent message's composer contents back into its draft, after any newer draft + * started meanwhile. Images are fresh copies, so their previews outlive the optimistic row. */ export function restoreComposerDraft( draftTarget: ScopedThreadRef | DraftId, snapshot: ComposerSendSnapshot, -): ComposerImageAttachment[] { +): void { const store = useComposerDraftStore.getState(); - const images = snapshot.images.map(cloneComposerImageForRetry); - store.setPrompt(draftTarget, snapshot.messageText); - store.addImages(draftTarget, images); - store.addFiles(draftTarget, snapshot.files); - store.setTerminalContexts(draftTarget, snapshot.terminalContexts); - store.setPreviewAnnotations(draftTarget, snapshot.previewAnnotations); - store.setReviewComments(draftTarget, snapshot.reviewComments); - store.setThreadContexts(draftTarget, snapshot.threadContexts); - store.setEnvironmentContexts(draftTarget, snapshot.environmentContexts); - return images; + const current = store.getComposerDraft(draftTarget); + const newer = composerDraftHasUserContent(current) ? current : null; + store.setPrompt( + draftTarget, + newer === null + ? snapshot.messageText + : snapshot.messageText.length === 0 + ? newer.prompt + : `${newer.prompt.trimEnd()}\n\n${snapshot.messageText}`, + ); + store.addImages(draftTarget, snapshot.images.map(cloneComposerImageForRetry), { + allowDuplicates: true, + }); + store.addFiles(draftTarget, snapshot.files, { allowDuplicates: true }); + store.setTerminalContexts(draftTarget, [ + ...(newer?.terminalContexts ?? []), + ...snapshot.terminalContexts, + ]); + store.setPreviewAnnotations(draftTarget, [ + ...(newer?.previewAnnotations ?? []), + ...snapshot.previewAnnotations, + ]); + store.setReviewComments(draftTarget, [ + ...(newer?.reviewComments ?? []), + ...snapshot.reviewComments, + ]); + store.setThreadContexts(draftTarget, [ + ...(newer?.threadContexts ?? []), + ...snapshot.threadContexts, + ]); + store.setEnvironmentContexts(draftTarget, [ + ...(newer?.environmentContexts ?? []), + ...snapshot.environmentContexts, + ]); } export interface MultipleModelTarget { diff --git a/apps/web/src/components/chat/useComposerSend.ts b/apps/web/src/components/chat/useComposerSend.ts index 24c0231c7ebda..dfa2c9591ea44 100644 --- a/apps/web/src/components/chat/useComposerSend.ts +++ b/apps/web/src/components/chat/useComposerSend.ts @@ -2,17 +2,18 @@ import { mergeDirectPreviewAnnotation, resolveBackgroundDraftWorkspaceOptions, shouldDockDraftHeroForSubmission, - restorePlanFollowUpComposer, buildExpiredTerminalContextToastCopy, deriveComposerSendState, boundAttachmentIds, formatOutgoingPrompt, optimisticMessageAttachments, + cloneComposerImageForRetry, planComposerTurn, sendTitleFallbackLabels, snapshotComposerSend, withMessageContext, makeThreadErrorSetter, + type ComposerSendSnapshot, } from "../ChatView.logic"; import { isUsageLimitsCommand } from "@t3tools/shared/usageLimits"; import { @@ -339,17 +340,6 @@ export function useComposerSend(input: ComposerSendInput) { reportFailure: false, }); const setComposerDraftPrompt = useComposerDraftStore((store) => store.setPrompt); - const setComposerDraftTerminalContexts = useComposerDraftStore( - (store) => store.setTerminalContexts, - ); - const setComposerDraftPreviewAnnotations = useComposerDraftStore( - (store) => store.setPreviewAnnotations, - ); - const setComposerDraftReviewComments = useComposerDraftStore((store) => store.setReviewComments); - const setComposerDraftThreadContexts = useComposerDraftStore((store) => store.setThreadContexts); - const setComposerDraftEnvironmentContexts = useComposerDraftStore( - (store) => store.setEnvironmentContexts, - ); const clearComposerDraftContent = useComposerDraftStore((store) => store.clearComposerContent); const sendToMultipleModels = useMultipleModelSend({ environmentId, @@ -394,6 +384,27 @@ export function useComposerSend(input: ComposerSendInput) { respondingRequestIdsRef.current = respondingRequestIds; }, [pendingApprovals, respondingRequestIds]); + /** Puts an unsent message back in its composer after any newer draft, so neither is lost. */ + const restoreUnsentComposer = ( + draftTarget: typeof composerDraftTarget, + sentRouteThreadKey: string, + snapshot: ComposerSendSnapshot, + ) => { + restoreComposerDraft(draftTarget, snapshot); + const restored = useComposerDraftStore.getState().getComposerDraft(draftTarget); + if (restored && currentRouteThreadKeyRef.current === sentRouteThreadKey) { + promptRef.current = restored.prompt; + composerImagesRef.current = restored.images; + composerFilesRef.current = restored.files; + composerTerminalContextsRef.current = restored.terminalContexts; + composerRef.current?.resetCursorState({ + cursor: collapseExpandedComposerCursor(restored.prompt, restored.prompt.length), + prompt: restored.prompt, + detectTrigger: true, + }); + } + }; + const onSend = async ( e?: { preventDefault: () => void }, dispatchMode: ComposerDispatchMode = "auto", @@ -680,12 +691,19 @@ export function useComposerSend(input: ComposerSendInput) { if (composerRef.current?.validateProviderInput(outgoingFollowUpText) === false) { return; } - const followUpPromptSnapshot = promptRef.current; - const followUpTerminalContexts = [...sendableComposerTerminalContexts]; - const followUpReviewComments = [...composerReviewComments]; - const followUpPreviewAnnotations = [...composerPreviewAnnotations]; - const followUpThreadContexts = [...composerThreadContexts]; - const followUpEnvironmentContexts = [...composerEnvironmentContexts]; + const followUpSnapshot: ComposerSendSnapshot = { + images: [], + files: [], + attachments: [], + terminalContexts: [...sendableComposerTerminalContexts], + previewAnnotations: [...composerPreviewAnnotations], + reviewComments: [...composerReviewComments], + threadContexts: [...composerThreadContexts], + environmentContexts: [...composerEnvironmentContexts], + messageText: promptRef.current, + }; + const restoreFollowUp = () => + restoreUnsentComposer(composerDraftTarget, routeThreadKey, followUpSnapshot); promptRef.current = ""; clearComposerDraftContent(composerDraftTarget); composerRef.current?.resetCursorState(); @@ -699,32 +717,10 @@ export function useComposerSend(input: ComposerSendInput) { environmentContexts: composerEnvironmentContexts, }), interactionMode: followUp.interactionMode, + restoreUnsent: restoreFollowUp, }); if (!followUpSent) { - promptRef.current = followUpPromptSnapshot; - composerTerminalContextsRef.current = [...followUpTerminalContexts]; - restorePlanFollowUpComposer({ - snapshot: { - prompt: followUpPromptSnapshot, - terminalContexts: followUpTerminalContexts, - reviewComments: followUpReviewComments, - previewAnnotations: followUpPreviewAnnotations, - threadContexts: followUpThreadContexts, - environmentContexts: followUpEnvironmentContexts, - }, - writePrompt: (prompt) => setComposerDraftPrompt(composerDraftTarget, prompt), - writeTerminalContexts: (contexts) => - setComposerDraftTerminalContexts(composerDraftTarget, [...contexts]), - writeReviewComments: (comments) => - setComposerDraftReviewComments(composerDraftTarget, [...comments]), - writePreviewAnnotations: (annotations) => - setComposerDraftPreviewAnnotations(composerDraftTarget, [...annotations]), - writeThreadContexts: (records) => - setComposerDraftThreadContexts(composerDraftTarget, [...records]), - writeEnvironmentContexts: (records) => - setComposerDraftEnvironmentContexts(composerDraftTarget, [...records]), - resetCursor: (options) => composerRef.current?.resetCursorState(options), - }); + restoreFollowUp(); } else if ( submissionIntent === "background" && currentRouteThreadKeyRef.current === routeThreadKey @@ -1274,37 +1270,15 @@ export function useComposerSend(input: ComposerSendInput) { ); clearBackgroundDraftSubmissionByRef(scopeThreadRef(environmentId, threadIdForSend)); } - if ( - !composerDraftHasUserContent( - useComposerDraftStore.getState().getComposerDraft(composerDraftTarget), - ) - ) { - setOptimisticUserMessages((existing) => { - const removed = existing.filter((message) => message.id === messageIdForSend); - for (const message of removed) { - revokeUserMessagePreviewUrls(message); - } - const next = existing.filter((message) => message.id !== messageIdForSend); - return next.length === existing.length ? existing : next; - }); - const retryComposerImages = restoreComposerDraft(composerDraftTarget, sendSnapshot); - if (currentRouteThreadKeyRef.current === routeThreadKey) { - promptRef.current = messageTextForSend; - composerImagesRef.current = retryComposerImages; - composerFilesRef.current = composerFilesSnapshot; - composerTerminalContextsRef.current = composerTerminalContextsSnapshot; - composerRef.current?.resetCursorState({ - cursor: collapseExpandedComposerCursor(messageTextForSend, messageTextForSend.length), - prompt: messageTextForSend, - detectTrigger: true, - }); + setOptimisticUserMessages((existing) => { + const removed = existing.filter((message) => message.id === messageIdForSend); + for (const message of removed) { + revokeUserMessagePreviewUrls(message); } - } else { - setOptimisticUserMessages((existing) => { - const next = existing.filter((message) => message.id !== messageIdForSend); - return next.length === existing.length ? existing : next; - }); - } + const next = existing.filter((message) => message.id !== messageIdForSend); + return next.length === existing.length ? existing : next; + }); + restoreUnsentComposer(composerDraftTarget, routeThreadKey, sendSnapshot); if (failure !== null && !isAtomCommandInterrupted(failure)) { const error = squashAtomCommandFailure(failure); setThreadError( @@ -1394,7 +1368,20 @@ export function useComposerSend(input: ComposerSendInput) { ...(bootstrap ? { bootstrap } : {}), createdAt: messageCreatedAt, }, - optimisticMessage, + // Its own previews, since this view revokes the row's when it switches threads. + () => + optimisticAttachments.length === 0 + ? optimisticMessage + : { + ...optimisticMessage, + attachments: optimisticMessageAttachments( + composerAttachmentsSnapshot.map((attachment) => + attachment.type === "image" + ? cloneComposerImageForRetry(attachment) + : attachment, + ), + ), + }, ); if (backgroundThreadRef) { try { diff --git a/apps/web/src/components/chat/useProposedPlanSend.ts b/apps/web/src/components/chat/useProposedPlanSend.ts index eb945e3e056a4..48bfc7fe23765 100644 --- a/apps/web/src/components/chat/useProposedPlanSend.ts +++ b/apps/web/src/components/chat/useProposedPlanSend.ts @@ -1,5 +1,6 @@ import { AuthOrchestrationOperateScope, + CommandId, type EnvironmentId, type ModelSelection, type ProviderInteractionMode, @@ -20,7 +21,7 @@ import { scopedThreadKey } from "@t3tools/client-runtime/environment"; import { truncate } from "@t3tools/shared/String"; import { useNavigate } from "@tanstack/react-router"; import type { Dispatch, SetStateAction } from "react"; -import { newMessageId, newThreadId } from "~/lib/utils"; +import { newMessageId, newThreadId, randomUUID } from "~/lib/utils"; import { useComposerDraftStore } from "../../composerDraftStore"; import type { ComposerHandleRef } from "../../composerHandleContext"; import type { buildMessageContext } from "../../lib/composerContextRecords"; @@ -31,6 +32,8 @@ import { import type { LatestProposedPlanState } from "../../session-logic"; import { readEnvironmentScope } from "../../state/session"; import { threadEnvironment } from "../../state/threads"; +import { turnDeliveries } from "../../state/turnDeliveries"; +import type { TurnDeliveryFailure } from "@t3tools/client-runtime/state/turn-delivery"; import { useOrchestrationCommand } from "../../state/use-orchestration-command"; import type { ChatMessage } from "../../types"; import { stackedThreadToast, toastManager } from "../ui/toast"; @@ -75,7 +78,8 @@ interface PlanFollowUpSendInput extends PlanSendBase { /** * Answers a proposed plan on this thread: refine it in plan mode, or implement it here - * (which links the turn to the plan). Resolves whether the turn started. + * (which links the turn to the plan). Resolves whether the turn started or is waiting to send; + * `restoreUnsent` runs if a waiting one is later refused or cancelled. */ export function usePlanFollowUpSend(input: PlanFollowUpSendInput) { const { @@ -95,9 +99,6 @@ export function usePlanFollowUpSend(input: PlanFollowUpSendInput) { anchorNewTurn, persistThreadSettingsForNextTurn, } = input; - const startThreadTurn = useOrchestrationCommand(threadEnvironment.startTurn, { - reportFailure: false, - }); const setComposerDraftInteractionMode = useComposerDraftStore( (store) => store.setInteractionMode, ); @@ -107,10 +108,12 @@ export function usePlanFollowUpSend(input: PlanFollowUpSendInput) { text, context, interactionMode: nextInteractionMode, + restoreUnsent, }: { text: string; context?: ReturnType; interactionMode: "default" | "plan"; + restoreUnsent: () => void; }) { if ( !readEnvironmentScope(environmentId, AuthOrchestrationOperateScope) || @@ -152,19 +155,29 @@ export function usePlanFollowUpSend(input: PlanFollowUpSendInput) { messageIdForSend, ); - setOptimisticUserMessages((existing) => [ - ...existing, - { - id: messageIdForSend, - role: "user", - text: outgoingMessageText, - ...(context ? { context } : {}), - runId: null, - createdAt: messageCreatedAt, - updatedAt: messageCreatedAt, - streaming: false, - }, - ]); + const optimisticMessage: ChatMessage = { + id: messageIdForSend, + role: "user", + text: outgoingMessageText, + ...(context ? { context } : {}), + runId: null, + createdAt: messageCreatedAt, + updatedAt: messageCreatedAt, + streaming: false, + }; + setOptimisticUserMessages((existing) => [...existing, optimisticMessage]); + const dropOptimisticMessage = () => + setOptimisticUserMessages((existing) => + existing.filter((message) => message.id !== messageIdForSend), + ); + const reportFailure = (failure: TurnDeliveryFailure) => { + if (isAtomCommandInterrupted(failure)) return; + const error = squashAtomCommandFailure(failure); + setThreadError( + threadIdForSend, + error instanceof Error ? error.message : "Failed to send plan follow-up.", + ); + }; const settingsResult = await persistThreadSettingsForNextTurn({ threadId: threadIdForSend, @@ -173,8 +186,9 @@ export function usePlanFollowUpSend(input: PlanFollowUpSendInput) { runtimeMode, interactionMode: nextInteractionMode, }); - let failure: AtomCommandResult | null = + let failure: TurnDeliveryFailure | null = settingsResult._tag === "Failure" ? settingsResult : null; + let cancelled = false; if (failure === null) { setComposerDraftInteractionMode( @@ -182,9 +196,10 @@ export function usePlanFollowUpSend(input: PlanFollowUpSendInput) { nextInteractionMode, ); - const startResult = await startThreadTurn({ + const outcome = await turnDeliveries.deliver( environmentId, - input: { + { + commandId: CommandId.make(randomUUID()), threadId: threadIdForSend, message: { messageId: messageIdForSend, @@ -207,26 +222,28 @@ export function usePlanFollowUpSend(input: PlanFollowUpSendInput) { : {}), createdAt: messageCreatedAt, }, - }); - failure = startResult._tag === "Failure" ? startResult : null; + () => optimisticMessage, + ); + if (outcome.type === "rejected") failure = outcome.failure; + if (outcome.type === "cancelled") cancelled = true; + if (outcome.type === "waiting") { + void outcome.settled.then((end) => { + if (end.type === "delivered") return; + dropOptimisticMessage(); + if (end.type === "rejected") reportFailure(end.failure); + restoreUnsent(); + }); + } } - if (failure === null) { + if (failure === null && !cancelled) { sendInFlightRef.current = false; resetLocalDispatch(); return true; } - setOptimisticUserMessages((existing) => - existing.filter((message) => message.id !== messageIdForSend), - ); - if (!isAtomCommandInterrupted(failure)) { - const error = squashAtomCommandFailure(failure); - setThreadError( - threadIdForSend, - error instanceof Error ? error.message : "Failed to send plan follow-up.", - ); - } + dropOptimisticMessage(); + if (failure !== null) reportFailure(failure); sendInFlightRef.current = false; resetLocalDispatch(); return false; diff --git a/apps/web/src/state/turnDeliveries.ts b/apps/web/src/state/turnDeliveries.ts index 217741d47c4d7..5ebd01915f600 100644 --- a/apps/web/src/state/turnDeliveries.ts +++ b/apps/web/src/state/turnDeliveries.ts @@ -1,4 +1,3 @@ -import { AVAILABLE_CONNECTION_STATE } from "@t3tools/client-runtime/connection"; import { EMPTY_ENVIRONMENT_THREAD_STATE } from "@t3tools/client-runtime/state/threads"; import { createTurnDeliveries, @@ -9,9 +8,10 @@ import * as Option from "effect/Option"; import { AsyncResult } from "effect/reactivity"; import { useSyncExternalStore } from "react"; -import { environmentCatalog } from "../connection/catalog"; +import { revokeUserMessagePreviewUrls } from "../components/ChatView.logic"; import { appAtomRegistry } from "../rpc/atomRegistry"; import type { ChatMessage } from "../types"; +import { environmentShell } from "./shell"; import { environmentThreadStateAtom, threadEnvironment } from "./threads"; import { runOrchestrationCommand } from "./use-orchestration-command"; @@ -30,13 +30,11 @@ export const turnDeliveries = createTurnDeliveries EMPTY_ENVIRONMENT_THREAD_STATE)), { immediate: true }, ), - observeConnection: (environmentId, listener) => - appAtomRegistry.subscribe( - environmentCatalog.stateAtom(environmentId), - (result) => - listener(Option.getOrElse(AsyncResult.value(result), () => AVAILABLE_CONNECTION_STATE)), - { immediate: true }, - ), + observeShell: (environmentId, listener) => + appAtomRegistry.subscribe(environmentShell.stateValueAtom(environmentId), listener, { + immediate: true, + }), + releaseDisplay: revokeUserMessagePreviewUrls, }); const EMPTY: ReadonlyArray> = []; diff --git a/docs/user/composer.md b/docs/user/composer.md index eedee4fba2212..24e8541e8a66b 100644 --- a/docs/user/composer.md +++ b/docs/user/composer.md @@ -87,8 +87,8 @@ device until you sign back into the same account. On web and desktop, a message sent as the connection drops shows **Waiting to send**. When the thread reconnects, T3 Code sends it once, or keeps it if the server already received it. Cancel -stops the send and restores the message to an empty composer. It can't unsend a message the -server already received. +stops the send and puts the message back in the composer, after anything you typed since. It +can't unsend a message the server already received. ## Custom models diff --git a/packages/client-runtime/src/state/threadCommands.test.ts b/packages/client-runtime/src/state/threadCommands.test.ts index 8ee50e280fc90..e215043673140 100644 --- a/packages/client-runtime/src/state/threadCommands.test.ts +++ b/packages/client-runtime/src/state/threadCommands.test.ts @@ -25,10 +25,10 @@ import * as Scope from "effect/Scope"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { Atom, AtomRegistry } from "effect/reactivity"; -import { AVAILABLE_CONNECTION_STATE } from "../connection/model.ts"; import * as EnvironmentRegistry from "../connection/registry.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import type { RpcSession } from "../rpc/session.ts"; +import { EMPTY_SHELL_STATE } from "./shell.ts"; import { createThreadEnvironmentAtoms } from "./threadCommands.ts"; import * as ThreadHistoryController from "./threadHistoryController.ts"; import { EMPTY_ENVIRONMENT_THREAD_STATE, type EnvironmentThreadState } from "./threadState.ts"; @@ -530,10 +530,11 @@ describe("turn delivery across a lost acknowledgement", () => { listener(threadState("live")); return () => threadListeners.delete(listener); }, - observeConnection: (_environmentId, listener) => { - listener({ ...AVAILABLE_CONNECTION_STATE, phase: "connected", generation: 1 }); + observeShell: (_environmentId, listener) => { + listener(EMPTY_SHELL_STATE); return () => {}; }, + releaseDisplay: () => {}, }); const emitThread = (status: EnvironmentThreadState["status"]) => { for (const listener of threadListeners) listener(threadState(status)); @@ -554,7 +555,7 @@ describe("turn delivery across a lost acknowledgement", () => { interactionMode: "default", dispatchMode: "start", }, - "optimistic row", + () => "optimistic row", ); const first = yield* Queue.take(h.requests); expect(first.command).toMatchObject({ @@ -590,6 +591,19 @@ describe("turn delivery across a lost acknowledgement", () => { type: "delivered", value: { sequence: 3 }, }); + expect(deliveries.getSnapshot()).toEqual([ + expect.objectContaining({ messageId: "send-message", state: "delivered" }), + ]); + for (const listener of threadListeners) { + listener({ + ...threadState("live"), + data: Option.some({ + messages: [{ id: "send-message", role: "user" }], + runs: [], + turnItems: [], + } as never), + }); + } expect(deliveries.getSnapshot()).toEqual([]); }), ); diff --git a/packages/client-runtime/src/state/turnDelivery.test.ts b/packages/client-runtime/src/state/turnDelivery.test.ts index 0c9aea8729fe7..460501ca25f87 100644 --- a/packages/client-runtime/src/state/turnDelivery.test.ts +++ b/packages/client-runtime/src/state/turnDelivery.test.ts @@ -5,6 +5,7 @@ import { ProjectId, ProviderInstanceId, ThreadId, + type OrchestrationV2ShellSnapshot, type OrchestrationV2ThreadProjection, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; @@ -12,8 +13,8 @@ import * as Option from "effect/Option"; import { AsyncResult } from "effect/reactivity"; import { describe, expect, it } from "vite-plus/test"; -import { AVAILABLE_CONNECTION_STATE, type SupervisorConnectionState } from "../connection/model.ts"; import type { AtomCommandResult } from "./runtime.ts"; +import { EMPTY_SHELL_STATE, type EnvironmentShellState } from "./shell.ts"; import { EMPTY_ENVIRONMENT_THREAD_STATE, type EnvironmentThreadState } from "./threadState.ts"; import { createTurnDeliveries, type TurnDeliveryInput } from "./turnDelivery.ts"; @@ -62,13 +63,26 @@ const REJECTED: SendResult = AsyncResult.failure( ); const ACCEPTED: SendResult = AsyncResult.success({ sequence: 7 }); -function connection( - phase: SupervisorConnectionState["phase"], - generation: number, -): SupervisorConnectionState { - return { ...AVAILABLE_CONNECTION_STATE, phase, generation }; +function shell( + status: EnvironmentShellState["status"], + threadIds: ReadonlyArray = [], +): EnvironmentShellState { + return { + ...EMPTY_SHELL_STATE, + status, + snapshot: Option.some({ + threads: threadIds.map((id) => ({ id })), + archivedThreads: [], + } as unknown as OrchestrationV2ShellSnapshot), + }; } +const HELD = thread("live", { + messages: [ + { id: MESSAGE_ID, role: "user" } as OrchestrationV2ThreadProjection["messages"][number], + ], +}); + function thread( status: EnvironmentThreadState["status"], projection: Partial> = {}, @@ -85,12 +99,17 @@ function thread( }; } -function setup(results: ReadonlyArray>) { +function setup( + results: ReadonlyArray>, + initialThread = thread("live"), +) { const queue = [...results]; const sent: Array = []; let threadListener: ((state: EnvironmentThreadState) => void) | null = null; - let connectionListener: ((state: SupervisorConnectionState) => void) | null = null; + let shellListener: ((state: EnvironmentShellState) => void) | null = null; let threadObservations = 0; + let displaysMade = 0; + const released: Array = []; const deliveries = createTurnDeliveries<{ sequence: number }, unknown, string>({ send: async (_environmentId, input) => { sent.push(input); @@ -99,30 +118,39 @@ function setup(results: ReadonlyArray>) { observeThread: (_environmentId, _threadId, listener) => { threadObservations += 1; threadListener = listener; - listener(thread("live")); + listener(initialThread); return () => { threadListener = null; }; }, - observeConnection: (_environmentId, listener) => { - connectionListener = listener; - listener(connection("connected", 1)); + observeShell: (_environmentId, listener) => { + shellListener = listener; + listener(shell("live")); return () => { - connectionListener = null; + shellListener = null; }; }, + releaseDisplay: (display) => { + released.push(display); + }, }); return { deliveries, sent, + released, threadObservations: () => threadObservations, + displaysMade: () => displaysMade, + makeDisplay: () => { + displaysMade += 1; + return "optimistic row"; + }, emitThread: (state: EnvironmentThreadState) => threadListener?.(state), - emitConnection: (state: SupervisorConnectionState) => connectionListener?.(state), + emitShell: (state: EnvironmentShellState) => shellListener?.(state), }; } async function deliverUntilWaiting(harness: ReturnType, input = INPUT) { - const outcome = await harness.deliveries.deliver(ENVIRONMENT_ID, input, "optimistic row"); + const outcome = await harness.deliveries.deliver(ENVIRONMENT_ID, input, harness.makeDisplay); expect(outcome.type).toBe("waiting"); if (outcome.type !== "waiting") throw new Error(`Expected waiting, got ${outcome.type}.`); // Wrapped: an async function returning the promise would wait for it to settle. @@ -164,7 +192,23 @@ describe("turn deliveries", () => { expect(await settled).toEqual({ type: "delivered", value: { sequence: 7 } }); expect(harness.sent).toEqual([INPUT, INPUT]); + }); + + it("keeps an acknowledged replay on screen until the thread shows the message", async () => { + const harness = setup([LOST_ACK, ACCEPTED]); + const { settled } = await deliverUntilWaiting(harness); + + harness.emitThread(thread("synchronizing")); + harness.emitThread(thread("live")); + await settled; + + expect(harness.deliveries.getSnapshot()).toEqual([ + expect.objectContaining({ state: "delivered", display: "optimistic row" }), + ]); + expect(harness.released).toEqual([]); + harness.emitThread(HELD); expect(harness.deliveries.getSnapshot()).toEqual([]); + expect(harness.released).toEqual(["optimistic row"]); }); it("settles without a resend when the resynced thread already holds the message", async () => { @@ -172,16 +216,11 @@ describe("turn deliveries", () => { const { settled } = await deliverUntilWaiting(harness); harness.emitThread(thread("synchronizing")); - harness.emitThread( - thread("live", { - messages: [ - { id: MESSAGE_ID, role: "user" } as OrchestrationV2ThreadProjection["messages"][number], - ], - }), - ); + harness.emitThread(HELD); expect(await settled).toEqual({ type: "delivered", value: null }); expect(harness.sent).toHaveLength(1); + expect(harness.deliveries.getSnapshot()).toEqual([]); }); it("settles without a resend when the resynced thread queued the message", async () => { @@ -216,6 +255,7 @@ describe("turn deliveries", () => { harness.emitThread(thread("live")); expect(harness.sent).toHaveLength(2); expect(harness.deliveries.getSnapshot()).toEqual([]); + expect(harness.released).toEqual(["optimistic row"]); }); it("stops replaying after a cancel", async () => { @@ -230,6 +270,7 @@ describe("turn deliveries", () => { expect(await settled).toEqual({ type: "cancelled" }); expect(harness.sent).toHaveLength(1); expect(harness.deliveries.getSnapshot()).toEqual([]); + expect(harness.released).toEqual(["optimistic row"]); }); it("keeps a replay the server accepted even when cancelled while it was in flight", async () => { @@ -252,23 +293,38 @@ describe("turn deliveries", () => { it("returns a rejection right away without holding the send", async () => { const harness = setup([REJECTED]); - const outcome = await harness.deliveries.deliver(ENVIRONMENT_ID, INPUT, "optimistic row"); + const outcome = await harness.deliveries.deliver(ENVIRONMENT_ID, INPUT, harness.makeDisplay); expect(outcome).toEqual({ type: "rejected", failure: REJECTED }); expect(harness.deliveries.getSnapshot()).toEqual([]); + expect(harness.displaysMade()).toBe(0); }); - it("replays a first message after reconnecting without watching the unborn thread", async () => { + it("replays a first message once the resynced shell lacks its thread, without watching it", async () => { const harness = setup([LOST_ACK, ACCEPTED]); const { settled } = await deliverUntilWaiting(harness, LAUNCH); - harness.emitConnection(connection("connected", 1)); + harness.emitShell(shell("live")); expect(harness.sent).toHaveLength(1); - harness.emitConnection(connection("connecting", 2)); - harness.emitConnection(connection("connected", 2)); + harness.emitShell(shell("synchronizing")); + harness.emitShell(shell("live")); expect(await settled).toEqual({ type: "delivered", value: { sequence: 7 } }); expect(harness.sent).toEqual([LAUNCH, LAUNCH]); + expect(harness.threadObservations()).toBe(1); + }); + + it("settles a launch the resynced shell already holds without replaying it", async () => { + const harness = setup([LOST_ACK], EMPTY_ENVIRONMENT_THREAD_STATE); + const { settled } = await deliverUntilWaiting(harness, LAUNCH); expect(harness.threadObservations()).toBe(0); + + harness.emitShell(shell("synchronizing")); + harness.emitShell(shell("live", [THREAD_ID])); + harness.emitThread(HELD); + + expect(await settled).toEqual({ type: "delivered", value: null }); + expect(harness.sent).toEqual([LAUNCH]); + expect(harness.deliveries.getSnapshot()).toEqual([]); }); }); diff --git a/packages/client-runtime/src/state/turnDelivery.ts b/packages/client-runtime/src/state/turnDelivery.ts index e806dbf889355..f800289c38e09 100644 --- a/packages/client-runtime/src/state/turnDelivery.ts +++ b/packages/client-runtime/src/state/turnDelivery.ts @@ -3,7 +3,6 @@ import * as Cause from "effect/Cause"; import * as Option from "effect/Option"; import { AsyncResult } from "effect/reactivity"; -import type { SupervisorConnectionState } from "../connection/model.ts"; import { isRetryableDeliveryFailure } from "../errors/transport.ts"; import type { StartThreadTurnInput } from "../operations/commands.ts"; import { @@ -11,6 +10,7 @@ import { isAtomCommandInterrupted, squashAtomCommandFailure, } from "./runtime.ts"; +import type { EnvironmentShellState } from "./shell.ts"; import type { EnvironmentThreadState } from "./threadState.ts"; export type TurnDeliveryInput = StartThreadTurnInput & { readonly commandId: CommandId }; @@ -33,11 +33,12 @@ export interface PendingTurnDelivery { readonly environmentId: EnvironmentId; readonly threadId: ThreadId; readonly messageId: MessageId; - readonly state: "waiting" | "retrying"; + /** `delivered` rows stay until the thread shows the message, since they may be its only copy. */ + readonly state: "waiting" | "retrying" | "delivered"; readonly display: M; } -interface TurnDeliveryDependencies { +interface TurnDeliveryDependencies { readonly send: ( environmentId: EnvironmentId, input: TurnDeliveryInput, @@ -49,24 +50,29 @@ interface TurnDeliveryDependencies { listener: (state: EnvironmentThreadState) => void, ) => () => void; /** Must call the listener with the current state before returning. */ - readonly observeConnection: ( + readonly observeShell: ( environmentId: EnvironmentId, - listener: (state: SupervisorConnectionState) => void, + listener: (state: EnvironmentShellState) => void, ) => () => void; + /** Frees a display made by `deliver`'s factory once no pending row shows it. */ + readonly releaseDisplay: (display: M) => void; } interface Entry { - readonly pending: Omit, "state">; + readonly key: string; + readonly environmentId: EnvironmentId; readonly input: TurnDeliveryInput; - readonly launch: boolean; - state: "sending" | "waiting"; - failedOnce: boolean; + readonly createDisplay: () => M; + /** Made on the first lost acknowledgement, so a send that lands at once owns nothing. */ + display: M | null; + /** Observing a thread that does not exist yet would park its state as deleted. */ + unborn: boolean; + state: "sending" | "waiting" | "delivered"; cancelRequested: boolean; - /** The thread left live sync since the last dispatch. */ + /** The thread, or for an unborn one the shell, left live sync since the last dispatch. */ resynced: boolean; thread: EnvironmentThreadState | null; - connection: SupervisorConnectionState | null; - dispatchGeneration: number; + shell: EnvironmentShellState | null; settle: (end: TurnDeliveryEnd) => void; stopObserving: Array<() => void>; } @@ -91,59 +97,115 @@ function threadHoldsUserMessage(state: EnvironmentThreadState, messageId: Messag }); } +function shellHoldsThread(state: EnvironmentShellState, threadId: ThreadId): boolean { + return Option.match(state.snapshot, { + onNone: () => false, + onSome: (snapshot) => + snapshot.threads.some((thread) => thread.id === threadId) || + snapshot.archivedThreads.some((thread) => thread.id === threadId), + }); +} + /** * Keeps a send whose acknowledgement was lost until the thread resyncs on a later session, then * settles it when the server already holds the message or replays the same command otherwise. * Waiting happens here, outside the thread's serial command lane, so Stop is never held behind it. */ -export function createTurnDeliveries(deps: TurnDeliveryDependencies) { +export function createTurnDeliveries(deps: TurnDeliveryDependencies) { const entries = new Map>(); const listeners = new Set<() => void>(); let snapshot: ReadonlyArray> = []; const publish = () => { snapshot = [...entries.values()].flatMap((entry) => - entry.failedOnce - ? [ + entry.display === null + ? [] + : [ { - ...entry.pending, - state: entry.state === "waiting" ? ("waiting" as const) : ("retrying" as const), + key: entry.key, + environmentId: entry.environmentId, + threadId: entry.input.threadId, + messageId: entry.input.message.messageId, + state: entry.state === "sending" ? ("retrying" as const) : entry.state, + display: entry.display, }, - ] - : [], + ], ); for (const listener of listeners) listener(); }; - const finish = (entry: Entry, end: TurnDeliveryEnd) => { - if (entries.get(entry.pending.key) !== entry) return; - entries.delete(entry.pending.key); + const remove = (entry: Entry) => { + if (entries.get(entry.key) !== entry) return; + entries.delete(entry.key); for (const stop of entry.stopObserving) stop(); - if (entry.failedOnce) publish(); - entry.settle(end); + entry.stopObserving = []; + if (entry.display === null) return; + publish(); + deps.releaseDisplay(entry.display); }; - // A launch's thread may not exist yet, and its command receipt makes a replay safe. - const readyToReplay = (entry: Entry) => - entry.launch - ? entry.connection?.phase === "connected" && - entry.connection.generation > entry.dispatchGeneration - : entry.resynced && entry.thread?.status === "live"; + const observeThread = (entry: Entry) => { + entry.stopObserving.push( + deps.observeThread(entry.environmentId, entry.input.threadId, (state) => { + entry.thread = state; + if (state.status !== "live") entry.resynced = true; + evaluate(entry); + }), + ); + }; - const evaluate = (entry: Entry) => { - if (entry.state !== "waiting") return; - if (!entry.launch && entry.thread?.status === "deleted") { - finish(entry, { - type: "rejected", - failure: AsyncResult.failure(Cause.fail(new Error("The thread was deleted."))), - }); + const finish = (entry: Entry, end: TurnDeliveryEnd) => { + if (entries.get(entry.key) !== entry) return; + entry.settle(end); + if (end.type !== "delivered" || entry.display === null) { + remove(entry); + return; + } + entry.state = "delivered"; + publish(); + if (entry.unborn) { + // Acknowledged, so the thread exists now. + entry.unborn = false; + observeThread(entry); return; } - if (!readyToReplay(entry)) return; - if (!entry.launch && threadHoldsUserMessage(entry.thread!, entry.pending.messageId)) { - finish(entry, { type: "delivered", value: null }); + evaluate(entry); + }; + + const evaluate = (entry: Entry) => { + const messageId = entry.input.message.messageId; + if (entry.state === "delivered") { + if ( + entry.thread !== null && + (entry.thread.status === "deleted" || threadHoldsUserMessage(entry.thread, messageId)) + ) { + remove(entry); + } return; } + if (entry.state !== "waiting") return; + if (entry.unborn) { + if (!entry.resynced || entry.shell?.status !== "live") return; + if (shellHoldsThread(entry.shell, entry.input.threadId)) { + // The launch committed; its pending uploads may have expired, so reconcile, never replay. + entry.unborn = false; + observeThread(entry); + return; + } + } else { + if (entry.thread?.status === "deleted") { + finish(entry, { + type: "rejected", + failure: AsyncResult.failure(Cause.fail(new Error("The thread was deleted."))), + }); + return; + } + if (!entry.resynced || entry.thread?.status !== "live") return; + if (threadHoldsUserMessage(entry.thread, messageId)) { + finish(entry, { type: "delivered", value: null }); + return; + } + } void dispatch(entry).then((end) => { if (end !== null) finish(entry, end); }); @@ -152,9 +214,8 @@ export function createTurnDeliveries(deps: TurnDeliveryDependencies): Promise | null> => { entry.state = "sending"; entry.resynced = false; - entry.dispatchGeneration = entry.connection?.generation ?? 0; - if (entry.failedOnce) publish(); - const result = await deps.send(entry.pending.environmentId, entry.input); + if (entry.display !== null) publish(); + const result = await deps.send(entry.environmentId, entry.input); if (result._tag === "Success") return { type: "delivered", value: result.value }; const retryable = isAtomCommandInterrupted(result) || @@ -162,7 +223,7 @@ export function createTurnDeliveries(deps: TurnDeliveryDependencies(deps: TurnDeliveryDependencies M, ): Promise> => { const key = turnDeliveryKey(environmentId, input.threadId, input.message.messageId); let settle!: (end: TurnDeliveryEnd) => void; @@ -179,43 +240,36 @@ export function createTurnDeliveries(deps: TurnDeliveryDependencies = { - pending: { - key, - environmentId, - threadId: input.threadId, - messageId: input.message.messageId, - display, - }, + key, + environmentId, input, - // Observing a thread that does not exist yet would park its state as deleted. - launch: input.bootstrap?.createThread !== undefined, + createDisplay, + display: null, + unborn: input.bootstrap?.createThread !== undefined, state: "sending", - failedOnce: false, cancelRequested: false, resynced: false, thread: null, - connection: null, - dispatchGeneration: 0, + shell: null, settle, stopObserving: [], }; const previous = entries.get(key); - if (previous !== undefined) finish(previous, { type: "cancelled" }); + if (previous !== undefined) { + previous.settle({ type: "cancelled" }); + remove(previous); + } entries.set(key, entry); - entry.stopObserving.push( - deps.observeConnection(environmentId, (state) => { - entry.connection = state; - evaluate(entry); - }), - ); - if (!entry.launch) { + if (entry.unborn) { entry.stopObserving.push( - deps.observeThread(environmentId, input.threadId, (state) => { - entry.thread = state; - if (state.status !== "live") entry.resynced = true; + deps.observeShell(environmentId, (state) => { + entry.shell = state; + if (entry.unborn && state.status !== "live") entry.resynced = true; evaluate(entry); }), ); + } else { + observeThread(entry); } const end = await dispatch(entry); if (end === null) return { type: "waiting", settled }; @@ -226,7 +280,7 @@ export function createTurnDeliveries(deps: TurnDeliveryDependencies { const entry = entries.get(key); - if (entry === undefined) return; + if (entry === undefined || entry.state === "delivered") return; if (entry.state === "waiting") { finish(entry, { type: "cancelled" }); return; From e5622c86671bea7f6513763ea539018ca919e89f Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Sun, 25 Oct 2026 03:50:02 +0000 Subject: [PATCH 04/11] fix(web): keep an unsent message that no longer fits its composer, and stop watching settled launches Restoring a cancelled or refused send into a composer that already holds the attachment limit used to drop its attachments, then remove its row. The restore now refuses and changes nothing, and the message stays in its thread as Not sent, owned by a small store that outlives ChatView, until it is restored or discarded. A launch reconciled through an immediate thread callback could settle and remove its delivery before the unsubscribe was registered, leaking one observer. Each observer is now released at once when its entry is already gone. Co-Authored-By: Claude Opus 5.5 --- apps/web/src/components/ChatView.tsx | 18 ++-- .../chat/TimelineUserMessageRow.tsx | 26 ++++++ .../src/components/chat/chatViewSend.test.ts | 31 ++++++- apps/web/src/components/chat/chatViewSend.ts | 13 ++- .../src/components/chat/useComposerSend.ts | 71 +++++++++------ apps/web/src/state/unsentMessages.test.ts | 90 +++++++++++++++++++ apps/web/src/state/unsentMessages.ts | 80 +++++++++++++++++ docs/user/composer.md | 3 +- .../src/state/turnDelivery.test.ts | 18 ++++ .../client-runtime/src/state/turnDelivery.ts | 12 ++- 10 files changed, 325 insertions(+), 37 deletions(-) create mode 100644 apps/web/src/state/unsentMessages.test.ts create mode 100644 apps/web/src/state/unsentMessages.ts diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 64728ba26f531..b535cdd7bdbe4 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -306,6 +306,7 @@ import { useThreadLinkedPullRequest } from "./pullRequest/useThreadLinkedPullReq import { useProactivePanels } from "../hooks/useProactivePanels"; import { useIncrementalTimelineEntries } from "../hooks/useIncrementalTimelineEntries"; import { usePendingTurnDeliveries } from "../state/turnDeliveries"; +import { useUnsentMessages } from "../state/unsentMessages"; import { useRightPanelSurfaceClose } from "../hooks/useRightPanelSurfaceClose"; import { ComposerBannerStack, type ComposerBannerStackItem } from "./chat/ComposerBannerStack"; import { modelFallbackBannerItem } from "./chat/ComposerUsageLimits"; @@ -656,16 +657,21 @@ export default function ChatView(props: ChatViewProps) { ); // Sends waiting on a reconnect live outside this view, so they stay visible across thread switches. const pendingTurnDeliveries = usePendingTurnDeliveries(environmentId, threadId); + const unsentMessageRecords = useUnsentMessages(environmentId, threadId); + const detachedUserMessages = [ + ...pendingTurnDeliveries.map((delivery) => delivery.display), + ...unsentMessageRecords.map((record) => record.display), + ]; const displayedOptimisticMessages = - pendingTurnDeliveries.length === 0 + detachedUserMessages.length === 0 ? optimisticUserMessages : [ ...optimisticUserMessages, - ...pendingTurnDeliveries.flatMap((delivery) => - serverAcknowledgedUserMessageIds.has(delivery.messageId) || - optimisticUserMessages.some((message) => message.id === delivery.messageId) - ? [] - : [delivery.display], + ...detachedUserMessages.filter( + (display, index) => + !serverAcknowledgedUserMessageIds.has(display.id) && + !optimisticUserMessages.some((message) => message.id === display.id) && + detachedUserMessages.findIndex((other) => other.id === display.id) === index, ), ]; // Draft ids survive promotion and expire on navigation, keeping the pending resend scoped correctly. diff --git a/apps/web/src/components/chat/TimelineUserMessageRow.tsx b/apps/web/src/components/chat/TimelineUserMessageRow.tsx index de55ffe064d29..ad8b87f45fc2f 100644 --- a/apps/web/src/components/chat/TimelineUserMessageRow.tsx +++ b/apps/web/src/components/chat/TimelineUserMessageRow.tsx @@ -12,6 +12,7 @@ import { resolveUserMessagePresentation } from "@t3tools/client-runtime/user-mes import { Link } from "@tanstack/react-router"; import { useQuietThreadShell } from "../../state/entities"; import { turnDeliveries, usePendingTurnDelivery } from "../../state/turnDeliveries"; +import { unsentMessages, useUnsentMessage } from "../../state/unsentMessages"; import { useEnvironmentQuery } from "../../state/query"; import { serverEnvironment } from "../../state/server"; import { replaceComposerContextReferences } from "@t3tools/shared/composerContextReferences"; @@ -293,9 +294,34 @@ function UserMessageProjectedStatus({ row }: { row: Extract + {blocked ? "Not sent. Make room for its attachments in the composer first." : "Not sent"} + setBlocked(!unsentMessages.restore(environmentId, messageId))} + > + Restore to composer + + unsentMessages.discard(environmentId, messageId)}> + Discard + +
+ ); +} + function UserMessageDeliveryStatus({ messageId }: { messageId: MessageId }) { const ctx = use(TimelineRowCtx); const delivery = usePendingTurnDelivery(ctx.activeThreadEnvironmentId, messageId); + const unsent = useUnsentMessage(ctx.activeThreadEnvironmentId, messageId); + if (unsent !== null) return ; if (delivery === null || delivery.state === "delivered") return null; return (
expect(draft?.prompt).toBe("Newer draft\n\nUnsent message"); expect(draft?.images.map((attachment) => attachment.id)).toEqual(["newer", "unsent"]); }); + +it("leaves a full newer draft untouched when the unsent attachments do not fit", () => { + useComposerDraftStore.setState({ draftsByThreadKey: {}, draftThreadsByThreadKey: {} }); + const store = useComposerDraftStore.getState(); + store.setPrompt(target, "Newer draft"); + store.addImages( + target, + Array.from({ length: PROVIDER_SEND_TURN_MAX_ATTACHMENTS }, (_, index) => + image(`newer-${index}`), + ), + ); + const unsent = snapshotComposerSend({ + prompt: "Unsent message", + images: [image("unsent")], + files: [], + terminalContexts: [], + sendableTerminalContexts: [], + previewAnnotations: [], + reviewComments: [], + threadContexts: [], + }); + + expect(restoreComposerDraft(target, unsent)).toBe(false); + + const draft = store.getComposerDraft(target); + expect(draft?.prompt).toBe("Newer draft"); + expect(draft?.images).toHaveLength(PROVIDER_SEND_TURN_MAX_ATTACHMENTS); + expect(unsent.images.map((attachment) => attachment.id)).toEqual(["unsent"]); +}); diff --git a/apps/web/src/components/chat/chatViewSend.ts b/apps/web/src/components/chat/chatViewSend.ts index 3df000650cf62..19aa0ea2996df 100644 --- a/apps/web/src/components/chat/chatViewSend.ts +++ b/apps/web/src/components/chat/chatViewSend.ts @@ -255,14 +255,24 @@ export async function rewindToMessage(input: { /** * Writes an unsent message's composer contents back into its draft, after any newer draft * started meanwhile. Images are fresh copies, so their previews outlive the optimistic row. + * Returns false, changing nothing, when its attachments would not fit beside the newer draft. */ export function restoreComposerDraft( draftTarget: ScopedThreadRef | DraftId, snapshot: ComposerSendSnapshot, -): void { +): boolean { const store = useComposerDraftStore.getState(); const current = store.getComposerDraft(draftTarget); const newer = composerDraftHasUserContent(current) ? current : null; + if ( + (newer?.images.length ?? 0) + + (newer?.files.length ?? 0) + + snapshot.images.length + + snapshot.files.length > + PROVIDER_SEND_TURN_MAX_ATTACHMENTS + ) { + return false; + } store.setPrompt( draftTarget, newer === null @@ -295,6 +305,7 @@ export function restoreComposerDraft( ...(newer?.environmentContexts ?? []), ...snapshot.environmentContexts, ]); + return true; } export interface MultipleModelTarget { diff --git a/apps/web/src/components/chat/useComposerSend.ts b/apps/web/src/components/chat/useComposerSend.ts index dfa2c9591ea44..4e500c6b7a532 100644 --- a/apps/web/src/components/chat/useComposerSend.ts +++ b/apps/web/src/components/chat/useComposerSend.ts @@ -23,6 +23,7 @@ import { type EnvironmentId, type ModelSelection, type PreviewAnnotationPayload, + PROVIDER_SEND_TURN_MAX_ATTACHMENTS, type ThreadId, ProviderInteractionMode, RuntimeMode, @@ -86,6 +87,7 @@ import { import { environmentServerConfigsAtom } from "../../state/server"; import { threadEnvironment } from "../../state/threads"; import { turnDeliveries } from "../../state/turnDeliveries"; +import { unsentMessages } from "../../state/unsentMessages"; import type { TurnDeliveryFailure } from "@t3tools/client-runtime/state/turn-delivery"; import { runMobileComposerTransition } from "./draftHeroTransition"; import { useClientSettingsHydrated, useEnvironmentSettings } from "../../hooks/useSettings"; @@ -384,13 +386,16 @@ export function useComposerSend(input: ComposerSendInput) { respondingRequestIdsRef.current = respondingRequestIds; }, [pendingApprovals, respondingRequestIds]); - /** Puts an unsent message back in its composer after any newer draft, so neither is lost. */ + /** + * Puts an unsent message back in its composer after any newer draft, so neither is lost. + * Returns false, changing nothing, when its attachments do not fit there. + */ const restoreUnsentComposer = ( draftTarget: typeof composerDraftTarget, sentRouteThreadKey: string, snapshot: ComposerSendSnapshot, - ) => { - restoreComposerDraft(draftTarget, snapshot); + ): boolean => { + if (!restoreComposerDraft(draftTarget, snapshot)) return false; const restored = useComposerDraftStore.getState().getComposerDraft(draftTarget); if (restored && currentRouteThreadKeyRef.current === sentRouteThreadKey) { promptRef.current = restored.prompt; @@ -403,6 +408,7 @@ export function useComposerSend(input: ComposerSendInput) { detectTrigger: true, }); } + return true; }; const onSend = async ( @@ -1153,6 +1159,18 @@ export function useComposerSend(input: ComposerSendInput) { } else { optimisticBlobPreviews.release(optimisticMessage); } + // Its own previews, since this view revokes the row's when it switches threads. + const detachedOptimisticMessage = (): ChatMessage => + optimisticAttachments.length === 0 + ? optimisticMessage + : { + ...optimisticMessage, + attachments: optimisticMessageAttachments( + composerAttachmentsSnapshot.map((attachment) => + attachment.type === "image" ? cloneComposerImageForRetry(attachment) : attachment, + ), + ), + }; setOptimisticUserMessages((existing) => [...existing, optimisticMessage]); setThreadError(threadIdForSend, null); if (expiredTerminalContextCount > 0) { @@ -1259,15 +1277,13 @@ export function useComposerSend(input: ComposerSendInput) { }; // `failure` is null when the user cancelled a send that was waiting on a reconnect. const restoreUnsentDraft = (failure: TurnDeliveryFailure | null) => { + let restoredThreadId = threadIdForSend; if (isLocalDraftThread && draftId && draftThread) { - restoreFailedDraftThread( - draftId, - draftThread, - // A cancelled launch may still have committed, so its draft must not reuse the thread id. - failure === null || wasBootstrapThreadDeleted(squashAtomCommandFailure(failure)) - ? newThreadId() - : threadIdForSend, - ); + // A cancelled launch may still have committed, so its draft must not reuse the thread id. + if (failure === null || wasBootstrapThreadDeleted(squashAtomCommandFailure(failure))) { + restoredThreadId = newThreadId(); + } + restoreFailedDraftThread(draftId, draftThread, restoredThreadId); clearBackgroundDraftSubmissionByRef(scopeThreadRef(environmentId, threadIdForSend)); } setOptimisticUserMessages((existing) => { @@ -1278,7 +1294,23 @@ export function useComposerSend(input: ComposerSendInput) { const next = existing.filter((message) => message.id !== messageIdForSend); return next.length === existing.length ? existing : next; }); - restoreUnsentComposer(composerDraftTarget, routeThreadKey, sendSnapshot); + const restore = () => + restoreUnsentComposer(composerDraftTarget, routeThreadKey, sendSnapshot); + if (!restore()) { + unsentMessages.keep({ + environmentId, + threadId: restoredThreadId, + display: detachedOptimisticMessage(), + restore, + }); + toastManager.add( + stackedThreadToast({ + type: "warning", + title: "Unsent message kept in its thread", + description: `Its attachments would put the composer over its ${PROVIDER_SEND_TURN_MAX_ATTACHMENTS}-attachment limit. Make room, then restore it from the thread.`, + }), + ); + } if (failure !== null && !isAtomCommandInterrupted(failure)) { const error = squashAtomCommandFailure(failure); setThreadError( @@ -1368,20 +1400,7 @@ export function useComposerSend(input: ComposerSendInput) { ...(bootstrap ? { bootstrap } : {}), createdAt: messageCreatedAt, }, - // Its own previews, since this view revokes the row's when it switches threads. - () => - optimisticAttachments.length === 0 - ? optimisticMessage - : { - ...optimisticMessage, - attachments: optimisticMessageAttachments( - composerAttachmentsSnapshot.map((attachment) => - attachment.type === "image" - ? cloneComposerImageForRetry(attachment) - : attachment, - ), - ), - }, + detachedOptimisticMessage, ); if (backgroundThreadRef) { try { diff --git a/apps/web/src/state/unsentMessages.test.ts b/apps/web/src/state/unsentMessages.test.ts new file mode 100644 index 0000000000000..6c0c335dce01d --- /dev/null +++ b/apps/web/src/state/unsentMessages.test.ts @@ -0,0 +1,90 @@ +import { + EnvironmentId, + MessageId, + PROVIDER_SEND_TURN_MAX_ATTACHMENTS, + ThreadId, +} from "@t3tools/contracts"; +import { expect, it } from "vite-plus/test"; + +import { useComposerDraftStore } from "../composerDraftStore"; +import { snapshotComposerSend } from "../components/ChatView.logic"; +import { restoreComposerDraft } from "../components/chat/chatViewSend"; +import type { ChatMessage } from "../types"; +import { unsentMessages } from "./unsentMessages"; + +const environmentId = EnvironmentId.make("environment"); +const threadId = ThreadId.make("t"); +const target = { environmentId, threadId }; +const messageId = MessageId.make("unsent-message"); + +function image(id: string) { + return { + type: "image" as const, + id, + name: `${id}.png`, + mimeType: "image/png", + sizeBytes: 5, + previewUrl: `data:image/png;base64,${id}`, + file: new File([id], `${id}.png`, { type: "image/png" }), + }; +} + +function keepUnsentBesideFullDraft() { + useComposerDraftStore.setState({ draftsByThreadKey: {}, draftThreadsByThreadKey: {} }); + const store = useComposerDraftStore.getState(); + store.setPrompt(target, "Newer draft"); + store.addImages( + target, + Array.from({ length: PROVIDER_SEND_TURN_MAX_ATTACHMENTS }, (_, index) => + image(`newer-${index}`), + ), + ); + const snapshot = snapshotComposerSend({ + prompt: "Unsent message", + images: [image("unsent")], + files: [], + terminalContexts: [], + sendableTerminalContexts: [], + previewAnnotations: [], + reviewComments: [], + threadContexts: [], + }); + const display: ChatMessage = { + id: messageId, + role: "user", + text: "Unsent message", + runId: null, + createdAt: "2026-10-25T00:00:00.000Z", + updatedAt: "2026-10-25T00:00:00.000Z", + streaming: false, + }; + const restore = () => restoreComposerDraft(target, snapshot); + expect(restore()).toBe(false); + unsentMessages.keep({ environmentId, threadId, display, restore }); + return store; +} + +it("keeps an unsent message that does not fit until the composer has room", () => { + const store = keepUnsentBesideFullDraft(); + + expect(unsentMessages.restore(environmentId, messageId)).toBe(false); + expect(unsentMessages.getSnapshot().map((record) => record.display.id)).toEqual([messageId]); + expect(store.getComposerDraft(target)?.prompt).toBe("Newer draft"); + + store.removeImage(target, "newer-0"); + expect(unsentMessages.restore(environmentId, messageId)).toBe(true); + + expect(unsentMessages.getSnapshot()).toEqual([]); + const draft = store.getComposerDraft(target); + expect(draft?.prompt).toBe("Newer draft\n\nUnsent message"); + expect(draft?.images.map((attachment) => attachment.id).at(-1)).toBe("unsent"); +}); + +it("drops a kept message only when it is discarded", () => { + const store = keepUnsentBesideFullDraft(); + + unsentMessages.discard(environmentId, messageId); + + expect(unsentMessages.getSnapshot()).toEqual([]); + expect(store.getComposerDraft(target)?.prompt).toBe("Newer draft"); +}); diff --git a/apps/web/src/state/unsentMessages.ts b/apps/web/src/state/unsentMessages.ts new file mode 100644 index 0000000000000..3cdb67c3aab12 --- /dev/null +++ b/apps/web/src/state/unsentMessages.ts @@ -0,0 +1,80 @@ +import type { EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts"; +import { useSyncExternalStore } from "react"; + +import { revokeUserMessagePreviewUrls } from "../components/ChatView.logic"; +import type { ChatMessage } from "../types"; + +interface UnsentMessage { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; + /** Owns its previews, which are revoked once the message is restored or discarded. */ + readonly display: ChatMessage; + /** Writes the message back into its composer; false while it still does not fit. */ + readonly restore: () => boolean; +} + +let records: ReadonlyArray = []; +const listeners = new Set<() => void>(); + +const isRecord = (record: UnsentMessage, environmentId: EnvironmentId, messageId: MessageId) => + record.environmentId === environmentId && record.display.id === messageId; + +function drop(environmentId: EnvironmentId, messageId: MessageId) { + const record = records.find((candidate) => isRecord(candidate, environmentId, messageId)); + if (record === undefined) return; + records = records.filter((candidate) => candidate !== record); + for (const listener of listeners) listener(); + revokeUserMessagePreviewUrls(record.display); +} + +/** + * Holds a message whose send ended unsent but which no longer fits in its composer, so it stays in + * its thread until restored or discarded. Outlives ChatView, like a send waiting on a reconnect. + */ +export const unsentMessages = { + keep: (record: UnsentMessage) => { + drop(record.environmentId, record.display.id); + records = [...records, record]; + for (const listener of listeners) listener(); + }, + restore: (environmentId: EnvironmentId, messageId: MessageId): boolean => { + const record = records.find((candidate) => isRecord(candidate, environmentId, messageId)); + if (record === undefined) return true; + if (!record.restore()) return false; + drop(environmentId, messageId); + return true; + }, + discard: drop, + getSnapshot: (): ReadonlyArray => records, + subscribe: (listener: () => void) => { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }, +}; + +const EMPTY: ReadonlyArray = []; + +function useUnsentMessageSnapshot() { + return useSyncExternalStore(unsentMessages.subscribe, unsentMessages.getSnapshot); +} + +export function useUnsentMessages( + environmentId: EnvironmentId, + threadId: ThreadId | null, +): ReadonlyArray { + const kept = useUnsentMessageSnapshot().filter( + (record) => record.environmentId === environmentId && record.threadId === threadId, + ); + return kept.length === 0 ? EMPTY : kept; +} + +export function useUnsentMessage( + environmentId: EnvironmentId, + messageId: MessageId, +): UnsentMessage | null { + return ( + useUnsentMessageSnapshot().find((record) => isRecord(record, environmentId, messageId)) ?? null + ); +} diff --git a/docs/user/composer.md b/docs/user/composer.md index 24e8541e8a66b..68ff4e48e098c 100644 --- a/docs/user/composer.md +++ b/docs/user/composer.md @@ -88,7 +88,8 @@ device until you sign back into the same account. On web and desktop, a message sent as the connection drops shows **Waiting to send**. When the thread reconnects, T3 Code sends it once, or keeps it if the server already received it. Cancel stops the send and puts the message back in the composer, after anything you typed since. It -can't unsend a message the server already received. +can't unsend a message the server already received. If its attachments no longer fit beside what +you added since, it stays in the thread as **Not sent** until you restore or discard it. ## Custom models diff --git a/packages/client-runtime/src/state/turnDelivery.test.ts b/packages/client-runtime/src/state/turnDelivery.test.ts index 460501ca25f87..279bad414ca7d 100644 --- a/packages/client-runtime/src/state/turnDelivery.test.ts +++ b/packages/client-runtime/src/state/turnDelivery.test.ts @@ -108,6 +108,7 @@ function setup( let threadListener: ((state: EnvironmentThreadState) => void) | null = null; let shellListener: ((state: EnvironmentShellState) => void) | null = null; let threadObservations = 0; + let activeObservers = 0; let displaysMade = 0; const released: Array = []; const deliveries = createTurnDeliveries<{ sequence: number }, unknown, string>({ @@ -117,16 +118,20 @@ function setup( }, observeThread: (_environmentId, _threadId, listener) => { threadObservations += 1; + activeObservers += 1; threadListener = listener; listener(initialThread); return () => { + activeObservers -= 1; threadListener = null; }; }, observeShell: (_environmentId, listener) => { + activeObservers += 1; shellListener = listener; listener(shell("live")); return () => { + activeObservers -= 1; shellListener = null; }; }, @@ -139,6 +144,7 @@ function setup( sent, released, threadObservations: () => threadObservations, + activeObservers: () => activeObservers, displaysMade: () => displaysMade, makeDisplay: () => { displaysMade += 1; @@ -327,4 +333,16 @@ describe("turn deliveries", () => { expect(harness.sent).toEqual([LAUNCH]); expect(harness.deliveries.getSnapshot()).toEqual([]); }); + + it("stops watching a committed launch whose thread already shows the message", async () => { + const harness = setup([LOST_ACK], HELD); + const { settled } = await deliverUntilWaiting(harness, LAUNCH); + + harness.emitShell(shell("synchronizing")); + harness.emitShell(shell("live", [THREAD_ID])); + + expect(await settled).toEqual({ type: "delivered", value: null }); + expect(harness.deliveries.getSnapshot()).toEqual([]); + expect(harness.activeObservers()).toBe(0); + }); }); diff --git a/packages/client-runtime/src/state/turnDelivery.ts b/packages/client-runtime/src/state/turnDelivery.ts index f800289c38e09..b5cf3a92fe941 100644 --- a/packages/client-runtime/src/state/turnDelivery.ts +++ b/packages/client-runtime/src/state/turnDelivery.ts @@ -144,8 +144,15 @@ export function createTurnDeliveries(deps: TurnDeliveryDependencies, stop: () => void) => { + if (entries.get(entry.key) === entry) entry.stopObserving.push(stop); + else stop(); + }; + const observeThread = (entry: Entry) => { - entry.stopObserving.push( + keepObserving( + entry, deps.observeThread(entry.environmentId, entry.input.threadId, (state) => { entry.thread = state; if (state.status !== "live") entry.resynced = true; @@ -261,7 +268,8 @@ export function createTurnDeliveries(deps: TurnDeliveryDependencies { entry.shell = state; if (entry.unborn && state.status !== "live") entry.resynced = true; From ed5fbd28a042eb45755bb91ac9fa1a60010a2a5c Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Sun, 25 Oct 2026 03:58:32 +0000 Subject: [PATCH 05/11] fix(web): restore a cancelled launch's kept message after its draft is sent A cancelled launch whose message did not fit its composer kept a restore that wrote to the draft id. Sending the newer draft promoted it and moved its composer to the thread, so restoring later wrote to an orphaned draft key, reported success and dropped the row. Restore to the reopened thread's ref, which resolves to the draft before promotion and to the thread after it. Co-Authored-By: Claude Opus 5.5 --- apps/web/src/components/chat/chatViewSend.ts | 15 ++++ .../src/components/chat/useComposerSend.ts | 12 ++- apps/web/src/state/unsentMessages.test.ts | 86 ++++++++++++++----- 3 files changed, 88 insertions(+), 25 deletions(-) diff --git a/apps/web/src/components/chat/chatViewSend.ts b/apps/web/src/components/chat/chatViewSend.ts index 19aa0ea2996df..5cb5321aa0808 100644 --- a/apps/web/src/components/chat/chatViewSend.ts +++ b/apps/web/src/components/chat/chatViewSend.ts @@ -26,7 +26,9 @@ import { newThreadId } from "~/lib/utils"; import { type ComposerFileAttachment, type ComposerImageAttachment, + type DraftThreadState, composerDraftHasUserContent, + restoreFailedDraftThread, useComposerDraftStore, DraftId, } from "../../composerDraftStore"; @@ -252,6 +254,19 @@ export async function rewindToMessage(input: { } } +/** + * Reopens a cancelled launch's draft as `threadRef` and returns where its unsent message + * restores. The thread ref still resolves after that draft is sent and promoted. + */ +export function reopenUnsentDraftThread( + draftId: DraftId, + draftThread: DraftThreadState, + threadRef: ScopedThreadRef, +): ScopedThreadRef { + restoreFailedDraftThread(draftId, draftThread, threadRef.threadId); + return threadRef; +} + /** * Writes an unsent message's composer contents back into its draft, after any newer draft * started meanwhile. Images are fresh copies, so their previews outlive the optimistic row. diff --git a/apps/web/src/components/chat/useComposerSend.ts b/apps/web/src/components/chat/useComposerSend.ts index 4e500c6b7a532..81fd3dce08b8d 100644 --- a/apps/web/src/components/chat/useComposerSend.ts +++ b/apps/web/src/components/chat/useComposerSend.ts @@ -70,7 +70,6 @@ import { clearBackgroundDraftSubmissionByRef, finalizePromotedDraftThreadByRef, markPromotedDraftThreadByRef, - restoreFailedDraftThread, composerDraftHasUserContent, type ComposerFileAttachment, type ComposerImageAttachment, @@ -123,6 +122,7 @@ import { type MultipleModelTarget, resolveMultipleModelTargets, prepareTurnAttachments, + reopenUnsentDraftThread, restoreComposerDraft, } from "./chatViewSend"; import type { @@ -1278,12 +1278,17 @@ export function useComposerSend(input: ComposerSendInput) { // `failure` is null when the user cancelled a send that was waiting on a reconnect. const restoreUnsentDraft = (failure: TurnDeliveryFailure | null) => { let restoredThreadId = threadIdForSend; + let restoreTarget = composerDraftTarget; if (isLocalDraftThread && draftId && draftThread) { // A cancelled launch may still have committed, so its draft must not reuse the thread id. if (failure === null || wasBootstrapThreadDeleted(squashAtomCommandFailure(failure))) { restoredThreadId = newThreadId(); } - restoreFailedDraftThread(draftId, draftThread, restoredThreadId); + restoreTarget = reopenUnsentDraftThread( + draftId, + draftThread, + scopeThreadRef(environmentId, restoredThreadId), + ); clearBackgroundDraftSubmissionByRef(scopeThreadRef(environmentId, threadIdForSend)); } setOptimisticUserMessages((existing) => { @@ -1294,8 +1299,7 @@ export function useComposerSend(input: ComposerSendInput) { const next = existing.filter((message) => message.id !== messageIdForSend); return next.length === existing.length ? existing : next; }); - const restore = () => - restoreUnsentComposer(composerDraftTarget, routeThreadKey, sendSnapshot); + const restore = () => restoreUnsentComposer(restoreTarget, routeThreadKey, sendSnapshot); if (!restore()) { unsentMessages.keep({ environmentId, diff --git a/apps/web/src/state/unsentMessages.test.ts b/apps/web/src/state/unsentMessages.test.ts index 6c0c335dce01d..856b73659a8e0 100644 --- a/apps/web/src/state/unsentMessages.test.ts +++ b/apps/web/src/state/unsentMessages.test.ts @@ -1,14 +1,21 @@ +import { scopeProjectRef, scopeThreadRef } from "@t3tools/client-runtime/environment"; import { EnvironmentId, MessageId, PROVIDER_SEND_TURN_MAX_ATTACHMENTS, + ProjectId, ThreadId, } from "@t3tools/contracts"; import { expect, it } from "vite-plus/test"; -import { useComposerDraftStore } from "../composerDraftStore"; +import { + DraftId, + finalizePromotedDraftThreadByRef, + markPromotedDraftThreadByRef, + useComposerDraftStore, +} from "../composerDraftStore"; import { snapshotComposerSend } from "../components/ChatView.logic"; -import { restoreComposerDraft } from "../components/chat/chatViewSend"; +import { reopenUnsentDraftThread, restoreComposerDraft } from "../components/chat/chatViewSend"; import type { ChatMessage } from "../types"; import { unsentMessages } from "./unsentMessages"; @@ -29,17 +36,14 @@ function image(id: string) { }; } -function keepUnsentBesideFullDraft() { - useComposerDraftStore.setState({ draftsByThreadKey: {}, draftThreadsByThreadKey: {} }); - const store = useComposerDraftStore.getState(); - store.setPrompt(target, "Newer draft"); - store.addImages( - target, - Array.from({ length: PROVIDER_SEND_TURN_MAX_ATTACHMENTS }, (_, index) => - image(`newer-${index}`), - ), +function newerImages() { + return Array.from({ length: PROVIDER_SEND_TURN_MAX_ATTACHMENTS }, (_, index) => + image(`newer-${index}`), ); - const snapshot = snapshotComposerSend({ +} + +function unsentSnapshot() { + return snapshotComposerSend({ prompt: "Unsent message", images: [image("unsent")], files: [], @@ -49,15 +53,24 @@ function keepUnsentBesideFullDraft() { reviewComments: [], threadContexts: [], }); - const display: ChatMessage = { - id: messageId, - role: "user", - text: "Unsent message", - runId: null, - createdAt: "2026-10-25T00:00:00.000Z", - updatedAt: "2026-10-25T00:00:00.000Z", - streaming: false, - }; +} + +const display: ChatMessage = { + id: messageId, + role: "user", + text: "Unsent message", + runId: null, + createdAt: "2026-10-25T00:00:00.000Z", + updatedAt: "2026-10-25T00:00:00.000Z", + streaming: false, +}; + +function keepUnsentBesideFullDraft() { + useComposerDraftStore.setState({ draftsByThreadKey: {}, draftThreadsByThreadKey: {} }); + const store = useComposerDraftStore.getState(); + store.setPrompt(target, "Newer draft"); + store.addImages(target, newerImages()); + const snapshot = unsentSnapshot(); const restore = () => restoreComposerDraft(target, snapshot); expect(restore()).toBe(false); unsentMessages.keep({ environmentId, threadId, display, restore }); @@ -88,3 +101,34 @@ it("drops a kept message only when it is discarded", () => { expect(unsentMessages.getSnapshot()).toEqual([]); expect(store.getComposerDraft(target)?.prompt).toBe("Newer draft"); }); + +it("restores a cancelled launch into its thread after the newer draft there is sent", () => { + useComposerDraftStore.setState({ draftsByThreadKey: {}, draftThreadsByThreadKey: {} }); + const store = useComposerDraftStore.getState(); + const draftId = DraftId.make("launch-draft"); + store.setProjectDraftThreadId(scopeProjectRef(environmentId, ProjectId.make("p")), draftId, { + threadId: ThreadId.make("launched"), + }); + const launched = store.getDraftSession(draftId)!; + markPromotedDraftThreadByRef(scopeThreadRef(environmentId, launched.threadId)); + store.setPrompt(draftId, "Newer draft"); + store.addImages(draftId, newerImages()); + + const reopenedRef = scopeThreadRef(environmentId, ThreadId.make("reopened")); + const restoreTarget = reopenUnsentDraftThread(draftId, launched, reopenedRef); + const snapshot = unsentSnapshot(); + const restore = () => restoreComposerDraft(restoreTarget, snapshot); + expect(restore()).toBe(false); + unsentMessages.keep({ environmentId, threadId: reopenedRef.threadId, display, restore }); + + markPromotedDraftThreadByRef(reopenedRef); + store.clearComposerContent(draftId); + finalizePromotedDraftThreadByRef(reopenedRef); + expect(store.getDraftSession(draftId)).toBeNull(); + + expect(unsentMessages.restore(environmentId, messageId)).toBe(true); + expect(unsentMessages.getSnapshot()).toEqual([]); + const draft = store.getComposerDraft(reopenedRef); + expect(draft?.prompt).toBe("Unsent message"); + expect(draft?.images.map((attachment) => attachment.id)).toEqual(["unsent"]); +}); From 16809b09b9a927a30a327845b97bd4bfac801e28 Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Sun, 25 Oct 2026 04:13:11 +0000 Subject: [PATCH 06/11] fix(web): show kept queued messages in their thread and keep restored terminal references resolvable A queued send kept after cancel or rejection stayed marked as queued, so the timeline hid it and the queue showed it as saving forever. Kept messages now drop the queued intent. When a restored draft carries a terminal selection the newer draft already has under another id, setTerminalContexts dropped the duplicate but left its prompt reference dangling. Its references now point at the surviving context. Co-Authored-By: Claude Opus 5.5 --- .../src/components/chat/chatViewSend.test.ts | 40 +++++++++++++++++++ apps/web/src/composerDraftStore.ts | 28 ++++++++++++- apps/web/src/state/unsentMessages.test.ts | 25 ++++++++++++ apps/web/src/state/unsentMessages.ts | 4 +- 4 files changed, 95 insertions(+), 2 deletions(-) diff --git a/apps/web/src/components/chat/chatViewSend.test.ts b/apps/web/src/components/chat/chatViewSend.test.ts index 7805ada1d65c0..66d0fac6d8b3d 100644 --- a/apps/web/src/components/chat/chatViewSend.test.ts +++ b/apps/web/src/components/chat/chatViewSend.test.ts @@ -2,6 +2,9 @@ import { EnvironmentId, PROVIDER_SEND_TURN_MAX_ATTACHMENTS, ThreadId } from "@t3 import { expect, it } from "vite-plus/test"; import { useComposerDraftStore } from "../../composerDraftStore"; +import { terminalContextReference } from "../../lib/composerContextRecords"; +import { formatInlineContextReference } from "../../lib/composerContextReferences"; +import type { TerminalContextDraft } from "../../lib/terminalContext"; import { snapshotComposerSend } from "../ChatView.logic"; import { restoreComposerDraft } from "./chatViewSend"; @@ -70,3 +73,40 @@ it("leaves a full newer draft untouched when the unsent attachments do not fit", expect(draft?.images).toHaveLength(PROVIDER_SEND_TURN_MAX_ATTACHMENTS); expect(unsent.images.map((attachment) => attachment.id)).toEqual(["unsent"]); }); + +it("points a restored terminal reference at the newer draft's copy of the same selection", () => { + useComposerDraftStore.setState({ draftsByThreadKey: {}, draftThreadsByThreadKey: {} }); + const store = useComposerDraftStore.getState(); + const selection = (id: string): TerminalContextDraft => ({ + id, + threadId: target.threadId, + createdAt: "2026-10-25T00:00:00.000Z", + terminalId: "default", + terminalLabel: "Terminal 1", + lineStart: 4, + lineEnd: 6, + text: "npm test failed", + }); + const newer = selection("newer-selection"); + const unsentSelection = selection("unsent-selection"); + const reference = (context: TerminalContextDraft) => + formatInlineContextReference(terminalContextReference(context)); + store.setPrompt(target, `Newer ${reference(newer)}`); + store.setTerminalContexts(target, [newer]); + const unsent = snapshotComposerSend({ + prompt: `Unsent ${reference(unsentSelection)}`, + images: [], + files: [], + terminalContexts: [unsentSelection], + sendableTerminalContexts: [unsentSelection], + previewAnnotations: [], + reviewComments: [], + threadContexts: [], + }); + + expect(restoreComposerDraft(target, unsent)).toBe(true); + + const draft = store.getComposerDraft(target); + expect(draft?.terminalContexts.map((context) => context.id)).toEqual(["newer-selection"]); + expect(draft?.prompt).toBe(`Newer ${reference(newer)}\n\nUnsent ${reference(newer)}`); +}); diff --git a/apps/web/src/composerDraftStore.ts b/apps/web/src/composerDraftStore.ts index e823d2ddf33a3..df403427b5a36 100644 --- a/apps/web/src/composerDraftStore.ts +++ b/apps/web/src/composerDraftStore.ts @@ -2904,8 +2904,34 @@ const composerDraftStore = create()( return; } const normalizedContexts = normalizeTerminalContextsForThread(threadId, contexts); + // A context deduplicated onto an earlier copy of its selection hands its references over. + const survivorByReferenceId = new Map(); + for (const context of contexts) { + const normalized = normalizeTerminalContextForThread(threadId, context); + if (!normalized) continue; + const dedupKey = terminalContextDedupKey(normalized); + const survivor = normalizedContexts.find( + (candidate) => + candidate.id === normalized.id || terminalContextDedupKey(candidate) === dedupKey, + ); + if (survivor && survivor.id !== normalized.id) { + survivorByReferenceId.set(terminalContextReference(normalized).contextId, survivor); + } + } set((state) => { const existing = state.draftsByThreadKey[threadKey] ?? createEmptyThreadDraft(); + const remappedPrompt = + survivorByReferenceId.size === 0 + ? existing.prompt + : replaceComposerContextReferences(existing.prompt, (occurrence) => { + const survivor = + occurrence.kind === "terminal" + ? survivorByReferenceId.get(occurrence.contextId) + : undefined; + return survivor + ? formatInlineContextReference(terminalContextReference(survivor)) + : occurrence.source; + }); const nextDraft: ComposerThreadDraftState = { ...existing, prompt: ensureInlineContextReferences( @@ -2917,7 +2943,7 @@ const composerDraftStore = create()( prompt, terminalContextReference(context).contextId, ).prompt, - existing.prompt, + remappedPrompt, ), normalizedContexts.map(terminalContextReference), ), diff --git a/apps/web/src/state/unsentMessages.test.ts b/apps/web/src/state/unsentMessages.test.ts index 856b73659a8e0..e41fd085e2ec3 100644 --- a/apps/web/src/state/unsentMessages.test.ts +++ b/apps/web/src/state/unsentMessages.test.ts @@ -16,6 +16,7 @@ import { } from "../composerDraftStore"; import { snapshotComposerSend } from "../components/ChatView.logic"; import { reopenUnsentDraftThread, restoreComposerDraft } from "../components/chat/chatViewSend"; +import { deriveTimelineEntriesFromVisibleTurnItems } from "../session-logic"; import type { ChatMessage } from "../types"; import { unsentMessages } from "./unsentMessages"; @@ -132,3 +133,27 @@ it("restores a cancelled launch into its thread after the newer draft there is s expect(draft?.prompt).toBe("Unsent message"); expect(draft?.images.map((attachment) => attachment.id)).toEqual(["unsent"]); }); + +it("shows a kept queued message in its timeline instead of as a queued run", () => { + useComposerDraftStore.setState({ draftsByThreadKey: {}, draftThreadsByThreadKey: {} }); + const store = useComposerDraftStore.getState(); + store.setPrompt(target, "Newer draft"); + store.addImages(target, newerImages()); + const snapshot = unsentSnapshot(); + const restore = () => restoreComposerDraft(target, snapshot); + unsentMessages.keep({ + environmentId, + threadId, + display: { ...display, inputIntent: "queued_turn" }, + restore, + }); + + const [kept] = unsentMessages.getSnapshot(); + expect(kept?.display.inputIntent).not.toBe("queued_turn"); + const entries = deriveTimelineEntriesFromVisibleTurnItems({ + visibleTurnItems: [], + optimisticMessages: unsentMessages.getSnapshot().map((record) => record.display), + }); + expect(entries.map((entry) => entry.id)).toEqual([messageId]); + unsentMessages.discard(environmentId, messageId); +}); diff --git a/apps/web/src/state/unsentMessages.ts b/apps/web/src/state/unsentMessages.ts index 3cdb67c3aab12..b59a2c1c54eb5 100644 --- a/apps/web/src/state/unsentMessages.ts +++ b/apps/web/src/state/unsentMessages.ts @@ -34,7 +34,9 @@ function drop(environmentId: EnvironmentId, messageId: MessageId) { export const unsentMessages = { keep: (record: UnsentMessage) => { drop(record.environmentId, record.display.id); - records = [...records, record]; + // It is no longer queued, so it belongs in the timeline rather than the queue. + const { inputIntent, ...display } = record.display; + records = [...records, inputIntent === "queued_turn" ? { ...record, display } : record]; for (const listener of listeners) listener(); }, restore: (environmentId: EnvironmentId, messageId: MessageId): boolean => { From 93a3ef758237f4b715c217d37ce4ed81d9d9e608 Mon Sep 17 00:00:00 2001 From: PR Batch Tester Date: Sun, 25 Oct 2026 04:18:30 +0000 Subject: [PATCH 07/11] fix(client-runtime): never replay a launch the shell shows as committed A launch whose acknowledgement was lost moved back to waiting once the resynced shell held its thread. If the thread then reported live without the first message yet, the still-set resync flag replayed the launch, which could hit expired pending uploads and restore a duplicate draft. Settle it as delivered instead, keeping the row until the thread shows the message. Co-Authored-By: Claude Opus 5.5 --- .../src/state/turnDelivery.test.ts | 21 +++++++++++++++++++ .../client-runtime/src/state/turnDelivery.ts | 5 ++--- 2 files changed, 23 insertions(+), 3 deletions(-) diff --git a/packages/client-runtime/src/state/turnDelivery.test.ts b/packages/client-runtime/src/state/turnDelivery.test.ts index 279bad414ca7d..70137d61fc50c 100644 --- a/packages/client-runtime/src/state/turnDelivery.test.ts +++ b/packages/client-runtime/src/state/turnDelivery.test.ts @@ -334,6 +334,27 @@ describe("turn deliveries", () => { expect(harness.deliveries.getSnapshot()).toEqual([]); }); + it("never replays a committed launch whose live thread does not show the message yet", async () => { + const harness = setup([LOST_ACK, ACCEPTED], thread("live")); + const { settled } = await deliverUntilWaiting(harness, LAUNCH); + + harness.emitShell(shell("synchronizing")); + harness.emitShell(shell("live", [THREAD_ID])); + + expect(await settled).toEqual({ type: "delivered", value: null }); + expect(harness.sent).toEqual([LAUNCH]); + expect(harness.deliveries.getSnapshot()).toEqual([ + expect.objectContaining({ state: "delivered", display: "optimistic row" }), + ]); + harness.emitThread(thread("synchronizing")); + harness.emitThread(thread("live")); + expect(harness.sent).toEqual([LAUNCH]); + harness.emitThread(HELD); + expect(harness.deliveries.getSnapshot()).toEqual([]); + expect(harness.released).toEqual(["optimistic row"]); + expect(harness.activeObservers()).toBe(0); + }); + it("stops watching a committed launch whose thread already shows the message", async () => { const harness = setup([LOST_ACK], HELD); const { settled } = await deliverUntilWaiting(harness, LAUNCH); diff --git a/packages/client-runtime/src/state/turnDelivery.ts b/packages/client-runtime/src/state/turnDelivery.ts index b5cf3a92fe941..bd7971c15a9bb 100644 --- a/packages/client-runtime/src/state/turnDelivery.ts +++ b/packages/client-runtime/src/state/turnDelivery.ts @@ -171,7 +171,7 @@ export function createTurnDeliveries(deps: TurnDeliveryDependencies(deps: TurnDeliveryDependencies Date: Sun, 25 Oct 2026 04:25:18 +0000 Subject: [PATCH 08/11] fix(client-runtime): keep a launch waiting until its thread shows the message The resynced shell holding a launch's thread proves only the thread.create. The server commits the first message separately, so a restart between the two left a launch marked as sent that never was, with no way to cancel it. Watch the thread without replaying the launch, and settle as delivered only once the thread holds the message. Cancel and a deleted thread still end it. Co-Authored-By: Claude Opus 5.5 --- .../src/state/turnDelivery.test.ts | 46 +++++++++++++++++-- .../client-runtime/src/state/turnDelivery.ts | 12 +++-- 2 files changed, 50 insertions(+), 8 deletions(-) diff --git a/packages/client-runtime/src/state/turnDelivery.test.ts b/packages/client-runtime/src/state/turnDelivery.test.ts index 70137d61fc50c..650913776c835 100644 --- a/packages/client-runtime/src/state/turnDelivery.test.ts +++ b/packages/client-runtime/src/state/turnDelivery.test.ts @@ -334,27 +334,63 @@ describe("turn deliveries", () => { expect(harness.deliveries.getSnapshot()).toEqual([]); }); - it("never replays a committed launch whose live thread does not show the message yet", async () => { + it("waits without replaying a launch the shell holds until its thread shows the message", async () => { const harness = setup([LOST_ACK, ACCEPTED], thread("live")); const { settled } = await deliverUntilWaiting(harness, LAUNCH); harness.emitShell(shell("synchronizing")); harness.emitShell(shell("live", [THREAD_ID])); - - expect(await settled).toEqual({ type: "delivered", value: null }); - expect(harness.sent).toEqual([LAUNCH]); + // The shell proves only the thread.create; the first message may still be missing. expect(harness.deliveries.getSnapshot()).toEqual([ - expect.objectContaining({ state: "delivered", display: "optimistic row" }), + expect.objectContaining({ state: "waiting", display: "optimistic row" }), ]); harness.emitThread(thread("synchronizing")); harness.emitThread(thread("live")); + harness.emitShell(shell("synchronizing", [THREAD_ID])); + harness.emitShell(shell("live", [THREAD_ID])); expect(harness.sent).toEqual([LAUNCH]); + expect(harness.threadObservations()).toBe(1); + expect(harness.deliveries.getSnapshot()).toEqual([ + expect.objectContaining({ state: "waiting" }), + ]); + harness.emitThread(HELD); + expect(await settled).toEqual({ type: "delivered", value: null }); + expect(harness.deliveries.getSnapshot()).toEqual([]); + expect(harness.released).toEqual(["optimistic row"]); + expect(harness.activeObservers()).toBe(0); + }); + + it("cancels a launch the shell holds while its thread lacks the message", async () => { + const harness = setup([LOST_ACK], thread("live")); + const { settled } = await deliverUntilWaiting(harness, LAUNCH); + + harness.emitShell(shell("synchronizing")); + harness.emitShell(shell("live", [THREAD_ID])); + const [pending] = harness.deliveries.getSnapshot(); + harness.deliveries.cancel(pending!.key); + + expect(await settled).toEqual({ type: "cancelled" }); + expect(harness.sent).toEqual([LAUNCH]); expect(harness.deliveries.getSnapshot()).toEqual([]); expect(harness.released).toEqual(["optimistic row"]); expect(harness.activeObservers()).toBe(0); }); + it("rejects a launch the shell held once its thread is deleted", async () => { + const harness = setup([LOST_ACK], thread("live")); + const { settled } = await deliverUntilWaiting(harness, LAUNCH); + + harness.emitShell(shell("synchronizing")); + harness.emitShell(shell("live", [THREAD_ID])); + harness.emitThread(thread("deleted")); + + expect((await settled).type).toBe("rejected"); + expect(harness.sent).toEqual([LAUNCH]); + expect(harness.deliveries.getSnapshot()).toEqual([]); + expect(harness.activeObservers()).toBe(0); + }); + it("stops watching a committed launch whose thread already shows the message", async () => { const harness = setup([LOST_ACK], HELD); const { settled } = await deliverUntilWaiting(harness, LAUNCH); diff --git a/packages/client-runtime/src/state/turnDelivery.ts b/packages/client-runtime/src/state/turnDelivery.ts index bd7971c15a9bb..4d51d0f2c4b20 100644 --- a/packages/client-runtime/src/state/turnDelivery.ts +++ b/packages/client-runtime/src/state/turnDelivery.ts @@ -67,6 +67,8 @@ interface Entry { display: M | null; /** Observing a thread that does not exist yet would park its state as deleted. */ unborn: boolean; + /** The shell showed the launch's thread, which proves the create but not the first message. */ + replayBlocked: boolean; state: "sending" | "waiting" | "delivered"; cancelRequested: boolean; /** The thread, or for an unborn one the shell, left live sync since the last dispatch. */ @@ -171,7 +173,7 @@ export function createTurnDeliveries(deps: TurnDeliveryDependencies(deps: TurnDeliveryDependencies(deps: TurnDeliveryDependencies { if (end !== null) finish(entry, end); @@ -252,6 +257,7 @@ export function createTurnDeliveries(deps: TurnDeliveryDependencies Date: Sun, 25 Oct 2026 04:37:47 +0000 Subject: [PATCH 09/11] fix(web): cancel a waiting queued send and give a deleted launch's draft a fresh thread A launch whose thread was deleted while its send waited now rejects with the typed bootstrap-deleted dispatch error, so its restored draft gets a fresh thread id instead of the tombstoned one. A queued send waiting on a reconnect has no server run yet, so its queue row now cancels the local delivery instead of staying disabled. Co-Authored-By: Claude Opus 5.5 --- .../chat/QueuedRunsControl.browser.test.tsx | 65 ++++++++++++++++++- .../src/components/chat/QueuedRunsControl.tsx | 30 +++++++-- .../src/components/chat/chatViewSend.test.ts | 55 +++++++++++++++- apps/web/src/components/chat/chatViewSend.ts | 14 +++- .../src/components/chat/useComposerSend.ts | 11 ++-- apps/web/src/state/unsentMessages.test.ts | 7 +- docs/user/composer.md | 5 +- .../src/state/turnDelivery.test.ts | 9 ++- .../client-runtime/src/state/turnDelivery.ts | 19 +++++- 9 files changed, 188 insertions(+), 27 deletions(-) diff --git a/apps/web/src/components/chat/QueuedRunsControl.browser.test.tsx b/apps/web/src/components/chat/QueuedRunsControl.browser.test.tsx index 7e25e84135ba5..42e5e8ffa0e20 100644 --- a/apps/web/src/components/chat/QueuedRunsControl.browser.test.tsx +++ b/apps/web/src/components/chat/QueuedRunsControl.browser.test.tsx @@ -1,4 +1,10 @@ -import { EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts"; +import { + CommandId, + EnvironmentId, + MessageId, + ProviderInstanceId, + ThreadId, +} from "@t3tools/contracts"; import { serializeAssistantCitation } from "@t3tools/shared/assistantCitations"; import * as DateTime from "effect/DateTime"; import type { ComponentProps } from "react"; @@ -41,7 +47,33 @@ vi.mock(import("../../assets/assetUrls"), async (importOriginal) => ({ })); import "~/index.css"; +vi.mock(import("../../state/turnDeliveries"), async () => { + const { createTurnDeliveries } = await import("@t3tools/client-runtime/state/turn-delivery"); + const { useSyncExternalStore } = await import("react"); + const Cause = await import("effect/Cause"); + const { AsyncResult } = await import("effect/reactivity"); + const turnDeliveries = createTurnDeliveries({ + send: async () => + AsyncResult.failure(Cause.fail({ _tag: "RpcClientError", message: "SocketCloseError" })), + // A thread that never resyncs keeps a lost send waiting. + observeThread: (_environmentId, _threadId, listener) => { + listener({ status: "live" } as never); + return () => undefined; + }, + observeShell: () => () => undefined, + releaseDisplay: () => undefined, + }); + return { + turnDeliveries, + usePendingTurnDeliveries: (environmentId: string, threadId: string) => + useSyncExternalStore(turnDeliveries.subscribe, turnDeliveries.getSnapshot).filter( + (delivery) => delivery.environmentId === environmentId && delivery.threadId === threadId, + ), + } as never; +}); + import { render as renderUi } from "../../../test/render"; +import { turnDeliveries } from "../../state/turnDeliveries"; import { QueuedRunsControl } from "./QueuedRunsControl"; function setup() { @@ -203,3 +235,34 @@ it("stays inside the timeline scroller instead of growing the chat column", asyn const column = screen.getByTestId("column").element(); expect(column.scrollHeight).toBe(column.clientHeight); }); + +it("cancels a queued send that is waiting on a reconnect", async () => { + setup(); + state.projection = { projection: { messages: [] } }; + state.workflow = { activeRun: { id: "run:active" }, canPromoteToSteer: true, queuedRuns: [] }; + const message = { + id: MessageId.make("message:queued"), + inputIntent: "queued_turn" as const, + text: "Run the tests next", + attachments: [], + }; + const outcome = await turnDeliveries.deliver( + baseProps.environmentId, + { + commandId: CommandId.make("command:queued"), + threadId: baseProps.threadId, + message: { messageId: message.id, role: "user", text: message.text, attachments: [] }, + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + }, + () => message as never, + ); + expect(outcome.type).toBe("waiting"); + await using _screen = await renderQueue({ optimisticMessages: [message] }); + + await page.getByRole("button", { name: "Cancel waiting queued message", exact: true }).click(); + + expect(outcome.type === "waiting" && (await outcome.settled)).toEqual({ type: "cancelled" }); + expect(turnDeliveries.getSnapshot()).toEqual([]); +}); diff --git a/apps/web/src/components/chat/QueuedRunsControl.tsx b/apps/web/src/components/chat/QueuedRunsControl.tsx index ca98e919ff93d..a0be8612cd964 100644 --- a/apps/web/src/components/chat/QueuedRunsControl.tsx +++ b/apps/web/src/components/chat/QueuedRunsControl.tsx @@ -28,6 +28,7 @@ import { useId, useImperativeHandle, useRef, useState, type ReactNode, type Ref import { useAssetUrls } from "../../assets/assetUrls"; import { threadEnvironment } from "../../state/threads"; +import { turnDeliveries, usePendingTurnDeliveries } from "../../state/turnDeliveries"; import { useQuietThreadShell, useThreadProjection } from "../../state/entities"; import { useAtomCommand } from "../../state/use-atom-command"; import { isImageAttachment, type ChatMessage } from "../../types"; @@ -99,6 +100,7 @@ export function QueuedRunsControl({ const promote = useAtomCommand(threadEnvironment.promoteQueuedRun); const resume = useAtomCommand(threadEnvironment.resumeThreadQueue); const cancel = useAtomCommand(threadEnvironment.cancelQueuedRun); + const pendingDeliveries = usePendingTurnDeliveries(props.environmentId, props.threadId); const [showAll, setShowAll] = useState(false); const queueListId = useId(); const [busyRunId, setBusyRunId] = useState(null); @@ -142,6 +144,7 @@ export function QueuedRunsControl({ url: queuedImageUrlById.get(attachment.id) ?? null, })), pending: false, + waitingDeliveryKey: null, })), ...optimisticQueued.map((message) => ({ key: message.id, @@ -160,6 +163,10 @@ export function QueuedRunsControl({ url: attachment.previewUrl ?? null, })), pending: true, + waitingDeliveryKey: + pendingDeliveries.find( + (delivery) => delivery.messageId === message.id && delivery.state === "waiting", + )?.key ?? null, })), ]; @@ -293,7 +300,7 @@ export function QueuedRunsControl({ {item.pending ? ( - Saving + {item.waitingDeliveryKey === null ? "Saving" : "Waiting to send"} ) : null} {isEditing ? ( @@ -348,17 +355,32 @@ export function QueuedRunsControl({