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/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts
index bef8f7ab405f1..68a877f6b1ba9 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";
@@ -707,6 +707,12 @@ export function collectUserMessageBlobPreviewUrls(message: ChatMessage): string[
   return previewUrls;
 }
 
+export function revokeUserMessagePreviewUrls(message: ChatMessage): void {
+  for (const previewUrl of collectUserMessageBlobPreviewUrls(message)) {
+    URL.revokeObjectURL(previewUrl);
+  }
+}
+
 export { readFileAsDataUrl } from "../lib/imageCompression";
 
 export { resolveSendEnvMode } from "@t3tools/shared/projectSettings";
@@ -1166,42 +1172,6 @@ export function shouldRefocusComposerOnWindowFocus(
   );
 }
 
-export interface PlanFollowUpComposerSnapshot {
-  readonly prompt: string;
-  readonly terminalContexts: ReadonlyArray<TerminalContextDraft>;
-  readonly reviewComments: ReadonlyArray<ReviewCommentContext>;
-  readonly previewAnnotations: ReadonlyArray<PreviewAnnotationPayload>;
-  readonly threadContexts: ReadonlyArray<ThreadContextRecord>;
-  readonly environmentContexts: ReadonlyArray<EnvironmentContextRecord>;
-}
-
-export function restorePlanFollowUpComposer(input: {
-  readonly snapshot: PlanFollowUpComposerSnapshot;
-  readonly writePrompt: (prompt: string) => void;
-  readonly writeTerminalContexts: (contexts: ReadonlyArray<TerminalContextDraft>) => void;
-  readonly writeReviewComments: (comments: ReadonlyArray<ReviewCommentContext>) => void;
-  readonly writePreviewAnnotations: (annotations: ReadonlyArray<PreviewAnnotationPayload>) => void;
-  readonly writeThreadContexts: (records: ReadonlyArray<ThreadContextRecord>) => void;
-  readonly writeEnvironmentContexts: (records: ReadonlyArray<EnvironmentContextRecord>) => 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<ComposerImageAttachment>;
diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx
index cf48ddcb842dd..b535cdd7bdbe4 100644
--- a/apps/web/src/components/ChatView.tsx
+++ b/apps/web/src/components/ChatView.tsx
@@ -305,6 +305,8 @@ 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 { useUnsentMessages } from "../state/unsentMessages";
 import { useRightPanelSurfaceClose } from "../hooks/useRightPanelSurfaceClose";
 import { ComposerBannerStack, type ComposerBannerStackItem } from "./chat/ComposerBannerStack";
 import { modelFallbackBannerItem } from "./chat/ComposerUsageLimits";
@@ -653,6 +655,25 @@ 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 unsentMessageRecords = useUnsentMessages(environmentId, threadId);
+  const detachedUserMessages = [
+    ...pendingTurnDeliveries.map((delivery) => delivery.display),
+    ...unsentMessageRecords.map((record) => record.display),
+  ];
+  const displayedOptimisticMessages =
+    detachedUserMessages.length === 0
+      ? optimisticUserMessages
+      : [
+          ...optimisticUserMessages,
+          ...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.
   const [workLocallyResendDraftId, setWorkLocallyResendDraftId] = useState<DraftId | null>(null);
   const [feedbackSubmissionsByThreadKey, setFeedbackSubmissionsByThreadKey] = useState<
@@ -1498,7 +1519,7 @@ export default function ChatView(props: ChatViewProps) {
         ],
   );
   const timelineEntriesInput = {
-    optimisticMessages: optimisticUserMessages,
+    optimisticMessages: displayedOptimisticMessages,
     anchoredMessages: anchoredTimelineMessages,
     attachmentUrlById: timelineAttachmentUrlById,
     ...(serverProjection === null
@@ -1519,7 +1540,7 @@ export default function ChatView(props: ChatViewProps) {
     ...timelineEntriesInput,
     visibleTurnItems: serverTimelineRows,
   });
-  const draftTimelineEntries = optimisticUserMessages.map(
+  const draftTimelineEntries = displayedOptimisticMessages.map(
     (message) =>
       ({
         id: message.id,
@@ -2939,7 +2960,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/QueuedRunsControl.browser.test.tsx b/apps/web/src/components/chat/QueuedRunsControl.browser.test.tsx
index 7e25e84135ba5..66499e23d402c 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,12 +47,44 @@ 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<unknown, unknown, unknown>({
+    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,
+    usePendingTurnDelivery: (environmentId: string, messageId: string) =>
+      useSyncExternalStore(turnDeliveries.subscribe, turnDeliveries.getSnapshot).find(
+        (delivery) => delivery.environmentId === environmentId && delivery.messageId === messageId,
+      ) ?? null,
+    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() {
   state.projection = null;
   state.workflow = null;
+  for (const delivery of turnDeliveries.getSnapshot()) turnDeliveries.cancel(delivery.key);
+  expect(turnDeliveries.getSnapshot()).toEqual([]);
 }
 
 const baseProps = {
@@ -203,3 +241,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<RunId | null>(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 ? (
                   <span className="flex items-center gap-1 px-1">
                     <Clock3Icon aria-hidden className="size-3" />
-                    Saving
+                    {item.waitingDeliveryKey === null ? "Saving" : "Waiting to send"}
                   </span>
                 ) : null}
                 {isEditing ? (
@@ -348,17 +355,32 @@ export function QueuedRunsControl({
                           <Button
                             size="icon-xs"
                             variant="ghost-muted"
-                            aria-label="Remove queued message"
-                            disabled={item.runId === null || busyRunId !== null}
+                            aria-label={
+                              item.waitingDeliveryKey === null
+                                ? "Remove queued message"
+                                : "Cancel waiting queued message"
+                            }
+                            disabled={
+                              item.runId === null
+                                ? item.waitingDeliveryKey === null
+                                : busyRunId !== null
+                            }
                             onClick={() => {
                               if (item.runId !== null) void remove(item.runId);
+                              else if (item.waitingDeliveryKey !== null) {
+                                turnDeliveries.cancel(item.waitingDeliveryKey);
+                              }
                             }}
                           />
                         }
                       >
                         <XIcon />
                       </TooltipTrigger>
-                      <TooltipPopup>Remove from queue</TooltipPopup>
+                      <TooltipPopup>
+                        {item.waitingDeliveryKey === null
+                          ? "Remove from queue"
+                          : "Cancel this send and restore it to the composer"}
+                      </TooltipPopup>
                     </Tooltip>
                   </>
                 )}
diff --git a/apps/web/src/components/chat/TimelineUserMessageRow.tsx b/apps/web/src/components/chat/TimelineUserMessageRow.tsx
index c4363a1b45983..ad8b87f45fc2f 100644
--- a/apps/web/src/components/chat/TimelineUserMessageRow.tsx
+++ b/apps/web/src/components/chat/TimelineUserMessageRow.tsx
@@ -11,6 +11,8 @@ 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 { unsentMessages, useUnsentMessage } from "../../state/unsentMessages";
 import { useEnvironmentQuery } from "../../state/query";
 import { serverEnvironment } from "../../state/server";
 import { replaceComposerContextReferences } from "@t3tools/shared/composerContextReferences";
@@ -292,6 +294,50 @@ function UserMessageProjectedStatus({ row }: { row: Extract<TimelineRow, { kind:
   );
 }
 
+function UserMessageUnsentStatus({ messageId }: { messageId: MessageId }) {
+  const ctx = use(TimelineRowCtx);
+  const environmentId = ctx.activeThreadEnvironmentId;
+  const [blocked, setBlocked] = useState(false);
+  return (
+    <div
+      className="me-1 flex items-center gap-1.5 text-xs text-muted-foreground"
+      data-unsent-message
+    >
+      {blocked ? "Not sent. Make room for its attachments in the composer first." : "Not sent"}
+      <InlineButton
+        tone="muted"
+        onClick={() => setBlocked(!unsentMessages.restore(environmentId, messageId))}
+      >
+        Restore to composer
+      </InlineButton>
+      <InlineButton tone="muted" onClick={() => unsentMessages.discard(environmentId, messageId)}>
+        Discard
+      </InlineButton>
+    </div>
+  );
+}
+
+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 <UserMessageUnsentStatus messageId={messageId} />;
+  if (delivery === null || delivery.state === "delivered") return null;
+  return (
+    <div
+      className="me-1 flex items-center gap-1.5 text-xs text-muted-foreground"
+      data-turn-delivery={delivery.state}
+    >
+      {delivery.state === "waiting" ? "Waiting to send" : "Sending"}
+      {delivery.state === "waiting" ? (
+        <InlineButton tone="muted" onClick={() => turnDeliveries.cancel(delivery.key)}>
+          Cancel
+        </InlineButton>
+      ) : null}
+    </div>
+  );
+}
+
 function UserTimelineRow({ row }: { row: Extract<TimelineRow, { kind: "message" }> }) {
   const ctx = use(TimelineRowCtx);
   const { onImageExpand, onFileOpen } = ctx;
@@ -711,6 +757,7 @@ function UserTimelineRow({ row }: { row: Extract<TimelineRow, { kind: "message"
         ) : null}
       </div>
       <UserMessageProjectedStatus row={row} />
+      <UserMessageDeliveryStatus messageId={row.message.id} />
       {!folded || expanded ? (
         <div className="flex w-full max-w-[80%] items-center justify-end pe-1 text-xs tabular-nums opacity-0 transition-opacity duration-200 pointer-coarse:opacity-100 focus-within:opacity-100 group-hover:opacity-100">
           <div className="flex shrink-0 items-center gap-2">
diff --git a/apps/web/src/components/chat/chatViewSend.test.ts b/apps/web/src/components/chat/chatViewSend.test.ts
new file mode 100644
index 0000000000000..dc228aaea4546
--- /dev/null
+++ b/apps/web/src/components/chat/chatViewSend.test.ts
@@ -0,0 +1,176 @@
+import {
+  EnvironmentId,
+  OrchestrationDispatchCommandError,
+  PROVIDER_SEND_TURN_MAX_ATTACHMENTS,
+  ProjectId,
+  ThreadId,
+} from "@t3tools/contracts";
+import * as Cause from "effect/Cause";
+import { AsyncResult } from "effect/reactivity";
+import { expect, it } from "vite-plus/test";
+
+import { DraftId, type DraftThreadState, useComposerDraftStore } from "../../composerDraftStore";
+import {
+  environmentContextRecord,
+  environmentContextReference,
+  terminalContextReference,
+} from "../../lib/composerContextRecords";
+import { formatInlineContextReference } from "../../lib/composerContextReferences";
+import type { TerminalContextDraft } from "../../lib/terminalContext";
+import { snapshotComposerSend } from "../ChatView.logic";
+import { reopenUnsentDraftThread, restoreComposerDraft } from "./chatViewSend";
+
+const target = { environmentId: EnvironmentId.make("environment"), threadId: ThreadId.make("t") };
+
+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" }),
+  };
+}
+
+it("restores an unsent message after a newer draft without replacing it", () => {
+  useComposerDraftStore.setState({ draftsByThreadKey: {}, draftThreadsByThreadKey: {} });
+  const store = useComposerDraftStore.getState();
+  store.setPrompt(target, "Newer draft");
+  store.addImages(target, [image("newer")]);
+  const newerEnvironment = environmentContextRecord(EnvironmentId.make("newer-host"), "Newer host");
+  const unsentEnvironment = environmentContextRecord(
+    EnvironmentId.make("unsent-host"),
+    "Unsent host",
+  );
+  store.setEnvironmentContexts(target, [newerEnvironment]);
+  const newerPrompt = store.getComposerDraft(target)?.prompt;
+  const unsent = snapshotComposerSend({
+    prompt: `Unsent message ${formatInlineContextReference(environmentContextReference(unsentEnvironment))}`,
+    images: [image("unsent")],
+    files: [],
+    terminalContexts: [],
+    sendableTerminalContexts: [],
+    previewAnnotations: [],
+    reviewComments: [],
+    threadContexts: [],
+    environmentContexts: [unsentEnvironment],
+  });
+
+  restoreComposerDraft(target, unsent);
+
+  const draft = store.getComposerDraft(target);
+  expect(draft?.environmentContexts).toEqual([newerEnvironment, unsentEnvironment]);
+  expect(draft?.prompt.trim()).toBe(`${newerPrompt?.trimEnd()}\n\n${unsent.messageText}`.trim());
+  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: [],
+    environmentContexts: [],
+  });
+
+  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"]);
+});
+
+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: [],
+    environmentContexts: [],
+  });
+
+  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)}`);
+});
+
+it("gives a deleted launch's restored draft a fresh thread id and keeps it for other rejections", () => {
+  const draftId = DraftId.make("draft");
+  const draftThread: DraftThreadState = {
+    threadId: target.threadId,
+    environmentId: target.environmentId,
+    projectId: ProjectId.make("project"),
+    logicalProjectKey: "project",
+    createdAt: "2026-10-25T00:00:00.000Z",
+    runtimeMode: "full-access",
+    interactionMode: "default",
+    branch: null,
+    worktreePath: null,
+    envMode: "local",
+    startFromOrigin: false,
+    promotedTo: target,
+  };
+  const reject = (bootstrapThreadDisposition?: "deleted") =>
+    AsyncResult.failure(
+      Cause.fail(
+        new OrchestrationDispatchCommandError({
+          message: "The thread was deleted.",
+          ...(bootstrapThreadDisposition ? { bootstrapThreadDisposition } : {}),
+        }),
+      ),
+    );
+  useComposerDraftStore.setState({ draftsByThreadKey: {}, draftThreadsByThreadKey: {} });
+
+  const deleted = reopenUnsentDraftThread(draftId, draftThread, target, reject("deleted"));
+
+  expect(deleted.threadId).not.toBe(target.threadId);
+  expect(useComposerDraftStore.getState().getDraftSession(draftId)).toMatchObject({
+    threadId: deleted.threadId,
+    promotedTo: null,
+  });
+
+  const rejected = reopenUnsentDraftThread(draftId, draftThread, target, reject());
+
+  expect(rejected).toEqual(target);
+  expect(useComposerDraftStore.getState().getDraftSession(draftId)?.threadId).toBe(target.threadId);
+});
diff --git a/apps/web/src/components/chat/chatViewSend.ts b/apps/web/src/components/chat/chatViewSend.ts
index 81a00823a98be..c9a9e33882603 100644
--- a/apps/web/src/components/chat/chatViewSend.ts
+++ b/apps/web/src/components/chat/chatViewSend.ts
@@ -16,6 +16,8 @@ import {
   squashAtomCommandFailure,
   type AtomCommandResult,
 } from "@t3tools/client-runtime/state/runtime";
+import type { TurnDeliveryFailure } from "@t3tools/client-runtime/state/turn-delivery";
+import { scopeThreadRef } from "@t3tools/client-runtime/environment";
 import { readLocalApi } from "../../localApi";
 import { getUploadedAttachments } from "../../lib/attachmentUploadQueue";
 import { findRevertCheckpoint } from "../../session-logic";
@@ -26,6 +28,9 @@ import { newThreadId } from "~/lib/utils";
 import {
   type ComposerFileAttachment,
   type ComposerImageAttachment,
+  type DraftThreadState,
+  composerDraftHasUserContent,
+  restoreFailedDraftThread,
   useComposerDraftStore,
   DraftId,
 } from "../../composerDraftStore";
@@ -252,25 +257,78 @@ 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.
+ * Reopens an unsent launch's draft and returns where its message restores. A cancelled launch
+ * (`failure` null) may have committed, and a deleted one is tombstoned, so either gets a fresh
+ * thread id. The returned ref still resolves after that draft is sent and promoted.
+ */
+export function reopenUnsentDraftThread(
+  draftId: DraftId,
+  draftThread: DraftThreadState,
+  launchRef: ScopedThreadRef,
+  failure: TurnDeliveryFailure | null,
+): ScopedThreadRef {
+  const threadRef =
+    failure === null || wasBootstrapThreadDeleted(squashAtomCommandFailure(failure))
+      ? scopeThreadRef(launchRef.environmentId, newThreadId())
+      : launchRef;
+  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.
+ * Returns false, changing nothing, when its attachments would not fit beside the newer draft.
  */
 export function restoreComposerDraft(
   draftTarget: ScopedThreadRef | DraftId,
   snapshot: ComposerSendSnapshot,
-): ComposerImageAttachment[] {
+): boolean {
   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;
+  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
+      ? 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,
+  ]);
+  return true;
 }
 
 export interface MultipleModelTarget {
diff --git a/apps/web/src/components/chat/useComposerSend.ts b/apps/web/src/components/chat/useComposerSend.ts
index a5fa30ffe15a9..689b46ff05a81 100644
--- a/apps/web/src/components/chat/useComposerSend.ts
+++ b/apps/web/src/components/chat/useComposerSend.ts
@@ -2,31 +2,33 @@ 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 {
   AuthOrchestrationOperateScope,
+  CommandId,
   DEFAULT_MODEL,
   type EnvironmentId,
   type ModelSelection,
   type PreviewAnnotationPayload,
+  PROVIDER_SEND_TURN_MAX_ATTACHMENTS,
   type ThreadId,
   ProviderInteractionMode,
   RuntimeMode,
 } from "@t3tools/contracts";
 import { deriveThreadTitleSeed } from "@t3tools/client-runtime/operations";
-import { wasBootstrapThreadDeleted } from "@t3tools/client-runtime/errors";
 import {
   declinePendingApprovals,
   sendApprovalFeedback,
@@ -59,7 +61,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, randomUUID } from "~/lib/utils";
 import { useNewThreadHandler } from "../../hooks/useHandleNewThread";
 import { buildThreadRouteParams } from "../../threadRoutes";
 import {
@@ -67,7 +69,6 @@ import {
   clearBackgroundDraftSubmissionByRef,
   finalizePromotedDraftThreadByRef,
   markPromotedDraftThreadByRef,
-  restoreFailedDraftThread,
   composerDraftHasUserContent,
   type ComposerFileAttachment,
   type ComposerImageAttachment,
@@ -83,6 +84,9 @@ import {
 } from "../../lib/composerContextRecords";
 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";
 import { environmentPresentations } from "../../state/presentation";
@@ -117,6 +121,7 @@ import {
   type MultipleModelTarget,
   resolveMultipleModelTargets,
   prepareTurnAttachments,
+  reopenUnsentDraftThread,
   restoreComposerDraft,
 } from "./chatViewSend";
 import type {
@@ -336,17 +341,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,
@@ -391,6 +385,31 @@ 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.
+   * Returns false, changing nothing, when its attachments do not fit there.
+   */
+  const restoreUnsentComposer = (
+    draftTarget: typeof composerDraftTarget,
+    sentRouteThreadKey: string,
+    snapshot: ComposerSendSnapshot,
+  ): boolean => {
+    if (!restoreComposerDraft(draftTarget, snapshot)) return false;
+    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,
+      });
+    }
+    return true;
+  };
+
   const onSend = async (
     e?: { preventDefault: () => void },
     dispatchMode: ComposerDispatchMode = "auto",
@@ -677,12 +696,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();
@@ -696,32 +722,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
@@ -1154,6 +1158,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) {
@@ -1189,7 +1205,7 @@ export function useComposerSend(input: ComposerSendInput) {
       ctxSelectedModelSelection.options,
     );
 
-    let failure: AtomCommandResult<unknown, unknown> | null = null;
+    let failure: TurnDeliveryFailure | null = null;
 
     if (failure === null && isServerThread) {
       const settingsResult = await persistThreadSettingsForNextTurn({
@@ -1242,6 +1258,86 @@ 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) => {
+      let restoredThreadId = threadIdForSend;
+      let restoreTarget = composerDraftTarget;
+      if (isLocalDraftThread && draftId && draftThread) {
+        restoreTarget = reopenUnsentDraftThread(
+          draftId,
+          draftThread,
+          scopeThreadRef(environmentId, threadIdForSend),
+          failure,
+        );
+        restoredThreadId = restoreTarget.threadId;
+        clearBackgroundDraftSubmissionByRef(scopeThreadRef(environmentId, threadIdForSend));
+      }
+      setOptimisticUserMessages((existing) => {
+        const next = existing.filter((message) => message.id !== messageIdForSend);
+        return next.length === existing.length ? existing : next;
+      });
+      optimisticBlobPreviews.release(optimisticMessage);
+      const restore = () => restoreUnsentComposer(restoreTarget, 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(
+          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 +1373,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 +1398,8 @@ export function useComposerSend(input: ComposerSendInput) {
           ...(bootstrap ? { bootstrap } : {}),
           createdAt: messageCreatedAt,
         },
-      });
+        detachedOptimisticMessage,
+      );
       if (backgroundThreadRef) {
         try {
           backgroundDraftOpened = Boolean(
@@ -1324,25 +1423,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 +1474,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/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<typeof buildMessageContext>;
     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<unknown, unknown> | 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/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<ComposerDraftStoreState>()(
             return;
           }
           const normalizedContexts = normalizeTerminalContextsForThread(threadId, contexts);
+          // A context deduplicated onto an earlier copy of its selection hands its references over.
+          const survivorByReferenceId = new Map<string, TerminalContextDraft>();
+          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<ComposerDraftStoreState>()(
                         prompt,
                         terminalContextReference(context).contextId,
                       ).prompt,
-                    existing.prompt,
+                    remappedPrompt,
                   ),
                 normalizedContexts.map(terminalContextReference),
               ),
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..5ebd01915f600
--- /dev/null
+++ b/apps/web/src/state/turnDeliveries.ts
@@ -0,0 +1,66 @@
+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 { 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";
+
+/** Outlives ChatView, so a send waiting on a reconnect survives navigating between threads. */
+export const turnDeliveries = createTurnDeliveries<unknown, unknown, ChatMessage>({
+  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 },
+    ),
+  observeShell: (environmentId, listener) =>
+    appAtomRegistry.subscribe(environmentShell.stateValueAtom(environmentId), listener, {
+      immediate: true,
+    }),
+  releaseDisplay: revokeUserMessagePreviewUrls,
+});
+
+const EMPTY: ReadonlyArray<PendingTurnDelivery<ChatMessage>> = [];
+
+function usePendingTurnDeliverySnapshot() {
+  return useSyncExternalStore(turnDeliveries.subscribe, turnDeliveries.getSnapshot);
+}
+
+export function usePendingTurnDeliveries(
+  environmentId: EnvironmentId,
+  threadId: ThreadId | null,
+): ReadonlyArray<PendingTurnDelivery<ChatMessage>> {
+  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<ChatMessage> | null {
+  const snapshot = usePendingTurnDeliverySnapshot();
+  return (
+    snapshot.find(
+      (delivery) => delivery.environmentId === environmentId && delivery.messageId === messageId,
+    ) ?? null
+  );
+}
diff --git a/apps/web/src/state/unsentMessages.test.ts b/apps/web/src/state/unsentMessages.test.ts
new file mode 100644
index 0000000000000..819168f206308
--- /dev/null
+++ b/apps/web/src/state/unsentMessages.test.ts
@@ -0,0 +1,161 @@
+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 {
+  DraftId,
+  finalizePromotedDraftThreadByRef,
+  markPromotedDraftThreadByRef,
+  useComposerDraftStore,
+} 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";
+
+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 newerImages() {
+  return Array.from({ length: PROVIDER_SEND_TURN_MAX_ATTACHMENTS }, (_, index) =>
+    image(`newer-${index}`),
+  );
+}
+
+function unsentSnapshot() {
+  return snapshotComposerSend({
+    prompt: "Unsent message",
+    images: [image("unsent")],
+    files: [],
+    terminalContexts: [],
+    sendableTerminalContexts: [],
+    previewAnnotations: [],
+    reviewComments: [],
+    threadContexts: [],
+    environmentContexts: [],
+  });
+}
+
+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 });
+  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");
+});
+
+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 launchedRef = scopeThreadRef(environmentId, launched.threadId);
+  const reopenedRef = reopenUnsentDraftThread(draftId, launched, launchedRef, null);
+  expect(reopenedRef.threadId).not.toBe(launched.threadId);
+  const snapshot = unsentSnapshot();
+  const restore = () => restoreComposerDraft(reopenedRef, 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"]);
+});
+
+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
new file mode 100644
index 0000000000000..b59a2c1c54eb5
--- /dev/null
+++ b/apps/web/src/state/unsentMessages.ts
@@ -0,0 +1,82 @@
+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<UnsentMessage> = [];
+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);
+    // 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 => {
+    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<UnsentMessage> => records,
+  subscribe: (listener: () => void) => {
+    listeners.add(listener);
+    return () => {
+      listeners.delete(listener);
+    };
+  },
+};
+
+const EMPTY: ReadonlyArray<UnsentMessage> = [];
+
+function useUnsentMessageSnapshot() {
+  return useSyncExternalStore(unsentMessages.subscribe, unsentMessages.getSnapshot);
+}
+
+export function useUnsentMessages(
+  environmentId: EnvironmentId,
+  threadId: ThreadId | null,
+): ReadonlyArray<UnsentMessage> {
+  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/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<W extends { readonly environmentId: EnvironmentId }, A, E>(
-  command: AtomCommand<W, A, E>,
-  options?: string | AtomCommandOptions,
-): (value: W) => Promise<AtomCommandResult<A, E | EnvironmentAuthorizationError>> {
-  const run = useAtomCommand(command, options);
-  return async (value: W): Promise<AtomCommandResult<A, E | EnvironmentAuthorizationError>> => {
-    if (!readEnvironmentScope(value.environmentId, AuthOrchestrationOperateScope)) {
-      return AsyncResult.failure(
+function operateDenied<A>(
+  environmentId: EnvironmentId,
+): AtomCommandResult<A, EnvironmentAuthorizationError> | null {
+  return readEnvironmentScope(environmentId, AuthOrchestrationOperateScope)
+    ? null
+    : AsyncResult.failure(
         Cause.fail(
           new EnvironmentAuthorizationError({
             requiredScope: AuthOrchestrationOperateScope,
@@ -31,7 +32,29 @@ export function useOrchestrationCommand<W extends { readonly environmentId: Envi
           }),
         ),
       );
-    }
-    return run(value);
-  };
+}
+
+export function useOrchestrationCommand<W extends { readonly environmentId: EnvironmentId }, A, E>(
+  command: AtomCommand<W, A, E>,
+  options?: string | AtomCommandOptions,
+): (value: W) => Promise<AtomCommandResult<A, E | EnvironmentAuthorizationError>> {
+  const run = useAtomCommand(command, options);
+  return async (value: W): Promise<AtomCommandResult<A, E | EnvironmentAuthorizationError>> =>
+    operateDenied<A>(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<W, A, E>,
+  value: W,
+  options?: AtomCommandOptions,
+): Promise<AtomCommandResult<A, E | EnvironmentAuthorizationError>> {
+  return (
+    operateDenied<A>(value.environmentId) ??
+    runAtomCommand(appAtomRegistry, command, value, options)
+  );
 }
diff --git a/docs/user/composer.md b/docs/user/composer.md
index 714572dfd5005..34b4fbe057ce0 100644
--- a/docs/user/composer.md
+++ b/docs/user/composer.md
@@ -85,6 +85,13 @@ 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 puts the message back in the composer, after anything you typed since. It
+can't unsend a message the server already received. A queued message waits in the queue instead,
+where its remove control cancels it. 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
 
 On web and desktop, use Settings → Providers → **Models** to add an unlisted model with a custom
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..e215043673140 100644
--- a/packages/client-runtime/src/state/threadCommands.test.ts
+++ b/packages/client-runtime/src/state/threadCommands.test.ts
@@ -28,8 +28,11 @@ import { Atom, AtomRegistry } from "effect/reactivity";
 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";
+import { createTurnDeliveries } from "./turnDelivery.ts";
 
 const ENVIRONMENT_ID = EnvironmentId.make("remote");
 const THREAD_ID = ThreadId.make("thread");
@@ -507,3 +510,101 @@ 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<unknown, unknown, string>({
+        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);
+        },
+        observeShell: (_environmentId, listener) => {
+          listener(EMPTY_SHELL_STATE);
+          return () => {};
+        },
+        releaseDisplay: () => {},
+      });
+      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([
+        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
new file mode 100644
index 0000000000000..ff829f582a346
--- /dev/null
+++ b/packages/client-runtime/src/state/turnDelivery.test.ts
@@ -0,0 +1,410 @@
+import {
+  CommandId,
+  EnvironmentId,
+  MessageId,
+  ProjectId,
+  ProviderInstanceId,
+  ThreadId,
+  type OrchestrationV2ShellSnapshot,
+  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 "vite-plus/test";
+
+import { wasBootstrapThreadDeleted } from "../errors/orchestration.ts";
+import { type AtomCommandResult, squashAtomCommandFailure } 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";
+
+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 shell(
+  status: EnvironmentShellState["status"],
+  threadIds: ReadonlyArray<ThreadId> = [],
+): 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<Pick<OrchestrationV2ThreadProjection, "messages" | "runs">> = {},
+): EnvironmentThreadState {
+  return {
+    ...EMPTY_ENVIRONMENT_THREAD_STATE,
+    status,
+    data: Option.some({
+      messages: [],
+      runs: [],
+      turnItems: [],
+      ...projection,
+    } as unknown as OrchestrationV2ThreadProjection),
+  };
+}
+
+function setup(
+  results: ReadonlyArray<SendResult | Promise<SendResult>>,
+  initialThread = thread("live"),
+) {
+  const queue = [...results];
+  const sent: Array<TurnDeliveryInput> = [];
+  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<string> = [];
+  const deliveries = createTurnDeliveries<{ sequence: number }, unknown, string>({
+    send: async (_environmentId, input) => {
+      sent.push(input);
+      return queue.shift() ?? ACCEPTED;
+    },
+    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;
+      };
+    },
+    releaseDisplay: (display) => {
+      released.push(display);
+    },
+  });
+  return {
+    deliveries,
+    sent,
+    released,
+    threadObservations: () => threadObservations,
+    activeObservers: () => activeObservers,
+    displaysMade: () => displaysMade,
+    makeDisplay: () => {
+      displaysMade += 1;
+      return "optimistic row";
+    },
+    emitThread: (state: EnvironmentThreadState) => threadListener?.(state),
+    emitShell: (state: EnvironmentShellState) => shellListener?.(state),
+  };
+}
+
+async function deliverUntilWaiting(harness: ReturnType<typeof setup>, input = INPUT) {
+  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.
+  return { settled: 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]);
+  });
+
+  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 () => {
+    const harness = setup([LOST_ACK]);
+    const { settled } = await deliverUntilWaiting(harness);
+
+    harness.emitThread(thread("synchronizing"));
+    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 () => {
+    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([]);
+    expect(harness.released).toEqual(["optimistic row"]);
+  });
+
+  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([]);
+    expect(harness.released).toEqual(["optimistic row"]);
+  });
+
+  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<SendResult>((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, harness.makeDisplay);
+
+    expect(outcome).toEqual({ type: "rejected", failure: REJECTED });
+    expect(harness.deliveries.getSnapshot()).toEqual([]);
+    expect(harness.displaysMade()).toBe(0);
+  });
+
+  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.emitShell(shell("live"));
+    expect(harness.sent).toHaveLength(1);
+    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([]);
+  });
+
+  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]));
+    // The shell proves only the thread.create; the first message may still be missing.
+    expect(harness.deliveries.getSnapshot()).toEqual([
+      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"));
+
+    const end = await settled;
+    // The composer gives a restored draft a fresh thread id only for this classification.
+    expect(
+      end.type === "rejected" && wasBootstrapThreadDeleted(squashAtomCommandFailure(end.failure)),
+    ).toBe(true);
+    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);
+
+    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
new file mode 100644
index 0000000000000..5db2d062d299d
--- /dev/null
+++ b/packages/client-runtime/src/state/turnDelivery.ts
@@ -0,0 +1,330 @@
+import {
+  type CommandId,
+  type EnvironmentId,
+  type MessageId,
+  OrchestrationDispatchCommandError,
+  type ThreadId,
+} from "@t3tools/contracts";
+import * as Cause from "effect/Cause";
+import * as Option from "effect/Option";
+import { AsyncResult } from "effect/reactivity";
+
+import { isRetryableDeliveryFailure } from "../errors/transport.ts";
+import type { StartThreadTurnInput } from "../operations/commands.ts";
+import {
+  type AtomCommandResult,
+  isAtomCommandInterrupted,
+  squashAtomCommandFailure,
+} from "./runtime.ts";
+import type { EnvironmentShellState } from "./shell.ts";
+import type { EnvironmentThreadState } from "./threadState.ts";
+
+export type TurnDeliveryInput = StartThreadTurnInput & { readonly commandId: CommandId };
+
+export type TurnDeliveryFailure = Extract<AtomCommandResult<unknown, unknown>, { _tag: "Failure" }>;
+
+type TurnDeliveryEnd<A> =
+  /** `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" };
+
+type TurnDeliveryOutcome<A> =
+  | TurnDeliveryEnd<A>
+  /** The server may have committed it; `settled` ends after a later resync or cancel. */
+  | { readonly type: "waiting"; readonly settled: Promise<TurnDeliveryEnd<A>> };
+
+export interface PendingTurnDelivery<M> {
+  readonly key: string;
+  readonly environmentId: EnvironmentId;
+  readonly threadId: ThreadId;
+  readonly messageId: MessageId;
+  /** `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<A, E, M> {
+  readonly send: (
+    environmentId: EnvironmentId,
+    input: TurnDeliveryInput,
+  ) => Promise<AtomCommandResult<A, E>>;
+  /** 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 observeShell: (
+    environmentId: EnvironmentId,
+    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<A, M> {
+  readonly key: string;
+  readonly environmentId: EnvironmentId;
+  readonly input: TurnDeliveryInput;
+  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;
+  /** 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. */
+  resynced: boolean;
+  thread: EnvironmentThreadState | null;
+  shell: EnvironmentShellState | null;
+  settle: (end: TurnDeliveryEnd<A>) => void;
+  stopObserving: Array<() => void>;
+}
+
+function turnDeliveryKey(
+  environmentId: EnvironmentId,
+  threadId: ThreadId,
+  messageId: MessageId,
+): string {
+  return JSON.stringify([environmentId, threadId, messageId]);
+}
+
+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,
+      ),
+  });
+}
+
+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<A, E, M>(deps: TurnDeliveryDependencies<A, E, M>) {
+  const entries = new Map<string, Entry<A, M>>();
+  const listeners = new Set<() => void>();
+  let snapshot: ReadonlyArray<PendingTurnDelivery<M>> = [];
+
+  const publish = () => {
+    snapshot = [...entries.values()].flatMap((entry) =>
+      entry.display === null
+        ? []
+        : [
+            {
+              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 remove = (entry: Entry<A, M>) => {
+    if (entries.get(entry.key) !== entry) return;
+    entries.delete(entry.key);
+    for (const stop of entry.stopObserving) stop();
+    entry.stopObserving = [];
+    if (entry.display === null) return;
+    publish();
+    deps.releaseDisplay(entry.display);
+  };
+
+  /** The first callback runs before `stop` exists and may already have removed the entry. */
+  const keepObserving = (entry: Entry<A, M>, stop: () => void) => {
+    if (entries.get(entry.key) === entry) entry.stopObserving.push(stop);
+    else stop();
+  };
+
+  const observeThread = (entry: Entry<A, M>) => {
+    keepObserving(
+      entry,
+      deps.observeThread(entry.environmentId, entry.input.threadId, (state) => {
+        entry.thread = state;
+        if (state.status !== "live") entry.resynced = true;
+        evaluate(entry);
+      }),
+    );
+  };
+
+  const finish = (entry: Entry<A, M>, end: TurnDeliveryEnd<A>) => {
+    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;
+    }
+    evaluate(entry);
+  };
+
+  const evaluate = (entry: Entry<A, M>) => {
+    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)) {
+        // Its pending uploads may have expired, so wait for the message instead of replaying.
+        entry.unborn = false;
+        entry.replayBlocked = true;
+        observeThread(entry);
+        return;
+      }
+    } else {
+      if (entry.thread?.status === "deleted") {
+        finish(entry, {
+          type: "rejected",
+          failure: AsyncResult.failure(
+            Cause.fail(
+              new OrchestrationDispatchCommandError({
+                message: "The thread was deleted.",
+                ...(entry.input.bootstrap?.createThread !== undefined
+                  ? { bootstrapThreadDisposition: "deleted" as const }
+                  : {}),
+              }),
+            ),
+          ),
+        });
+        return;
+      }
+      if (!entry.resynced || entry.thread?.status !== "live") return;
+      if (threadHoldsUserMessage(entry.thread, messageId)) {
+        finish(entry, { type: "delivered", value: null });
+        return;
+      }
+      if (entry.replayBlocked) return;
+    }
+    void dispatch(entry).then((end) => {
+      if (end !== null) finish(entry, end);
+    });
+  };
+
+  const dispatch = async (entry: Entry<A, M>): Promise<TurnDeliveryEnd<A> | null> => {
+    entry.state = "sending";
+    entry.resynced = false;
+    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) ||
+      isRetryableDeliveryFailure(squashAtomCommandFailure(result));
+    if (entry.cancelRequested) return { type: "cancelled" };
+    if (!retryable) return { type: "rejected", failure: result };
+    entry.state = "waiting";
+    entry.display ??= entry.createDisplay();
+    publish();
+    evaluate(entry);
+    return null;
+  };
+
+  const deliver = async (
+    environmentId: EnvironmentId,
+    input: TurnDeliveryInput,
+    createDisplay: () => M,
+  ): Promise<TurnDeliveryOutcome<A>> => {
+    const key = turnDeliveryKey(environmentId, input.threadId, input.message.messageId);
+    let settle!: (end: TurnDeliveryEnd<A>) => void;
+    const settled = new Promise<TurnDeliveryEnd<A>>((resolve) => {
+      settle = resolve;
+    });
+    const entry: Entry<A, M> = {
+      key,
+      environmentId,
+      input,
+      createDisplay,
+      display: null,
+      unborn: input.bootstrap?.createThread !== undefined,
+      replayBlocked: false,
+      state: "sending",
+      cancelRequested: false,
+      resynced: false,
+      thread: null,
+      shell: null,
+      settle,
+      stopObserving: [],
+    };
+    const previous = entries.get(key);
+    if (previous !== undefined) {
+      previous.settle({ type: "cancelled" });
+      remove(previous);
+    }
+    entries.set(key, entry);
+    if (entry.unborn) {
+      keepObserving(
+        entry,
+        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 };
+    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 || entry.state === "delivered") return;
+    if (entry.state === "waiting") {
+      finish(entry, { type: "cancelled" });
+      return;
+    }
+    entry.cancelRequested = true;
+  };
+
+  return {
+    deliver,
+    cancel,
+    getSnapshot: (): ReadonlyArray<PendingTurnDelivery<M>> => snapshot,
+    subscribe: (listener: () => void) => {
+      listeners.add(listener);
+      return () => {
+        listeners.delete(listener);
+      };
+    },
+  };
+}