// @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,  ProviderSetupError,} from "@t3tools/contracts";import { createModelSelection } from "@t3tools/shared/model";import {  ApprovalRequestId,  CommandId,  DEFAULT_PROVIDER_INTERACTION_MODE,  EnvironmentId,  EventId,  MessageId,  ProjectId,  ThreadId,  TurnId,} from "@t3tools/contracts";import { serializeAssistantCitation } from "@t3tools/shared/assistantCitations";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 Option from "effect/Option";import * as PubSub from "effect/PubSub";import * as Scope from "effect/Scope";import * as Stream from "effect/Stream";import * as SqlClient from "effect/unstable/sql/SqlClient";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,  ProviderWorkspaceMissingError,  type ProviderServiceError,} 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 { ProviderAuthService } from "../../provider/Services/ProviderAuthService.ts";import { makeProviderRegistryLayer } from "../../provider/testUtils/providerRegistryMock.ts";import { TextGeneration } 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 * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";import {  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 { ServerActivation } from "../../serverActivation.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 assistantQuoteText = "Retain the reconnect backoff.";const assistantCitation = {  version: 1 as const,  environmentId: EnvironmentId.make("source-environment"),  threadId: ThreadId.make("source-thread"),  messageId: asMessageId("source-message"),  text: assistantQuoteText,  start: 0,  end: assistantQuoteText.length,  prefix: "",  suffix: "",}; 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    | SqlClient.SqlClient,    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");    });  });   async function createHarness(input?: {    readonly baseDir?: string;    readonly threadModelSelection?: ModelSelection;    readonly sessionModelSwitch?: "unsupported" | "in-session";    readonly requiresNewThreadForModelChange?: boolean;    readonly unreadableHistory?: boolean;    readonly titleRegenerationCompletionDispatchFailures?: number;    readonly titleRegenerationBeforeStart?: "one" | "two";    readonly serverActivation?: Effect.Effect<void>;    readonly beforeReadySessionDispatch?: () => Effect.Effect<void>;    readonly beforeTurnStartDispatch?: () => Effect.Effect<void>;    readonly afterTurnStartDispatch?: () => Effect.Effect<void>;    readonly compactThreadEffect?: () => Effect.Effect<void, ProviderAdapterRequestError>;    readonly interruptTurnEffect?: () => Effect.Effect<void, ProviderAdapterRequestError>;    readonly stopSessionEffect?: () => Effect.Effect<void, ProviderAdapterRequestError>;    readonly startSessionEffect?: (      session: ProviderSession,    ) => Effect.Effect<ProviderSession, ProviderServiceError>;    readonly tryHandlePromptCommandEffect?: ProviderAuthService["Service"]["tryHandlePromptCommand"];  }) {    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>());    const tryHandlePromptCommand = vi.fn<ProviderAuthService["Service"]["tryHandlePromptCommand"]>(      input?.tryHandlePromptCommandEffect ?? (() => Effect.succeed(false)),    );    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 compactThread = vi.fn((_: ThreadId) => input?.compactThreadEffect?.() ?? Effect.void);    const interruptTurn = vi.fn((_: unknown) => input?.interruptTurnEffect?.() ?? Effect.void);    const respondToRequest = vi.fn<ProviderServiceShape["respondToRequest"]>(() => Effect.void);    const respondToUserInput = vi.fn<ProviderServiceShape["respondToUserInput"]>(() => Effect.void);    const stopSession = vi.fn((stopInput: unknown) =>      (input?.stopSessionEffect?.() ?? Effect.void).pipe(        Effect.tap(() =>          Effect.sync(() => {            const threadId =              typeof stopInput === "object" && stopInput !== null && "threadId" in stopInput                ? (stopInput 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 pruneWorktrees = vi.fn((_: { readonly cwd: string }) => Effect.void);    const createWorktree = vi.fn(      (input: { readonly refName: string; readonly path: string | null }) =>        Effect.succeed({ worktree: { path: input.path ?? "", refName: input.refName } }),    );    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<TextGeneration["Service"]["generateBranchName"]>((_) =>      Effect.fail(        new TextGenerationError({          operation: "generateBranchName",          detail: "disabled in test harness",        }),      ),    );    const generateThreadTitle = vi.fn<TextGeneration["Service"]["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"],      compactThread,      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",        }),      assertConversationRollbackSupported: () => unsupported(),      getInstanceInfo: (instanceId) => {        const raw = String(instanceId);        const driverKind = ProviderDriverKind.make(          raw.startsWith("claude")            ? "claudeAgent"            : raw.startsWith("codex")              ? "codex"              : raw.startsWith("antigravity")                ? "antigravity"                : 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(),      uploadFeedback: () => unsupported(),      get streamEvents() {        return Stream.fromPubSub(runtimeEventPubSub);      },    };     const orchestrationLayer = OrchestrationEngineLive.pipe(      Layer.provide(OrchestrationProjectionSnapshotQueryLive),      Layer.provide(ThreadBackgroundLiveness.layer),      Layer.provide(ThreadPlanProgress.layer),      Layer.provide(OrchestrationProjectionPipelineLive),      Layer.provide(OrchestrationEventStoreLive),      Layer.provide(OrchestrationCommandReceiptRepositoryLive),      Layer.provide(RepositoryIdentityResolver.layer),      Layer.provide(SqlitePersistenceMemory),    );    const projectionSnapshotLayer = OrchestrationProjectionSnapshotQueryLive.pipe(      Layer.provide(ThreadBackgroundLiveness.layer),      Layer.provide(ThreadPlanProgress.layer),      Layer.provide(RepositoryIdentityResolver.layer),      Layer.provide(SqlitePersistenceMemory),    );    let titleRegenerationCompletionDispatchAttempts = 0;    const reactorOrchestrationLayer = Layer.effect(      OrchestrationEngineService,      Effect.gen(function* () {        const engine = yield* OrchestrationEngineService;        return {          readEvents: engine.readEvents,          readThreadEvents: engine.readThreadEvents,          getThreadReplayStats: engine.getThreadReplayStats,          dispatch: (command) => {            if (command.type === "thread.title.regeneration.complete") {              titleRegenerationCompletionDispatchAttempts += 1;              if (                titleRegenerationCompletionDispatchAttempts <=                (input?.titleRegenerationCompletionDispatchFailures ?? 0)              ) {                return Effect.die(new Error("Injected title regeneration completion failure"));              }            }            const isReplay =              command.type === "thread.turn.start" &&              command.commandId.startsWith("server:after-compaction:");            const before =              command.type === "thread.session.set" && command.session.status === "ready"                ? input?.beforeReadySessionDispatch                : isReplay                  ? input?.beforeTurnStartDispatch                  : undefined;            return (before?.() ?? Effect.void).pipe(              Effect.andThen(engine.dispatch(command)),              Effect.tap(() =>                isReplay ? (input?.afterTurnStartDispatch?.() ?? Effect.void) : Effect.void,              ),            );          },          get streamDomainEvents() {            return engine.streamDomainEvents;          },          subscribeDomainEvents: engine.subscribeDomainEvents,          latestSequence: engine.latestSequence,        } satisfies OrchestrationEngineService["Service"];      }),    ).pipe(Layer.provide(orchestrationLayer));    const layer = ProviderCommandReactorLive.pipe(      Layer.provideMerge(reactorOrchestrationLayer),      Layer.provideMerge(projectionSnapshotLayer),      Layer.provideMerge(Layer.succeed(ProviderService, service)),      Layer.provide(Layer.mock(ProviderAuthService, { tryHandlePromptCommand })),      Layer.provideMerge(makeProviderRegistryLayer(providerSnapshots as never)),      Layer.provideMerge(        Layer.mock(GitWorkflowService.GitWorkflowService)({          renameBranch,          pruneWorktrees,          createWorktree,        } 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,          refreshPullRequestStatus: () =>            Effect.die("refreshPullRequestStatus should not be called in this test"),          streamStatus: () => Stream.die("streamStatus should not be called in this test"),        }),      ),      Layer.provideMerge(        Layer.mock(TextGeneration, {          generateBranchName,          generateThreadTitle,        }),      ),      Layer.provideMerge(ServerSettingsService.layerTest()),      Layer.provideMerge(SqlitePersistenceMemory),      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));    const runEffect = <A, E>(effect: Effect.Effect<A, E>) => runtime!.runPromise(effect);

Lines 1–500 of 4178. Load more