// @effect-diagnostics nodeBuiltinImport:offimport * as NodeFS from "node:fs";import * as NodeOS from "node:os";import * as NodePath from "node:path"; import {  ModelSelection,  ProviderRuntimeEvent,  ProviderSession,  ProviderDriverKind,  ProviderInstanceId,} from "@t3tools/contracts";import { createModelSelection } from "@t3tools/shared/model";import {  ApprovalRequestId,  CommandId,  DEFAULT_PROVIDER_INTERACTION_MODE,  EventId,  MessageId,  ProjectId,  ThreadId,  TurnId,} from "@t3tools/contracts";import * as Effect from "effect/Effect";import * as Deferred from "effect/Deferred";import * as Exit from "effect/Exit";import * as Layer from "effect/Layer";import * as ManagedRuntime from "effect/ManagedRuntime";import * as PubSub from "effect/PubSub";import * as Scope from "effect/Scope";import * as Stream from "effect/Stream";import { it as effectIt } from "@effect/vitest";import { afterEach, describe, expect, it, vi } from "vite-plus/test"; import { deriveServerPaths, ServerConfig } from "../../config.ts";import { TextGenerationError } from "@t3tools/contracts";import { ProviderAdapterRequestError } from "../../provider/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 { makeProviderRegistryLayer } from "../../provider/testUtils/providerRegistryMock.ts";import { TextGeneration, type TextGenerationShape } from "../../textGeneration/TextGeneration.ts";import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts";import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";import {  providerErrorLabel,  providerErrorLabelFromInstanceHint,  ProviderCommandReactorLive,} from "./ProviderCommandReactor.ts";import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts";import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";import * as NodeServices from "@effect/platform-node/NodeServices";import * as Clock from "effect/Clock";import { ServerSettingsService } from "../../serverSettings.ts";import { VcsStatusBroadcaster } from "../../vcs/VcsStatusBroadcaster.ts";import * as GitWorkflowService from "../../git/GitWorkflowService.ts"; const asProjectId = (value: string): ProjectId => ProjectId.make(value);const asApprovalRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.make(value);const asMessageId = (value: string): MessageId => MessageId.make(value);const asTurnId = (value: string): TurnId => TurnId.make(value); const deriveServerPathsSync = (baseDir: string, devUrl: URL | undefined) =>  Effect.runSync(deriveServerPaths(baseDir, devUrl).pipe(Effect.provide(NodeServices.layer))); async function waitFor(  predicate: () => boolean | Promise<boolean>,  timeoutMs = 10_000,): Promise<void> {  const deadline = (await Effect.runPromise(Clock.currentTimeMillis)) + timeoutMs;  const poll = async (): Promise<void> => {    if (await predicate()) {      return;    }    if ((await Effect.runPromise(Clock.currentTimeMillis)) >= deadline) {      throw new Error("Timed out waiting for expectation.");    }    await Effect.runPromise(Effect.yieldNow);    return poll();  };   return poll();} describe("ProviderCommandReactor", () => {  let runtime: ManagedRuntime.ManagedRuntime<    OrchestrationEngineService | ProviderCommandReactor | ProjectionSnapshotQuery,    unknown  > | null = null;  let scope: Scope.Closeable | null = null;  const createdStateDirs = new Set<string>();  const createdBaseDirs = 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) {      NodeFS.rmSync(stateDir, { recursive: true, force: true });    }    createdStateDirs.clear();    for (const baseDir of createdBaseDirs) {      NodeFS.rmSync(baseDir, { recursive: true, force: true });    }    createdBaseDirs.clear();  });   describe("provider error attribution", () => {    it("uses the current provider instance slug when current instance lookup fails", () => {      expect(        providerErrorLabelFromInstanceHint({          instanceId: "codex_personal",          modelSelectionInstanceId: "codex",          sessionProvider: "codex",        }),      ).toBe("codex_personal");    });     it("uses the desired provider instance slug when desired instance lookup fails", () => {      expect(        providerErrorLabelFromInstanceHint({          instanceId: "claude_openrouter",        }),      ).toBe("claude_openrouter");    });     it("uses the unknown driver kind when the resolved driver is not registered locally", () => {      expect(providerErrorLabel("third_party_driver")).toBe("third_party_driver");    });  });   async function createHarness(input?: {    readonly baseDir?: string;    readonly threadModelSelection?: ModelSelection;    readonly sessionModelSwitch?: "unsupported" | "in-session";    readonly requiresNewThreadForModelChange?: boolean;    readonly startSessionEffect?: (      session: ProviderSession,    ) => Effect.Effect<ProviderSession, ProviderAdapterRequestError>;  }) {    const now = "2026-01-01T00:00:00.000Z";    const baseDir =      input?.baseDir ?? NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3code-reactor-"));    createdBaseDirs.add(baseDir);    const { stateDir } = deriveServerPathsSync(baseDir, undefined);    createdStateDirs.add(stateDir);    const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());    let nextSessionIndex = 1;    const runtimeSessions: Array<ProviderSession> = [];    const modelSelection = input?.threadModelSelection ?? {      instanceId: ProviderInstanceId.make("codex"),      model: "gpt-5-codex",    };    const startSessionEffect = input?.startSessionEffect;    const startSession = vi.fn((_: unknown, input: unknown) => {      const sessionIndex = nextSessionIndex++;      const resumeCursor =        typeof input === "object" && input !== null && "resumeCursor" in input          ? input.resumeCursor          : undefined;      const threadId =        typeof input === "object" &&        input !== null &&        "threadId" in input &&        typeof input.threadId === "string"          ? ThreadId.make(input.threadId)          : ThreadId.make(`thread-${sessionIndex}`);      const inputModelSelection =        typeof input === "object" && input !== null && "modelSelection" in input          ? (input.modelSelection as ModelSelection | undefined)          : undefined;      const providerInstanceId =        typeof input === "object" && input !== null && "providerInstanceId" in input          ? (input.providerInstanceId as ProviderInstanceId | undefined)          : inputModelSelection?.instanceId;      const provider =        typeof input === "object" &&        input !== null &&        "provider" in input &&        typeof input.provider === "string"          ? (input.provider as ProviderSession["provider"])          : ProviderDriverKind.make(inputModelSelection?.instanceId ?? modelSelection.instanceId);      const session: ProviderSession = {        provider,        ...(providerInstanceId ? { providerInstanceId } : {}),        status: "ready" as const,        runtimeMode:          typeof input === "object" &&          input !== null &&          "runtimeMode" in input &&          (input.runtimeMode === "approval-required" || input.runtimeMode === "full-access")            ? input.runtimeMode            : "full-access",        ...(typeof input === "object" &&        input !== null &&        "cwd" in input &&        typeof input.cwd === "string"          ? { cwd: input.cwd }          : {}),        ...((inputModelSelection?.model ?? modelSelection.model)          ? { model: inputModelSelection?.model ?? modelSelection.model }          : {}),        threadId,        resumeCursor: resumeCursor ?? { opaque: `resume-${sessionIndex}` },        createdAt: now,        updatedAt: now,      };      return (startSessionEffect?.(session) ?? Effect.succeed(session)).pipe(        Effect.tap((startedSession) =>          Effect.sync(() => {            runtimeSessions.push(startedSession);          }),        ),      );    });    const sendTurn = vi.fn((_: unknown) =>      Effect.succeed({        threadId: ThreadId.make("thread-1"),        turnId: asTurnId("turn-1"),      }),    );    const interruptTurn = vi.fn((_: unknown) => Effect.void);    const respondToRequest = vi.fn<ProviderServiceShape["respondToRequest"]>(() => Effect.void);    const respondToUserInput = vi.fn<ProviderServiceShape["respondToUserInput"]>(() => Effect.void);    const stopSession = vi.fn((input: unknown) =>      Effect.sync(() => {        const threadId =          typeof input === "object" && input !== null && "threadId" in input            ? (input as { threadId?: ThreadId }).threadId            : undefined;        if (!threadId) {          return;        }        const index = runtimeSessions.findIndex((session) => session.threadId === threadId);        if (index >= 0) {          runtimeSessions.splice(index, 1);        }      }),    );    const renameBranch = vi.fn((input: unknown) =>      Effect.succeed({        branch:          typeof input === "object" &&          input !== null &&          "newBranch" in input &&          typeof input.newBranch === "string"            ? input.newBranch            : "renamed-branch",      }),    );    const refreshStatus = vi.fn((_: string) =>      Effect.succeed({        isRepo: true,        hasPrimaryRemote: true,        isDefaultRef: false,        refName: "renamed-branch",        hasWorkingTreeChanges: false,        workingTree: {          files: [],          insertions: 0,          deletions: 0,        },        hasUpstream: true,        aheadCount: 0,        behindCount: 0,        pr: null,      }),    );    const generateBranchName = vi.fn<TextGenerationShape["generateBranchName"]>((_) =>      Effect.fail(        new TextGenerationError({          operation: "generateBranchName",          detail: "disabled in test harness",        }),      ),    );    const generateThreadTitle = vi.fn<TextGenerationShape["generateThreadTitle"]>((_) =>      Effect.fail(        new TextGenerationError({          operation: "generateThreadTitle",          detail: "disabled in test harness",        }),      ),    );    const providerSnapshots = [      {        instanceId: modelSelection.instanceId,        ...(input?.requiresNewThreadForModelChange === true          ? { requiresNewThreadForModelChange: true }          : {}),      },    ];     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"],      respondToUserInput: respondToUserInput as ProviderServiceShape["respondToUserInput"],      stopSession: stopSession as ProviderServiceShape["stopSession"],      listSessions: () => Effect.succeed(runtimeSessions),      getCapabilities: (_provider) =>        Effect.succeed({          sessionModelSwitch: input?.sessionModelSwitch ?? "in-session",        }),      getInstanceInfo: (instanceId) => {        const raw = String(instanceId);        const driverKind = ProviderDriverKind.make(          raw.startsWith("claude") ? "claudeAgent" : raw.startsWith("codex") ? "codex" : raw,        );        return Effect.succeed({          instanceId,          driverKind,          displayName: undefined,          enabled: true,          continuationIdentity: {            driverKind,            continuationKey:              driverKind === ProviderDriverKind.make("codex")                ? "codex:home:/shared-codex"                : `${driverKind}:instance:${instanceId}`,          },        });      },      rollbackConversation: () => unsupported(),      get streamEvents() {        return Stream.fromPubSub(runtimeEventPubSub);      },    };     const orchestrationLayer = OrchestrationEngineLive.pipe(      Layer.provide(OrchestrationProjectionSnapshotQueryLive),      Layer.provide(OrchestrationProjectionPipelineLive),      Layer.provide(OrchestrationEventStoreLive),      Layer.provide(OrchestrationCommandReceiptRepositoryLive),      Layer.provide(RepositoryIdentityResolver.layer),      Layer.provide(SqlitePersistenceMemory),    );    const projectionSnapshotLayer = OrchestrationProjectionSnapshotQueryLive.pipe(      Layer.provide(RepositoryIdentityResolver.layer),      Layer.provide(SqlitePersistenceMemory),    );    const layer = ProviderCommandReactorLive.pipe(      Layer.provideMerge(orchestrationLayer),      Layer.provideMerge(projectionSnapshotLayer),      Layer.provideMerge(Layer.succeed(ProviderService, service)),      Layer.provideMerge(makeProviderRegistryLayer(providerSnapshots as never)),      Layer.provideMerge(        Layer.mock(GitWorkflowService.GitWorkflowService)({          renameBranch,        } satisfies Partial<GitWorkflowService.GitWorkflowService["Service"]>),      ),      Layer.provideMerge(        Layer.succeed(VcsStatusBroadcaster, {          getStatus: () => Effect.die("getStatus should not be called in this test"),          refreshLocalStatus: () =>            Effect.die("refreshLocalStatus should not be called in this test"),          refreshStatus,          streamStatus: () => Stream.die("streamStatus should not be called in this test"),        }),      ),      Layer.provideMerge(        Layer.mock(TextGeneration, {          generateBranchName,          generateThreadTitle,        }),      ),      Layer.provideMerge(ServerSettingsService.layerTest()),      Layer.provideMerge(ServerConfig.layerTest(process.cwd(), baseDir)),      Layer.provideMerge(NodeServices.layer),    );    runtime = ManagedRuntime.make(layer);     const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));    const snapshotQuery = await runtime.runPromise(Effect.service(ProjectionSnapshotQuery));    const reactor = await runtime.runPromise(Effect.service(ProviderCommandReactor));    scope = await Effect.runPromise(Scope.make("sequential"));    await Effect.runPromise(reactor.start().pipe(Scope.provide(scope)));    const drain = () => Effect.runPromise(reactor.drain);     await Effect.runPromise(      engine.dispatch({        type: "project.create",        commandId: CommandId.make("cmd-project-create"),        projectId: asProjectId("project-1"),        title: "Provider Project",        workspaceRoot: "/tmp/provider-project",        defaultModelSelection: modelSelection,        createdAt: now,      }),    );    await Effect.runPromise(      engine.dispatch({        type: "thread.create",        commandId: CommandId.make("cmd-thread-create"),        threadId: ThreadId.make("thread-1"),        projectId: asProjectId("project-1"),        title: "Thread",        modelSelection: modelSelection,        interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,        runtimeMode: "approval-required",        branch: null,        worktreePath: null,        createdAt: now,      }),    );     return {      engine,      readModel: () => Effect.runPromise(snapshotQuery.getSnapshot()),      startSession,      sendTurn,      interruptTurn,      respondToRequest,      respondToUserInput,      stopSession,      renameBranch,      refreshStatus,      generateBranchName,      generateThreadTitle,      runtimeSessions,      stateDir,      drain,    };  }   it("reacts to thread.turn.start by ensuring session and sending provider turn", async () => {    const harness = await createHarness();    const now = "2026-01-01T00:00:00.000Z";     await Effect.runPromise(      harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.make("cmd-turn-start-1"),        threadId: ThreadId.make("thread-1"),        message: {          messageId: asMessageId("user-message-1"),          role: "user",          text: "hello reactor",          attachments: [],        },        interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,        runtimeMode: "approval-required",        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.make("thread-1"));    expect(harness.startSession.mock.calls[0]?.[1]).toMatchObject({      cwd: "/tmp/provider-project",      modelSelection: {        instanceId: ProviderInstanceId.make("codex"),        model: "gpt-5-codex",      },      runtimeMode: "approval-required",    });     const readModel = await harness.readModel();    const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));    expect(thread?.session?.threadId).toBe("thread-1");    expect(thread?.session?.status).toBe("starting");    expect(thread?.session?.runtimeMode).toBe("approval-required");  });   effectIt.effect("projects starting before a slow provider session finishes", () =>    Effect.gen(function* () {      const releaseStart = yield* Deferred.make<void>();      const harness = yield* Effect.promise(() =>        createHarness({          startSessionEffect: (session) => Deferred.await(releaseStart).pipe(Effect.as(session)),        }),      );      const now = "2026-01-01T00:00:00.000Z";       yield* harness.engine.dispatch({        type: "thread.turn.start",        commandId: CommandId.make("cmd-turn-start-slow-provider"),        threadId: ThreadId.make("thread-1"),        message: {          messageId: asMessageId("user-message-slow-provider"),          role: "user",          text: "start slowly",          attachments: [],        },        interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,

Lines 1–500 of 2218. Load more