1import { useAtomValue } from "@effect/atom-react";
2import {
3 threadRuntimeIsActive,
4 type EnvironmentProject,
5 type EnvironmentThreadShell,
6} from "@t3tools/client-runtime/state/shell";
7import type { AtomCommandResult } from "@t3tools/client-runtime/state/runtime";
8import { deriveThreadTitleSeed } from "@t3tools/client-runtime/operations";
9import {
10 AuthOrchestrationOperateScope,
11 CommandId,
12 DEFAULT_PROVIDER_INTERACTION_MODE,
13 DEFAULT_RUNTIME_MODE,
14 PROVIDER_SEND_TURN_MAX_ATTACHMENTS,
15 type MessageId,
16} from "@t3tools/contracts";
17import { buildTemporaryWorktreeBranchName } from "@t3tools/shared/git";
18import * as Cause from "effect/Cause";
19import { AsyncResult } from "effect/reactivity";
20import { useCallback, useEffect, useRef, useState } from "react";
21import { Alert } from "react-native";
22
23import { createDebugLogger } from "../lib/debugLog";
24import { scopedThreadKey } from "../lib/scopedEntities";
25import { buildProjectThreadStartTurnInput } from "../lib/projectThreadStartTurn";
26import { serializeComposerMessageForServer, uploadedComposerContext } from "../lib/composerContext";
27import { prepareTurnAttachments, type PreparedTurnAttachments } from "../lib/attachmentUpload";
28import { randomHex } from "../lib/uuid";
29import {
30 describeUnavailableModelSelection,
31 isModelSelectionUnavailable,
32} from "../lib/modelOptions";
33import {
34 retainAcknowledgedThreadMessage,
35 forgetAcknowledgedThreadMessage,
36} from "./acknowledged-thread-messages";
37import { appAtomRegistry } from "./atom-registry";
38import { restoredNewTaskDraftKey } from "./new-task-draft-key";
39import { useProjects, useServerConfigs, useThreadShells } from "./entities";
40import {
41 clearPendingThreadCreationOutcome,
42 pendingThreadCreationOutcomesAtom,
43 recordPendingThreadCreationOutcome,
44} from "./pending-thread-creation";
45import { serverEnvironment } from "./server";
46import { gitEnvironment } from "./git";
47import { readEnvironmentScope, useEnvironmentsWithScope } from "./session";
48import {
49 confirmThreadOutboxMessageQueued,
50 threadOutboxManager,
51 threadOutboxRevision,
52 updateThreadOutboxMessage,
53} from "./thread-outbox";
54import { removeThreadOutboxMessage } from "./thread-outbox-removal";
55import {
56 isQueuedThreadCreationSendable,
57 resolveThreadOutboxDeliveryAction,
58 resolveThreadOutboxDispatchStep,
59 resolveThreadOutboxFailureAction,
60 resolveQueuedThreadSettings,
61 shouldRetryThreadOutboxDelivery,
62 threadOutboxRetryDelayMs,
63 type QueuedThreadCreation,
64 type QueuedThreadMessage,
65 type ThreadOutboxCommandStage,
66 type ThreadOutboxFailureAction,
67} from "./thread-outbox-model";
68import { environmentThreadShells, threadEnvironment } from "./threads";
69import {
70 appendComposerDraftAttachments,
71 composerDraftsAtom,
72 flushComposerDrafts,
73 type ComposerDraft,
74 getComposerDraftSnapshot,
75 mergeComposerDraftContent,
76 replaceComposerDraftAttachments,
77 removeDeliveredCloudQueuedMessage,
78 undoComposerDraftMerge,
79 updateComposerDraftSettings,
80 waitForComposerDraftsLoaded,
81} from "./use-composer-drafts";
82import { useAtomCommand } from "./use-atom-command";
83import {
84 dispatchingQueuedMessageIdAtom,
85 editingQueuedMessageIdsAtom,
86 useThreadOutboxMessages,
87 useThreadOutboxShellStatuses,
88} from "./use-thread-outbox";
89import { clearThreadComposerError, setThreadComposerError } from "./thread-composer-error";
90import { useRemoteConnectionStatus } from "./use-remote-environment-registry";
91
92// Ordinary offline behavior (a socket dropping mid-request, a retryable
93// attachment upload failure) must not spam `console.warn` on every backoff
94// retry; it goes to the filterable `[t3-thread-outbox]` debug log instead.
95// Failures the server decided stay on `console.warn`.
96const threadOutboxDebug = createDebugLogger("thread-outbox");
97
98/**
99 * On the queued-request path (settings sync, startTurn) the RPC client
100 * reports ordinary transport drops as the raw socket/worker reason tags, and
101 * reserves `RpcClientDefect` for client-side protocol violations and decoding
102 * failures — unlike the shared config-subscription stream, which
103 * deliberately re-wraps transport causes under that tag. Defects still retry,
104 * but they are not ordinary offline behavior and must not hide behind the
105 * offline debug log.
106 */
107function isRpcClientDecodeDefect(error: unknown): boolean {
108 if (
109 typeof error !== "object" ||
110 error === null ||
111 !("_tag" in error) ||
112 error._tag !== "RpcClientError"
113 ) {
114 return false;
115 }
116 const reason: unknown = (error as { readonly reason?: unknown }).reason;
117 return (
118 typeof reason === "object" &&
119 reason !== null &&
120 "_tag" in reason &&
121 reason._tag === "RpcClientDefect"
122 );
123}
124
125function isOrdinaryThreadOutboxTransportFailure(error: unknown): boolean {
126 return shouldRetryThreadOutboxDelivery(error) && !isRpcClientDecodeDefect(error);
127}
128
129/**
130 * Logs one queued-message delivery failure and returns the retry-or-restore
131 * decision for the caller. Ordinary transport retries — what an offline
132 * device or a flapping socket produces on every backoff attempt — go to the
133 * debug log. Server-decided failures warn. Settings-sync failures always
134 * resolve to a retry even when the server rejected the command, so the
135 * error, not the resolved action, must decide the log level there; routing
136 * every retry to debug could hide a permanently rejected update forever.
137 */
138function logThreadOutboxDeliveryFailure(input: {
139 readonly stage: ThreadOutboxCommandStage;
140 readonly error: unknown;
141 readonly interrupted: boolean;
142 readonly context: Record<string, unknown>;
143}): ThreadOutboxFailureAction {
144 const action = resolveThreadOutboxFailureAction({
145 stage: input.stage,
146 error: input.error,
147 interrupted: input.interrupted,
148 });
149 const details = { ...input.context, stage: input.stage, action };
150 const ordinaryTransportRetry =
151 action === "retry" &&
152 !isRpcClientDecodeDefect(input.error) &&
153 (input.interrupted ||
154 input.stage !== "settings-sync" ||
155 shouldRetryThreadOutboxDelivery(input.error));
156 if (ordinaryTransportRetry) {
157 threadOutboxDebug.log("queued message delivery failed", details);
158 } else {
159 console.warn("[thread-outbox] queued message delivery failed", details);
160 }
161 return action;
162}
163
164/** Attachment uploads retry like delivery: transport failures are ordinary offline noise. */
165function logThreadOutboxUploadFailure(queuedMessage: QueuedThreadMessage, error: unknown): void {
166 const context = {
167 environmentId: queuedMessage.environmentId,
168 threadId: queuedMessage.threadId,
169 messageId: queuedMessage.messageId,
170 };
171 if (isOrdinaryThreadOutboxTransportFailure(error)) {
172 threadOutboxDebug.log("attachment upload failed; retrying", { ...context, error });
173 } else {
174 console.warn("[thread-outbox] failed to upload attachments", { ...context, error });
175 }
176}
177
178function beginDispatchingQueuedMessage(queuedMessageId: MessageId): void {
179 appAtomRegistry.set(dispatchingQueuedMessageIdAtom, queuedMessageId);
180}
181
182function finishDispatchingQueuedMessage(queuedMessageId: MessageId): void {
183 const current = appAtomRegistry.get(dispatchingQueuedMessageIdAtom);
184 appAtomRegistry.set(dispatchingQueuedMessageIdAtom, current === queuedMessageId ? null : current);
185}
186
187function findThread(
188 threads: ReadonlyArray<EnvironmentThreadShell>,
189 message: QueuedThreadMessage,
190): EnvironmentThreadShell | undefined {
191 return threads.find(
192 (candidate) =>
193 candidate.environmentId === message.environmentId && candidate.id === message.threadId,
194 );
195}
196
197function findCreationProject(
198 projects: ReadonlyArray<EnvironmentProject>,
199 message: QueuedThreadMessage,
200): EnvironmentProject | undefined {
201 return projects.find(
202 (candidate) =>
203 candidate.environmentId === message.environmentId &&
204 candidate.id === message.creation?.projectId,
205 );
206}
207
208function settingsCommandId(message: QueuedThreadMessage, setting: string): CommandId {
209 return CommandId.make(`${message.commandId}:${setting}`);
210}
211
212/**
213 * Uploads a queued message's attachments and persists the uploaded ids back
214 * onto the queued message. The revision-checked update means an edit accepted
215 * while the bytes uploaded wins: this attempt abandons and the next drain pass
216 * re-reads the message.
217 * `deliveryRevision` is the revision of the payload this attempt will send,
218 * used for the delivery removal's compare-and-set.
219 */
220export async function prepareQueuedMessageAttachments(
221 queuedMessage: QueuedThreadMessage,
222 supportsImageUploads = false,
223): Promise<
224 | {
225 readonly status: "ready";
226 readonly prepared: PreparedTurnAttachments;
227 readonly persistedMessage: QueuedThreadMessage;
228 readonly deliveryRevision: number;
229 }
230 | { readonly status: "abandoned" }
231> {
232 if (!(await confirmThreadOutboxMessageQueued(queuedMessage))) {
233 return { status: "abandoned" };
234 }
235 if (!readEnvironmentScope(queuedMessage.environmentId, AuthOrchestrationOperateScope)) {
236 return { status: "abandoned" };
237 }
238 const revision = threadOutboxRevision(queuedMessage.messageId);
239 if (!isQueuedMessagePayloadCurrent(queuedMessage, revision)) {
240 return { status: "abandoned" };
241 }
242 let persistedMessage = queuedMessage;
243 let deliveryRevision = revision;
244 const result = await prepareTurnAttachments({
245 environmentId: queuedMessage.environmentId,
246 attachments: queuedMessage.attachments,
247 supportsImageUploads,
248 persistUploadedReferences: async (draftAttachments) => {
249 if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId]) {
250 return "abandon";
251 }
252 const updatedMessage = { ...queuedMessage, attachments: draftAttachments };
253 if (!(await updateThreadOutboxMessage(updatedMessage, revision))) {
254 return "abandon";
255 }
256 persistedMessage = updatedMessage;
257 deliveryRevision = revision + 1;
258 return "persisted";
259 },
260 });
261 if (
262 result.status === "abandoned" ||
263 !isQueuedMessagePayloadCurrent(persistedMessage, deliveryRevision)
264 ) {
265 return { status: "abandoned" };
266 }
267 return { status: "ready", prepared: result, persistedMessage, deliveryRevision };
268}
269
270function isQueuedMessagePayloadCurrent(
271 message: QueuedThreadMessage,
272 expectedRevision: number,
273): boolean {
274 return (
275 threadOutboxRevision(message.messageId) === expectedRevision &&
276 Object.values(appAtomRegistry.get(threadOutboxManager.queuedMessagesByThreadKeyAtom))
277 .flat()
278 .some((candidate) => candidate === message)
279 );
280}
281
282/**
283 * Removes a delivered message from the outbox. The revision and editor checks
284 * preserve a creation payload when its pending-task editor owns newer work.
285 * The outcome tells the caller whether removal completed, ownership changed,
286 * or storage cleanup failed. Exported for tests.
287 */
288export async function completeQueuedMessageDelivery(
289 queuedMessage: QueuedThreadMessage,
290 deliveryRevision: number,
291): Promise<"removed" | "edited" | "failed"> {
292 // The server took it after all: an error left by an earlier failed recovery
293 // of this same message no longer applies.
294 clearThreadComposerError(
295 scopedThreadKey(queuedMessage.environmentId, queuedMessage.threadId),
296 queuedMessage.messageId,
297 );
298 try {
299 await removeDeliveredCloudQueuedMessage(queuedMessage).catch((error) => {
300 console.warn("[thread-outbox] could not update sign-out snapshot after delivery", {
301 messageId: queuedMessage.messageId,
302 error,
303 });
304 });
305 // The editor may have taken the entry while startTurn was in flight; its
306 // unsaved edits have not bumped the revision yet, so the CAS alone would
307 // let removal win and the editor would lose them once it saves.
308 if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId]) {
309 return "edited";
310 }
311 retainAcknowledgedThreadMessage(queuedMessage);
312 // Removal also releases the message's local attachment files.
313 const removed = await removeThreadOutboxMessage(
314 queuedMessage,
315 deliveryRevision,
316 () => !appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId],
317 );
318 if (!removed) {
319 forgetAcknowledgedThreadMessage(queuedMessage);
320 // Losing the cleanup race to a user edit is an expected outcome the
321 // caller handles by keeping the newer message; it is not a warning.
322 threadOutboxDebug.log("delivered message was edited before cleanup", {
323 environmentId: queuedMessage.environmentId,
324 threadId: queuedMessage.threadId,
325 messageId: queuedMessage.messageId,
326 });
327 return "edited";
328 }
329 return "removed";
330 } catch (error) {
331 forgetAcknowledgedThreadMessage(queuedMessage);
332 console.warn("[thread-outbox] failed to remove delivered queued message", {
333 environmentId: queuedMessage.environmentId,
334 threadId: queuedMessage.threadId,
335 messageId: queuedMessage.messageId,
336 error,
337 });
338 return "failed";
339 }
340}
341
342/** Retries local cleanup for an existing-thread send acknowledged in this drain lifetime. */
343export async function removeAcknowledgedExistingThreadMessage(
344 queuedMessage: QueuedThreadMessage,
345 acknowledgedMessageIds: Set<MessageId>,
346): Promise<boolean> {
347 try {
348 await removeDeliveredCloudQueuedMessage(queuedMessage).catch((error) => {
349 console.warn("[thread-outbox] could not update sign-out snapshot after delivery", {
350 messageId: queuedMessage.messageId,
351 error,
352 });
353 });
354 const removed = await removeThreadOutboxMessage(queuedMessage);
355 if (removed) {
356 acknowledgedMessageIds.delete(queuedMessage.messageId);
357 }
358 return removed;
359 } catch (error) {
360 console.warn("[thread-outbox] failed to remove acknowledged queued message", {
361 environmentId: queuedMessage.environmentId,
362 threadId: queuedMessage.threadId,
363 messageId: queuedMessage.messageId,
364 error,
365 });
366 return false;
367 }
368}
369
370/**
371 * A creation delivered its startTurn but an edit won the cleanup race, so the
372 * edited payload is still queued. The next drain would see the created thread
373 * and take the creation "remove" path, silently discarding the edit; hand the
374 * edited content to the new thread's composer instead and remove the entry.
375 * Returns true when recovery is complete or an open editor owns the next
376 * action, and false when the drain should retry with backoff.
377 * Exported for tests; the drain is the only production caller.
378 */
379export async function recoverEditedCreationAfterDelivery(
380 queuedMessage: QueuedThreadMessage,
381): Promise<boolean> {
382 const kept = Object.values(appAtomRegistry.get(threadOutboxManager.queuedMessagesByThreadKeyAtom))
383 .flat()
384 .find((candidate) => candidate.messageId === queuedMessage.messageId);
385 if (!kept) {
386 return true;
387 }
388 const keptRevision = threadOutboxRevision(kept.messageId);
389 if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[kept.messageId]) {
390 return true;
391 }
392 const draftKey = scopedThreadKey(kept.environmentId, kept.threadId);
393 try {
394 // Merge before removing: the draft's reference keeps the removal sweep
395 // from deleting the attachment files. allowOverflow mirrors the
396 // send-failure restore; the send path refuses over-cap drafts, so the
397 // state stays recoverable.
398 await mergeComposerDraftContent(draftKey, {
399 text: kept.text,
400 context: kept.context,
401 attachments: [],
402 });
403 if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[kept.messageId]) {
404 return true;
405 }
406 if (threadOutboxRevision(kept.messageId) !== keptRevision) {
407 return false;
408 }
409 const existingAttachmentIds = new Set(
410 getComposerDraftSnapshot(draftKey).attachments.map((attachment) => attachment.id),
411 );
412 appendComposerDraftAttachments(
413 draftKey,
414 kept.attachments.filter((attachment) => !existingAttachmentIds.has(attachment.id)),
415 { allowOverflow: true },
416 );
417 // Only settings the queued message actually carries: spreading explicit
418 // undefined would clear choices the user already made on the draft.
419 updateComposerDraftSettings(draftKey, {
420 ...(kept.modelSelection !== undefined ? { modelSelection: kept.modelSelection } : {}),
421 ...(kept.runtimeMode !== undefined ? { runtimeMode: kept.runtimeMode } : {}),
422 ...(kept.interactionMode !== undefined ? { interactionMode: kept.interactionMode } : {}),
423 });
424 // The append only schedules a debounced write; the queue entry is the
425 // only durable copy until the draft lands, so flush before removing.
426 await flushComposerDrafts();
427 } catch (error) {
428 // Keep the entry queued. The drain retries with backoff, and the merge is
429 // idempotent so content that persisted before the failure is not repeated.
430 console.warn("[thread-outbox] could not hand an edited pending task to the composer", error);
431 return false;
432 }
433 if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[kept.messageId]) {
434 return true;
435 }
436 try {
437 return await removeThreadOutboxMessage(
438 kept,
439 keptRevision,
440 () => !appAtomRegistry.get(editingQueuedMessageIdsAtom)[kept.messageId],
441 );
442 } catch (error) {
443 console.warn("[thread-outbox] could not remove recovered pending task", error);
444 return false;
445 }
446}
447
448/** Exported for tests; the drain is the only production caller. */
449export async function restoreRejectedQueuedMessage(
450 queuedMessage: QueuedThreadMessage,
451 message: string,
452): Promise<"restored" | "deferred" | "blocked" | "retry"> {
453 const draftKey = recoveryDraftKey(queuedMessage);
454 const threadKey = scopedThreadKey(queuedMessage.environmentId, queuedMessage.threadId);
455 // Set once the merge publishes, cleared once the queued message is removed.
456 // The catch below uses it to take the merged content back out, so a retry
457 // after a mid-recovery failure cannot append the recovered text again.
458 let rollback: { readonly snapshot: ComposerDraft; readonly merged: ComposerDraft } | null = null;
459 try {
460 if (
461 appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId] ||
462 !(await confirmThreadOutboxMessageQueued(queuedMessage)) ||
463 appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId]
464 ) {
465 return "deferred";
466 }
467 // The confirmation above checked this exact payload is what is queued, so
468 // the current revision guards the removal at the end against an edit
469 // accepted while this recovery ran.
470 const revision = threadOutboxRevision(queuedMessage.messageId);
471
472 await waitForComposerDraftsLoaded();
473 if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId]) {
474 return "deferred";
475 }
476 const originalDraft = getComposerDraftSnapshot(draftKey);
477 const existingAttachmentIds = new Set(
478 originalDraft.attachments.map((attachment) => attachment.id),
479 );
480 const addedAttachmentCount = queuedMessage.attachments.filter(
481 (attachment) => !existingAttachmentIds.has(attachment.id),
482 ).length;
483 if (existingAttachmentIds.size + addedAttachmentCount > PROVIDER_SEND_TURN_MAX_ATTACHMENTS) {
484 setThreadComposerError(
485 threadKey,
486 `Remove attachments from the draft before restoring this message. Messages can contain at most ${PROVIDER_SEND_TURN_MAX_ATTACHMENTS} attachments.`,
487 queuedMessage.messageId,
488 );
489 return "blocked";
490 }
491
492 // Shown before the merge publishes the text, so a resend of that text,
493 // which can happen while this recovery still awaits persistence, clears
494 // it. Withdrawn below wherever the recovery backs out.
495 const withdrawError = () => clearThreadComposerError(threadKey, queuedMessage.messageId);
496 if (!queuedMessage.creation) {
497 setThreadComposerError(threadKey, message, queuedMessage.messageId);
498 }
499 let mergedDraft: ComposerDraft;
500 try {
501 stampRecoveryDraftProject(queuedMessage, draftKey);
502 await mergeComposerDraftContent(draftKey, {
503 text: queuedMessage.text,
504 context: queuedMessage.context,
505 attachments: queuedMessage.attachments,
506 });
507 } finally {
508 // Snapshots for the rollbacks below: undoComposerDraftMerge restores
509 // the original draft only while it is untouched, and otherwise takes
510 // out just what this recovery inserted so edits typed during the awaits
511 // survive. Captured in a finally because mergeComposerDraftContent
512 // publishes before its persistence await: even its failure leaves the
513 // merged content in the draft.
514 mergedDraft = getComposerDraftSnapshot(draftKey);
515 rollback = { snapshot: originalDraft, merged: mergedDraft };
516 }
517 if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId]) {
518 withdrawError();
519 await undoComposerDraftMerge(draftKey, originalDraft, mergedDraft);
520 return "deferred";
521 }
522 updateComposerDraftSettings(draftKey, {
523 ...(queuedMessage.modelSelection ? { modelSelection: queuedMessage.modelSelection } : {}),
524 ...(queuedMessage.runtimeMode ? { runtimeMode: queuedMessage.runtimeMode } : {}),
525 ...(queuedMessage.interactionMode ? { interactionMode: queuedMessage.interactionMode } : {}),
526 ...(queuedMessage.creation
527 ? {
528 workspaceSelection: {
529 mode: queuedMessage.creation.workspaceMode,
530 branch: queuedMessage.creation.branch,
531 worktreePath: queuedMessage.creation.worktreePath,
532 pullRequestReference: queuedMessage.creation.pullRequestReference,
533 ...(queuedMessage.creation.startFromOrigin !== undefined
534 ? { startFromOrigin: queuedMessage.creation.startFromOrigin }
535 : {}),
536 },
537 }
538 : {}),
539 });
540 const restoredDraft = getComposerDraftSnapshot(draftKey);
541 rollback = { snapshot: originalDraft, merged: restoredDraft };
542 await flushComposerDrafts();
543 if (
544 appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId] ||
545 !(await confirmThreadOutboxMessageQueued(queuedMessage)) ||
546 appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId]
547 ) {
548 withdrawError();
549 await undoComposerDraftMerge(draftKey, originalDraft, restoredDraft);
550 return "deferred";
551 }
552 // Revision-checked: an edit that landed after the confirmation above
553 // must not be deleted with the pre-edit payload this recovery restored.
554 if (
555 !(await removeThreadOutboxMessage(
556 queuedMessage,
557 revision,
558 () => !appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId],
559 ))
560 ) {
561 withdrawError();
562 await undoComposerDraftMerge(draftKey, originalDraft, restoredDraft);
563 return "deferred";
564 }
565 // The queued message is gone; from here the draft owns the content and
566 // must never be rolled back.
567 rollback = null;
568 if (queuedMessage.creation) {
569 // The failure card shows the reason, so an error left by an earlier
570 // failed attempt at this recovery no longer applies.
571 withdrawError();
572 // The thread screen for this creation is likely open; it reads the
573 // outcome to offer reopening the restored draft, and shows the reason.
574 recordPendingThreadCreationOutcome({
575 kind: "failed",
576 message: queuedMessage,
577 reason: message,
578 });
579 }
580 return "restored";
581 } catch (error) {
582 if (rollback !== null) {
583 // Take the recovered content back out (keeping edits typed since) so
584 // the retry's merge starts clean instead of appending a duplicate. The
585 // in-memory rollback lands even when its own persistence write fails.
586 await undoComposerDraftMerge(draftKey, rollback.snapshot, rollback.merged).catch(
587 (undoError) => {
588 console.warn("[thread-outbox] failed to persist a recovery rollback", undoError);
589 },
590 );
591 }
592 console.warn("[thread-outbox] failed to restore an undeliverable message", error);
593 setThreadComposerError(
594 threadKey,
595 error instanceof Error ? error.message : "The unsent message could not be restored.",
596 queuedMessage.messageId,
597 );
598 return "retry";
599 }
600}
601
602/**
603 * A rejected creation becomes its own new-task draft rather than merging into
604 * whatever the user is typing for that project. The key derives from the
605 * message id so a retry after a mid-recovery failure lands on the same draft
606 * instead of minting another.
607 */
608function recoveryDraftKey(queuedMessage: QueuedThreadMessage): string {
609 return queuedMessage.creation
610 ? restoredNewTaskDraftKey(queuedMessage.messageId)
611 : scopedThreadKey(queuedMessage.environmentId, queuedMessage.threadId);
612}
613
614function stampRecoveryDraftProject(queuedMessage: QueuedThreadMessage, draftKey: string): void {
615 if (!queuedMessage.creation) {
616 return;
617 }
618 updateComposerDraftSettings(draftKey, {
619 project: {
620 environmentId: queuedMessage.environmentId,
621 projectId: queuedMessage.creation.projectId,
622 createdAt: queuedMessage.createdAt,
623 },
624 });
625}
626
627async function preserveUploadedAttachmentsForEditor(
628 originalMessage: QueuedThreadMessage,
629 uploadedMessage: QueuedThreadMessage,
630): Promise<void> {
631 if (!originalMessage.creation) {
632 return;
633 }
634
635 const draftKey = `pending-task:${originalMessage.messageId}`;
636 const draft = getComposerDraftSnapshot(draftKey);
637 const uploadedById = new Map(
638 uploadedMessage.attachments.map((attachment) => [attachment.id, attachment] as const),
639 );
640 let changed = false;
641 const nextAttachments = draft.attachments.map((attachment) => {
642 const uploaded = uploadedById.get(attachment.id);
643 if (
644 !uploaded?.uploadedAttachmentId ||
645 uploaded.uploadEnvironmentId !== originalMessage.environmentId ||
646 (attachment.uploadedAttachmentId === uploaded.uploadedAttachmentId &&
647 attachment.uploadEnvironmentId === uploaded.uploadEnvironmentId)
648 ) {
649 return attachment;
650 }
651 changed = true;
652 return {
653 ...attachment,
654 uploadedAttachmentId: uploaded.uploadedAttachmentId,
655 uploadEnvironmentId: uploaded.uploadEnvironmentId,
656 };
657 });
658 if (changed) {
659 replaceComposerDraftAttachments(draftKey, nextAttachments);
660 await flushComposerDrafts();
661 }
662}
663
664export function useThreadOutboxDrain(): void {
665 const startTurn = useAtomCommand(threadEnvironment.startTurn, { reportFailure: false });
666 const preparePullRequest = useAtomCommand(gitEnvironment.preparePullRequestThread, {
667 reportFailure: false,
668 });
669 const setThreadRuntimeMode = useAtomCommand(threadEnvironment.setRuntimeMode, {
670 reportFailure: false,
671 });
672 const setThreadInteractionMode = useAtomCommand(threadEnvironment.setInteractionMode, {
673 reportFailure: false,
674 });
675 const dispatchingQueuedMessageId = useAtomValue(dispatchingQueuedMessageIdAtom);
676 const editingQueuedMessageIds = useAtomValue(editingQueuedMessageIdsAtom);
677 const queuedMessagesByThreadKey = useThreadOutboxMessages();
678 const shellStatuses = useThreadOutboxShellStatuses();
679 const threads = useThreadShells();
680 const creationOutcomes = useAtomValue(pendingThreadCreationOutcomesAtom);
681 const projects = useProjects();
682 const serverConfigs = useServerConfigs();
683 const { connectedEnvironments } = useRemoteConnectionStatus();
684 const operableEnvironments = useEnvironmentsWithScope(
685 connectedEnvironments,
686 AuthOrchestrationOperateScope,
687 );
688 const [retryTick, setRetryTick] = useState(0);
689 const retryAttemptRef = useRef(new Map<MessageId, number>());
690 const retryNotBeforeRef = useRef(new Map<MessageId, number>());
691 const retryTimersRef = useRef(new Map<MessageId, ReturnType<typeof setTimeout>>());
692 const acknowledgedExistingThreadMessageIdsRef = useRef(new Set<MessageId>());
693 const blockedRecoverySubscriptionsRef = useRef(
694 new Map<
695 MessageId,
696 { readonly message: QueuedThreadMessage; readonly unsubscribe: () => void }
697 >(),
698 );
699
700 const scheduleQueuedMessageRetry = useCallback((messageId: MessageId) => {
701 const retryAttempt = (retryAttemptRef.current.get(messageId) ?? 0) + 1;
702 retryAttemptRef.current.set(messageId, retryAttempt);
703 const retryDelayMs = threadOutboxRetryDelayMs(retryAttempt);
704 retryNotBeforeRef.current.set(messageId, Date.now() + retryDelayMs);
705 const pendingTimer = retryTimersRef.current.get(messageId);
706 if (pendingTimer !== undefined) {
707 clearTimeout(pendingTimer);
708 }
709 const retryTimer = setTimeout(() => {
710 retryTimersRef.current.delete(messageId);
711 setRetryTick((current) => current + 1);
712 }, retryDelayMs);
713 retryTimersRef.current.set(messageId, retryTimer);
714 }, []);
715
716 const restoreQueuedMessage = useCallback(
717 async (queuedMessage: QueuedThreadMessage, message: string): Promise<boolean> => {
718 const result = await restoreRejectedQueuedMessage(queuedMessage, message);
719 if (result !== "blocked") {
720 return result !== "retry";
721 }
722
723 if (!blockedRecoverySubscriptionsRef.current.has(queuedMessage.messageId)) {
724 const draftKey = recoveryDraftKey(queuedMessage);
725 const editorDraftKey = queuedMessage.creation
726 ? `pending-task:${queuedMessage.messageId}`
727 : null;
728 const currentDrafts = appAtomRegistry.get(composerDraftsAtom);
729 const blockedAttachments = currentDrafts[draftKey]?.attachments;
730 const editorAttachments =
731 editorDraftKey === null ? undefined : currentDrafts[editorDraftKey]?.attachments;
732 const unsubscribe = appAtomRegistry.subscribe(composerDraftsAtom, (drafts) => {
733 if (
734 drafts[draftKey]?.attachments === blockedAttachments &&
735 (editorDraftKey === null || drafts[editorDraftKey]?.attachments === editorAttachments)
736 ) {
737 return;
738 }
739 const active = blockedRecoverySubscriptionsRef.current.get(queuedMessage.messageId);
740 if (!active) {
741 return;
742 }
743 blockedRecoverySubscriptionsRef.current.delete(queuedMessage.messageId);
744 active.unsubscribe();
745 setRetryTick((current) => current + 1);
746 });
747 blockedRecoverySubscriptionsRef.current.set(queuedMessage.messageId, {
748 message: queuedMessage,
749 unsubscribe,
750 });
751 }
752 return true;
753 },
754 [],
755 );
756
757 useEffect(() => {
758 let mounted = true;
759 const load = async () => {
760 if ((await threadOutboxManager.load()) || !mounted) return;
761 Alert.alert(
762 "Some queued messages could not be loaded",
763 "Unreadable records and attachment files are still saved. Other messages can still be sent.",
764 [
765 { text: "Dismiss", style: "cancel" },
766 { text: "Retry", onPress: () => void load() },
767 ],
768 );
769 };
770 void load();
771 return () => {
772 mounted = false;
773 for (const timer of retryTimersRef.current.values()) {
774 clearTimeout(timer);
775 }
776 retryTimersRef.current.clear();
777 for (const blocked of blockedRecoverySubscriptionsRef.current.values()) {
778 blocked.unsubscribe();
779 }
780 blockedRecoverySubscriptionsRef.current.clear();
781 };
782 }, []);
783
784 const makeDeliveryHelpers = useCallback((queuedMessage: QueuedThreadMessage) => {
785 const reportFailure = (
786 commandResult: AtomCommandResult<unknown, unknown>,
787 stage: ThreadOutboxCommandStage,
788 ): { readonly action: "retry" | "restore"; readonly message: string } | null => {
789 if (!AsyncResult.isFailure(commandResult)) {
790 return null;
791 }
792 const error = Cause.squash(commandResult.cause);
793 const action = logThreadOutboxDeliveryFailure({
794 stage,
795 error,
796 interrupted: Cause.hasInterruptsOnly(commandResult.cause),
797 context: {
798 environmentId: queuedMessage.environmentId,
799 threadId: queuedMessage.threadId,
800 messageId: queuedMessage.messageId,
801 cause: commandResult.cause,
802 },
803 });
804 return {
805 action,
806 message: error instanceof Error ? error.message : "The message could not be sent.",
807 };
808 };
809 return { reportFailure };
810 }, []);
811
812 const sendQueuedMessage = useCallback(
813 async (queuedMessage: QueuedThreadMessage, thread: EnvironmentThreadShell) => {
814 const hasAccess = () =>
815 readEnvironmentScope(queuedMessage.environmentId, AuthOrchestrationOperateScope);
816 if (!hasAccess()) return true;
817 const serverConfig = appAtomRegistry.get(
818 serverEnvironment.configValueAtom(queuedMessage.environmentId),
819 );
820 if (!serverConfig) return false;
821 const settings = resolveQueuedThreadSettings(queuedMessage, thread, serverConfig.providers);
822 if (isModelSelectionUnavailable(serverConfig, settings.modelSelection, thread.projectId)) {
823 const { title, detail } = describeUnavailableModelSelection(
824 serverConfig,
825 settings.modelSelection,
826 thread.projectId,
827 );
828 return restoreQueuedMessage(queuedMessage, `${title}. ${detail}`);
829 }
830 const { reportFailure } = makeDeliveryHelpers(queuedMessage);
831
832 if (settings.runtimeMode !== thread.runtimeMode) {
833 if (!hasAccess()) return true;
834 const runtimeResult = await setThreadRuntimeMode({
835 environmentId: queuedMessage.environmentId,
836 input: {
837 commandId: settingsCommandId(queuedMessage, "runtime-mode"),
838 threadId: queuedMessage.threadId,
839 runtimeMode: settings.runtimeMode,
840 createdAt: queuedMessage.createdAt,
841 },
842 });
843 if (AsyncResult.isFailure(runtimeResult)) {
844 reportFailure(runtimeResult, "settings-sync");
845 return false;
846 }
847 }
848
849 if (settings.interactionMode !== thread.interactionMode) {
850 if (!hasAccess()) return true;
851 const interactionResult = await setThreadInteractionMode({
852 environmentId: queuedMessage.environmentId,
853 input: {
854 commandId: settingsCommandId(queuedMessage, "interaction-mode"),
855 threadId: queuedMessage.threadId,
856 interactionMode: settings.interactionMode,
857 createdAt: queuedMessage.createdAt,
858 },
859 });
860 if (AsyncResult.isFailure(interactionResult)) {
861 reportFailure(interactionResult, "settings-sync");
862 return false;
863 }
864 }
865
866 let prepared: PreparedTurnAttachments;
867 let persistedMessage: QueuedThreadMessage;
868 let deliveryRevision: number;
869 if (!hasAccess()) return true;
870 try {
871 const preparedResult = await prepareQueuedMessageAttachments(
872 queuedMessage,
873 serverConfig.environment.capabilities.attachmentUploads === true,
874 );
875 if (preparedResult.status === "abandoned") {
876 return true;
877 }
878 prepared = preparedResult.prepared;
879 persistedMessage = preparedResult.persistedMessage;
880 deliveryRevision = preparedResult.deliveryRevision;
881 if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId]) {
882 await preserveUploadedAttachmentsForEditor(
883 queuedMessage,
884 preparedResult.persistedMessage,
885 );
886 return true;
887 }
888 } catch (error) {
889 if (!hasAccess()) return true;
890 logThreadOutboxUploadFailure(queuedMessage, error);
891 if (!shouldRetryThreadOutboxDelivery(error)) {
892 return restoreQueuedMessage(
893 queuedMessage,
894 error instanceof Error ? error.message : "An attachment could not upload.",
895 );
896 }
897 return false;
898 }
899 if (!isQueuedMessagePayloadCurrent(persistedMessage, deliveryRevision)) {
900 return true;
901 }
902 const currentConfig = appAtomRegistry.get(
903 serverEnvironment.configValueAtom(queuedMessage.environmentId),
904 );
905 if (!currentConfig) return false;
906 if (isModelSelectionUnavailable(currentConfig, settings.modelSelection, thread.projectId)) {
907 const { title, detail } = describeUnavailableModelSelection(
908 currentConfig,
909 settings.modelSelection,
910 thread.projectId,
911 );
912 return restoreQueuedMessage(persistedMessage, `${title}. ${detail}`);
913 }
914 const sendSettings = resolveQueuedThreadSettings(
915 queuedMessage,
916 settings,
917 currentConfig.providers,
918 );
919 if (!hasAccess()) return true;
920 const deliveryResult = await startTurn({
921 environmentId: queuedMessage.environmentId,
922 input: {
923 commandId: queuedMessage.commandId,
924 creationSource: "mobile",
925 threadId: queuedMessage.threadId,
926 message: {
927 messageId: queuedMessage.messageId,
928 role: "user",
929 ...serializeComposerMessageForServer(
930 queuedMessage.text,
931 uploadedComposerContext(
932 queuedMessage.context,
933 queuedMessage.attachments,
934 prepared.attachments,
935 ),
936 currentConfig.environment.capabilities.inlineMessageContext === true,
937 ),
938 attachments: prepared.attachments,
939 },
940 modelSelection: sendSettings.modelSelection,
941 titleSeed: deriveThreadTitleSeed({
942 text: queuedMessage.text,
943 attachments: queuedMessage.attachments,
944 }),
945 runtimeMode: sendSettings.runtimeMode,
946 interactionMode: sendSettings.interactionMode,
947 createdAt: queuedMessage.createdAt,
948 // Rows written before follow-up behavior existed keep the previous
949 // delivery, which the server turns into a queued run when a turn is
950 // already active.
951 dispatchMode: queuedMessage.dispatchMode ?? "start",
952 },
953 });
954 if (AsyncResult.isFailure(deliveryResult) && !hasAccess()) return true;
955 const failure = reportFailure(deliveryResult, "start-turn");
956 if (failure?.action === "retry") {
957 return false;
958 }
959 if (failure?.action === "restore") {
960 return restoreQueuedMessage(persistedMessage, failure.message);
961 }
962 acknowledgedExistingThreadMessageIdsRef.current.add(persistedMessage.messageId);
963 const delivered =
964 (await completeQueuedMessageDelivery(persistedMessage, deliveryRevision)) === "removed";
965 if (delivered) {
966 acknowledgedExistingThreadMessageIdsRef.current.delete(persistedMessage.messageId);
967 }
968 return delivered;
969 },
970 [
971 makeDeliveryHelpers,
972 setThreadInteractionMode,
973 setThreadRuntimeMode,
974 startTurn,
975 restoreQueuedMessage,
976 ],
977 );
978
979 const sendQueuedCreation = useCallback(
980 async (
981 queuedMessage: QueuedThreadMessage,
982 creation: QueuedThreadCreation,
983 projectCwd: string,
984 ) => {
985 const hasAccess = () =>
986 readEnvironmentScope(queuedMessage.environmentId, AuthOrchestrationOperateScope);
987 if (!hasAccess()) return true;
988 const modelSelection = queuedMessage.modelSelection;
989 if (modelSelection === undefined) {
990 return false;
991 }
992 const serverConfig = appAtomRegistry.get(
993 serverEnvironment.configValueAtom(queuedMessage.environmentId),
994 );
995 if (!serverConfig) return false;
996 const settings = resolveQueuedThreadSettings(
997 queuedMessage,
998 {
999 modelSelection,
1000 runtimeMode: DEFAULT_RUNTIME_MODE,
1001 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
1002 },
1003 serverConfig.providers,
1004 );
1005 if (isModelSelectionUnavailable(serverConfig, settings.modelSelection, creation.projectId)) {
1006 const { title, detail } = describeUnavailableModelSelection(
1007 serverConfig,
1008 settings.modelSelection,
1009 creation.projectId,
1010 );
1011 return restoreQueuedMessage(queuedMessage, `${title}. ${detail}`);
1012 }
1013 let prepared: PreparedTurnAttachments;
1014 let persistedMessage: QueuedThreadMessage;
1015 let deliveryRevision: number;
1016 try {
1017 const preparedResult = await prepareQueuedMessageAttachments(
1018 queuedMessage,
1019 serverConfig.environment.capabilities.attachmentUploads === true,
1020 );
1021 if (preparedResult.status === "abandoned") {
1022 return true;
1023 }
1024 prepared = preparedResult.prepared;
1025 persistedMessage = preparedResult.persistedMessage;
1026 deliveryRevision = preparedResult.deliveryRevision;
1027 if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId]) {
1028 await preserveUploadedAttachmentsForEditor(
1029 queuedMessage,
1030 preparedResult.persistedMessage,
1031 );
1032 return true;
1033 }
1034 } catch (error) {
1035 if (!hasAccess()) return true;
1036 logThreadOutboxUploadFailure(queuedMessage, error);
1037 if (!shouldRetryThreadOutboxDelivery(error)) {
1038 return restoreQueuedMessage(
1039 queuedMessage,
1040 error instanceof Error ? error.message : "An attachment could not upload.",
1041 );
1042 }
1043 return false;
1044 }
1045 if (!isQueuedMessagePayloadCurrent(persistedMessage, deliveryRevision)) {
1046 return true;
1047 }
1048 const currentConfig = appAtomRegistry.get(
1049 serverEnvironment.configValueAtom(queuedMessage.environmentId),
1050 );
1051 if (!currentConfig) return false;
1052 if (isModelSelectionUnavailable(currentConfig, settings.modelSelection, creation.projectId)) {
1053 const { title, detail } = describeUnavailableModelSelection(
1054 currentConfig,
1055 settings.modelSelection,
1056 creation.projectId,
1057 );
1058 return restoreQueuedMessage(persistedMessage, `${title}. ${detail}`);
1059 }
1060 const sendSettings = resolveQueuedThreadSettings(
1061 queuedMessage,
1062 settings,
1063 currentConfig.providers,
1064 );
1065 if (!hasAccess()) return true;
1066 const { reportFailure } = makeDeliveryHelpers(queuedMessage);
1067 let workspace = creation;
1068 if (creation.pullRequestReference) {
1069 // Keep opening the composer and queuing offline immediate. Only Send
1070 // resolves the provider's actual PR head, including forks, in a worktree.
1071 const checkout = await preparePullRequest({
1072 environmentId: queuedMessage.environmentId,
1073 input: { cwd: projectCwd, reference: creation.pullRequestReference, mode: "worktree" },
1074 });
1075 if (!isQueuedMessagePayloadCurrent(persistedMessage, deliveryRevision) || !hasAccess())
1076 return true;
1077 const failure = reportFailure(checkout, "start-turn");
1078 if (failure?.action === "retry") return false;
1079 if (failure?.action === "restore")
1080 return restoreQueuedMessage(persistedMessage, failure.message);
1081 if (!AsyncResult.isSuccess(checkout)) return false;
1082 if (!checkout.value.worktreePath || !checkout.value.isOnPullRequestHead)
1083 return restoreQueuedMessage(
1084 persistedMessage,
1085 "The PR worktree has local changes and could not update to the PR head. Resolve them before trying again.",
1086 );
1087 workspace = {
1088 ...creation,
1089 workspaceMode: "local",
1090 branch: checkout.value.branch,
1091 worktreePath: checkout.value.worktreePath,
1092 startFromOrigin: false,
1093 };
1094 }
1095 const deliveryResult = await startTurn({
1096 environmentId: queuedMessage.environmentId,
1097 input: buildProjectThreadStartTurnInput({
1098 projectId: creation.projectId,
1099 projectCwd,
1100 threadId: queuedMessage.threadId,
1101 commandId: queuedMessage.commandId,
1102 messageId: queuedMessage.messageId,
1103 createdAt: queuedMessage.createdAt,
1104 ...serializeComposerMessageForServer(
1105 queuedMessage.text.trim(),
1106 uploadedComposerContext(
1107 queuedMessage.context,
1108 queuedMessage.attachments,
1109 prepared.attachments,
1110 ),
1111 currentConfig.environment.capabilities.inlineMessageContext === true,
1112 ),
1113 uploadedAttachments: prepared.attachments,
1114 modelSelection: sendSettings.modelSelection,
1115 runtimeMode: sendSettings.runtimeMode,
1116 interactionMode: sendSettings.interactionMode,
1117 workspaceMode: workspace.workspaceMode,
1118 branch: workspace.branch,
1119 worktreePath: workspace.worktreePath,
1120 startFromOrigin: workspace.startFromOrigin ?? false,
1121 worktreeBranchName: buildTemporaryWorktreeBranchName(randomHex),
1122 }),
1123 });
1124 if (AsyncResult.isFailure(deliveryResult) && !hasAccess()) return true;
1125 const failure = reportFailure(deliveryResult, "start-turn");
1126 if (failure?.action === "retry") {
1127 return false;
1128 }
1129 if (failure?.action === "restore") {
1130 return restoreQueuedMessage(persistedMessage, failure.message);
1131 }
1132 // Recorded before the queue entry goes so the thread screen never sees a
1133 // gap between the queued creation and the server's shell.
1134 recordPendingThreadCreationOutcome({ kind: "delivered", message: persistedMessage });
1135 const outcome = await completeQueuedMessageDelivery(persistedMessage, deliveryRevision);
1136 if (outcome === "edited") {
1137 if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId]) {
1138 // The editor holds the entry with unsaved edits; merging the queue
1139 // payload now would duplicate the delivered turn. Once the editor
1140 // saves, the duplicate-creation removal below recovers the edits.
1141 return true;
1142 }
1143 // The thread exists now, so the next drain would remove the edited
1144 // payload as a duplicate creation. Hand it to the thread's composer.
1145 return recoverEditedCreationAfterDelivery(persistedMessage);
1146 }
1147 return outcome === "removed";
1148 },
1149 [makeDeliveryHelpers, preparePullRequest, restoreQueuedMessage, startTurn],
1150 );
1151
1152 // A creation outcome bridges setup until the server's shell has a turn.
1153 // Drop it once that happens so the map cannot grow for a whole session; a
1154 // failed outcome stays until its thread screen consumes it.
1155 // Subscribed, not read once: the shell often lands before the outcome is
1156 // recorded, and a non-reactive read would leave that entry uncollected
1157 // because `threads` never changes again.
1158 useEffect(() => {
1159 for (const [threadKey, outcome] of Object.entries(creationOutcomes)) {
1160 if (
1161 outcome.kind === "delivered" &&
1162 threads.some(
1163 (thread) =>
1164 scopedThreadKey(thread.environmentId, thread.id) === threadKey &&
1165 (thread.latestRun !== null ||
1166 thread.runtime?.status === "failed" ||
1167 thread.runtime?.status === "cancelled" ||
1168 thread.runtime?.status === "interrupted"),
1169 )
1170 ) {
1171 clearPendingThreadCreationOutcome(threadKey);
1172 }
1173 }
1174 }, [creationOutcomes, threads]);
1175
1176 useEffect(() => {
1177 if (dispatchingQueuedMessageId !== null) {
1178 return;
1179 }
1180
1181 const queuedMessageIds = new Set(
1182 Object.values(queuedMessagesByThreadKey)
1183 .flat()
1184 .map((message) => message.messageId),
1185 );
1186 for (const messageId of acknowledgedExistingThreadMessageIdsRef.current) {
1187 if (!queuedMessageIds.has(messageId)) {
1188 acknowledgedExistingThreadMessageIdsRef.current.delete(messageId);
1189 }
1190 }
1191
1192 for (const [threadKey, queuedMessages] of Object.entries(queuedMessagesByThreadKey)) {
1193 const nextQueuedMessage = queuedMessages[0];
1194 if (!nextQueuedMessage) {
1195 continue;
1196 }
1197 if (
1198 nextQueuedMessage.creation === undefined &&
1199 acknowledgedExistingThreadMessageIdsRef.current.has(nextQueuedMessage.messageId)
1200 ) {
1201 if ((retryNotBeforeRef.current.get(nextQueuedMessage.messageId) ?? 0) > Date.now()) {
1202 continue;
1203 }
1204 beginDispatchingQueuedMessage(nextQueuedMessage.messageId);
1205 void removeAcknowledgedExistingThreadMessage(
1206 nextQueuedMessage,
1207 acknowledgedExistingThreadMessageIdsRef.current,
1208 )
1209 .then((removed) => {
1210 if (!removed) {
1211 scheduleQueuedMessageRetry(nextQueuedMessage.messageId);
1212 return;
1213 }
1214 retryAttemptRef.current.delete(nextQueuedMessage.messageId);
1215 retryNotBeforeRef.current.delete(nextQueuedMessage.messageId);
1216 const pendingTimer = retryTimersRef.current.get(nextQueuedMessage.messageId);
1217 if (pendingTimer !== undefined) {
1218 clearTimeout(pendingTimer);
1219 retryTimersRef.current.delete(nextQueuedMessage.messageId);
1220 }
1221 })
1222 .finally(() => finishDispatchingQueuedMessage(nextQueuedMessage.messageId));
1223 return;
1224 }
1225 if (editingQueuedMessageIds[nextQueuedMessage.messageId]) {
1226 continue;
1227 }
1228 const blockedRecovery = blockedRecoverySubscriptionsRef.current.get(
1229 nextQueuedMessage.messageId,
1230 );
1231 if (blockedRecovery) {
1232 if (blockedRecovery.message === nextQueuedMessage) {
1233 continue;
1234 }
1235 blockedRecoverySubscriptionsRef.current.delete(nextQueuedMessage.messageId);
1236 blockedRecovery.unsubscribe();
1237 }
1238 if ((retryNotBeforeRef.current.get(nextQueuedMessage.messageId) ?? 0) > Date.now()) {
1239 continue;
1240 }
1241
1242 const thread = findThread(threads, nextQueuedMessage);
1243 if (thread && scopedThreadKey(thread.environmentId, thread.id) !== threadKey) {
1244 continue;
1245 }
1246
1247 const creation = nextQueuedMessage.creation;
1248 const environment = connectedEnvironments.find(
1249 (candidate) => candidate.environmentId === nextQueuedMessage.environmentId,
1250 );
1251 const shellStatus = shellStatuses.get(nextQueuedMessage.environmentId) ?? "empty";
1252 const deliveryAction = resolveThreadOutboxDeliveryAction({
1253 isCreation: creation !== undefined,
1254 threadExists: thread !== undefined,
1255 shellStatus,
1256 environmentConnected: environment?.connectionState === "connected",
1257 threadBusy: threadRuntimeIsActive(thread?.runtime),
1258 });
1259 if (deliveryAction === "send" && !operableEnvironments.has(nextQueuedMessage.environmentId)) {
1260 continue;
1261 }
1262 // The delivery action resolves first; capability checks apply only to
1263 // a message that will send. Checking earlier would restore a
1264 // creation whose startTurn already made the thread as a duplicate draft
1265 // instead of removing it.
1266 const serverConfig = serverConfigs.get(nextQueuedMessage.environmentId);
1267 const dispatchStep = resolveThreadOutboxDispatchStep({
1268 deliveryAction,
1269 fileAttachments: nextQueuedMessage.attachments.filter(
1270 (attachment) => attachment.type === "file",
1271 ),
1272 serverConfig: serverConfig
1273 ? {
1274 maxFileUploadBytes:
1275 serverConfig.environment.capabilities.fileAttachments?.maxUploadBytes,
1276 }
1277 : null,
1278 });
1279 if (dispatchStep.step === "wait") {
1280 continue;
1281 }
1282 if (dispatchStep.step === "retry") {
1283 // The environment is connected but its config has not synced yet.
1284 // Back off and retry instead of parking the message forever.
1285 scheduleQueuedMessageRetry(nextQueuedMessage.messageId);
1286 continue;
1287 }
1288 if (dispatchStep.step === "restore") {
1289 const attachmentError = dispatchStep.reason;
1290 beginDispatchingQueuedMessage(nextQueuedMessage.messageId);
1291 void confirmThreadOutboxMessageQueued(nextQueuedMessage)
1292 .then((queued) => {
1293 if (
1294 !queued ||
1295 appAtomRegistry.get(editingQueuedMessageIdsAtom)[nextQueuedMessage.messageId]
1296 ) {
1297 return true;
1298 }
1299 return restoreQueuedMessage(nextQueuedMessage, attachmentError);
1300 })
1301 .then((restored) => {
1302 if (!restored) {
1303 scheduleQueuedMessageRetry(nextQueuedMessage.messageId);
1304 }
1305 })
1306 .finally(() => finishDispatchingQueuedMessage(nextQueuedMessage.messageId));
1307 return;
1308 }
1309 // The live project shell is preferred for the workspace path, with the
1310 // snapshot taken at enqueue time as the fallback so a task never dies
1311 // just because its project shell is not loaded.
1312 const creationProjectCwd =
1313 creation !== undefined
1314 ? (findCreationProject(projects, nextQueuedMessage)?.workspaceRoot ??
1315 creation.projectCwd ??
1316 null)
1317 : null;
1318 // An incomplete pending task (e.g. worktree mode without a branch) stays
1319 // queued until the user finishes it in the editor.
1320 if (deliveryAction === "send" && creation !== undefined) {
1321 if (!isQueuedThreadCreationSendable(nextQueuedMessage)) {
1322 continue;
1323 }
1324 if (creationProjectCwd === null && shellStatus !== "live") {
1325 continue;
1326 }
1327 }
1328
1329 beginDispatchingQueuedMessage(nextQueuedMessage.messageId);
1330 const removeQueuedMessage = (warning: string) =>
1331 removeThreadOutboxMessage(nextQueuedMessage).then(
1332 () => true,
1333 (error) => {
1334 console.warn(warning, {
1335 environmentId: nextQueuedMessage.environmentId,
1336 threadId: nextQueuedMessage.threadId,
1337 messageId: nextQueuedMessage.messageId,
1338 error,
1339 });
1340 return false;
1341 },
1342 );
1343 // Enqueues publish optimistically before their durable write settles.
1344 // Confirm the write landed (and the message wasn't rolled back) before
1345 // sending, so a failed write can never chase an already-delivered turn.
1346 const delivery = confirmThreadOutboxMessageQueued(nextQueuedMessage).then((queued) => {
1347 if (!queued) {
1348 // Rolled back by a failed write; nothing to deliver or retry.
1349 return true;
1350 }
1351 // The guards evaluated before the confirmation await are stale by now:
1352 // the user may have opened this message in the editor. Re-read that
1353 // guard and defer to the next drain pass (returning true skips the
1354 // failure/backoff path) rather than sending a payload being edited.
1355 if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[nextQueuedMessage.messageId]) {
1356 return true;
1357 }
1358 // The shell state is equally stale. Re-run the same delivery policy
1359 // against the live thread snapshot so a vanished thread or newly
1360 // created target defers, while busy existing threads can still steer.
1361 if (deliveryAction === "send") {
1362 const liveThread = findThread(
1363 appAtomRegistry.get(environmentThreadShells.threadShellsAtom),
1364 nextQueuedMessage,
1365 );
1366 const liveThreadBusy = threadRuntimeIsActive(liveThread?.runtime);
1367 const liveDeliveryAction = resolveThreadOutboxDeliveryAction({
1368 isCreation: creation !== undefined,
1369 threadExists: liveThread !== undefined,
1370 shellStatus,
1371 environmentConnected: environment?.connectionState === "connected",
1372 threadBusy: liveThreadBusy,
1373 });
1374 if (liveDeliveryAction !== "send") {
1375 return true;
1376 }
1377 }
1378 return deliveryAction === "remove"
1379 ? creation !== undefined
1380 ? // A creation entry that survived its delivery cleanup either
1381 // holds edits (recover them) or the delivered payload (a
1382 // recovered duplicate the user can delete). Restart loses any
1383 // in-memory distinction, and losing edits is the worse failure,
1384 // so recovery is unconditional here.
1385 recoverEditedCreationAfterDelivery(nextQueuedMessage)
1386 : removeQueuedMessage("[thread-outbox] failed to remove message for a missing thread")
1387 : creation !== undefined
1388 ? creationProjectCwd !== null
1389 ? sendQueuedCreation(nextQueuedMessage, creation, creationProjectCwd)
1390 : removeQueuedMessage("[thread-outbox] dropped pending task for a missing project")
1391 : thread !== undefined
1392 ? sendQueuedMessage(nextQueuedMessage, thread)
1393 : Promise.resolve(false);
1394 });
1395 void delivery
1396 .then((sent) => {
1397 if (sent) {
1398 retryAttemptRef.current.delete(nextQueuedMessage.messageId);
1399 retryNotBeforeRef.current.delete(nextQueuedMessage.messageId);
1400 const pendingTimer = retryTimersRef.current.get(nextQueuedMessage.messageId);
1401 if (pendingTimer !== undefined) {
1402 clearTimeout(pendingTimer);
1403 retryTimersRef.current.delete(nextQueuedMessage.messageId);
1404 }
1405 return;
1406 }
1407
1408 scheduleQueuedMessageRetry(nextQueuedMessage.messageId);
1409 })
1410 .finally(() => {
1411 finishDispatchingQueuedMessage(nextQueuedMessage.messageId);
1412 });
1413 return;
1414 }
1415 }, [
1416 connectedEnvironments,
1417 dispatchingQueuedMessageId,
1418 editingQueuedMessageIds,
1419 operableEnvironments,
1420 projects,
1421 queuedMessagesByThreadKey,
1422 retryTick,
1423 restoreQueuedMessage,
1424 scheduleQueuedMessageRetry,
1425 sendQueuedCreation,
1426 sendQueuedMessage,
1427 serverConfigs,
1428 shellStatuses,
1429 threads,
1430 ]);
1431}