import { describe, expect, it } from "@effect/vitest";import { EnvironmentNotRegisteredError } from "@t3tools/client-runtime/connection";import { isTransportConnectionErrorMessage } from "@t3tools/client-runtime/errors";import { EnvironmentRpcUnavailableError } from "@t3tools/client-runtime/rpc";import {  CommandId,  EnvironmentAuthorizationError,  EnvironmentId,  MessageId,  OrchestrationDispatchCommandError,  ProjectId,  ProviderInstanceId,  ThreadId,} from "@t3tools/contracts";import { AtomRegistry } from "effect/unstable/reactivity";import * as RpcClientError from "effect/unstable/rpc/RpcClientError";import * as Socket from "effect/unstable/socket/Socket";import { onTestFinished, vi } from "vite-plus/test"; const outboxFiles = vi.hoisted(() => new Map<string, string | Error>()); vi.mock("expo-file-system", () => {  class Directory {    create() {}     list() {      return Array.from(outboxFiles.keys(), (name) => new File(name));    }  }   class File {    readonly name: string;    readonly parentDirectory = new Directory();     constructor(...parts: [string] | [Directory, string]) {      this.name = parts.length === 1 ? parts[0] : parts[1];    }     get exists() {      return outboxFiles.has(this.name);    }     create() {      outboxFiles.set(this.name, "");    }     write(contents: string) {      outboxFiles.set(this.name, contents);    }     moveSync(file: File) {      const contents = outboxFiles.get(this.name);      if (contents === undefined) throw new Error("Missing file");      outboxFiles.set(file.name, contents);      outboxFiles.delete(this.name);    }     delete() {      outboxFiles.delete(this.name);    }     async text(): Promise<string> {      const contents = outboxFiles.get(this.name);      if (contents instanceof Error) throw contents;      if (contents === undefined) throw new Error("Missing file");      return contents;    }  }   return {    File,    Directory,    Paths: { document: "/documents" },  };}); import {  decodeQueuedThreadMessage,  encodeQueuedThreadMessage,  groupQueuedThreadMessages,  isQueuedThreadCreationSendable,  modelSelectionsEqual,  resolveThreadOutboxDeliveryAction,  resolveThreadOutboxDispatchStep,  resolveThreadOutboxFailureAction,  resolveQueuedThreadSettings,  shouldRetryThreadOutboxDelivery,  threadOutboxRetryDelayMs,  type QueuedThreadMessage,} from "./thread-outbox-model";import { createThreadOutboxManager, ThreadOutboxManagerError } from "./thread-outbox-manager";import {  expoThreadOutboxStorage,  ThreadOutboxStorageError,  type ThreadOutboxLoadResult,  type ThreadOutboxStorage,} from "./thread-outbox-storage"; function queuedMessage(input: {  readonly environmentId?: string;  readonly threadId?: string;  readonly messageId: string;  readonly createdAt: string;}): QueuedThreadMessage {  return {    environmentId: EnvironmentId.make(input.environmentId ?? "environment-1"),    threadId: ThreadId.make(input.threadId ?? "thread-1"),    messageId: MessageId.make(input.messageId),    commandId: CommandId.make(`command-${input.messageId}`),    text: input.messageId,    attachments: [],    createdAt: input.createdAt,  };} describe("thread outbox", () => {  it.each(["read", "json", "schema"] as const)(    "recovers usable messages without permitting cleanup after a record %s failure",    async (failure) => {      onTestFinished(() => outboxFiles.clear());      const first = queuedMessage({        messageId: "message-1",        createdAt: "2026-06-08T10:00:01.000Z",      });      const second = queuedMessage({        environmentId: "environment-2",        messageId: "message-2",        createdAt: "2026-06-08T10:00:02.000Z",      });      // Put the unreadable record first to check that later records still load.      outboxFiles.set(        "message-2.json",        failure === "read"          ? new Error("storage unavailable")          : failure === "json"            ? "{"            : JSON.stringify({ ...second, schemaVersion: 999 }),      );      outboxFiles.set("message-1.json", JSON.stringify(encodeQueuedThreadMessage(first)));      const unreadable = outboxFiles.get("message-2.json");       await expect(expoThreadOutboxStorage.load()).resolves.toMatchObject({        messages: [first],        errors: [{ operation: "read-message", fileName: "message-2.json" }],      });       const registry = AtomRegistry.make();      onTestFinished(() => registry.dispose());      const manager = createThreadOutboxManager({        registry,        storage: expoThreadOutboxStorage,        warn: () => {},      });      await expect(manager.load()).resolves.toBe(false);      const recovered = registry.get(manager.queuedMessagesByThreadKeyAtom)[        "environment-1:thread-1"      ]![0]!;      expect(recovered).toEqual(first);      await expect(manager.confirmQueued(recovered)).resolves.toBe(true);      const edited = { ...recovered, text: "Edited after recovery" };      await expect(manager.update(edited)).resolves.toBe(true);      const beforeRetry = registry.get(manager.queuedMessagesByThreadKeyAtom);      await expect(manager.load()).resolves.toBe(false);      expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toBe(beforeRetry);      await expect(manager.clearEnvironment(first.environmentId)).rejects.toMatchObject({        operation: "clear-environment-load",      });      expect(outboxFiles.get("message-2.json")).toBe(unreadable);      expect(outboxFiles.has("message-1.json")).toBe(true);       // A delivered readable message can leave the queue while the failed      // record stays intact. Its attachment cleanup has a separate guard.      await expect(manager.remove(edited)).resolves.toBe(edited);      expect(outboxFiles.has("message-1.json")).toBe(false);       outboxFiles.set("message-2.json", JSON.stringify(encodeQueuedThreadMessage(second)));      await expect(manager.load()).resolves.toBe(true);      expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({        "environment-2:thread-1": [second],      });      await expect(expoThreadOutboxStorage.load()).resolves.toEqual({        messages: [second],        errors: [],      });    },  );   it("preserves queued messages when environment cleanup cannot read the outbox", async () => {    const registry = AtomRegistry.make();    onTestFinished(() => registry.dispose());    const message = queuedMessage({      messageId: "message-1",      createdAt: "2026-06-08T10:00:01.000Z",    });    const stored = new Map<MessageId, QueuedThreadMessage>();    const manager = createThreadOutboxManager({      registry,      warn: () => {},      storage: {        load: async () => {          throw new Error("storage unavailable");        },        write: async (entry) => {          stored.set(entry.messageId, entry);        },        remove: async (entry) => {          stored.delete(entry.messageId);        },      },    });    await manager.enqueue(message);     await expect(manager.clearEnvironment(message.environmentId)).rejects.toMatchObject({      operation: "clear-environment-load",    });    expect([...stored.values()]).toEqual([message]);    expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({      "environment-1:thread-1": [message],    });  });   it("keeps in-session edits and removals when an incomplete load is retried", async () => {    const registry = AtomRegistry.make();    onTestFinished(() => registry.dispose());    const message = queuedMessage({      messageId: "message-retried",      createdAt: "2026-06-08T10:00:01.000Z",    });    const started = Promise.withResolvers<void>();    const response = Promise.withResolvers<ThreadOutboxLoadResult>();    const load = vi.fn<ThreadOutboxStorage["load"]>(async () => ({      messages: [message],      errors: [],    }));    load.mockImplementationOnce(() => {      started.resolve();      return response.promise;    });    const manager = createThreadOutboxManager({      registry,      storage: { load, write: async () => {}, remove: async () => {} },      warn: () => {},    });    const loading = manager.load();    await started.promise;    const edited = { ...message, text: "Accepted while storage was being read" };    const writing = manager.enqueue(edited);    response.resolve({      messages: [message],      errors: [        new ThreadOutboxStorageError({          operation: "read-message",          environmentId: null,          threadId: null,          messageId: null,          fileName: "unreadable.json",          cause: new Error("unreadable record"),        }),      ],    });    await expect(loading).resolves.toBe(false);    await writing;    expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({      "environment-1:thread-1": [edited],    });    await expect(manager.confirmQueued(edited)).resolves.toBe(true);     await manager.remove(edited);    // A later repaired record can contain a stale copy of a removed message.    await expect(manager.load()).resolves.toBe(true);    expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({});    expect(load).toHaveBeenCalledTimes(2);  });   it("groups messages by scoped thread and preserves creation order", () => {    const later = queuedMessage({      messageId: "message-2",      createdAt: "2026-06-08T10:00:02.000Z",    });    const earlier = queuedMessage({      messageId: "message-1",      createdAt: "2026-06-08T10:00:01.000Z",    });     expect(groupQueuedThreadMessages([later, earlier])).toEqual({      "environment-1:thread-1": [earlier, later],    });  });   it("decodes the persisted schema and rejects incomplete messages", () => {    const message = queuedMessage({      messageId: "message-1",      createdAt: "2026-06-08T10:00:01.000Z",    });     expect(      decodeQueuedThreadMessage({        schemaVersion: 1,        ...message,      }),    ).toEqual(message);    expect(() =>      decodeQueuedThreadMessage({        schemaVersion: 1,        environmentId: "environment-1",      }),    ).toThrow();  });   it("persists generic attachment paths without embedding their contents", () => {    const message = {      ...queuedMessage({        messageId: "message-file",        createdAt: "2026-06-08T10:00:01.000Z",      }),      attachments: [        {          id: "file-1",          type: "file" as const,          name: "report.pdf",          mimeType: "application/pdf",          sizeBytes: 42,          fileUri: "file:///documents/report.pdf",          uploadedAttachmentId: "pending-report-pdf",          uploadEnvironmentId: EnvironmentId.make("environment-1"),        },      ],    } satisfies QueuedThreadMessage;     expect(decodeQueuedThreadMessage(encodeQueuedThreadMessage(message))).toEqual(message);  });   it("reads file-backed images from v4 queued messages", () => {    const message = {      ...queuedMessage({        messageId: "message-image",        createdAt: "2026-06-08T10:00:01.000Z",      }),      attachments: [        {          id: "image-1",          type: "image" as const,          name: "photo.png",          mimeType: "image/png",          sizeBytes: 3,          fileUri: "file:///documents/t3-composer-attachments/photo.png",          previewUri: "file:///documents/t3-composer-attachments/photo.png",          uploadedAttachmentId: "pending-photo-png",          uploadEnvironmentId: EnvironmentId.make("environment-1"),        },      ],    } satisfies QueuedThreadMessage;     expect(decodeQueuedThreadMessage({ ...message, schemaVersion: 4 })).toEqual(message);  });   it("persists the exact selector snapshot while remaining compatible with v1 messages", () => {    const legacyMessage = queuedMessage({      messageId: "message-1",      createdAt: "2026-06-08T10:00:01.000Z",    });    const selectedMessage = {      ...legacyMessage,      modelSelection: {        instanceId: ProviderInstanceId.make("codex"),        model: "gpt-5.4",        options: [{ id: "reasoningEffort", value: "xhigh" }],      },      runtimeMode: "approval-required",      interactionMode: "plan",    } satisfies QueuedThreadMessage;     expect(decodeQueuedThreadMessage(encodeQueuedThreadMessage(selectedMessage))).toEqual(      selectedMessage,    );    expect(      resolveQueuedThreadSettings(legacyMessage, {        modelSelection: selectedMessage.modelSelection,        runtimeMode: selectedMessage.runtimeMode,        interactionMode: selectedMessage.interactionMode,      }),    ).toEqual({      modelSelection: selectedMessage.modelSelection,      runtimeMode: selectedMessage.runtimeMode,      interactionMode: selectedMessage.interactionMode,    });  });   it("compares model options as part of the queued settings change", () => {    const base = {      instanceId: ProviderInstanceId.make("codex"),      model: "gpt-5.4",      options: [{ id: "reasoningEffort", value: "medium" }],    } as const;     expect(modelSelectionsEqual(base, base)).toBe(true);    expect(      modelSelectionsEqual(base, {        ...base,        options: [{ id: "reasoningEffort", value: "xhigh" }],      }),    ).toBe(false);  });   it("normalizes queued plan mode against the queued provider, not the current thread", () => {    const codex = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.6-sol" };    const antigravity = {      instanceId: ProviderInstanceId.make("google-personal"),      model: "gemini-test-thinking",      options: [{ id: "native-option", value: "keep-this-choice" }],    };    const providers = [      { instanceId: codex.instanceId, showInteractionModeToggle: true },      { instanceId: antigravity.instanceId, showInteractionModeToggle: false },    ];    const message = {      ...queuedMessage({ messageId: "queued-plan", createdAt: "2026-09-02T10:00:00.000Z" }),      text: "/plan inspect the project",      modelSelection: antigravity,      interactionMode: "plan",    } satisfies QueuedThreadMessage;     expect(      resolveQueuedThreadSettings(        message,        { modelSelection: codex, runtimeMode: "approval-required", interactionMode: "plan" },        providers,      ),    ).toEqual({      modelSelection: antigravity,      runtimeMode: "approval-required",      interactionMode: "default",    });    expect(      resolveQueuedThreadSettings(        { ...message, modelSelection: codex },        {          modelSelection: antigravity,          runtimeMode: "approval-required",          interactionMode: "default",        },        providers,      ).interactionMode,    ).toBe("plan");  });   it("normalizes a legacy queued message that inherits unsupported plan mode", () => {    const modelSelection = {      instanceId: ProviderInstanceId.make("google-personal"),      model: "gemini-test-thinking",    };    expect(      resolveQueuedThreadSettings(        queuedMessage({ messageId: "legacy-plan", createdAt: "2026-09-02T10:00:00.000Z" }),        { modelSelection, runtimeMode: "approval-required", interactionMode: "plan" },        [{ instanceId: modelSelection.instanceId, showInteractionModeToggle: false }],      ).interactionMode,    ).toBe("default");  });   it("backs off queued delivery retries and caps them at sixteen seconds", () => {    expect([1, 2, 3, 4, 5, 6].map(threadOutboxRetryDelayMs)).toEqual([      1_000, 2_000, 4_000, 8_000, 16_000, 16_000,    ]);  });   it("serializes mutations even when an earlier mutation is slower", async () => {    const registry = AtomRegistry.make();    const manager = createThreadOutboxManager({      registry,      storage: {        load: async () => ({ messages: [], errors: [] }),        write: async () => undefined,        remove: async () => undefined,      },    });    const order: string[] = [];    let releaseFirst!: () => void;    const firstBlocked = new Promise<void>((resolve) => {      releaseFirst = resolve;    });     const first = manager.serialize(async () => {      order.push("first:start");      await firstBlocked;      order.push("first:end");    });    const second = manager.serialize(async () => {      order.push("second");    });     await Promise.resolve();    expect(order).toEqual(["first:start"]);    releaseFirst();    await Promise.all([first, second]);    expect(order).toEqual(["first:start", "first:end", "second"]);    registry.dispose();  });   it("holds the mutation queue while persisted messages are loading", async () => {

Lines 1–500 of 1485. Load more