import fs from "node:fs";import os from "node:os";import path from "node:path";import { execFileSync } from "node:child_process"; import * as NodeServices from "@effect/platform-node/NodeServices";import {  ApprovalRequestId,  type OrchestrationEvent,  type OrchestrationThread,} from "@t3tools/contracts";import {  Effect,  Exit,  Layer,  ManagedRuntime,  Option,  Schedule,  Schema,  Scope,  Stream,} from "effect"; import { CheckpointStoreLive } from "../src/checkpointing/Layers/CheckpointStore.ts";import { CheckpointStore } from "../src/checkpointing/Services/CheckpointStore.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 { ProviderUnsupportedError } from "../src/provider/Errors.ts";import { ProviderAdapterRegistry } from "../src/provider/Services/ProviderAdapterRegistry.ts";import { ProviderSessionDirectoryLive } from "../src/provider/Layers/ProviderSessionDirectory.ts";import { makeProviderServiceLive } from "../src/provider/Layers/ProviderService.ts";import { ProviderService } from "../src/provider/Services/ProviderService.ts";import { CheckpointReactorLive } from "../src/orchestration/Layers/CheckpointReactor.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 { 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 { OrchestrationReactor } from "../src/orchestration/Services/OrchestrationReactor.ts";import { ProjectionSnapshotQuery } from "../src/orchestration/Services/ProjectionSnapshotQuery.ts"; import {  makeTestProviderAdapterHarness,  type TestProviderAdapterHarness,} from "./TestProviderAdapter.integration.ts"; function runGit(cwd: string, args: ReadonlyArray<string>) {  return execFileSync("git", args, {    cwd,    stdio: ["ignore", "pipe", "pipe"],    encoding: "utf8",  });} function initializeGitWorkspace(cwd: string) {  runGit(cwd, ["init", "--initial-branch=main"]);  runGit(cwd, ["config", "user.email", "test@example.com"]);  runGit(cwd, ["config", "user.name", "Test User"]);  fs.writeFileSync(path.join(cwd, "README.md"), "v1\n", "utf8");  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 = 3000,): 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;  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 dispose: Effect.Effect<void, never>;} export const makeOrchestrationIntegrationHarness = Effect.gen(function* () {  const sleep = (ms: number) => Effect.sleep(ms);  const adapterHarness = yield* makeTestProviderAdapterHarness;   const rootDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-orchestration-integration-"));  const workspaceDir = path.join(rootDir, "workspace");  const stateDir = path.join(rootDir, "state");  const dbPath = path.join(stateDir, "state.sqlite");  fs.mkdirSync(workspaceDir, { recursive: true });  fs.mkdirSync(stateDir, { recursive: true });  initializeGitWorkspace(workspaceDir);   const registry: typeof ProviderAdapterRegistry.Service = {    getByProvider: (provider) =>      provider === "codex"        ? Effect.succeed(adapterHarness.adapter)        : Effect.fail(new ProviderUnsupportedError({ provider })),    listProviders: () => Effect.succeed(["codex"]),  };   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 providerLayer = makeProviderServiceLive().pipe(    Layer.provide(providerSessionDirectoryLayer),    Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)),  );  const checkpointStoreLayer = CheckpointStoreLive.pipe(Layer.provide(NodeServices.layer));   const runtimeServicesLayer = Layer.mergeAll(    orchestrationLayer,    OrchestrationProjectionSnapshotQueryLive,    ProjectionCheckpointRepositoryLive,    ProjectionPendingApprovalRepositoryLive,    checkpointStoreLayer,    providerLayer,  );  const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe(    Layer.provideMerge(runtimeServicesLayer),  );  const providerCommandReactorLayer = ProviderCommandReactorLive.pipe(    Layer.provideMerge(runtimeServicesLayer),  );  const checkpointReactorLayer = CheckpointReactorLive.pipe(    Layer.provideMerge(runtimeServicesLayer),  );  const orchestrationReactorLayer = OrchestrationReactorLive.pipe(    Layer.provideMerge(runtimeIngestionLayer),    Layer.provideMerge(providerCommandReactorLayer),    Layer.provideMerge(checkpointReactorLayer),  );  const layer = orchestrationReactorLayer.pipe(    Layer.provide(persistenceLayer),    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 scope = yield* Scope.make("sequential");  yield* tryRuntimePromise("start OrchestrationReactor", () =>    runtime.runPromise(reactor.start.pipe(Scope.provide(scope))),  ).pipe(Effect.orDie);  yield* 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.makeUnsafe(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    >;   let disposed = false;  const dispose = Effect.gen(function* () {    if (disposed) {      return;    }    disposed = true;     const shutdown = Effect.gen(function* () {      const stopAllExit = yield* Effect.exit(        Effect.promise(() => runtime.runPromise(providerService.stopAll())),      );      const closeScopeExit = yield* Effect.exit(        Effect.promise(() => Effect.runPromise(Scope.close(scope, Exit.void))),      );      const disposeRuntimeExit = yield* Effect.exit(Effect.promise(() => runtime.dispose()));       const failureCause = Exit.isFailure(stopAllExit)        ? stopAllExit.cause        : Exit.isFailure(closeScopeExit)          ? closeScopeExit.cause          : Exit.isFailure(disposeRuntimeExit)            ? disposeRuntimeExit.cause            : null;       if (failureCause) {        return yield* Effect.failCause(failureCause);      }    });     yield* shutdown.pipe(      Effect.ensuring(        Effect.sync(() => {          fs.rmSync(rootDir, { recursive: true, force: true });        }),      ),    );  });   return {    rootDir,    workspaceDir,    dbPath,    adapterHarness,    engine,    snapshotQuery,    providerService,    checkpointStore,    checkpointRepository,    pendingApprovalRepository,    waitForThread,    waitForDomainEvent,    waitForPendingApproval,    dispose,  } satisfies OrchestrationIntegrationHarness;});