import {  EnvironmentId,  EventId,  ORCHESTRATION_V2_WS_METHODS,  ThreadId,  TurnItemId,  type OrchestrationV2ThreadDetailSnapshot,  type OrchestrationV2ThreadProjection,  type OrchestrationV2ThreadStreamItem,  type OrchestrationV2TurnItem,} from "@t3tools/contracts";import { describe, expect, it } from "@effect/vitest";import * as DateTime from "effect/DateTime";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 { v2Projection, v2ThreadId } from "./orchestrationV2TestFixtures.ts";import {  EMPTY_ENVIRONMENT_THREAD_STATE,  makeEnvironmentThreadState,  ThreadSnapshotLoader,  type EnvironmentThreadState,  type ThreadSnapshotLoadResult,} 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 = v2ThreadId;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_PROJECTION: OrchestrationV2ThreadProjection = {  ...v2Projection,  thread: { ...v2Projection.thread, title: "Cached thread" },}; type TestThreadInput = OrchestrationV2ThreadStreamItem | Error; function testSession(  client: WsRpcProtocolClient,  config?: { readonly completionMarker?: boolean },): RpcSession.RpcSession {  return {    client,    initialConfig: Effect.succeed(      config?.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?: OrchestrationV2ThreadProjection;  readonly cachedHistory?: {    readonly historyCursor: string | null;    readonly hasMoreHistory: boolean;    readonly latestLocalTurnOrdinal?: number | null;  };  readonly httpSnapshot?: ThreadSnapshotLoadResult;  readonly completionMarker?: boolean;  readonly resumeCache?: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]>;  readonly loadCached?: Effect.Effect<Option.Option<OrchestrationV2ThreadDetailSnapshot>>;  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(false);  const wakeups = yield* Queue.unbounded<ConnectionWakeups.ConnectionWakeup>();  const savedThreads = yield* Ref.make<ReadonlyArray<OrchestrationV2ThreadDetailSnapshot>>([]);  const removedThreads = yield* Ref.make<ReadonlyArray<ThreadId>>([]);  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_V2_WS_METHODS.subscribeThread]: (input: {      readonly afterSequence?: number;      readonly requestCompletionMarker?: true;    }) =>      Stream.unwrap(        Ref.updateAndGet(subscriptionCount, (count) => count + 1).pipe(          Effect.andThen(Ref.set(lastSubscribeAfterSequence, input.afterSequence)),          Effect.andThen(            Ref.set(lastRequestCompletionMarker, input.requestCompletionMarker === true),          ),          Effect.as(streamFrom(inputs)),        ),      ),  } as unknown as WsRpcProtocolClient;  const supervisorSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(    Option.some(testSession(client, options)),  );  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 ??                ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult))            : ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult),        ),      ),  });  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,              projection: options.cached,              ...(options.cachedHistory === undefined                ? {}                : {                    historyCursor: options.cachedHistory.historyCursor,                    hasMoreHistory: options.cachedHistory.hasMoreHistory,                    ...(options.cachedHistory.latestLocalTurnOrdinal === undefined                      ? {}                      : {                          latestLocalTurnOrdinal: options.cachedHistory.latestLocalTurnOrdinal,                        }),                  }),            })          : 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)),    ),  };}); const snapshot = (  projection: OrchestrationV2ThreadProjection,  snapshotSequence = 1,): OrchestrationV2ThreadStreamItem => ({  kind: "snapshot",  snapshotSequence,  projection,}); const synchronized = (): OrchestrationV2ThreadStreamItem => ({ kind: "synchronized" }); const titleUpdated = (title: string, sequence = 2): OrchestrationV2ThreadStreamItem => {  const occurredAt = DateTime.makeUnsafe("2026-06-20T01:00:00.000Z");  return {    kind: "event",    sequence,    event: {      id: EventId.make("event-title"),      type: "thread.metadata-updated",      threadId: THREAD_ID,      occurredAt,      payload: { ...v2Projection.thread, title, updatedAt: occurredAt },    },  };}; const deleted = (sequence = 3): OrchestrationV2ThreadStreamItem => {  const occurredAt = DateTime.makeUnsafe("2026-06-20T02:00:00.000Z");  return {    kind: "event",    sequence,    event: {      id: EventId.make("event-deleted"),      type: "thread.deleted",      threadId: THREAD_ID,      occurredAt,      payload: { ...v2Projection.thread, updatedAt: occurredAt, deletedAt: occurredAt },    },  };}; 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_PROJECTION }                : {                    httpSnapshot: {                      _tag: "present",                      snapshot: { snapshotSequence: 7, projection: BASE_PROJECTION },                    },                  }),            });            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("persists a complete bounded HTTP window only once", () =>    Effect.gen(function* () {      const saved = yield* Effect.scoped(        Effect.gen(function* () {          const h = yield* makeHarness({            httpSnapshot: {              _tag: "present",              snapshot: { snapshotSequence: 7, projection: BASE_PROJECTION },              history: {                historyCursor: null,                hasMoreHistory: false,                latestLocalTurnOrdinal: 12,              },            },          });          yield* awaitThreadState(h.observed, (value) => value.status === "live");          yield* TestClock.adjust("500 millis");          return h.savedThreads;        }),      );      expect(yield* Ref.get(saved)).toEqual([{ snapshotSequence: 7, projection: BASE_PROJECTION }]);    }),  );   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: {              _tag: "present",              snapshot: { snapshotSequence: 7, projection: BASE_PROJECTION },            },            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_PROJECTION,            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)?.thread.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)?.thread.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.projection.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: {          _tag: "present",          snapshot: {            projection: BASE_PROJECTION,            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");      let sequenceReads = 0;      Object.defineProperty(update, "sequence", {        get: () => {          if (++sequenceReads === 2) Deferred.doneUnsafe(applying, Exit.void);          return 8;        },      });      // Hold the projection write while the event advances its cursor. Closing      // the scope must resume from the last fully applied projection.      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).thread.title).toBe(BASE_PROJECTION.thread.title);      expect(yield* Ref.get(resumed.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE);      expect(yield* Ref.get(first.savedThreads)).toEqual([        {          projection: BASE_PROJECTION,          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_PROJECTION, 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));      yield* awaitThreadState(        successor.observed,        (value) => Option.getOrNull(value.data)?.thread.title === "Successor title",      );      yield* Queue.offer(old.inputs, titleUpdated("Old scope title", 8));

Lines 1–500 of 1525. Load more