import { EnvironmentId, EventId, ORCHESTRATION_WS_METHODS, ProjectId, ProviderInstanceId, ThreadId, type OrchestrationThread, type OrchestrationThreadStreamItem,} from "@t3tools/contracts";import { describe, expect, it } from "@effect/vitest";import * as Effect from "effect/Effect";import * as Option from "effect/Option";import * as Queue from "effect/Queue";import * as Ref from "effect/Ref";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 { EnvironmentSupervisor, type EnvironmentSupervisorService,} from "../connection/supervisor.ts";import { EnvironmentCacheStore } from "../platform/persistence.ts";import type { RpcSession } from "../rpc/session.ts";import { EMPTY_ENVIRONMENT_THREAD_STATE, makeEnvironmentThreadState, 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 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, deletedAt: null, messages: [], proposedPlans: [], activities: [], checkpoints: [], session: null,}; type TestThreadInput = OrchestrationThreadStreamItem | Error; function testSession(client: WsRpcProtocolClient): RpcSession { return { client, initialConfig: Effect.never, 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;}) { 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 savedThreads = yield* Ref.make<ReadonlyArray<OrchestrationThread>>([]); 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_WS_METHODS.subscribeThread]: () => Stream.unwrap( Ref.updateAndGet(subscriptionCount, (count) => count + 1).pipe( Effect.map(() => streamFrom(inputs)), ), ), } as unknown as WsRpcProtocolClient; const supervisorSession = yield* SubscriptionRef.make<Option.Option<RpcSession>>( Option.some(testSession(client)), ); const prepared = yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(Option.none()); const supervisor = EnvironmentSupervisor.of({ target: TARGET, state: supervisorState, session: supervisorSession, prepared, connect: Effect.void, disconnect: Effect.void, retryNow: Ref.update(retryCount, (count) => count + 1), } satisfies EnvironmentSupervisorService); const cache = EnvironmentCacheStore.of({ loadShell: () => Effect.succeed(Option.none()), saveShell: () => Effect.void, loadThread: (_environmentId, threadId) => Effect.succeed( threadId === THREAD_ID && options?.cached !== undefined ? Option.some(options.cached) : Option.none(), ), saveThread: (_environmentId, thread) => Ref.update(savedThreads, (current) => [...current, thread]), removeThread: (_environmentId, threadId) => Ref.update(removedThreads, (current) => [...current, threadId]), clear: () => Effect.void, }); const threadState = yield* makeEnvironmentThreadState(THREAD_ID).pipe( Effect.provideService(EnvironmentSupervisor, supervisor), Effect.provideService(EnvironmentCacheStore, cache), ); yield* SubscriptionRef.changes(threadState).pipe( Stream.runForEach((state) => Ref.set(latest, state).pipe(Effect.andThen(Queue.offer(observed, state))), ), Effect.forkScoped, ); return { inputs, observed, latest, retryCount, subscriptionCount, supervisorState, supervisorSession, savedThreads, removedThreads, replaceSession: SubscriptionRef.set(supervisorSession, Option.some(testSession(client))), };}); const snapshot = (thread: OrchestrationThread): OrchestrationThreadStreamItem => ({ kind: "snapshot", snapshot: { snapshotSequence: 1, thread, },}); 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", () => { it.effect("publishes cached data before a live snapshot arrives", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD }); const state = yield* awaitThreadState( harness.observed, (value) => value.status === "cached" && Option.isSome(value.data), ); expect(Option.getOrThrow(state.data)).toEqual(BASE_THREAD); expect(Option.isNone(state.error)).toBe(true); }), ); it.effect("reduces live events and persists the latest thread", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD }); yield* Queue.offer(harness.inputs, snapshot(BASE_THREAD)); yield* Queue.offer(harness.inputs, titleUpdated("Live title")); const state = yield* awaitThreadState( harness.observed, (value) => value.status === "live" && Option.isSome(value.data) && value.data.value.title === "Live title", ); yield* TestClock.adjust("500 millis"); yield* Effect.yieldNow; expect(Option.getOrThrow(state.data).title).toBe("Live title"); expect((yield* Ref.get(harness.savedThreads)).at(-1)?.title).toBe("Live title"); }), ); it.effect("ignores replayed thread events at or below the snapshot sequence", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD }); yield* Queue.offer(harness.inputs, snapshot(BASE_THREAD)); yield* Queue.offer(harness.inputs, titleUpdated("Replayed title", 1)); yield* Queue.offer(harness.inputs, titleUpdated("Live title", 2)); const state = yield* awaitThreadState( harness.observed, (value) => value.status === "live" && Option.isSome(value.data) && value.data.value.title === "Live title", ); expect(Option.getOrThrow(state.data).title).toBe("Live title"); }), ); it.effect("removes cached data when the thread is deleted", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD }); yield* Queue.offer(harness.inputs, snapshot(BASE_THREAD)); yield* Queue.offer(harness.inputs, deleted()); const state = yield* awaitThreadState( harness.observed, (value) => value.status === "deleted", ); expect(Option.isNone(state.data)).toBe(true); expect(yield* Ref.get(harness.removedThreads)).toEqual([THREAD_ID]); }), ); it.effect("preserves data after a domain failure and resumes on a replacement session", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD }); yield* Queue.offer(harness.inputs, snapshot(BASE_THREAD)); yield* Queue.offer(harness.inputs, new Error("stream failed")); const state = yield* awaitThreadState(harness.observed, (value) => Option.isSome(value.error), ); expect(Option.getOrThrow(state.data)).toEqual(BASE_THREAD); expect(Option.getOrThrow(state.error)).toBe("stream failed"); expect(yield* Ref.get(harness.retryCount)).toBe(0); yield* harness.replaceSession; for (let attempt = 0; attempt < 100; attempt += 1) { if ((yield* Ref.get(harness.subscriptionCount)) >= 2) { break; } yield* Effect.yieldNow; } yield* Queue.offer( harness.inputs, snapshot({ ...BASE_THREAD, title: "Recovered thread", }), ); const recovered = yield* awaitThreadState( harness.observed, (value) => value.status === "live" && Option.isSome(value.data) && value.data.value.title === "Recovered thread", ); expect(Option.isNone(recovered.error)).toBe(true); expect(yield* Ref.get(harness.subscriptionCount)).toBe(2); }), ); it.effect("recovers from a transient domain failure without replacing the session", () => Effect.gen(function* () { const harness = yield* makeHarness(); yield* Queue.offer(harness.inputs, new Error("thread not found yet")); const failed = yield* awaitThreadState(harness.observed, (value) => Option.isSome(value.error), ); expect(Option.getOrThrow(failed.error)).toBe("thread not found yet"); expect(yield* Ref.get(harness.subscriptionCount)).toBe(1); yield* TestClock.adjust("250 millis"); for (let attempt = 0; attempt < 100; attempt += 1) { if ((yield* Ref.get(harness.subscriptionCount)) >= 2) { break; } yield* Effect.yieldNow; } yield* Queue.offer( harness.inputs, snapshot({ ...BASE_THREAD, title: "Materialized thread", }), ); const recovered = yield* awaitThreadState( harness.observed, (value) => value.status === "live" && Option.isSome(value.data) && value.data.value.title === "Materialized thread", ); expect(Option.isNone(recovered.error)).toBe(true); expect(yield* Ref.get(harness.subscriptionCount)).toBe(2); expect(yield* Ref.get(harness.retryCount)).toBe(0); }), ); it.effect("does not overwrite a live snapshot when the supervisor becomes ready", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD }); yield* SubscriptionRef.set(harness.supervisorState, { desired: true, network: "online", phase: "connecting", stage: "synchronizing", attempt: 1, generation: 0, lastFailure: null, retryAt: null, }); yield* Queue.offer(harness.inputs, snapshot(BASE_THREAD)); yield* awaitThreadState(harness.observed, (value) => value.status === "live"); yield* SubscriptionRef.set(harness.supervisorState, { desired: true, network: "online", phase: "connected", stage: null, attempt: 1, generation: 1, lastFailure: null, retryAt: null, }); for (let index = 0; index < 10; index += 1) { yield* Effect.yieldNow; } expect((yield* Ref.get(harness.latest)).status).toBe("live"); }), );});
import {
EnvironmentId,
EventId,
ORCHESTRATION_WS_METHODS,
ProjectId,
ProviderInstanceId,
ThreadId,
type OrchestrationThread,
type OrchestrationThreadStreamItem,
} from "@t3tools/contracts";
import { describe, expect, it } from "@effect/vitest";
import * as Effect from "effect/Effect";
import * as Option from "effect/Option";
import * as Queue from "effect/Queue";
import * as Ref from "effect/Ref";
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 {
EnvironmentSupervisor,
type EnvironmentSupervisorService,
} from "../connection/supervisor.ts";
import { EnvironmentCacheStore } from "../platform/persistence.ts";
import type { RpcSession } from "../rpc/session.ts";
import {
EMPTY_ENVIRONMENT_THREAD_STATE,
makeEnvironmentThreadState,
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 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,
deletedAt: null,
messages: [],
proposedPlans: [],
activities: [],
checkpoints: [],
session: null,
};
type TestThreadInput = OrchestrationThreadStreamItem | Error;
function testSession(client: WsRpcProtocolClient): RpcSession {
return {
client,
initialConfig: Effect.never,
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;
}) {
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 savedThreads = yield* Ref.make<ReadonlyArray<OrchestrationThread>>([]);
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_WS_METHODS.subscribeThread]: () =>
Stream.unwrap(
Ref.updateAndGet(subscriptionCount, (count) => count + 1).pipe(
Effect.map(() => streamFrom(inputs)),
),
),
} as unknown as WsRpcProtocolClient;
const supervisorSession = yield* SubscriptionRef.make<Option.Option<RpcSession>>(
Option.some(testSession(client)),
);
const prepared = yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(Option.none());
const supervisor = EnvironmentSupervisor.of({
target: TARGET,
state: supervisorState,
session: supervisorSession,
prepared,
connect: Effect.void,
disconnect: Effect.void,
retryNow: Ref.update(retryCount, (count) => count + 1),
} satisfies EnvironmentSupervisorService);
const cache = EnvironmentCacheStore.of({
loadShell: () => Effect.succeed(Option.none()),
saveShell: () => Effect.void,
loadThread: (_environmentId, threadId) =>
Effect.succeed(
threadId === THREAD_ID && options?.cached !== undefined
? Option.some(options.cached)
: Option.none(),
),
saveThread: (_environmentId, thread) =>
Ref.update(savedThreads, (current) => [...current, thread]),
removeThread: (_environmentId, threadId) =>
Ref.update(removedThreads, (current) => [...current, threadId]),
clear: () => Effect.void,
});
const threadState = yield* makeEnvironmentThreadState(THREAD_ID).pipe(
Effect.provideService(EnvironmentSupervisor, supervisor),
Effect.provideService(EnvironmentCacheStore, cache),
);
yield* SubscriptionRef.changes(threadState).pipe(
Stream.runForEach((state) =>
Ref.set(latest, state).pipe(Effect.andThen(Queue.offer(observed, state))),
),
Effect.forkScoped,
);
return {
inputs,
observed,
latest,
retryCount,
subscriptionCount,
supervisorState,
supervisorSession,
savedThreads,
removedThreads,
replaceSession: SubscriptionRef.set(supervisorSession, Option.some(testSession(client))),
};
});
const snapshot = (thread: OrchestrationThread): OrchestrationThreadStreamItem => ({
kind: "snapshot",
snapshot: {
snapshotSequence: 1,
thread,
},
});
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", () => {
it.effect("publishes cached data before a live snapshot arrives", () =>
Effect.gen(function* () {
const harness = yield* makeHarness({ cached: BASE_THREAD });
const state = yield* awaitThreadState(
harness.observed,
(value) => value.status === "cached" && Option.isSome(value.data),
);
expect(Option.getOrThrow(state.data)).toEqual(BASE_THREAD);
expect(Option.isNone(state.error)).toBe(true);
}),
);
it.effect("reduces live events and persists the latest thread", () =>
Effect.gen(function* () {
const harness = yield* makeHarness({ cached: BASE_THREAD });
yield* Queue.offer(harness.inputs, snapshot(BASE_THREAD));
yield* Queue.offer(harness.inputs, titleUpdated("Live title"));
const state = yield* awaitThreadState(
harness.observed,
(value) =>
value.status === "live" &&
Option.isSome(value.data) &&
value.data.value.title === "Live title",
);
yield* TestClock.adjust("500 millis");
yield* Effect.yieldNow;
expect(Option.getOrThrow(state.data).title).toBe("Live title");
expect((yield* Ref.get(harness.savedThreads)).at(-1)?.title).toBe("Live title");
}),
);
it.effect("ignores replayed thread events at or below the snapshot sequence", () =>
Effect.gen(function* () {
const harness = yield* makeHarness({ cached: BASE_THREAD });
yield* Queue.offer(harness.inputs, snapshot(BASE_THREAD));
yield* Queue.offer(harness.inputs, titleUpdated("Replayed title", 1));
yield* Queue.offer(harness.inputs, titleUpdated("Live title", 2));
const state = yield* awaitThreadState(
harness.observed,
(value) =>
value.status === "live" &&
Option.isSome(value.data) &&
value.data.value.title === "Live title",
);
expect(Option.getOrThrow(state.data).title).toBe("Live title");
}),
);
it.effect("removes cached data when the thread is deleted", () =>
Effect.gen(function* () {
const harness = yield* makeHarness({ cached: BASE_THREAD });
yield* Queue.offer(harness.inputs, snapshot(BASE_THREAD));
yield* Queue.offer(harness.inputs, deleted());
const state = yield* awaitThreadState(
harness.observed,
(value) => value.status === "deleted",
);
expect(Option.isNone(state.data)).toBe(true);
expect(yield* Ref.get(harness.removedThreads)).toEqual([THREAD_ID]);
}),
);
it.effect("preserves data after a domain failure and resumes on a replacement session", () =>
Effect.gen(function* () {
const harness = yield* makeHarness({ cached: BASE_THREAD });
yield* Queue.offer(harness.inputs, snapshot(BASE_THREAD));
yield* Queue.offer(harness.inputs, new Error("stream failed"));
const state = yield* awaitThreadState(harness.observed, (value) =>
Option.isSome(value.error),
);
expect(Option.getOrThrow(state.data)).toEqual(BASE_THREAD);
expect(Option.getOrThrow(state.error)).toBe("stream failed");
expect(yield* Ref.get(harness.retryCount)).toBe(0);
yield* harness.replaceSession;
for (let attempt = 0; attempt < 100; attempt += 1) {
if ((yield* Ref.get(harness.subscriptionCount)) >= 2) {
break;
}
yield* Effect.yieldNow;
}
yield* Queue.offer(
harness.inputs,
snapshot({
...BASE_THREAD,
title: "Recovered thread",
}),
);
const recovered = yield* awaitThreadState(
harness.observed,
(value) =>
value.status === "live" &&
Option.isSome(value.data) &&
value.data.value.title === "Recovered thread",
);
expect(Option.isNone(recovered.error)).toBe(true);
expect(yield* Ref.get(harness.subscriptionCount)).toBe(2);
}),
);
it.effect("recovers from a transient domain failure without replacing the session", () =>
Effect.gen(function* () {
const harness = yield* makeHarness();
yield* Queue.offer(harness.inputs, new Error("thread not found yet"));
const failed = yield* awaitThreadState(harness.observed, (value) =>
Option.isSome(value.error),
);
expect(Option.getOrThrow(failed.error)).toBe("thread not found yet");
expect(yield* Ref.get(harness.subscriptionCount)).toBe(1);
yield* TestClock.adjust("250 millis");
for (let attempt = 0; attempt < 100; attempt += 1) {
if ((yield* Ref.get(harness.subscriptionCount)) >= 2) {
break;
}
yield* Effect.yieldNow;
}
yield* Queue.offer(
harness.inputs,
snapshot({
...BASE_THREAD,
title: "Materialized thread",
}),
);
const recovered = yield* awaitThreadState(
harness.observed,
(value) =>
value.status === "live" &&
Option.isSome(value.data) &&
value.data.value.title === "Materialized thread",
);
expect(Option.isNone(recovered.error)).toBe(true);
expect(yield* Ref.get(harness.subscriptionCount)).toBe(2);
expect(yield* Ref.get(harness.retryCount)).toBe(0);
}),
);
it.effect("does not overwrite a live snapshot when the supervisor becomes ready", () =>
Effect.gen(function* () {
const harness = yield* makeHarness({ cached: BASE_THREAD });
yield* SubscriptionRef.set(harness.supervisorState, {
desired: true,
network: "online",
phase: "connecting",
stage: "synchronizing",
attempt: 1,
generation: 0,
lastFailure: null,
retryAt: null,
});
yield* Queue.offer(harness.inputs, snapshot(BASE_THREAD));
yield* awaitThreadState(harness.observed, (value) => value.status === "live");
yield* SubscriptionRef.set(harness.supervisorState, {
desired: true,
network: "online",
phase: "connected",
stage: null,
attempt: 1,
generation: 1,
lastFailure: null,
retryAt: null,
});
for (let index = 0; index < 10; index += 1) {
yield* Effect.yieldNow;
}
expect((yield* Ref.get(harness.latest)).status).toBe("live");
}),
);
});