import { useAtomValue } from "@effect/atom-react";import type {  EnvironmentProject,  EnvironmentThreadShell,} from "@t3tools/client-runtime/state/shell";import type { AtomCommandResult } from "@t3tools/client-runtime/state/runtime";import {  CommandId,  DEFAULT_PROVIDER_INTERACTION_MODE,  DEFAULT_RUNTIME_MODE,  type MessageId,} from "@t3tools/contracts";import { buildTemporaryWorktreeBranchName } from "@t3tools/shared/git";import * as Cause from "effect/Cause";import { AsyncResult, Atom } from "effect/unstable/reactivity";import { useCallback, useEffect, useRef, useState } from "react"; import { scopedThreadKey } from "../lib/scopedEntities";import { buildProjectThreadStartTurnInput } from "../lib/projectThreadStartTurn";import { toUploadChatImageAttachments } from "../lib/composerImages";import { randomHex } from "../lib/uuid";import { appAtomRegistry } from "./atom-registry";import { useProjects, useThreadShells } from "./entities";import {  confirmThreadOutboxMessageQueued,  ensureThreadOutboxLoaded,  removeThreadOutboxMessage,} from "./thread-outbox";import {  isQueuedThreadCreationSendable,  modelSelectionsEqual,  resolveThreadOutboxDeliveryAction,  resolveThreadOutboxFailureAction,  resolveQueuedThreadSettings,  threadOutboxRetryDelayMs,  type QueuedThreadCreation,  type QueuedThreadMessage,  type ThreadOutboxCommandStage,} from "./thread-outbox-model";import { threadEnvironment } from "./threads";import { useAtomCommand } from "./use-atom-command";import {  editingQueuedMessageIdsAtom,  useThreadOutboxMessages,  useThreadOutboxShellStatuses,} from "./use-thread-outbox";import { useRemoteConnectionStatus } from "./use-remote-environment-registry"; export const dispatchingQueuedMessageIdAtom = Atom.make<MessageId | null>(null).pipe(  Atom.keepAlive,  Atom.withLabel("mobile:thread-outbox:dispatching-message-id"),); function beginDispatchingQueuedMessage(queuedMessageId: MessageId): void {  appAtomRegistry.set(dispatchingQueuedMessageIdAtom, queuedMessageId);} function finishDispatchingQueuedMessage(queuedMessageId: MessageId): void {  const current = appAtomRegistry.get(dispatchingQueuedMessageIdAtom);  appAtomRegistry.set(dispatchingQueuedMessageIdAtom, current === queuedMessageId ? null : current);} function findThread(  threads: ReadonlyArray<EnvironmentThreadShell>,  message: QueuedThreadMessage,): EnvironmentThreadShell | undefined {  return threads.find(    (candidate) =>      candidate.environmentId === message.environmentId && candidate.id === message.threadId,  );} function findCreationProject(  projects: ReadonlyArray<EnvironmentProject>,  message: QueuedThreadMessage,): EnvironmentProject | undefined {  return projects.find(    (candidate) =>      candidate.environmentId === message.environmentId &&      candidate.id === message.creation?.projectId,  );} function settingsCommandId(message: QueuedThreadMessage, setting: string): CommandId {  return CommandId.make(`${message.commandId}:${setting}`);} export function useThreadOutboxDrain(): void {  const startTurn = useAtomCommand(threadEnvironment.startTurn, { reportFailure: false });  const updateThreadMetadata = useAtomCommand(threadEnvironment.updateMetadata, {    reportFailure: false,  });  const setThreadRuntimeMode = useAtomCommand(threadEnvironment.setRuntimeMode, {    reportFailure: false,  });  const setThreadInteractionMode = useAtomCommand(threadEnvironment.setInteractionMode, {    reportFailure: false,  });  const dispatchingQueuedMessageId = useAtomValue(dispatchingQueuedMessageIdAtom);  const editingQueuedMessageIds = useAtomValue(editingQueuedMessageIdsAtom);  const queuedMessagesByThreadKey = useThreadOutboxMessages();  const shellStatuses = useThreadOutboxShellStatuses();  const threads = useThreadShells();  const projects = useProjects();  const { connectedEnvironments } = useRemoteConnectionStatus();  const [retryTick, setRetryTick] = useState(0);  const retryAttemptRef = useRef(new Map<MessageId, number>());  const retryNotBeforeRef = useRef(new Map<MessageId, number>());  const retryTimersRef = useRef(new Map<MessageId, ReturnType<typeof setTimeout>>());   useEffect(() => {    ensureThreadOutboxLoaded();    return () => {      for (const timer of retryTimersRef.current.values()) {        clearTimeout(timer);      }      retryTimersRef.current.clear();    };  }, []);   const makeDeliveryHelpers = useCallback((queuedMessage: QueuedThreadMessage) => {    const reportFailure = (      commandResult: AtomCommandResult<unknown, unknown>,      stage: ThreadOutboxCommandStage,    ): boolean => {      if (!AsyncResult.isFailure(commandResult)) {        return false;      }      const action = resolveThreadOutboxFailureAction({        stage,        error: Cause.squash(commandResult.cause),        interrupted: Cause.hasInterruptsOnly(commandResult.cause),      });      const retry = action === "retry";      console.warn("[thread-outbox] queued message delivery failed", {        environmentId: queuedMessage.environmentId,        threadId: queuedMessage.threadId,        messageId: queuedMessage.messageId,        stage,        cause: commandResult.cause,        retry,      });      return retry;    };    const completeDelivery = async (      deliveryResult: AtomCommandResult<unknown, unknown>,    ): Promise<boolean> => {      if (reportFailure(deliveryResult, "start-turn")) {        return false;      }       try {        await removeThreadOutboxMessage(queuedMessage);        return true;      } catch (error) {        console.warn("[thread-outbox] failed to remove delivered queued message", {          environmentId: queuedMessage.environmentId,          threadId: queuedMessage.threadId,          messageId: queuedMessage.messageId,          error,        });        return false;      }    };    return { reportFailure, completeDelivery };  }, []);   const sendQueuedMessage = useCallback(    async (queuedMessage: QueuedThreadMessage, thread: EnvironmentThreadShell) => {      const settings = resolveQueuedThreadSettings(queuedMessage, thread);      const { reportFailure, completeDelivery } = makeDeliveryHelpers(queuedMessage);       if (!modelSelectionsEqual(settings.modelSelection, thread.modelSelection)) {        const updateResult = await updateThreadMetadata({          environmentId: queuedMessage.environmentId,          input: {            commandId: settingsCommandId(queuedMessage, "model-selection"),            threadId: queuedMessage.threadId,            modelSelection: settings.modelSelection,          },        });        if (AsyncResult.isFailure(updateResult)) {          reportFailure(updateResult, "settings-sync");          return false;        }      }       if (settings.runtimeMode !== thread.runtimeMode) {        const runtimeResult = await setThreadRuntimeMode({          environmentId: queuedMessage.environmentId,          input: {            commandId: settingsCommandId(queuedMessage, "runtime-mode"),            threadId: queuedMessage.threadId,            runtimeMode: settings.runtimeMode,            createdAt: queuedMessage.createdAt,          },        });        if (AsyncResult.isFailure(runtimeResult)) {          reportFailure(runtimeResult, "settings-sync");          return false;        }      }       if (settings.interactionMode !== thread.interactionMode) {        const interactionResult = await setThreadInteractionMode({          environmentId: queuedMessage.environmentId,          input: {            commandId: settingsCommandId(queuedMessage, "interaction-mode"),            threadId: queuedMessage.threadId,            interactionMode: settings.interactionMode,            createdAt: queuedMessage.createdAt,          },        });        if (AsyncResult.isFailure(interactionResult)) {          reportFailure(interactionResult, "settings-sync");          return false;        }      }       const deliveryResult = await startTurn({        environmentId: queuedMessage.environmentId,        input: {          commandId: queuedMessage.commandId,          threadId: queuedMessage.threadId,          message: {            messageId: queuedMessage.messageId,            role: "user",            text: queuedMessage.text,            attachments: toUploadChatImageAttachments(queuedMessage.attachments),          },          modelSelection: settings.modelSelection,          runtimeMode: settings.runtimeMode,          interactionMode: settings.interactionMode,          createdAt: queuedMessage.createdAt,        },      });      return completeDelivery(deliveryResult);    },    [      makeDeliveryHelpers,      setThreadInteractionMode,      setThreadRuntimeMode,      startTurn,      updateThreadMetadata,    ],  );   const sendQueuedCreation = useCallback(    async (      queuedMessage: QueuedThreadMessage,      creation: QueuedThreadCreation,      projectCwd: string,    ) => {      const modelSelection = queuedMessage.modelSelection;      if (modelSelection === undefined) {        return false;      }      const { completeDelivery } = makeDeliveryHelpers(queuedMessage);      const deliveryResult = await startTurn({        environmentId: queuedMessage.environmentId,        input: buildProjectThreadStartTurnInput({          projectId: creation.projectId,          projectCwd,          threadId: queuedMessage.threadId,          commandId: queuedMessage.commandId,          messageId: queuedMessage.messageId,          createdAt: queuedMessage.createdAt,          text: queuedMessage.text.trim(),          attachments: queuedMessage.attachments,          modelSelection,          runtimeMode: queuedMessage.runtimeMode ?? DEFAULT_RUNTIME_MODE,          interactionMode: queuedMessage.interactionMode ?? DEFAULT_PROVIDER_INTERACTION_MODE,          workspaceMode: creation.workspaceMode,          branch: creation.branch,          worktreePath: creation.worktreePath,          startFromOrigin: creation.startFromOrigin ?? false,          worktreeBranchName: buildTemporaryWorktreeBranchName(randomHex),        }),      });      return completeDelivery(deliveryResult);    },    [makeDeliveryHelpers, startTurn],  );   useEffect(() => {    if (dispatchingQueuedMessageId !== null) {      return;    }     for (const [threadKey, queuedMessages] of Object.entries(queuedMessagesByThreadKey)) {      const nextQueuedMessage = queuedMessages[0];      if (!nextQueuedMessage) {        continue;      }      if (editingQueuedMessageIds[nextQueuedMessage.messageId]) {        continue;      }      if ((retryNotBeforeRef.current.get(nextQueuedMessage.messageId) ?? 0) > Date.now()) {        continue;      }       const thread = findThread(threads, nextQueuedMessage);      if (thread && scopedThreadKey(thread.environmentId, thread.id) !== threadKey) {        continue;      }       const creation = nextQueuedMessage.creation;      const environment = connectedEnvironments.find(        (candidate) => candidate.environmentId === nextQueuedMessage.environmentId,      );      const shellStatus = shellStatuses.get(nextQueuedMessage.environmentId) ?? "empty";      const deliveryAction = resolveThreadOutboxDeliveryAction({        isCreation: creation !== undefined,        threadExists: thread !== undefined,        shellStatus,        environmentConnected: environment?.connectionState === "connected",        threadBusy: thread?.session?.status === "running" || thread?.session?.status === "starting",      });      if (deliveryAction === "wait") {        continue;      }      // The live project shell is preferred for the workspace path, with the      // snapshot taken at enqueue time as the fallback so a task never dies      // just because its project shell is not loaded.      const creationProjectCwd =        creation !== undefined          ? (findCreationProject(projects, nextQueuedMessage)?.workspaceRoot ??            creation.projectCwd ??            null)          : null;      // An incomplete pending task (e.g. worktree mode without a branch) stays      // queued until the user finishes it in the editor.      if (deliveryAction === "send" && creation !== undefined) {        if (!isQueuedThreadCreationSendable(nextQueuedMessage)) {          continue;        }        if (creationProjectCwd === null && shellStatus !== "live") {          continue;        }      }       beginDispatchingQueuedMessage(nextQueuedMessage.messageId);      const removeQueuedMessage = (warning: string) =>        removeThreadOutboxMessage(nextQueuedMessage).then(          () => true,          (error) => {            console.warn(warning, {              environmentId: nextQueuedMessage.environmentId,              threadId: nextQueuedMessage.threadId,              messageId: nextQueuedMessage.messageId,              error,            });            return false;          },        );      // Enqueues publish optimistically before their durable write settles.      // Confirm the write landed (and the message wasn't rolled back) before      // sending, so a failed write can never chase an already-delivered turn.      const delivery = confirmThreadOutboxMessageQueued(nextQueuedMessage).then((queued) => {        if (!queued) {          // Rolled back by a failed write; nothing to deliver or retry.          return true;        }        // The guards evaluated before the confirmation await are stale by now:        // the user may have opened this message in the editor. Re-read that        // guard and defer to the next drain pass (returning true skips the        // failure/backoff path) rather than sending a payload being edited.        if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[nextQueuedMessage.messageId]) {          return true;        }        return deliveryAction === "remove"          ? removeQueuedMessage("[thread-outbox] failed to remove message for a missing thread")          : creation !== undefined            ? creationProjectCwd !== null              ? sendQueuedCreation(nextQueuedMessage, creation, creationProjectCwd)              : removeQueuedMessage("[thread-outbox] dropped pending task for a missing project")            : thread !== undefined              ? sendQueuedMessage(nextQueuedMessage, thread)              : Promise.resolve(false);      });      void delivery        .then((sent) => {          if (sent) {            retryAttemptRef.current.delete(nextQueuedMessage.messageId);            retryNotBeforeRef.current.delete(nextQueuedMessage.messageId);            const pendingTimer = retryTimersRef.current.get(nextQueuedMessage.messageId);            if (pendingTimer !== undefined) {              clearTimeout(pendingTimer);              retryTimersRef.current.delete(nextQueuedMessage.messageId);            }            return;          }           const retryAttempt = (retryAttemptRef.current.get(nextQueuedMessage.messageId) ?? 0) + 1;          retryAttemptRef.current.set(nextQueuedMessage.messageId, retryAttempt);          const retryDelayMs = threadOutboxRetryDelayMs(retryAttempt);          retryNotBeforeRef.current.set(nextQueuedMessage.messageId, Date.now() + retryDelayMs);          const pendingTimer = retryTimersRef.current.get(nextQueuedMessage.messageId);          if (pendingTimer !== undefined) {            clearTimeout(pendingTimer);          }          const retryTimer = setTimeout(() => {            retryTimersRef.current.delete(nextQueuedMessage.messageId);            setRetryTick((current) => current + 1);          }, retryDelayMs);          retryTimersRef.current.set(nextQueuedMessage.messageId, retryTimer);        })        .finally(() => {          finishDispatchingQueuedMessage(nextQueuedMessage.messageId);        });      return;    }  }, [    connectedEnvironments,    dispatchingQueuedMessageId,    editingQueuedMessageIds,    projects,    queuedMessagesByThreadKey,    retryTick,    sendQueuedCreation,    sendQueuedMessage,    shellStatuses,    threads,  ]);}