import * as NodeServices from "@effect/platform-node/NodeServices";import { assert, it } from "@effect/vitest";import { vi } from "vite-plus/test";import {  type ApplicationStoredEvent,  CommandId,  ContextTransferId,  EventId,  MessageId,  NodeId,  RuntimeRequestId,  TurnItemId,  type ModelSelection,  ProjectId,  ProviderDriverKind,  ProviderInstanceId,  ProviderThreadId,  RunId,  ThreadId,} from "@t3tools/contracts";import * as Effect from "effect/Effect";import * as DateTime from "effect/DateTime";import * as Deferred from "effect/Deferred";import * as Fiber from "effect/Fiber";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Queue from "effect/Queue";import * as Stream from "effect/Stream";import * as TestClock from "effect/testing/TestClock";import * as SqlClient from "effect/unstable/sql/SqlClient"; import * as CheckpointStore from "../checkpointing/CheckpointStore.ts";import * as GitWorkflow from "../git/GitWorkflowService.ts";import { ServerConfig } from "../config.ts";import { OrchestrationEngineService } from "../orchestration/Services/OrchestrationEngine.ts";import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts";import { OrchestrationLayerLive } from "../orchestration/runtimeLayer.ts";import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";import { OrchestrationEventStore } from "../persistence/Services/OrchestrationEventStore.ts";import { ProjectEnrichmentService } from "../project/ProjectEnrichmentService.ts";import * as ProjectService from "../project/ProjectService.ts";import { ServerSettingsService } from "../serverSettings.ts";import { layer as mcpSessionRegistryTestLayer } from "../mcp/McpSessionRegistry.testkit.ts";import { ProviderInstanceRegistry } from "../provider/Services/ProviderInstanceRegistry.ts";import type { ProviderInstance } from "../provider/ProviderDriver.ts";import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts";import * as VcsProcess from "../vcs/VcsProcess.ts";import { WorkspacePaths } from "../workspace/WorkspacePaths.ts";import { LegacyV1ThreadImporter, LegacyV1ThreadImportError } from "./LegacyV1ThreadImporter.ts";import {  OrchestratorDispatchError,  OrchestratorProjectionError,  OrchestratorV2,} from "./Orchestrator.ts";import { OrchestrationEffectWorkerV2 } from "./EffectWorker.ts";import { EventSinkV2 } from "./EventSink.ts";import { ProjectionMaintenanceV2 } from "./ProjectionMaintenance.ts";import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts";import {  OrchestrationV2EventSinkLayerLive,  OrchestrationV2LayerLive,  ProjectServiceLayerLive,} from "./runtimeLayer.ts";import { shellStreamItemFromThreadShell } from "./ShellStream.ts";import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";import { ThreadManagementService } from "./ThreadManagementService.ts";import {  ThreadCommandExecutor,  layer as threadCommandExecutorLayer,} from "./ThreadCommandExecutor.ts"; const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), {  prefix: "t3-orchestration-v2-runtime-layer-",}); const modelSelection = {  instanceId: ProviderInstanceId.make("codex"),  model: "gpt-5.4",} satisfies ModelSelection; const VcsDriverRegistryTestLayer = VcsDriverRegistry.layer.pipe(  Layer.provide(VcsProcess.layer),  Layer.provide(ServerConfigLayer),  Layer.provide(NodeServices.layer),); const CheckpointStoreTestLayer = CheckpointStore.layer.pipe(  Layer.provide(VcsDriverRegistryTestLayer),);const GitWorkflowTestLayer = Layer.mock(GitWorkflow.GitWorkflowService)({  pruneWorktrees: () => Effect.void,  createWorktree: () => Effect.succeed({} as never),});const ProjectServiceTestLayer = Layer.mock(ProjectService.ProjectService)({  getById: () => Effect.succeed(Option.none()),}); const driver = ProviderDriverKind.make("codex");const orchestrationAdapter = {  instanceId: modelSelection.instanceId,  driver,  getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),  planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),  openSession: () => Effect.die("sessions are not used by lifecycle tests"),} as ProviderAdapterV2Shape;const providerInstance = {  instanceId: modelSelection.instanceId,  driverKind: driver,  continuationIdentity: {    driverKind: driver,    continuationKey: "codex:test",  },  displayName: "Codex test",  enabled: true,  snapshot: {} as ProviderInstance["snapshot"],  orchestrationAdapter,  textGeneration: {} as ProviderInstance["textGeneration"],} satisfies ProviderInstance; const TestProviderInstanceRegistry = Layer.succeed(ProviderInstanceRegistry, {  getInstance: (instanceId) =>    Effect.succeed(instanceId === providerInstance.instanceId ? providerInstance : undefined),  listInstances: Effect.succeed([providerInstance]),  listUnavailable: Effect.succeed([]),  streamChanges: Stream.empty,  subscribeChanges: Effect.never,}); const TestLayer = Layer.merge(OrchestrationV2LayerLive, OrchestrationV2EventSinkLayerLive).pipe(  Layer.provide(mcpSessionRegistryTestLayer),  Layer.provide(SqlitePersistenceMemory),  Layer.provide(CheckpointStoreTestLayer),  Layer.provide(ServerConfigLayer),  Layer.provide(ServerSettingsService.layerTest()),  Layer.provide(TestProviderInstanceRegistry),  Layer.provide(GitWorkflowTestLayer),  Layer.provide(ProjectServiceTestLayer),  Layer.provide(NodeServices.layer),); const LegacyImportTestLayer = OrchestrationV2LayerLive.pipe(  Layer.provide(mcpSessionRegistryTestLayer),  Layer.provideMerge(SqlitePersistenceMemory),  Layer.provide(CheckpointStoreTestLayer),  Layer.provide(ServerConfigLayer),  Layer.provide(ServerSettingsService.layerTest()),  Layer.provide(TestProviderInstanceRegistry),  Layer.provide(GitWorkflowTestLayer),  Layer.provide(ProjectServiceTestLayer),  Layer.provide(NodeServices.layer),); const ProjectDeletionTestLayer = Layer.mergeAll(  OrchestrationV2LayerLive.pipe(Layer.provide(ProjectServiceLayerLive)),  ProjectServiceLayerLive,  OrchestrationV2EventSinkLayerLive,  threadCommandExecutorLayer,).pipe(  Layer.provide(    Layer.mock(ProjectEnrichmentService)({      peek: () =>        Effect.succeed({          repositoryIdentity: null,          faviconPath: null,          repositoryIdentityResolved: false,        }),      getAvailable: () =>        Effect.succeed({          repositoryIdentity: null,          faviconPath: null,          repositoryIdentityResolved: false,        }),      invalidate: () => Effect.void,    }),  ),  Layer.provide(    Layer.mock(WorkspacePaths)({      normalizeWorkspaceRoot: (workspaceRoot) => Effect.succeed(workspaceRoot),    }),  ),  Layer.provide(mcpSessionRegistryTestLayer),  Layer.provide(SqlitePersistenceMemory),  Layer.provide(CheckpointStoreTestLayer),  Layer.provide(ServerConfigLayer),  Layer.provide(ServerSettingsService.layerTest()),  Layer.provide(TestProviderInstanceRegistry),  Layer.provide(GitWorkflowTestLayer),  Layer.provide(NodeServices.layer),); it.layer(ProjectDeletionTestLayer)("project deletion during thread commands", (it) => {  it.effect("waits for an in-flight thread update before planning deletion", () =>    Effect.gen(function* () {      const projects = yield* ProjectService.ProjectService;      const orchestrator = yield* OrchestratorV2;      const eventSink = yield* EventSinkV2;      const executor = yield* ThreadCommandExecutor;      const projectId = ProjectId.make("runtime-project-delete-concurrent");      const threadId = ThreadId.make("runtime-thread-delete-concurrent");      const updateCommandId = CommandId.make("runtime-thread-delete-concurrent-update");      yield* projects.create({        commandId: CommandId.make("runtime-project-delete-concurrent-create"),        projectId,        title: "Concurrent deletion",        workspaceRoot: "/work/concurrent-deletion",      });      yield* orchestrator.dispatch({        type: "thread.create",        commandId: CommandId.make("runtime-thread-delete-concurrent-create"),        createdBy: "user",        creationSource: "web",        threadId,        projectId,        title: "Original title",        modelSelection,        runtimeMode: "full-access",        interactionMode: "default",        branch: null,        worktreePath: null,      });       const updateReady = yield* Deferred.make<void>();      const releaseUpdate = yield* Deferred.make<void>();      const deletionQueued = yield* Deferred.make<void>();      const commitCommand = eventSink.commitCommand;      const withLock = executor.withLock;      let threadLockRequests = 0;      const commitSpy = vi        .spyOn(eventSink, "commitCommand")        .mockImplementation((input) =>          input.commandId === updateCommandId            ? Deferred.succeed(updateReady, undefined).pipe(                Effect.andThen(Deferred.await(releaseUpdate)),                Effect.andThen(commitCommand(input)),              )            : commitCommand(input),        );      const observeLock: ThreadCommandExecutor["Service"]["withLock"] = (key, effect) => {        if (key !== threadId || ++threadLockRequests !== 2) return withLock(key, effect);        return Deferred.succeed(deletionQueued, undefined).pipe(          Effect.andThen(withLock(key, effect)),        );      };      const lockSpy = vi.spyOn(executor, "withLock").mockImplementation(observeLock);      yield* Effect.gen(function* () {        const updateFiber = yield* orchestrator          .dispatch({            type: "thread.metadata.update",            commandId: updateCommandId,            threadId,            title: "Updated before deletion",          })          .pipe(Effect.forkChild({ startImmediately: true }));        yield* Effect.raceFirst(          Deferred.await(updateReady),          Fiber.join(updateFiber).pipe(            Effect.andThen(Effect.die("The update completed before reaching its commit barrier.")),          ),        );        const deleteFiber = yield* projects          .delete({            commandId: CommandId.make("runtime-project-delete-concurrent-delete"),            projectId,            force: true,          })          .pipe(Effect.forkChild({ startImmediately: true }));        yield* Effect.raceFirst(          Deferred.await(deletionQueued),          Fiber.join(deleteFiber).pipe(            Effect.andThen(Effect.die("Project deletion bypassed the in-flight thread command.")),          ),        );        yield* Deferred.succeed(releaseUpdate, undefined);        yield* Fiber.join(updateFiber);        const deletedProject = yield* Fiber.join(deleteFiber);        const projection = yield* orchestrator.getThreadProjection(threadId);        assert.isNotNull(deletedProject.deletedAt);        assert.isNotNull(projection.thread.deletedAt);        assert.equal(projection.thread.title, "Updated before deletion");      }).pipe(        Effect.ensuring(          Effect.sync(() => {            commitSpy.mockRestore();            lockSpy.mockRestore();          }),        ),      );    }),  );}); const SharedApplicationDataPlaneTestLayer = Layer.merge(  OrchestrationLayerLive,  OrchestrationV2LayerLive,).pipe(  Layer.provide(    Layer.succeed(ProjectEnrichmentService, {      peek: () =>        Effect.succeed({          repositoryIdentity: null,          faviconPath: null,          repositoryIdentityResolved: false,        }),      request: () => Effect.void,      getAvailable: () =>        Effect.succeed({          repositoryIdentity: null,          faviconPath: null,          repositoryIdentityResolved: false,        }),      invalidate: () => Effect.void,      subscribeChanges: Effect.never,    }),  ),  Layer.provide(mcpSessionRegistryTestLayer),  Layer.provideMerge(SqlitePersistenceMemory),  Layer.provide(CheckpointStoreTestLayer),  Layer.provide(ServerConfigLayer),  Layer.provide(ServerSettingsService.layerTest()),  Layer.provide(TestProviderInstanceRegistry),  Layer.provide(GitWorkflowTestLayer),  Layer.provide(ProjectServiceTestLayer),  Layer.provide(NodeServices.layer),); it.layer(TestLayer)("OrchestrationV2LayerLive", (it) => {  it.effect("creates and reads a thread through the production V2 composition", () =>    Effect.gen(function* () {      const orchestrator = yield* OrchestratorV2;      const threadId = ThreadId.make("runtime-layer-thread");      const projectId = ProjectId.make("runtime-layer-project");       const result = yield* orchestrator.dispatch({        type: "thread.create",        createdBy: "user",        creationSource: "web",        commandId: CommandId.make("runtime-layer-create"),        threadId,        projectId,        title: "Runtime layer thread",        modelSelection: modelSelection,        runtimeMode: "full-access",        interactionMode: "default",        branch: null,        worktreePath: null,      });       const projection = yield* orchestrator.getThreadProjection(threadId);       assert.equal(result.sequence, 1);      assert.equal(projection.thread.id, threadId);      assert.equal(projection.thread.projectId, projectId);      assert.equal(projection.thread.providerInstanceId, "codex");      assert.deepEqual(projection.runs, []);    }),  );   it.effect("answers an async question after its provider exits and commits the answer once", () =>    Effect.gen(function* () {      const orchestrator = yield* OrchestratorV2;      const eventSink = yield* EventSinkV2;      const now = yield* DateTime.now;      const threadId = ThreadId.make("runtime-async-question");      const requestId = RuntimeRequestId.make("runtime-async-question-request");      const nodeId = NodeId.make("runtime-async-question-node");      const itemId = TurnItemId.make("runtime-async-question-item");      yield* orchestrator.dispatch({        type: "thread.create",        commandId: CommandId.make("runtime-async-question-create"),        createdBy: "user",        creationSource: "web",        threadId,        projectId: ProjectId.make("runtime-async-question-project"),        title: "Async question",        modelSelection,        runtimeMode: "full-access",        interactionMode: "default",        branch: null,        worktreePath: process.cwd(),      });      yield* eventSink.write({        commandId: CommandId.make("runtime-async-question-seed"),        events: [          {            id: EventId.make("runtime-async-question-node-event"),            type: "node.updated",            threadId,            nodeId,            occurredAt: now,            payload: {              id: nodeId,              threadId,              runId: null,              parentNodeId: null,              rootNodeId: nodeId,              kind: "user_input_request",              status: "waiting",              countsForRun: false,              providerThreadId: null,              providerTurnId: null,              nativeItemRef: null,              runtimeRequestId: requestId,              checkpointScopeId: null,              startedAt: now,              completedAt: null,            },          },          {            id: EventId.make("runtime-async-question-request-event"),            type: "runtime-request.updated",            threadId,            nodeId,            occurredAt: now,            payload: {              id: requestId,              nodeId,              providerTurnId: null,              nativeRequestRef: null,              kind: "user_input",              status: "pending",              responseCapability: { type: "message" },              createdAt: now,              resolvedAt: null,            },          },          {            id: EventId.make("runtime-async-question-item-event"),            type: "turn-item.updated",            threadId,            nodeId,            occurredAt: now,            payload: {              id: itemId,              type: "user_input_request",              threadId,              runId: null,              nodeId,              providerThreadId: null,              providerTurnId: null,              nativeItemRef: null,              parentItemId: null,              ordinal: 0,              status: "waiting",              title: null,              startedAt: now,              completedAt: null,              updatedAt: now,              requestId,              responseMode: "message",              questions: [{ id: "color", header: "Color", question: "Which color?", options: [] }],            },          },        ],      });      const invalid = yield* orchestrator        .dispatch({          type: "runtime-request.respond",          commandId: CommandId.make("runtime-async-question-blank"),          threadId,          requestId,          answers: { color: " " },        })        .pipe(Effect.result);      assert.equal(invalid._tag, "Failure");      const unanswered = yield* orchestrator.getThreadProjection(threadId);      assert.equal(unanswered.runtimeRequests[0]?.status, "pending");      assert.deepEqual(unanswered.messages, []);       const command = {        type: "runtime-request.respond" as const,        commandId: CommandId.make("runtime-async-question-answer"),        threadId,        requestId,        answers: { color: "  Blue  " },      };      const accepted = yield* orchestrator.dispatch(command);      const repeated = yield* orchestrator.dispatch(command);      assert.equal(repeated.sequence, accepted.sequence);      const answered = yield* orchestrator.getThreadProjection(threadId);      assert.equal(answered.runtimeRequests[0]?.status, "resolved");      assert.deepEqual(answered.runtimeRequests[0]?.answers, command.answers);      assert.equal(answered.nodes.find((node) => node.id === nodeId)?.status, "completed");      assert.equal(answered.turnItems.find((item) => item.id === itemId)?.status, "completed");      assert.equal(answered.messages.length, 1);      assert.equal(answered.messages[0]?.text, "Which color?\nBlue");      assert.equal(answered.messages[0]?.role, "user");      assert.equal(answered.runs.length, 1);       const duplicate = yield* orchestrator        .dispatch({          ...command,          commandId: CommandId.make("runtime-async-question-duplicate"),        })        .pipe(Effect.result);      assert.equal(duplicate._tag, "Failure");      assert.equal((yield* orchestrator.getThreadProjection(threadId)).messages.length, 1);    }),  );   it.effect("merges an explicit provider-finished run while checkpoint capture is pending", () =>

Lines 1–500 of 2147. Load more