import { EnvironmentId, ORCHESTRATION_WS_METHODS, ThreadId, type EnvironmentId as EnvironmentIdType, type OrchestrationThread, type OrchestrationThreadStreamItem, type ThreadId as ThreadIdType,} from "@t3tools/contracts";import * as Cause from "effect/Cause";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 { Atom } from "effect/unstable/reactivity"; import { EnvironmentRegistry } from "../connection/registry.ts";import { connectionProjectionPhase } from "../connection/model.ts";import { EnvironmentSupervisor } from "../connection/supervisor.ts";import { EnvironmentCacheStore } from "../platform/persistence.ts";import { subscribe } from "../rpc/client.ts";import { applyThreadDetailEvent } from "./threadReducer.ts";import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts";import { followStreamInEnvironment } from "./runtime.ts"; export type EnvironmentThreadStatus = "empty" | "cached" | "synchronizing" | "live" | "deleted"; export interface EnvironmentThreadState { readonly data: Option.Option<OrchestrationThread>; readonly status: EnvironmentThreadStatus; readonly error: Option.Option<string>;} export const EMPTY_ENVIRONMENT_THREAD_STATE: EnvironmentThreadState = { data: Option.none(), status: "empty", error: Option.none(),}; function statusWithoutLiveData(data: Option.Option<OrchestrationThread>): EnvironmentThreadStatus { return Option.isSome(data) ? "cached" : "empty";} function formatThreadError(cause: Cause.Cause<unknown>): string { const error = Cause.squash(cause); return error instanceof Error && error.message.trim().length > 0 ? error.message : "Could not synchronize the thread.";} export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* ( threadId: ThreadIdType,) { const supervisor = yield* EnvironmentSupervisor; const cache = yield* EnvironmentCacheStore; const environmentId = supervisor.target.environmentId; const cached = yield* cache.loadThread(environmentId, threadId).pipe( Effect.catch((error) => Effect.logWarning("Could not load cached thread.").pipe( Effect.annotateLogs({ environmentId, threadId, error: error.message, }), Effect.as(Option.none<OrchestrationThread>()), ), ), ); const state = yield* SubscriptionRef.make<EnvironmentThreadState>({ data: cached, status: statusWithoutLiveData(cached), error: Option.none(), }); const lastSequence = yield* SubscriptionRef.make(0); const persistence = yield* Queue.sliding<OrchestrationThread>(1); const persist = Effect.fn("EnvironmentThreadState.persist")(function* ( thread: OrchestrationThread, ) { yield* cache.saveThread(environmentId, thread).pipe( Effect.catch((error) => Effect.logWarning("Could not persist the thread cache.").pipe( Effect.annotateLogs({ environmentId, threadId, error: error.message, }), ), ), ); }); yield* Stream.fromQueue(persistence).pipe( Stream.debounce("500 millis"), Stream.runForEach(persist), Effect.forkScoped, ); const setSynchronizing = SubscriptionRef.update(state, (current) => ({ ...current, status: "synchronizing" as const, error: Option.none(), })); const setReady = SubscriptionRef.update(state, (current) => current.status === "live" || current.status === "deleted" ? current : { ...current, status: "synchronizing" as const, error: Option.none(), }, ); const setDisconnected = SubscriptionRef.update(state, (current) => ({ ...current, status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data), })); const setStreamError = (cause: Cause.Cause<unknown>) => SubscriptionRef.update(state, (current) => ({ ...current, status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data), error: Option.some(formatThreadError(cause)), })); const setThread = Effect.fn("EnvironmentThreadState.setThread")(function* ( thread: OrchestrationThread, ) { yield* SubscriptionRef.set(state, { data: Option.some(thread), status: "live", error: Option.none(), }); yield* Queue.offer(persistence, thread); }); const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () { yield* SubscriptionRef.set(state, { data: Option.none(), status: "deleted", error: Option.none(), }); yield* cache.removeThread(environmentId, threadId).pipe( Effect.catch((error) => Effect.logWarning("Could not remove the cached thread.").pipe( Effect.annotateLogs({ environmentId, threadId, error: error.message, }), ), ), ); }); const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* ( item: OrchestrationThreadStreamItem, ) { if (item.kind === "snapshot") { yield* SubscriptionRef.set(lastSequence, item.snapshot.snapshotSequence); yield* setThread(item.snapshot.thread); return; } const sequence = yield* SubscriptionRef.get(lastSequence); if (item.event.sequence <= sequence) { return; } yield* SubscriptionRef.set(lastSequence, item.event.sequence); const current = yield* SubscriptionRef.get(state); if (Option.isNone(current.data)) { if (item.event.type === "thread.deleted") { yield* setDeleted(); } return; } const result = applyThreadDetailEvent(current.data.value, item.event); if (result.kind === "updated") { yield* setThread(result.thread); } else if (result.kind === "deleted") { yield* setDeleted(); } }); yield* SubscriptionRef.changes(supervisor.state).pipe( Stream.runForEach((connectionState) => { switch (connectionProjectionPhase(connectionState)) { case "synchronizing": return setSynchronizing; case "disconnected": return setDisconnected; case "ready": return setReady; } }), Effect.forkScoped, ); yield* setSynchronizing; yield* subscribe( ORCHESTRATION_WS_METHODS.subscribeThread, { threadId }, { onExpectedFailure: setStreamError, retryExpectedFailureAfter: "250 millis", }, ).pipe(Stream.runForEach(applyItem), Effect.forkScoped); yield* Effect.addFinalizer(() => SubscriptionRef.get(state).pipe( Effect.flatMap((current) => Option.match(current.data, { onNone: () => Effect.void, onSome: persist, }), ), ), ); return state;}); export function threadStateChanges(environmentId: EnvironmentIdType, threadId: ThreadIdType) { return followStreamInEnvironment( environmentId, Stream.unwrap(makeEnvironmentThreadState(threadId).pipe(Effect.map(SubscriptionRef.changes))), );} function threadAtomKey(environmentId: EnvironmentIdType, threadId: ThreadIdType): string { return `${environmentId}\u0000${threadId}`;} function parseThreadAtomKey(key: string): { readonly environmentId: EnvironmentIdType; readonly threadId: ThreadIdType;} { const separator = key.indexOf("\u0000"); if (separator < 0) { throw new Error("Invalid environment thread atom key."); } return { environmentId: EnvironmentId.make(key.slice(0, separator)), threadId: ThreadId.make(key.slice(separator + 1)), };} export function createEnvironmentThreadStateAtoms<R, E>( runtime: Atom.AtomRuntime<EnvironmentRegistry | EnvironmentCacheStore | R, E>,) { const family = Atom.family((key: string) => { const { environmentId, threadId } = parseThreadAtomKey(key); return runtime .atom(threadStateChanges(environmentId, threadId), { initialValue: EMPTY_ENVIRONMENT_THREAD_STATE, }) .pipe( Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS), Atom.withLabel(`environment-thread-state:${key}`), ); }); return { stateAtom: (environmentId: EnvironmentIdType, threadId: ThreadIdType) => family(threadAtomKey(environmentId, threadId)), };} export * from "./archivedThreads.ts";export * from "./checkpointDiff.ts";export * from "./composerPathSearch.ts";export * from "./threadCommands.ts";export * from "./threadDetail.ts";export * from "./threadReducer.ts";export * from "./threadShell.ts";
import {
EnvironmentId,
ORCHESTRATION_WS_METHODS,
ThreadId,
type EnvironmentId as EnvironmentIdType,
type OrchestrationThread,
type OrchestrationThreadStreamItem,
type ThreadId as ThreadIdType,
} from "@t3tools/contracts";
import * as Cause from "effect/Cause";
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 { Atom } from "effect/unstable/reactivity";
import { EnvironmentRegistry } from "../connection/registry.ts";
import { connectionProjectionPhase } from "../connection/model.ts";
import { EnvironmentSupervisor } from "../connection/supervisor.ts";
import { EnvironmentCacheStore } from "../platform/persistence.ts";
import { subscribe } from "../rpc/client.ts";
import { applyThreadDetailEvent } from "./threadReducer.ts";
import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts";
import { followStreamInEnvironment } from "./runtime.ts";
export type EnvironmentThreadStatus = "empty" | "cached" | "synchronizing" | "live" | "deleted";
export interface EnvironmentThreadState {
readonly data: Option.Option<OrchestrationThread>;
readonly status: EnvironmentThreadStatus;
readonly error: Option.Option<string>;
}
export const EMPTY_ENVIRONMENT_THREAD_STATE: EnvironmentThreadState = {
data: Option.none(),
status: "empty",
error: Option.none(),
};
function statusWithoutLiveData(data: Option.Option<OrchestrationThread>): EnvironmentThreadStatus {
return Option.isSome(data) ? "cached" : "empty";
}
function formatThreadError(cause: Cause.Cause<unknown>): string {
const error = Cause.squash(cause);
return error instanceof Error && error.message.trim().length > 0
? error.message
: "Could not synchronize the thread.";
}
export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* (
threadId: ThreadIdType,
) {
const supervisor = yield* EnvironmentSupervisor;
const cache = yield* EnvironmentCacheStore;
const environmentId = supervisor.target.environmentId;
const cached = yield* cache.loadThread(environmentId, threadId).pipe(
Effect.catch((error) =>
Effect.logWarning("Could not load cached thread.").pipe(
Effect.annotateLogs({
environmentId,
threadId,
error: error.message,
}),
Effect.as(Option.none<OrchestrationThread>()),
),
),
);
const state = yield* SubscriptionRef.make<EnvironmentThreadState>({
data: cached,
status: statusWithoutLiveData(cached),
error: Option.none(),
});
const lastSequence = yield* SubscriptionRef.make(0);
const persistence = yield* Queue.sliding<OrchestrationThread>(1);
const persist = Effect.fn("EnvironmentThreadState.persist")(function* (
thread: OrchestrationThread,
) {
yield* cache.saveThread(environmentId, thread).pipe(
Effect.catch((error) =>
Effect.logWarning("Could not persist the thread cache.").pipe(
Effect.annotateLogs({
environmentId,
threadId,
error: error.message,
}),
),
),
);
});
yield* Stream.fromQueue(persistence).pipe(
Stream.debounce("500 millis"),
Stream.runForEach(persist),
Effect.forkScoped,
);
const setSynchronizing = SubscriptionRef.update(state, (current) => ({
...current,
status: "synchronizing" as const,
error: Option.none(),
}));
const setReady = SubscriptionRef.update(state, (current) =>
current.status === "live" || current.status === "deleted"
? current
: {
...current,
status: "synchronizing" as const,
error: Option.none(),
},
);
const setDisconnected = SubscriptionRef.update(state, (current) => ({
...current,
status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data),
}));
const setStreamError = (cause: Cause.Cause<unknown>) =>
SubscriptionRef.update(state, (current) => ({
...current,
status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data),
error: Option.some(formatThreadError(cause)),
}));
const setThread = Effect.fn("EnvironmentThreadState.setThread")(function* (
thread: OrchestrationThread,
) {
yield* SubscriptionRef.set(state, {
data: Option.some(thread),
status: "live",
error: Option.none(),
});
yield* Queue.offer(persistence, thread);
});
const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () {
yield* SubscriptionRef.set(state, {
data: Option.none(),
status: "deleted",
error: Option.none(),
});
yield* cache.removeThread(environmentId, threadId).pipe(
Effect.catch((error) =>
Effect.logWarning("Could not remove the cached thread.").pipe(
Effect.annotateLogs({
environmentId,
threadId,
error: error.message,
}),
),
),
);
});
const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* (
item: OrchestrationThreadStreamItem,
) {
if (item.kind === "snapshot") {
yield* SubscriptionRef.set(lastSequence, item.snapshot.snapshotSequence);
yield* setThread(item.snapshot.thread);
return;
}
const sequence = yield* SubscriptionRef.get(lastSequence);
if (item.event.sequence <= sequence) {
return;
}
yield* SubscriptionRef.set(lastSequence, item.event.sequence);
const current = yield* SubscriptionRef.get(state);
if (Option.isNone(current.data)) {
if (item.event.type === "thread.deleted") {
yield* setDeleted();
}
return;
}
const result = applyThreadDetailEvent(current.data.value, item.event);
if (result.kind === "updated") {
yield* setThread(result.thread);
} else if (result.kind === "deleted") {
yield* setDeleted();
}
});
yield* SubscriptionRef.changes(supervisor.state).pipe(
Stream.runForEach((connectionState) => {
switch (connectionProjectionPhase(connectionState)) {
case "synchronizing":
return setSynchronizing;
case "disconnected":
return setDisconnected;
case "ready":
return setReady;
}
}),
Effect.forkScoped,
);
yield* setSynchronizing;
yield* subscribe(
ORCHESTRATION_WS_METHODS.subscribeThread,
{ threadId },
{
onExpectedFailure: setStreamError,
retryExpectedFailureAfter: "250 millis",
},
).pipe(Stream.runForEach(applyItem), Effect.forkScoped);
yield* Effect.addFinalizer(() =>
SubscriptionRef.get(state).pipe(
Effect.flatMap((current) =>
Option.match(current.data, {
onNone: () => Effect.void,
onSome: persist,
}),
),
),
);
return state;
});
export function threadStateChanges(environmentId: EnvironmentIdType, threadId: ThreadIdType) {
return followStreamInEnvironment(
environmentId,
Stream.unwrap(makeEnvironmentThreadState(threadId).pipe(Effect.map(SubscriptionRef.changes))),
);
}
function threadAtomKey(environmentId: EnvironmentIdType, threadId: ThreadIdType): string {
return `${environmentId}\u0000${threadId}`;
}
function parseThreadAtomKey(key: string): {
readonly environmentId: EnvironmentIdType;
readonly threadId: ThreadIdType;
} {
const separator = key.indexOf("\u0000");
if (separator < 0) {
throw new Error("Invalid environment thread atom key.");
}
return {
environmentId: EnvironmentId.make(key.slice(0, separator)),
threadId: ThreadId.make(key.slice(separator + 1)),
};
}
export function createEnvironmentThreadStateAtoms<R, E>(
runtime: Atom.AtomRuntime<EnvironmentRegistry | EnvironmentCacheStore | R, E>,
) {
const family = Atom.family((key: string) => {
const { environmentId, threadId } = parseThreadAtomKey(key);
return runtime
.atom(threadStateChanges(environmentId, threadId), {
initialValue: EMPTY_ENVIRONMENT_THREAD_STATE,
})
.pipe(
Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
Atom.withLabel(`environment-thread-state:${key}`),
);
});
return {
stateAtom: (environmentId: EnvironmentIdType, threadId: ThreadIdType) =>
family(threadAtomKey(environmentId, threadId)),
};
}
export * from "./archivedThreads.ts";
export * from "./checkpointDiff.ts";
export * from "./composerPathSearch.ts";
export * from "./threadCommands.ts";
export * from "./threadDetail.ts";
export * from "./threadReducer.ts";
export * from "./threadShell.ts";