import {  EnvironmentId,  EventId,  ORCHESTRATION_WS_METHODS,  ProjectId,  ProviderInstanceId,  ThreadId,  TurnId,  type OrchestrationThread,  type OrchestrationThreadDetailSnapshot,  type OrchestrationThreadStreamItem,} from "@t3tools/contracts";import { describe, expect, it } from "@effect/vitest";import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Fiber from "effect/Fiber";import * as Option from "effect/Option";import * as Queue from "effect/Queue";import * as Ref from "effect/Ref";import * as Scope from "effect/Scope";import * as Stream from "effect/Stream";import * as SubscriptionRef from "effect/SubscriptionRef";import * as TestClock from "effect/testing/TestClock"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts";import {  AVAILABLE_CONNECTION_STATE,  PrimaryConnectionTarget,  type PreparedConnection,  type SupervisorConnectionState,} from "../connection/model.ts";import * as ConnectionWakeups from "../connection/wakeups.ts";import * as EnvironmentSupervisor from "../connection/supervisor.ts";import * as Persistence from "../platform/persistence.ts";import * as RpcSession from "../rpc/session.ts";import {  EMPTY_ENVIRONMENT_THREAD_STATE,  makeEnvironmentThreadState,  ThreadSnapshotLoader,  type EnvironmentThreadState,} from "./threads.ts"; const TARGET = new PrimaryConnectionTarget({  environmentId: EnvironmentId.make("environment-1"),  label: "Test environment",  httpBaseUrl: "https://environment.example.test",  wsBaseUrl: "wss://environment.example.test",});const THREAD_ID = ThreadId.make("thread-1");const CACHED_SNAPSHOT_SEQUENCE = 7;const PREPARED: PreparedConnection = {  environmentId: TARGET.environmentId,  label: TARGET.label,  httpBaseUrl: TARGET.httpBaseUrl,  socketUrl: TARGET.wsBaseUrl,  httpAuthorization: null,  target: TARGET,};const BASE_THREAD: OrchestrationThread = {  id: THREAD_ID,  projectId: ProjectId.make("project-1"),  title: "Cached thread",  modelSelection: {    instanceId: ProviderInstanceId.make("codex"),    model: "gpt-5.4",  },  runtimeMode: "full-access",  interactionMode: "default",  branch: "main",  worktreePath: null,  latestTurn: null,  createdAt: "2026-04-01T00:00:00.000Z",  updatedAt: "2026-04-01T00:00:00.000Z",  archivedAt: null,  settledOverride: null,  settledAt: null,  pullRequests: [],  deletedAt: null,  messages: [],  proposedPlans: [],  activities: [],  checkpoints: [],  session: null,};const ACTIVE_THREAD: OrchestrationThread = {  ...BASE_THREAD,  latestTurn: {    turnId: TurnId.make("turn-1"),    state: "running",    requestedAt: "2026-04-01T00:01:00.000Z",    startedAt: "2026-04-01T00:01:00.000Z",    completedAt: null,    assistantMessageId: null,  },  session: {    threadId: THREAD_ID,    status: "running",    providerName: "codex",    runtimeMode: "full-access",    activeTurnId: TurnId.make("turn-1"),    lastError: null,    updatedAt: "2026-04-01T00:01:00.000Z",  },}; type TestThreadInput = OrchestrationThreadStreamItem | Error; function testSession(  client: WsRpcProtocolClient,  options?: { readonly completionMarker?: boolean },): RpcSession.RpcSession {  return {    client,    initialConfig: Effect.succeed(      options?.completionMarker === true        ? ({ threadResumeCompletionMarker: true } as never)        : ({} as never),    ),    subscribeServerConfig: (input) => client.subscribeServerConfig(input),    ready: Effect.void,    probe: Effect.void,    closed: Effect.never,  };} function awaitThreadState(  observed: Queue.Queue<EnvironmentThreadState>,  predicate: (state: EnvironmentThreadState) => boolean,) {  return Queue.take(observed).pipe(    Effect.repeat({      until: predicate,    }),  );} const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (options?: {  readonly cached?: OrchestrationThread;  readonly httpSnapshot?: Option.Option<OrchestrationThreadDetailSnapshot>;  readonly completionMarker?: boolean;  readonly resumeCache?: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]>;  readonly loadCached?: Effect.Effect<Option.Option<OrchestrationThreadDetailSnapshot>>;  readonly saveThread?: Persistence.EnvironmentCacheStore["Service"]["saveThread"];}) {  const inputs = yield* Queue.unbounded<TestThreadInput>();  const observed = yield* Queue.unbounded<EnvironmentThreadState>();  const latest = yield* Ref.make<EnvironmentThreadState>(EMPTY_ENVIRONMENT_THREAD_STATE);  const retryCount = yield* Ref.make(0);  const subscriptionCount = yield* Ref.make(0);  const loaderCalls = yield* Ref.make(0);  const lastSubscribeAfterSequence = yield* Ref.make<number | undefined>(undefined);  const lastRequestCompletionMarker = yield* Ref.make<boolean | undefined>(undefined);  const savedThreads = yield* Ref.make<ReadonlyArray<OrchestrationThreadDetailSnapshot>>([]);  const removedThreads = yield* Ref.make<ReadonlyArray<ThreadId>>([]);  const wakeups = yield* Queue.unbounded<ConnectionWakeups.ConnectionWakeup>();  const supervisorState = yield* SubscriptionRef.make<SupervisorConnectionState>(    AVAILABLE_CONNECTION_STATE,  );  const streamFrom = (queue: Queue.Queue<TestThreadInput>) =>    Stream.fromQueue(queue).pipe(      Stream.mapEffect((input) =>        input instanceof Error ? Effect.fail(input) : Effect.succeed(input),      ),    );  const client = {    [ORCHESTRATION_WS_METHODS.subscribeThread]: (input: {      readonly afterSequence?: number;      readonly requestCompletionMarker?: boolean;    }) =>      Stream.unwrap(        Ref.updateAndGet(subscriptionCount, (count) => count + 1).pipe(          Effect.andThen(Ref.set(lastSubscribeAfterSequence, input.afterSequence)),          Effect.andThen(Ref.set(lastRequestCompletionMarker, input.requestCompletionMarker)),          Effect.as(streamFrom(inputs)),        ),      ),  } as unknown as WsRpcProtocolClient;  const supervisorSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(    Option.some(      testSession(        client,        options?.completionMarker === true ? { completionMarker: true } : undefined,      ),    ),  );  const prepared = yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(    Option.some(PREPARED),  );  const snapshotLoader = ThreadSnapshotLoader.of({    load: (_prepared, threadId) =>      Ref.update(loaderCalls, (count) => count + 1).pipe(        Effect.as(          threadId === THREAD_ID            ? (options?.httpSnapshot ?? Option.none<OrchestrationThreadDetailSnapshot>())            : Option.none<OrchestrationThreadDetailSnapshot>(),        ),      ),  });  const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({    target: TARGET,    state: supervisorState,    session: supervisorSession,    prepared,    connect: Effect.void,    disconnect: Effect.void,    retryNow: Ref.update(retryCount, (count) => count + 1),  } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);  const cache = Persistence.EnvironmentCacheStore.of({    loadShell: () => Effect.succeed(Option.none()),    saveShell: () => Effect.void,    loadThread: (_environmentId, threadId) =>      options?.loadCached ??      Effect.succeed(        threadId === THREAD_ID && options?.cached !== undefined          ? Option.some({              snapshotSequence: CACHED_SNAPSHOT_SEQUENCE,              thread: options.cached,            })          : Option.none(),      ),    saveThread: (environmentId, thread) =>      Ref.update(savedThreads, (current) => [...current, thread]).pipe(        Effect.andThen(options?.saveThread?.(environmentId, thread) ?? Effect.void),      ),    removeThread: (_environmentId, threadId) =>      Ref.update(removedThreads, (current) => [...current, threadId]),    loadServerConfig: () => Effect.succeed(Option.none()),    saveServerConfig: () => Effect.void,    loadVcsRefs: () => Effect.succeed(Option.none()),    saveVcsRefs: () => Effect.void,    removeVcsRefs: () => Effect.void,    clearVcsRefs: () => Effect.void,    clear: () => Effect.void,  });  const threadState = yield* makeEnvironmentThreadState(THREAD_ID, options?.resumeCache).pipe(    Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),    Effect.provideService(Persistence.EnvironmentCacheStore, cache),    Effect.provideService(ThreadSnapshotLoader, snapshotLoader),    Effect.provideService(      ConnectionWakeups.ConnectionWakeups,      ConnectionWakeups.ConnectionWakeups.of({ changes: Stream.fromQueue(wakeups) }),    ),  );  yield* SubscriptionRef.changes(threadState).pipe(    Stream.runForEach((state) =>      Ref.set(latest, state).pipe(Effect.andThen(Queue.offer(observed, state))),    ),    Effect.forkScoped,  );   return {    threadState,    inputs,    observed,    latest,    retryCount,    subscriptionCount,    loaderCalls,    lastSubscribeAfterSequence,    lastRequestCompletionMarker,    supervisorState,    supervisorSession,    savedThreads,    removedThreads,    wakeups,    replaceSession: SubscriptionRef.set(      supervisorSession,      Option.some(        testSession(          client,          options?.completionMarker === true ? { completionMarker: true } : undefined,        ),      ),    ),  };}); const snapshot = (thread: OrchestrationThread): OrchestrationThreadStreamItem => ({  kind: "snapshot",  snapshot: {    snapshotSequence: 1,    thread,  },}); const synchronized = (): OrchestrationThreadStreamItem => ({ kind: "synchronized" }); const titleUpdated = (title: string, sequence = 2): OrchestrationThreadStreamItem => ({  kind: "event",  event: {    eventId: EventId.make("event-title"),    sequence,    occurredAt: "2026-04-01T01:00:00.000Z",    commandId: null,    causationEventId: null,    correlationId: null,    metadata: {},    aggregateKind: "thread",    aggregateId: THREAD_ID,    type: "thread.meta-updated",    payload: {      threadId: THREAD_ID,      title,      updatedAt: "2026-04-01T01:00:00.000Z",    },  },}); const deleted = (): OrchestrationThreadStreamItem => ({  kind: "event",  event: {    eventId: EventId.make("event-deleted"),    sequence: 3,    occurredAt: "2026-04-01T02:00:00.000Z",    commandId: null,    causationEventId: null,    correlationId: null,    metadata: {},    aggregateKind: "thread",    aggregateId: THREAD_ID,    type: "thread.deleted",    payload: {      threadId: THREAD_ID,      deletedAt: "2026-04-01T02:00:00.000Z",    },  },}); describe("EnvironmentThreads", () => {  for (const source of ["disk", "HTTP"] as const) {    it.effect(`does not rewrite an unchanged ${source} snapshot on navigation or warm return`, () =>      Effect.gen(function* () {        const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {          snapshot: undefined,          owner: undefined,        };        const firstSaved = yield* Effect.scoped(          Effect.gen(function* () {            const h = yield* makeHarness({              resumeCache,              ...(source === "disk"                ? { cached: BASE_THREAD }                : { httpSnapshot: Option.some({ snapshotSequence: 7, thread: BASE_THREAD }) }),            });            yield* awaitThreadState(h.observed, (value) => value.status === "live");            if (source === "HTTP") yield* TestClock.adjust("500 millis");            return h.savedThreads;          }),        );        expect(yield* Ref.get(firstSaved)).toHaveLength(source === "disk" ? 0 : 1);        const nextSaved = yield* Effect.scoped(          Effect.gen(function* () {            const h = yield* makeHarness({ resumeCache });            yield* awaitThreadState(h.observed, (value) => value.status === "live");            return h.savedThreads;          }),        );        expect(yield* Ref.get(nextSaved)).toEqual([]);      }),    );  }   it.effect("retries a failed background cache write when the scope closes", () =>    Effect.gen(function* () {      let attempts = 0;      yield* Effect.scoped(        Effect.gen(function* () {          const h = yield* makeHarness({            httpSnapshot: Option.some({ snapshotSequence: 7, thread: BASE_THREAD }),            saveThread: () =>              Effect.suspend(() => {                attempts += 1;                return attempts === 1                  ? Effect.fail(                      new Persistence.ConnectionPersistenceError({                        operation: "save-thread",                        message: "Test storage failure",                      }),                    )                  : Effect.void;              }),          });          yield* awaitThreadState(h.observed, (value) => value.status === "live");          yield* TestClock.adjust("500 millis");          expect(attempts).toBe(1);        }),      );      expect(attempts).toBe(2);    }),  );   it.effect("flushes newer data after an older background write completes", () =>    Effect.gen(function* () {      const writing = yield* Deferred.make<void>();      const written = yield* Deferred.make<void>();      const release = yield* Deferred.make<void>();      const saved = yield* Effect.scoped(        Effect.gen(function* () {          const h = yield* makeHarness({            cached: BASE_THREAD,            saveThread: (_environmentId, snapshot) =>              snapshot.snapshotSequence === 8                ? Deferred.succeed(writing, undefined).pipe(                    Effect.andThen(Deferred.await(release)),                    Effect.andThen(Deferred.succeed(written, undefined)),                  )                : Effect.void,          });          yield* Queue.offer(h.inputs, titleUpdated("First update", 8));          yield* awaitThreadState(            h.observed,            (value) => Option.getOrNull(value.data)?.title === "First update",          );          yield* TestClock.adjust("500 millis");          yield* Deferred.await(writing);          yield* Queue.offer(h.inputs, titleUpdated("Newer update", 9));          yield* awaitThreadState(            h.observed,            (value) => Option.getOrNull(value.data)?.title === "Newer update",          );          yield* Deferred.succeed(release, undefined);          yield* Deferred.await(written);          return h.savedThreads;        }),      );      expect(        (yield* Ref.get(saved)).map((snapshot) => [          snapshot.snapshotSequence,          snapshot.thread.title,        ]),      ).toEqual([        [8, "First update"],        [9, "Newer update"],      ]);    }),  );   it.effect("does not resume past a canceled event whose data was not applied", () =>    Effect.gen(function* () {      const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {        snapshot: undefined,        owner: undefined,      };      const scope = yield* Effect.acquireRelease(Scope.make(), (scope) =>        Scope.close(scope, Exit.void),      );      const first = yield* makeHarness({        httpSnapshot: Option.some({          thread: BASE_THREAD,          snapshotSequence: CACHED_SNAPSHOT_SEQUENCE,        }),        resumeCache,      }).pipe(Effect.provideService(Scope.Scope, scope));      yield* awaitThreadState(first.observed, (value) => value.status === "live");      const applying = yield* Deferred.make<void>();      const update = titleUpdated("Not applied", 8);      if (update.kind !== "event") return yield* Effect.die("Expected an event");      Object.defineProperty(update.event.payload, "title", {        get: () => {          Deferred.doneUnsafe(applying, Exit.void);          return "Not applied";        },      });      // The reducer reads the title after it advances the cursor. Hold only      // the data write so closing the scope interrupts that exact interval.      yield* first.threadState.semaphore.take(1);      yield* Queue.offer(first.inputs, update);      yield* Deferred.await(applying);      yield* Scope.close(scope, Exit.void);      yield* first.threadState.semaphore.release(1);       const resumed = yield* makeHarness({ resumeCache });      const state = yield* awaitThreadState(resumed.observed, (value) => value.status === "live");      expect(Option.getOrThrow(state.data).title).toBe(BASE_THREAD.title);      expect(yield* Ref.get(resumed.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE);      expect(yield* Ref.get(first.savedThreads)).toEqual([        {          thread: BASE_THREAD,          snapshotSequence: CACHED_SNAPSHOT_SEQUENCE,        },      ]);    }),  );   it.effect("prevents an old scope from replacing its successor's cached data", () =>    Effect.gen(function* () {      const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {        snapshot: undefined,        owner: undefined,      };      const oldScope = yield* Effect.acquireRelease(Scope.make(), (scope) =>        Scope.close(scope, Exit.void),      );      const old = yield* makeHarness({ cached: BASE_THREAD, resumeCache }).pipe(        Effect.provideService(Scope.Scope, oldScope),      );      yield* awaitThreadState(old.observed, (value) => value.status === "live");      const successor = yield* makeHarness({ resumeCache });      yield* Queue.offer(successor.inputs, titleUpdated("Successor title", 9));

Lines 1–500 of 992. Load more