import { execFileSync } from "node:child_process"; import * as NodeServices from "@effect/platform-node/NodeServices";import {  ApprovalRequestId,  CodexSettings,  ProviderDriverKind,  type OrchestrationEvent,  type OrchestrationThread,} from "@t3tools/contracts";import {  Effect,  Exit,  FileSystem,  Layer,  ManagedRuntime,  Option,  Path,  Ref,  Schedule,  Schema,  Scope,  Stream,} from "effect"; import { CheckpointStoreLive } from "../src/checkpointing/Layers/CheckpointStore.ts";import { CheckpointStore } from "../src/checkpointing/Services/CheckpointStore.ts";import { TextGeneration, type TextGenerationShape } from "../src/textGeneration/TextGeneration.ts";import { OrchestrationCommandReceiptRepositoryLive } from "../src/persistence/Layers/OrchestrationCommandReceipts.ts";import { OrchestrationEventStoreLive } from "../src/persistence/Layers/OrchestrationEventStore.ts";import { ProjectionCheckpointRepositoryLive } from "../src/persistence/Layers/ProjectionCheckpoints.ts";import { ProjectionPendingApprovalRepositoryLive } from "../src/persistence/Layers/ProjectionPendingApprovals.ts";import { ProviderSessionRuntimeRepositoryLive } from "../src/persistence/Layers/ProviderSessionRuntime.ts";import { makeSqlitePersistenceLive } from "../src/persistence/Layers/Sqlite.ts";import { ProjectionCheckpointRepository } from "../src/persistence/Services/ProjectionCheckpoints.ts";import { ProjectionPendingApprovalRepository } from "../src/persistence/Services/ProjectionPendingApprovals.ts";import { makeAdapterRegistryMock } from "../src/provider/testUtils/providerAdapterRegistryMock.ts";import { ProviderAdapterRegistry } from "../src/provider/Services/ProviderAdapterRegistry.ts";import { ProviderSessionDirectoryLive } from "../src/provider/Layers/ProviderSessionDirectory.ts";import { ServerSettingsService } from "../src/serverSettings.ts";import { makeProviderServiceLive } from "../src/provider/Layers/ProviderService.ts";import { makeCodexAdapter } from "../src/provider/Layers/CodexAdapter.ts";import {  NoOpProviderEventLoggers,  ProviderEventLoggers,} from "../src/provider/Layers/ProviderEventLoggers.ts";import { ProviderService } from "../src/provider/Services/ProviderService.ts";import { AnalyticsService } from "../src/telemetry/Services/AnalyticsService.ts";import { CheckpointReactorLive } from "../src/orchestration/Layers/CheckpointReactor.ts";import { RepositoryIdentityResolverLive } from "../src/project/Layers/RepositoryIdentityResolver.ts";import { OrchestrationEngineLive } from "../src/orchestration/Layers/OrchestrationEngine.ts";import { OrchestrationProjectionPipelineLive } from "../src/orchestration/Layers/ProjectionPipeline.ts";import { OrchestrationProjectionSnapshotQueryLive } from "../src/orchestration/Layers/ProjectionSnapshotQuery.ts";import { RuntimeReceiptBusTest } from "../src/orchestration/Layers/RuntimeReceiptBus.ts";import { OrchestrationReactorLive } from "../src/orchestration/Layers/OrchestrationReactor.ts";import { ProviderCommandReactorLive } from "../src/orchestration/Layers/ProviderCommandReactor.ts";import { ProviderRuntimeIngestionLive } from "../src/orchestration/Layers/ProviderRuntimeIngestion.ts";import {  OrchestrationEngineService,  type OrchestrationEngineShape,} from "../src/orchestration/Services/OrchestrationEngine.ts";import { ThreadDeletionReactor } from "../src/orchestration/Services/ThreadDeletionReactor.ts";import { OrchestrationReactor } from "../src/orchestration/Services/OrchestrationReactor.ts";import { ProjectionSnapshotQuery } from "../src/orchestration/Services/ProjectionSnapshotQuery.ts";import {  RuntimeReceiptBus,  type OrchestrationRuntimeReceipt,} from "../src/orchestration/Services/RuntimeReceiptBus.ts"; import {  makeTestProviderAdapterHarness,  type TestProviderAdapterHarness,} from "./TestProviderAdapter.integration.ts";import { deriveServerPaths, ServerConfig } from "../src/config.ts";import { WorkspaceEntriesLive } from "../src/workspace/Layers/WorkspaceEntries.ts";import { WorkspacePathsLive } from "../src/workspace/Layers/WorkspacePaths.ts";import * as GitVcsDriver from "../src/vcs/GitVcsDriver.ts";import * as VcsDriverRegistry from "../src/vcs/VcsDriverRegistry.ts";import { VcsStatusBroadcaster } from "../src/vcs/VcsStatusBroadcaster.ts";import { GitWorkflowService } from "../src/git/GitWorkflowService.ts";import * as VcsProcess from "../src/vcs/VcsProcess.ts"; function runGit(cwd: string, args: ReadonlyArray<string>) {  return execFileSync("git", args, {    cwd,    stdio: ["ignore", "pipe", "pipe"],    encoding: "utf8",  });} const initializeGitWorkspace = Effect.fn(function* (cwd: string) {  runGit(cwd, ["init", "--initial-branch=main"]);  runGit(cwd, ["config", "user.email", "test@example.com"]);  runGit(cwd, ["config", "user.name", "Test User"]);  const fileSystem = yield* FileSystem.FileSystem;  const { join } = yield* Path.Path;  yield* fileSystem.writeFileString(join(cwd, "README.md"), "v1\n");  runGit(cwd, ["add", "."]);  runGit(cwd, ["commit", "-m", "Initial"]);}); export function gitRefExists(cwd: string, ref: string): boolean {  try {    runGit(cwd, ["show-ref", "--verify", "--quiet", ref]);    return true;  } catch {    return false;  }} export function gitShowFileAtRef(cwd: string, ref: string, filePath: string): string {  return runGit(cwd, ["show", `${ref}:${filePath}`]);} class WaitForTimeoutError extends Schema.TaggedErrorClass<WaitForTimeoutError>()(  "WaitForTimeoutError",  {    description: Schema.String,  },) {} function waitFor<A, E>(  read: Effect.Effect<A, E>,  predicate: (value: A) => boolean,  description: string,  timeoutMs?: number,): Effect.Effect<A, never>;function waitFor<A, B extends A, E>(  read: Effect.Effect<A, E>,  predicate: (value: A) => value is B,  description: string,  timeoutMs?: number,): Effect.Effect<B, never>;function waitFor<A, E>(  read: Effect.Effect<A, E>,  predicate: (value: A) => boolean,  description: string,  timeoutMs = 40_000,): Effect.Effect<A, never> {  const RETRY_SIGNAL = "wait_for_retry";  const retryIntervalMs = 10;  const maxRetries = Math.max(0, Math.floor(timeoutMs / retryIntervalMs));  const retrySchedule = Schedule.spaced(`${retryIntervalMs} millis`);   return read.pipe(    Effect.filterOrFail(predicate, () => RETRY_SIGNAL),    Effect.retry({      schedule: retrySchedule,      times: maxRetries,      while: (error) => error === RETRY_SIGNAL,    }),    Effect.mapError((error) =>      error === RETRY_SIGNAL ? new WaitForTimeoutError({ description }) : error,    ),    Effect.orDie,  );} class OrchestrationHarnessRuntimeError extends Schema.TaggedErrorClass<OrchestrationHarnessRuntimeError>()(  "OrchestrationHarnessRuntimeError",  {    operation: Schema.String,    cause: Schema.optional(Schema.Defect),  },) {} const tryRuntimePromise = <A>(operation: string, run: () => Promise<A>) =>  Effect.tryPromise({    try: run,    catch: (cause) => new OrchestrationHarnessRuntimeError({ operation, cause }),  }); export interface OrchestrationIntegrationHarness {  readonly rootDir: string;  readonly workspaceDir: string;  readonly dbPath: string;  readonly adapterHarness: TestProviderAdapterHarness | null;  readonly engine: OrchestrationEngineShape;  readonly snapshotQuery: ProjectionSnapshotQuery["Service"];  readonly providerService: ProviderService["Service"];  readonly checkpointStore: CheckpointStore["Service"];  readonly checkpointRepository: ProjectionCheckpointRepository["Service"];  readonly pendingApprovalRepository: ProjectionPendingApprovalRepository["Service"];  readonly waitForThread: (    threadId: string,    predicate: (thread: OrchestrationThread) => boolean,    timeoutMs?: number,  ) => Effect.Effect<OrchestrationThread, never>;  readonly waitForDomainEvent: (    predicate: (event: OrchestrationEvent) => boolean,    timeoutMs?: number,  ) => Effect.Effect<ReadonlyArray<OrchestrationEvent>, never>;  readonly waitForPendingApproval: (    requestId: string,    predicate: (row: {      readonly status: "pending" | "resolved";      readonly decision: "accept" | "acceptForSession" | "decline" | "cancel" | null;      readonly resolvedAt: string | null;    }) => boolean,    timeoutMs?: number,  ) => Effect.Effect<    {      readonly status: "pending" | "resolved";      readonly decision: "accept" | "acceptForSession" | "decline" | "cancel" | null;      readonly resolvedAt: string | null;    },    never  >;  readonly waitForReceipt: {    (      predicate: (receipt: OrchestrationRuntimeReceipt) => boolean,      timeoutMs?: number,    ): Effect.Effect<OrchestrationRuntimeReceipt, never>;    <Receipt extends OrchestrationRuntimeReceipt>(      predicate: (receipt: OrchestrationRuntimeReceipt) => receipt is Receipt,      timeoutMs?: number,    ): Effect.Effect<Receipt, never>;  };  readonly dispose: Effect.Effect<void, never>;} interface MakeOrchestrationIntegrationHarnessOptions {  readonly provider?: ProviderDriverKind;  readonly realCodex?: boolean;} export const makeOrchestrationIntegrationHarness = (  options?: MakeOrchestrationIntegrationHarnessOptions,) =>  Effect.gen(function* () {    const path = yield* Path.Path;    const fileSystem = yield* FileSystem.FileSystem;     const provider = options?.provider ?? ProviderDriverKind.make("codex");    const useRealCodex = options?.realCodex === true;    const adapterHarness = useRealCodex      ? null      : yield* makeTestProviderAdapterHarness({          provider,        });    const fakeRegistry = adapterHarness      ? Layer.succeed(          ProviderAdapterRegistry,          makeAdapterRegistryMock({ [adapterHarness.provider]: adapterHarness.adapter }),        )      : null;    const rootDir = yield* fileSystem.makeTempDirectoryScoped({      prefix: "t3-orchestration-integration-",    });    const workspaceDir = path.join(rootDir, "workspace");    const { stateDir, dbPath } = yield* deriveServerPaths(rootDir, undefined).pipe(      Effect.provideService(Path.Path, path),    );    yield* fileSystem.makeDirectory(workspaceDir, { recursive: true });    yield* fileSystem.makeDirectory(stateDir, { recursive: true });    yield* initializeGitWorkspace(workspaceDir);     const persistenceLayer = makeSqlitePersistenceLive(dbPath);    const orchestrationLayer = OrchestrationEngineLive.pipe(      Layer.provide(OrchestrationProjectionPipelineLive),      Layer.provide(OrchestrationEventStoreLive),      Layer.provide(OrchestrationCommandReceiptRepositoryLive),    );    const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe(      Layer.provide(ProviderSessionRuntimeRepositoryLive),    );    const realCodexRegistry = Layer.effect(      ProviderAdapterRegistry,      Effect.gen(function* () {        const codexSettings = Schema.decodeSync(CodexSettings)({});        const codexAdapter = yield* makeCodexAdapter(codexSettings);        return makeAdapterRegistryMock({          [ProviderDriverKind.make("codex")]: codexAdapter,        });      }),    ).pipe(      Layer.provideMerge(ServerConfig.layerTest(workspaceDir, rootDir)),      Layer.provideMerge(NodeServices.layer),      Layer.provideMerge(providerSessionDirectoryLayer),    );    const providerEventLoggersLayer = Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers);    const providerLayer = useRealCodex      ? makeProviderServiceLive().pipe(          Layer.provide(providerSessionDirectoryLayer),          Layer.provide(realCodexRegistry),          Layer.provide(AnalyticsService.layerTest),          Layer.provide(providerEventLoggersLayer),        )      : makeProviderServiceLive().pipe(          Layer.provide(providerSessionDirectoryLayer),          Layer.provide(fakeRegistry!),          Layer.provide(AnalyticsService.layerTest),          Layer.provide(providerEventLoggersLayer),        );     const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(VcsDriverRegistry.layer));    const projectionSnapshotQueryLayer = OrchestrationProjectionSnapshotQueryLive;    const runtimeServicesLayer = Layer.mergeAll(      projectionSnapshotQueryLayer,      orchestrationLayer.pipe(Layer.provide(projectionSnapshotQueryLayer)),      ProjectionCheckpointRepositoryLive,      ProjectionPendingApprovalRepositoryLive,      checkpointStoreLayer,      providerLayer,      RuntimeReceiptBusTest,    );    const serverSettingsLayer = ServerSettingsService.layerTest();    const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe(      Layer.provideMerge(runtimeServicesLayer),      Layer.provideMerge(serverSettingsLayer),    );    const gitWorkflowLayer = Layer.mock(GitWorkflowService)({      renameBranch: (input: Parameters<GitVcsDriver.GitVcsDriverShape["renameBranch"]>[0]) =>        Effect.succeed({ branch: input.newBranch }),    });    const textGenerationLayer = Layer.succeed(TextGeneration, {      generateBranchName: () => Effect.succeed({ branch: "update" }),      generateThreadTitle: () => Effect.succeed({ title: "New thread" }),    } as unknown as TextGenerationShape);    const providerCommandReactorLayer = ProviderCommandReactorLive.pipe(      Layer.provideMerge(runtimeServicesLayer),      Layer.provideMerge(gitWorkflowLayer),      Layer.provideMerge(textGenerationLayer),      Layer.provideMerge(serverSettingsLayer),    );    const checkpointReactorLayer = CheckpointReactorLive.pipe(      Layer.provideMerge(runtimeServicesLayer),      Layer.provideMerge(        Layer.succeed(VcsStatusBroadcaster, {          getStatus: () => Effect.die("getStatus should not be called in this test"),          refreshLocalStatus: () =>            Effect.succeed({              isRepo: true,              hasPrimaryRemote: false,              isDefaultRef: true,              refName: "main",              hasWorkingTreeChanges: false,              workingTree: { files: [], insertions: 0, deletions: 0 },            }),          refreshStatus: () => Effect.die("refreshStatus should not be called in this test"),          streamStatus: () => Stream.empty,        }),      ),      Layer.provideMerge(        WorkspaceEntriesLive.pipe(          Layer.provide(WorkspacePathsLive),          Layer.provideMerge(VcsDriverRegistry.layer),          Layer.provide(NodeServices.layer),        ),      ),      Layer.provideMerge(WorkspacePathsLive),      Layer.provideMerge(VcsProcess.layer),    );    const orchestrationReactorLayer = OrchestrationReactorLive.pipe(      Layer.provideMerge(runtimeIngestionLayer),      Layer.provideMerge(providerCommandReactorLayer),      Layer.provideMerge(checkpointReactorLayer),      Layer.provideMerge(        Layer.succeed(ThreadDeletionReactor, {          start: () => Effect.void,          drain: Effect.void,        }),      ),    );    const layer = Layer.empty.pipe(      Layer.provideMerge(runtimeServicesLayer),      Layer.provideMerge(orchestrationReactorLayer),      Layer.provide(persistenceLayer),      Layer.provideMerge(RepositoryIdentityResolverLive),      Layer.provideMerge(ServerSettingsService.layerTest()),      Layer.provideMerge(ServerConfig.layerTest(workspaceDir, rootDir)),      Layer.provideMerge(NodeServices.layer),    );     const runtime = ManagedRuntime.make(layer);    const engine = yield* tryRuntimePromise("load OrchestrationEngine service", () =>      runtime.runPromise(Effect.service(OrchestrationEngineService)),    ).pipe(Effect.orDie);    const reactor = yield* tryRuntimePromise("load OrchestrationReactor service", () =>      runtime.runPromise(Effect.service(OrchestrationReactor)),    ).pipe(Effect.orDie);    const snapshotQuery = yield* tryRuntimePromise("load ProjectionSnapshotQuery service", () =>      runtime.runPromise(Effect.service(ProjectionSnapshotQuery)),    ).pipe(Effect.orDie);    const providerService = yield* tryRuntimePromise("load ProviderService service", () =>      runtime.runPromise(Effect.service(ProviderService)),    ).pipe(Effect.orDie);    const checkpointStore = yield* tryRuntimePromise("load CheckpointStore service", () =>      runtime.runPromise(Effect.service(CheckpointStore)),    ).pipe(Effect.orDie);    const checkpointRepository = yield* tryRuntimePromise(      "load ProjectionCheckpointRepository service",      () => runtime.runPromise(Effect.service(ProjectionCheckpointRepository)),    ).pipe(Effect.orDie);    const pendingApprovalRepository = yield* tryRuntimePromise(      "load ProjectionPendingApprovalRepository service",      () => runtime.runPromise(Effect.service(ProjectionPendingApprovalRepository)),    ).pipe(Effect.orDie);    const runtimeReceiptBus = yield* tryRuntimePromise("load RuntimeReceiptBus service", () =>      runtime.runPromise(Effect.service(RuntimeReceiptBus)),    ).pipe(Effect.orDie);     const scope = yield* Scope.make("sequential");    yield* tryRuntimePromise("start OrchestrationReactor", () =>      runtime.runPromise(reactor.start().pipe(Scope.provide(scope))),    ).pipe(Effect.orDie);    const receiptHistory = yield* Ref.make<ReadonlyArray<OrchestrationRuntimeReceipt>>([]);    yield* Stream.runForEach(runtimeReceiptBus.streamEventsForTest, (receipt) =>      Ref.update(receiptHistory, (history) => [...history, receipt]).pipe(Effect.asVoid),    ).pipe(Effect.forkIn(scope));    yield* Effect.sleep(10);     const waitForThread: OrchestrationIntegrationHarness["waitForThread"] = (      threadId,      predicate,      timeoutMs,    ) =>      waitFor(        snapshotQuery          .getSnapshot()          .pipe(            Effect.map(              (snapshot) => snapshot.threads.find((thread) => thread.id === threadId) ?? null,            ),          ),        (thread): thread is OrchestrationThread => thread !== null && predicate(thread),        `projected thread '${threadId}'`,        timeoutMs,      ) as Effect.Effect<OrchestrationThread, never>;     const waitForDomainEvent: OrchestrationIntegrationHarness["waitForDomainEvent"] = (      predicate,      timeoutMs,    ) =>      waitFor(        Stream.runCollect(engine.readEvents(0)).pipe(          Effect.map((chunk): ReadonlyArray<OrchestrationEvent> => Array.from(chunk)),        ),        (events) => events.some(predicate),        "domain event",        timeoutMs,      );     const waitForPendingApproval: OrchestrationIntegrationHarness["waitForPendingApproval"] = (      requestId,      predicate,      timeoutMs,    ) =>      waitFor(        pendingApprovalRepository          .getByRequestId({ requestId: ApprovalRequestId.make(requestId) })          .pipe(            Effect.map((row) =>              Option.match(row, {                onNone: () => null,                onSome: (value) => ({                  status: value.status,                  decision: value.decision,                  resolvedAt: value.resolvedAt,                }),              }),            ),          ),        (          row,        ): row is {          readonly status: "pending" | "resolved";          readonly decision: "accept" | "acceptForSession" | "decline" | "cancel" | null;          readonly resolvedAt: string | null;        } => row !== null && predicate(row),        `pending approval '${requestId}'`,        timeoutMs,      ) as Effect.Effect<        {          readonly status: "pending" | "resolved";          readonly decision: "accept" | "acceptForSession" | "decline" | "cancel" | null;          readonly resolvedAt: string | null;        },        never      >;     function waitForReceipt(      predicate: (receipt: OrchestrationRuntimeReceipt) => boolean,      timeoutMs?: number,    ): Effect.Effect<OrchestrationRuntimeReceipt, never>;    function waitForReceipt<Receipt extends OrchestrationRuntimeReceipt>(      predicate: (receipt: OrchestrationRuntimeReceipt) => receipt is Receipt,      timeoutMs?: number,    ): Effect.Effect<Receipt, never>;    function waitForReceipt(      predicate: (receipt: OrchestrationRuntimeReceipt) => boolean,      timeoutMs?: number,    ) {      const readMatchingReceipt = Ref.get(receiptHistory).pipe(        Effect.map((history) => history.find(predicate)),      );       return waitFor(        readMatchingReceipt,        (receipt): receipt is OrchestrationRuntimeReceipt => receipt !== undefined,

Lines 1–500 of 548. Load more