import type { ProviderRuntimeEvent, ProviderSession } from "@t3tools/contracts";import {  ApprovalRequestId,  CommandId,  MessageId,  ProjectId,  ProviderSessionId,  ProviderThreadId,  ProviderTurnId,  ThreadId,  TurnId,} from "@t3tools/contracts";import { Effect, Exit, Layer, ManagedRuntime, PubSub, Scope, Stream } from "effect";import { afterEach, describe, expect, it, vi } from "vitest"; import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";import {  ProviderService,  type ProviderServiceShape,} from "../../provider/Services/ProviderService.ts";import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";import { ProviderCommandReactorLive } from "./ProviderCommandReactor.ts";import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts"; const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value);const asApprovalRequestId = (value: string): ApprovalRequestId =>  ApprovalRequestId.makeUnsafe(value);const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value);const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); async function waitFor(predicate: () => boolean, timeoutMs = 2000): Promise<void> {  const deadline = Date.now() + timeoutMs;  const poll = async (): Promise<void> => {    if (predicate()) {      return;    }    if (Date.now() >= deadline) {      throw new Error("Timed out waiting for expectation.");    }    await new Promise((resolve) => setTimeout(resolve, 10));    return poll();  };   return poll();} describe("ProviderCommandReactor", () => {  let runtime: ManagedRuntime.ManagedRuntime<    OrchestrationEngineService | ProviderCommandReactor,    unknown  > | null = null;  let scope: Scope.Closeable | null = null;   afterEach(async () => {    if (scope) {      await Effect.runPromise(Scope.close(scope, Exit.void));    }    scope = null;    if (runtime) {      await runtime.dispose();    }    runtime = null;  });   async function createHarness() {    const now = new Date().toISOString();    const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());    let nextSessionIndex = 1;    const runtimeSessions: Array<ProviderSession> = [];    const startSession = vi.fn((_: unknown, input: unknown) => {      const sessionIndex = nextSessionIndex++;      const provider =        typeof input === "object" &&        input !== null &&        "provider" in input &&        (input.provider === "codex" ||          input.provider === "claudeCode" ||          input.provider === "cursor")          ? input.provider          : "codex";      const resumeCursor =        typeof input === "object" && input !== null && "resumeCursor" in input          ? input.resumeCursor          : undefined;      const session: ProviderSession = {        sessionId: asSessionId(`sess-${sessionIndex}`),        provider,        status: "ready" as const,        threadId: ProviderThreadId.makeUnsafe(`provider-thread-${sessionIndex}`),        resumeCursor: resumeCursor ?? { opaque: `cursor-${sessionIndex}` },        createdAt: now,        updatedAt: now,      };      runtimeSessions.push(session);      return Effect.succeed(session);    });    const sendTurn = vi.fn((_: unknown) =>      Effect.succeed({        threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),        turnId: asProviderTurnId("provider-turn-1"),      }),    );    const interruptTurn = vi.fn((_: unknown) => Effect.void);    const respondToRequest = vi.fn((_: unknown) => Effect.void);    const stopSession = vi.fn((input: unknown) =>      Effect.sync(() => {        const sessionId =          typeof input === "object" && input !== null && "sessionId" in input            ? (input as { sessionId?: ProviderSessionId }).sessionId            : undefined;        if (!sessionId) {          return;        }        const index = runtimeSessions.findIndex((session) => session.sessionId === sessionId);        if (index >= 0) {          runtimeSessions.splice(index, 1);        }      }),    );     const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never;    const service: ProviderServiceShape = {      startSession: startSession as ProviderServiceShape["startSession"],      sendTurn: sendTurn as ProviderServiceShape["sendTurn"],      interruptTurn: interruptTurn as ProviderServiceShape["interruptTurn"],      respondToRequest: respondToRequest as ProviderServiceShape["respondToRequest"],      stopSession: stopSession as ProviderServiceShape["stopSession"],      listSessions: () => Effect.succeed(runtimeSessions),      rollbackConversation: () => unsupported(),      stopAll: () => Effect.void,      streamEvents: Stream.fromPubSub(runtimeEventPubSub),    };     const orchestrationLayer = OrchestrationEngineLive.pipe(      Layer.provide(OrchestrationProjectionPipelineLive),      Layer.provide(OrchestrationEventStoreLive),      Layer.provide(OrchestrationCommandReceiptRepositoryLive),      Layer.provide(SqlitePersistenceMemory),    );    const layer = ProviderCommandReactorLive.pipe(      Layer.provideMerge(orchestrationLayer),      Layer.provideMerge(Layer.succeed(ProviderService, service)),    );    runtime = ManagedRuntime.make(layer);    const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));    const reactor = await runtime.runPromise(Effect.service(ProviderCommandReactor));    scope = await Effect.runPromise(Scope.make("sequential"));    await Effect.runPromise(reactor.start.pipe(Scope.provide(scope)));    await Effect.runPromise(Effect.sleep("10 millis"));     await Effect.runPromise(      engine.dispatch({        type: "project.create",        commandId: CommandId.makeUnsafe("cmd-project-create"),        projectId: asProjectId("project-1"),        title: "Provider Project",        workspaceRoot: "/tmp/provider-project",        defaultModel: "gpt-5-codex",        createdAt: now,      }),    );    await Effect.runPromise(      engine.dispatch({        type: "thread.create",        commandId: CommandId.makeUnsafe("cmd-thread-create"),        threadId: ThreadId.makeUnsafe("thread-1"),        projectId: asProjectId("project-1"),        title: "Thread",        model: "gpt-5-codex",        branch: null,        worktreePath: null,        createdAt: now,      }),    );     return {      engine,      startSession,      sendTurn,      interruptTurn,      respondToRequest,      stopSession,    };  }   it("reacts to thread.turn.start by ensuring session and sending provider turn", async () => {    const harness = await createHarness();    const now = new Date().toISOString();     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-1"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: asMessageId("user-message-1"),          role: "user",          text: "hello reactor",          attachments: [],        },        approvalPolicy: "on-request",        sandboxMode: "workspace-write",        createdAt: now,      }),    );     await waitFor(() => harness.startSession.mock.calls.length === 1);    await waitFor(() => harness.sendTurn.mock.calls.length === 1);    expect(harness.startSession.mock.calls[0]?.[0]).toEqual(ThreadId.makeUnsafe("thread-1"));    expect(harness.startSession.mock.calls[0]?.[1]).toMatchObject({      cwd: "/tmp/provider-project",      model: "gpt-5-codex",      approvalPolicy: "on-request",      sandboxMode: "workspace-write",    });     const readModel = await Effect.runPromise(harness.engine.getReadModel());    const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));    expect(thread?.session?.providerSessionId).toBe("sess-1");    expect(thread?.session?.approvalPolicy).toBe("on-request");    expect(thread?.session?.sandboxMode).toBe("workspace-write");  });   it("starts first turn with requested provider when provider is specified", async () => {    const harness = await createHarness();    const now = new Date().toISOString();     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-provider-first"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: asMessageId("user-message-provider-first"),          role: "user",          text: "hello claude",          attachments: [],        },        provider: "claudeCode",        approvalPolicy: "on-request",        sandboxMode: "workspace-write",        createdAt: now,      }),    );     await waitFor(() => harness.startSession.mock.calls.length === 1);    await waitFor(() => harness.sendTurn.mock.calls.length === 1);    expect(harness.startSession.mock.calls[0]?.[1]).toMatchObject({      provider: "claudeCode",      cwd: "/tmp/provider-project",      model: "gpt-5-codex",      approvalPolicy: "on-request",      sandboxMode: "workspace-write",    });     const readModel = await Effect.runPromise(harness.engine.getReadModel());    const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));    expect(thread?.session?.providerName).toBe("claudeCode");    expect(thread?.session?.providerSessionId).toBe("sess-1");  });   it("starts first turn with cursor provider when provider is specified", async () => {    const harness = await createHarness();    const now = new Date().toISOString();     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-provider-cursor"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: asMessageId("user-message-provider-cursor"),          role: "user",          text: "hello cursor",          attachments: [],        },        provider: "cursor",        approvalPolicy: "on-request",        sandboxMode: "workspace-write",        createdAt: now,      }),    );     await waitFor(() => harness.startSession.mock.calls.length === 1);    await waitFor(() => harness.sendTurn.mock.calls.length === 1);    expect(harness.startSession.mock.calls[0]?.[1]).toMatchObject({      provider: "cursor",      cwd: "/tmp/provider-project",      model: "gpt-5-codex",      approvalPolicy: "on-request",      sandboxMode: "workspace-write",    });     const readModel = await Effect.runPromise(harness.engine.getReadModel());    const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));    expect(thread?.session?.providerName).toBe("cursor");    expect(thread?.session?.providerSessionId).toBe("sess-1");  });   it("reuses the same provider session when runtime mode is unchanged", async () => {    const harness = await createHarness();    const now = new Date().toISOString();     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-unchanged-1"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: asMessageId("user-message-unchanged-1"),          role: "user",          text: "first",          attachments: [],        },        approvalPolicy: "on-request",        sandboxMode: "workspace-write",        createdAt: now,      }),    );     await waitFor(() => harness.startSession.mock.calls.length === 1);    await waitFor(() => harness.sendTurn.mock.calls.length === 1);     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-unchanged-2"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: asMessageId("user-message-unchanged-2"),          role: "user",          text: "second",          attachments: [],        },        approvalPolicy: "on-request",        sandboxMode: "workspace-write",        createdAt: now,      }),    );     await waitFor(() => harness.sendTurn.mock.calls.length === 2);    expect(harness.startSession.mock.calls.length).toBe(1);    expect(harness.stopSession.mock.calls.length).toBe(0);  });   it("restarts the provider session when runtime mode changes", async () => {    const harness = await createHarness();    const now = new Date().toISOString();     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-runtime-mode-1"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: asMessageId("user-message-runtime-mode-1"),          role: "user",          text: "first",          attachments: [],        },        approvalPolicy: "never",        sandboxMode: "danger-full-access",        createdAt: now,      }),    );     await waitFor(() => harness.startSession.mock.calls.length === 1);    await waitFor(() => harness.sendTurn.mock.calls.length === 1);     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-runtime-mode-2"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: asMessageId("user-message-runtime-mode-2"),          role: "user",          text: "second",          attachments: [],        },        approvalPolicy: "on-request",        sandboxMode: "workspace-write",        createdAt: now,      }),    );     await waitFor(() => harness.stopSession.mock.calls.length === 1);    await waitFor(() => harness.startSession.mock.calls.length === 2);    await waitFor(() => harness.sendTurn.mock.calls.length === 2);     expect(harness.stopSession.mock.calls[0]?.[0]).toEqual({ sessionId: asSessionId("sess-1") });    expect(harness.startSession.mock.calls[1]?.[1]).toMatchObject({      resumeCursor: { opaque: "cursor-1" },      approvalPolicy: "on-request",      sandboxMode: "workspace-write",    });    expect(harness.sendTurn.mock.calls[1]?.[0]).toMatchObject({      sessionId: asSessionId("sess-2"),    });     const readModel = await Effect.runPromise(harness.engine.getReadModel());    const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));    expect(thread?.session?.providerSessionId).toBe("sess-2");    expect(thread?.session?.approvalPolicy).toBe("on-request");    expect(thread?.session?.sandboxMode).toBe("workspace-write");  });   it("switches provider by restarting the session when turn request provider changes", async () => {    const harness = await createHarness();    const now = new Date().toISOString();     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-provider-switch-1"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: asMessageId("user-message-provider-switch-1"),          role: "user",          text: "first",          attachments: [],        },        approvalPolicy: "on-request",        sandboxMode: "workspace-write",        createdAt: now,      }),    );     await waitFor(() => harness.startSession.mock.calls.length === 1);    await waitFor(() => harness.sendTurn.mock.calls.length === 1);     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-provider-switch-2"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: asMessageId("user-message-provider-switch-2"),          role: "user",          text: "second",          attachments: [],        },        provider: "claudeCode",        approvalPolicy: "on-request",        sandboxMode: "workspace-write",        createdAt: now,      }),    );     await waitFor(() => harness.stopSession.mock.calls.length === 1);    await waitFor(() => harness.startSession.mock.calls.length === 2);    await waitFor(() => harness.sendTurn.mock.calls.length === 2);     expect(harness.stopSession.mock.calls[0]?.[0]).toEqual({ sessionId: asSessionId("sess-1") });    expect(harness.startSession.mock.calls[1]?.[1]).toMatchObject({      provider: "claudeCode",      approvalPolicy: "on-request",      sandboxMode: "workspace-write",    });    expect(harness.startSession.mock.calls[1]?.[1]).not.toHaveProperty("resumeCursor");     const readModel = await Effect.runPromise(harness.engine.getReadModel());    const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));    expect(thread?.session?.providerSessionId).toBe("sess-2");    expect(thread?.session?.providerName).toBe("claudeCode");    expect(thread?.session?.approvalPolicy).toBe("on-request");    expect(thread?.session?.sandboxMode).toBe("workspace-write");  });   it("does not stop the active session when restart fails before rebind", async () => {    const harness = await createHarness();    const now = new Date().toISOString();     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-restart-failure-1"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: asMessageId("user-message-restart-failure-1"),          role: "user",          text: "first",          attachments: [],        },        approvalPolicy: "never",        sandboxMode: "danger-full-access",        createdAt: now,      }),    );     await waitFor(() => harness.startSession.mock.calls.length === 1);    await waitFor(() => harness.sendTurn.mock.calls.length === 1);     harness.startSession.mockImplementationOnce((_: unknown, __: unknown) =>

Lines 1–500 of 662. Load more