import { describe, expect, it } from "@effect/vitest";import {  CommandId,  EnvironmentId,  MessageId,  ProjectId,  ProviderInstanceId,  ThreadId,} from "@t3tools/contracts";import { AtomRegistry } from "effect/unstable/reactivity"; import {  decodeQueuedThreadMessage,  encodeQueuedThreadMessage,  groupQueuedThreadMessages,  isQueuedThreadCreationSendable,  modelSelectionsEqual,  resolveThreadOutboxDeliveryAction,  resolveThreadOutboxFailureAction,  resolveQueuedThreadSettings,  shouldRetryThreadOutboxDelivery,  threadOutboxRetryDelayMs,  type QueuedThreadMessage,} from "./thread-outbox-model";import { createThreadOutboxManager, ThreadOutboxManagerError } from "./thread-outbox-manager";import 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("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 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("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 () => [],        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 () => {    const registry = AtomRegistry.make();    const message = queuedMessage({      messageId: "message-1",      createdAt: "2026-06-08T10:00:01.000Z",    });    const stored = new Map([[message.messageId, message]]);    let loadCalls = 0;    let removeCalls = 0;    let releaseInitialLoad!: () => void;    const initialLoadBlocked = new Promise<void>((resolve) => {      releaseInitialLoad = resolve;    });    const storage: ThreadOutboxStorage = {      load: async () => {        loadCalls += 1;        if (loadCalls === 1) {          await initialLoadBlocked;        }        return [...stored.values()];      },      write: async () => undefined,      remove: async (candidate) => {        removeCalls += 1;        stored.delete(candidate.messageId);      },    };    const manager = createThreadOutboxManager({ registry, storage });     const loading = manager.load();    await Promise.resolve();    const clearing = manager.clearEnvironment(message.environmentId);    await Promise.resolve();    await Promise.resolve();     expect(loadCalls).toBe(1);    expect(removeCalls).toBe(0);     releaseInitialLoad();    await Promise.all([loading, clearing]);    expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({});    registry.dispose();  });   it("reports structured load failures and permits a retry", async () => {    const registry = AtomRegistry.make();    const loadCause = new Error("storage unavailable");    const warnings: Array<{ message: string; error: unknown }> = [];    let loadCalls = 0;    const manager = createThreadOutboxManager({      registry,      storage: {        load: async () => {          loadCalls += 1;          if (loadCalls === 1) throw loadCause;          return [];        },        write: async () => undefined,        remove: async () => undefined,      },      warn: (message, error) => warnings.push({ message, error }),    });     await manager.load();    expect(warnings).toEqual([      {        message: "[thread-outbox] failed to load persisted messages",        error: new ThreadOutboxManagerError({          operation: "load",          environmentId: null,          threadId: null,          messageId: null,          cause: loadCause,        }),      },    ]);     await manager.load();    expect(loadCalls).toBe(2);    registry.dispose();  });   it("keeps atom state aligned with durable writes and removals", async () => {    const registry = AtomRegistry.make();    const stored = new Map<MessageId, QueuedThreadMessage>();    const removalCause = new Error("remove failed");    let failRemoval = true;    const storage: ThreadOutboxStorage = {      load: async () => [...stored.values()],      write: async (message) => {        stored.set(message.messageId, message);      },      remove: async (message) => {        if (failRemoval) {          throw removalCause;        }        stored.delete(message.messageId);      },    };    const manager = createThreadOutboxManager({ registry, storage });    const message = queuedMessage({      messageId: "message-1",      createdAt: "2026-06-08T10:00:01.000Z",    });     await manager.enqueue(message);    expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({      "environment-1:thread-1": [message],    });     await expect(manager.remove(message)).rejects.toEqual(      new ThreadOutboxManagerError({        operation: "remove",        environmentId: message.environmentId,        threadId: message.threadId,        messageId: message.messageId,        cause: removalCause,      }),    );    expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({      "environment-1:thread-1": [message],    });     failRemoval = false;    await manager.remove(message);    expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({});    registry.dispose();  });   it("publishes an enqueued message before the durable write resolves", async () => {    const registry = AtomRegistry.make();    let releaseWrite!: () => void;    const writeBlocked = new Promise<void>((resolve) => {      releaseWrite = resolve;    });    const manager = createThreadOutboxManager({      registry,      storage: {        load: async () => [],        write: async () => writeBlocked,        remove: async () => undefined,      },    });    const message = queuedMessage({      messageId: "message-1",      createdAt: "2026-06-08T10:00:01.000Z",    });     const enqueueing = manager.enqueue(message);    expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({      "environment-1:thread-1": [message],    });     releaseWrite();    await enqueueing;    expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({      "environment-1:thread-1": [message],    });    registry.dispose();  });   it("rolls an enqueued message back out when the durable write fails", async () => {    const registry = AtomRegistry.make();    const writeCause = new Error("disk full");    const manager = createThreadOutboxManager({      registry,      storage: {        load: async () => [],        write: async () => {          throw writeCause;        },        remove: async () => undefined,      },    });    const message = queuedMessage({      messageId: "message-1",      createdAt: "2026-06-08T10:00:01.000Z",    });     await expect(manager.enqueue(message)).rejects.toEqual(      new ThreadOutboxManagerError({        operation: "enqueue",        environmentId: message.environmentId,        threadId: message.threadId,        messageId: message.messageId,        cause: writeCause,      }),    );    expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({});    registry.dispose();  });   it("keeps a same-id retry queued when the first attempt's write fails", async () => {    const registry = AtomRegistry.make();    let failNextWrite = true;    let releaseFirstWrite!: () => void;    const firstWriteBlocked = new Promise<void>((resolve) => {      releaseFirstWrite = resolve;    });    const manager = createThreadOutboxManager({      registry,      storage: {        load: async () => [],        write: async () => {          if (failNextWrite) {            failNextWrite = false;            await firstWriteBlocked;            throw new Error("disk full");          }        },        remove: async () => undefined,      },    });    const message = queuedMessage({      messageId: "message-1",      createdAt: "2026-06-08T10:00:01.000Z",    });    const retried = { ...message, text: "retried" };     const first = manager.enqueue(message);    const second = manager.enqueue(retried);    releaseFirstWrite();    await expect(first).rejects.toBeInstanceOf(ThreadOutboxManagerError);    await second;     // The failed first attempt must not roll back the retry that replaced it.    expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({      "environment-1:thread-1": [retried],    });    await expect(manager.confirmQueued(retried)).resolves.toBe(true);    await expect(manager.confirmQueued(message)).resolves.toBe(false);    registry.dispose();  });   it("replaces an existing message when an enqueue retry uses the same id", async () => {    const registry = AtomRegistry.make();    const manager = createThreadOutboxManager({      registry,      storage: {        load: async () => [],        write: async () => undefined,        remove: async () => undefined,      },    });    const message = queuedMessage({      messageId: "message-1",      createdAt: "2026-06-08T10:00:01.000Z",    });    const retried = { ...message, text: "retried" };     await manager.enqueue(message);    await manager.enqueue(retried);     expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({      "environment-1:thread-1": [retried],    });    registry.dispose();  });   it("updates a queued message in place but never resurrects a removed one", async () => {    const registry = AtomRegistry.make();    const stored = new Map<MessageId, QueuedThreadMessage>();    const storage: ThreadOutboxStorage = {      load: async () => [...stored.values()],      write: async (message) => {        stored.set(message.messageId, message);      },      remove: async (message) => {        stored.delete(message.messageId);      },    };    const manager = createThreadOutboxManager({ registry, storage });    const message = queuedMessage({      messageId: "message-1",      createdAt: "2026-06-08T10:00:01.000Z",    });     await manager.enqueue(message);    const edited = { ...message, text: "edited" };    await expect(manager.update(edited)).resolves.toBe(true);    expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({      "environment-1:thread-1": [edited],    });    expect(stored.get(message.messageId)).toEqual(edited);     await manager.remove(edited);    await expect(manager.update({ ...message, text: "stale flush" })).resolves.toBe(false);    expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({});    expect(stored.size).toBe(0);    registry.dispose();  });   it("only removes a missing-thread message after shell synchronization is live", () => {    expect(      resolveThreadOutboxDeliveryAction({        isCreation: false,        threadExists: false,        shellStatus: "synchronizing",        environmentConnected: true,        threadBusy: false,      }),    ).toBe("wait");    expect(      resolveThreadOutboxDeliveryAction({        isCreation: false,        threadExists: false,        shellStatus: "live",        environmentConnected: true,        threadBusy: false,      }),    ).toBe("remove");    expect(      resolveThreadOutboxDeliveryAction({        isCreation: false,        threadExists: true,        shellStatus: "live",        environmentConnected: true,        threadBusy: false,      }),    ).toBe("send");  });   it("sends queued creations once connected and live, removing already-created ones", () => {    expect(      resolveThreadOutboxDeliveryAction({        isCreation: true,        threadExists: false,        shellStatus: "cached",        environmentConnected: false,        threadBusy: false,      }),    ).toBe("wait");    // Connected but not yet synchronized: a previously delivered creation may

Lines 1–500 of 602. Load more