import { EnvironmentId, ORCHESTRATION_WS_METHODS, type OrchestrationShellSnapshot, type OrchestrationShellStreamItem,} 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 Stream from "effect/Stream";import * as SubscriptionRef from "effect/SubscriptionRef"; import { AVAILABLE_CONNECTION_STATE, PrimaryConnectionTarget, type PreparedConnection,} from "../connection/model.ts";import * as EnvironmentSupervisor from "../connection/supervisor.ts";import * as Persistence from "../platform/persistence.ts";import * as RpcSession from "../rpc/session.ts";import type { WsRpcProtocolClient } from "../rpc/protocol.ts";import { makeEnvironmentShellState, ShellSnapshotLoader } from "./shell.ts"; const TARGET = new PrimaryConnectionTarget({ environmentId: EnvironmentId.make("environment-1"), label: "Test environment", httpBaseUrl: "https://environment.example.test", wsBaseUrl: "wss://environment.example.test",}); const PREPARED: PreparedConnection = { environmentId: TARGET.environmentId, label: TARGET.label, httpBaseUrl: TARGET.httpBaseUrl, socketUrl: TARGET.wsBaseUrl, httpAuthorization: null, target: TARGET,}; const LIVE_SHELL_SNAPSHOT: OrchestrationShellSnapshot = { snapshotSequence: 1, projects: [], threads: [], updatedAt: "2026-06-06T00:00:00.000Z",}; function session(client: WsRpcProtocolClient): RpcSession.RpcSession { return { client, initialConfig: Effect.never, ready: Effect.void, probe: Effect.void, closed: Effect.never, };} describe("environment shell synchronization", () => { it.effect("publishes live state before persistence and preserves it when ready", () => Effect.gen(function* () { const events = yield* Queue.unbounded<OrchestrationShellStreamItem>(); const client = { [ORCHESTRATION_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events), } as unknown as WsRpcProtocolClient; const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>( Option.some(session(client)), ); const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ target: TARGET, state: supervisorState, session: activeSession, prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), connect: Effect.void, disconnect: Effect.void, retryNow: Effect.void, } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); const cache = Persistence.EnvironmentCacheStore.of({ loadShell: () => Effect.succeed(Option.none()), saveShell: () => Effect.never, loadThread: () => Effect.succeed(Option.none()), saveThread: () => Effect.void, removeThread: () => Effect.void, clear: () => Effect.void, }); // Cold cache with no HTTP snapshot available → falls back to the // socket-embedded snapshot. const snapshotLoader = ShellSnapshotLoader.of({ load: () => Effect.succeed(Option.none()), }); const shellState = yield* makeEnvironmentShellState().pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService(Persistence.EnvironmentCacheStore, cache), Effect.provideService(ShellSnapshotLoader, snapshotLoader), ); yield* SubscriptionRef.set(supervisorState, { desired: true, network: "online", phase: "connecting", stage: "synchronizing", attempt: 1, generation: 0, lastFailure: null, retryAt: null, }); yield* Queue.offer(events, { kind: "snapshot", snapshot: LIVE_SHELL_SNAPSHOT, }); yield* SubscriptionRef.changes(shellState).pipe( Stream.filter((state) => state.status === "live"), Stream.runHead, ); yield* SubscriptionRef.set(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; } const state = yield* SubscriptionRef.get(shellState); expect(state.status).toBe("live"); expect(Option.getOrThrow(state.snapshot)).toEqual(LIVE_SHELL_SNAPSHOT); }), ); it.effect("resumes a warm shell cache via afterSequence without an HTTP fetch", () => Effect.gen(function* () { const cachedSnapshot: OrchestrationShellSnapshot = { snapshotSequence: 5, projects: [], threads: [], updatedAt: "2026-06-06T00:00:00.000Z", }; const events = yield* Queue.unbounded<OrchestrationShellStreamItem>(); const capturedAfterSequence = yield* SubscriptionRef.make<number | undefined>(undefined); const loaderCalls = yield* SubscriptionRef.make(0); const client = { [ORCHESTRATION_WS_METHODS.subscribeShell]: (input: { readonly afterSequence?: number }) => Stream.unwrap( SubscriptionRef.set(capturedAfterSequence, input.afterSequence).pipe( Effect.as(Stream.fromQueue(events)), ), ), } as unknown as WsRpcProtocolClient; const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>( Option.some(session(client)), ); const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ target: TARGET, state: supervisorState, session: activeSession, prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), connect: Effect.void, disconnect: Effect.void, retryNow: Effect.void, } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); const cache = Persistence.EnvironmentCacheStore.of({ loadShell: () => Effect.succeed(Option.some(cachedSnapshot)), saveShell: () => Effect.void, loadThread: () => Effect.succeed(Option.none()), saveThread: () => Effect.void, removeThread: () => Effect.void, clear: () => Effect.void, }); const snapshotLoader = ShellSnapshotLoader.of({ load: () => SubscriptionRef.update(loaderCalls, (count) => count + 1).pipe(Effect.as(Option.none())), }); yield* makeEnvironmentShellState().pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService(Persistence.EnvironmentCacheStore, cache), Effect.provideService(ShellSnapshotLoader, snapshotLoader), ); // Wait until the subscription is established from the warm cache. yield* SubscriptionRef.changes(capturedAfterSequence).pipe( Stream.filter((value) => value !== undefined), Stream.runHead, ); expect(yield* SubscriptionRef.get(capturedAfterSequence)).toBe(5); expect(yield* SubscriptionRef.get(loaderCalls)).toBe(0); }), );});
import {
EnvironmentId,
ORCHESTRATION_WS_METHODS,
type OrchestrationShellSnapshot,
type OrchestrationShellStreamItem,
} 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 Stream from "effect/Stream";
import * as SubscriptionRef from "effect/SubscriptionRef";
import {
AVAILABLE_CONNECTION_STATE,
PrimaryConnectionTarget,
type PreparedConnection,
} from "../connection/model.ts";
import * as EnvironmentSupervisor from "../connection/supervisor.ts";
import * as Persistence from "../platform/persistence.ts";
import * as RpcSession from "../rpc/session.ts";
import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
import { makeEnvironmentShellState, ShellSnapshotLoader } from "./shell.ts";
const TARGET = new PrimaryConnectionTarget({
environmentId: EnvironmentId.make("environment-1"),
label: "Test environment",
httpBaseUrl: "https://environment.example.test",
wsBaseUrl: "wss://environment.example.test",
});
const PREPARED: PreparedConnection = {
environmentId: TARGET.environmentId,
label: TARGET.label,
httpBaseUrl: TARGET.httpBaseUrl,
socketUrl: TARGET.wsBaseUrl,
httpAuthorization: null,
target: TARGET,
};
const LIVE_SHELL_SNAPSHOT: OrchestrationShellSnapshot = {
snapshotSequence: 1,
projects: [],
threads: [],
updatedAt: "2026-06-06T00:00:00.000Z",
};
function session(client: WsRpcProtocolClient): RpcSession.RpcSession {
return {
client,
initialConfig: Effect.never,
ready: Effect.void,
probe: Effect.void,
closed: Effect.never,
};
}
describe("environment shell synchronization", () => {
it.effect("publishes live state before persistence and preserves it when ready", () =>
Effect.gen(function* () {
const events = yield* Queue.unbounded<OrchestrationShellStreamItem>();
const client = {
[ORCHESTRATION_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events),
} as unknown as WsRpcProtocolClient;
const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
Option.some(session(client)),
);
const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
target: TARGET,
state: supervisorState,
session: activeSession,
prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
connect: Effect.void,
disconnect: Effect.void,
retryNow: Effect.void,
} satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
const cache = Persistence.EnvironmentCacheStore.of({
loadShell: () => Effect.succeed(Option.none()),
saveShell: () => Effect.never,
loadThread: () => Effect.succeed(Option.none()),
saveThread: () => Effect.void,
removeThread: () => Effect.void,
clear: () => Effect.void,
});
// Cold cache with no HTTP snapshot available → falls back to the
// socket-embedded snapshot.
const snapshotLoader = ShellSnapshotLoader.of({
load: () => Effect.succeed(Option.none()),
});
const shellState = yield* makeEnvironmentShellState().pipe(
Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
Effect.provideService(Persistence.EnvironmentCacheStore, cache),
Effect.provideService(ShellSnapshotLoader, snapshotLoader),
);
yield* SubscriptionRef.set(supervisorState, {
desired: true,
network: "online",
phase: "connecting",
stage: "synchronizing",
attempt: 1,
generation: 0,
lastFailure: null,
retryAt: null,
});
yield* Queue.offer(events, {
kind: "snapshot",
snapshot: LIVE_SHELL_SNAPSHOT,
});
yield* SubscriptionRef.changes(shellState).pipe(
Stream.filter((state) => state.status === "live"),
Stream.runHead,
);
yield* SubscriptionRef.set(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;
}
const state = yield* SubscriptionRef.get(shellState);
expect(state.status).toBe("live");
expect(Option.getOrThrow(state.snapshot)).toEqual(LIVE_SHELL_SNAPSHOT);
}),
);
it.effect("resumes a warm shell cache via afterSequence without an HTTP fetch", () =>
Effect.gen(function* () {
const cachedSnapshot: OrchestrationShellSnapshot = {
snapshotSequence: 5,
projects: [],
threads: [],
updatedAt: "2026-06-06T00:00:00.000Z",
};
const events = yield* Queue.unbounded<OrchestrationShellStreamItem>();
const capturedAfterSequence = yield* SubscriptionRef.make<number | undefined>(undefined);
const loaderCalls = yield* SubscriptionRef.make(0);
const client = {
[ORCHESTRATION_WS_METHODS.subscribeShell]: (input: { readonly afterSequence?: number }) =>
Stream.unwrap(
SubscriptionRef.set(capturedAfterSequence, input.afterSequence).pipe(
Effect.as(Stream.fromQueue(events)),
),
),
} as unknown as WsRpcProtocolClient;
const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
Option.some(session(client)),
);
const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
target: TARGET,
state: supervisorState,
session: activeSession,
prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
connect: Effect.void,
disconnect: Effect.void,
retryNow: Effect.void,
} satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
const cache = Persistence.EnvironmentCacheStore.of({
loadShell: () => Effect.succeed(Option.some(cachedSnapshot)),
saveShell: () => Effect.void,
loadThread: () => Effect.succeed(Option.none()),
saveThread: () => Effect.void,
removeThread: () => Effect.void,
clear: () => Effect.void,
});
const snapshotLoader = ShellSnapshotLoader.of({
load: () =>
SubscriptionRef.update(loaderCalls, (count) => count + 1).pipe(Effect.as(Option.none())),
});
yield* makeEnvironmentShellState().pipe(
Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
Effect.provideService(Persistence.EnvironmentCacheStore, cache),
Effect.provideService(ShellSnapshotLoader, snapshotLoader),
);
// Wait until the subscription is established from the warm cache.
yield* SubscriptionRef.changes(capturedAfterSequence).pipe(
Stream.filter((value) => value !== undefined),
Stream.runHead,
);
expect(yield* SubscriptionRef.get(capturedAfterSequence)).toBe(5);
expect(yield* SubscriptionRef.get(loaderCalls)).toBe(0);
}),
);
});