import fs from "node:fs";import os from "node:os";import path from "node:path"; import type { ProviderRuntimeEvent } 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 { ServerConfig } from "../../config.ts";import { TextGenerationError } from "../../git/Errors.ts";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 { GitCore, type GitCoreShape } from "../../git/Services/GitCore.ts";import { TextGeneration, type TextGenerationShape } from "../../git/Services/TextGeneration.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";import * as NodeServices from "@effect/platform-node/NodeServices"; 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;  const createdStateDirs = new Set<string>();   afterEach(async () => {    if (scope) {      await Effect.runPromise(Scope.close(scope, Exit.void));    }    scope = null;    if (runtime) {      await runtime.dispose();    }    runtime = null;    for (const stateDir of createdStateDirs) {      fs.rmSync(stateDir, { recursive: true, force: true });    }    createdStateDirs.clear();  });   async function createHarness(input?: { readonly stateDir?: string }) {    const now = new Date().toISOString();    const stateDir = input?.stateDir ?? fs.mkdtempSync(path.join(os.tmpdir(), "t3code-reactor-"));    createdStateDirs.add(stateDir);    const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());    let nextSessionIndex = 1;    const startSession = vi.fn((_: unknown, __: unknown) => {      const sessionIndex = nextSessionIndex++;      return Effect.succeed({        sessionId: asSessionId(`sess-${sessionIndex}`),        provider: "codex" as const,        status: "ready" as const,        threadId: ProviderThreadId.makeUnsafe(`provider-thread-${sessionIndex}`),        createdAt: now,        updatedAt: now,      });    });    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((_: unknown) => Effect.void);    const renameBranch = vi.fn((_: unknown) =>      Effect.succeed({        branch: "t3code/generated-name",      }),    );    const generateBranchName = vi.fn<TextGenerationShape["generateBranchName"]>(() =>      Effect.succeed({        branch: "generated-name",      }),    );     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([]),      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)),      Layer.provideMerge(Layer.succeed(GitCore, { renameBranch } as unknown as GitCoreShape)),      Layer.provideMerge(        Layer.succeed(TextGeneration, { generateBranchName } as unknown as TextGenerationShape),      ),      Layer.provideMerge(ServerConfig.layerTest(process.cwd(), stateDir)),      Layer.provideMerge(NodeServices.layer),    );    const 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,      renameBranch,      generateBranchName,      stateDir,    };  }   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("generates and renames temporary worktree branch on first turn", async () => {    const harness = await createHarness();    const now = new Date().toISOString();     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.meta.update",        commandId: CommandId.makeUnsafe("cmd-thread-meta-set-temp-branch"),        threadId: ThreadId.makeUnsafe("thread-1"),        branch: "t3code/89abc123",        worktreePath: "/tmp/provider-project/.t3/worktrees/t3code-89abc123",      }),    );     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-worktree-rename"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: asMessageId("user-message-worktree-rename"),          role: "user",          text: "Fix visual bug from screenshot",          attachments: [            {              type: "image",              id: "thread-1-att-rename",              name: "bug.png",              mimeType: "image/png",              sizeBytes: 5,            },          ],        },        approvalPolicy: "on-request",        sandboxMode: "workspace-write",        createdAt: now,      }),    );     await waitFor(() => harness.generateBranchName.mock.calls.length === 1);    await waitFor(() => harness.renameBranch.mock.calls.length === 1);     expect(harness.generateBranchName.mock.calls[0]?.[0]).toEqual({      cwd: "/tmp/provider-project/.t3/worktrees/t3code-89abc123",      message: "Fix visual bug from screenshot",      attachments: [        {          type: "image",          id: "thread-1-att-rename",          name: "bug.png",          mimeType: "image/png",          sizeBytes: 5,        },      ],    });    expect(harness.renameBranch.mock.calls[0]?.[0]).toEqual({      cwd: "/tmp/provider-project/.t3/worktrees/t3code-89abc123",      oldBranch: "t3code/89abc123",      newBranch: "t3code/generated-name",    });     await waitFor(() => {      const readModel = Effect.runSync(harness.engine.getReadModel());      const thread = readModel.threads.find(        (entry) => entry.id === ThreadId.makeUnsafe("thread-1"),      );      return thread?.branch === "t3code/generated-name";    });  });   it("passes persisted attachment references to branch generation and turn start", async () => {    const harness = await createHarness();    const now = new Date().toISOString();     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.meta.update",        commandId: CommandId.makeUnsafe("cmd-thread-meta-set-temp-branch-persisted"),        threadId: ThreadId.makeUnsafe("thread-1"),        branch: "t3code/abcdef12",        worktreePath: "/tmp/provider-project/.t3/worktrees/t3code-abcdef12",      }),    );     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-persisted-attachments"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: asMessageId("user-message-persisted-attachments"),          role: "user",          text: "Fix visual bug from screenshot",          attachments: [            {              type: "image",              id: "thread-1-att-persisted",              name: "bug.png",              mimeType: "image/png",              sizeBytes: 5,            },          ],        },        approvalPolicy: "on-request",        sandboxMode: "workspace-write",        createdAt: now,      }),    );     await waitFor(() => harness.generateBranchName.mock.calls.length === 1);    await waitFor(() => harness.sendTurn.mock.calls.length === 1);     expect(harness.generateBranchName.mock.calls[0]?.[0]).toEqual({      cwd: "/tmp/provider-project/.t3/worktrees/t3code-abcdef12",      message: "Fix visual bug from screenshot",      attachments: [        {          type: "image",          id: "thread-1-att-persisted",          name: "bug.png",          mimeType: "image/png",          sizeBytes: 5,        },      ],    });    expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({      attachments: [        {          type: "image",          id: "thread-1-att-persisted",          name: "bug.png",          mimeType: "image/png",          sizeBytes: 5,        },      ],    });  });   it("skips worktree branch generation after the first user turn", async () => {    const harness = await createHarness();    const now = new Date().toISOString();     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.meta.update",        commandId: CommandId.makeUnsafe("cmd-thread-meta-set-temp-branch-2"),        threadId: ThreadId.makeUnsafe("thread-1"),        branch: "t3code/1234abcd",        worktreePath: "/tmp/provider-project/.t3/worktrees/t3code-1234abcd",      }),    );     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-first"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: asMessageId("user-message-first"),          role: "user",          text: "first",          attachments: [],        },        approvalPolicy: "on-request",        sandboxMode: "workspace-write",        createdAt: now,      }),    );     await waitFor(() => harness.generateBranchName.mock.calls.length === 1);     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-second"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: asMessageId("user-message-second"),          role: "user",          text: "second",          attachments: [],        },        approvalPolicy: "on-request",        sandboxMode: "workspace-write",        createdAt: now,      }),    );     await waitFor(() => harness.sendTurn.mock.calls.length === 2);    expect(harness.generateBranchName.mock.calls.length).toBe(1);    expect(harness.renameBranch.mock.calls.length).toBe(1);  });   it("skips worktree rename when branch-name generation fails", async () => {    const harness = await createHarness();    const now = new Date().toISOString();    harness.generateBranchName.mockImplementationOnce(() =>      Effect.fail(        new TextGenerationError({          operation: "generateBranchName",          detail: "model returned invalid payload",        }),      ),    );     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.meta.update",        commandId: CommandId.makeUnsafe("cmd-thread-meta-set-temp-branch-null"),        threadId: ThreadId.makeUnsafe("thread-1"),        branch: "t3code/0000abcd",        worktreePath: "/tmp/provider-project/.t3/worktrees/t3code-0000abcd",      }),    );     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.makeUnsafe("cmd-turn-start-null-branch"),        threadId: ThreadId.makeUnsafe("thread-1"),        message: {          messageId: asMessageId("user-message-null-branch"),          role: "user",          text: "Fix visual regression",          attachments: [],        },        approvalPolicy: "on-request",        sandboxMode: "workspace-write",        createdAt: now,      }),    );     await waitFor(() => harness.generateBranchName.mock.calls.length === 1);    await Effect.runPromise(Effect.sleep("20 millis"));    expect(harness.renameBranch.mock.calls.length).toBe(0);     const readModel = await Effect.runPromise(harness.engine.getReadModel());    const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));    expect(thread?.branch).toBe("t3code/0000abcd");  });   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(

Lines 1–500 of 770. Load more