import { type EnvironmentId, type ServerConfig, type ServerConfigStreamEvent, type ServerLifecycleWelcomePayload, WS_METHODS,} from "@t3tools/contracts";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 Result from "effect/Result";import * as Stream from "effect/Stream";import * as SubscriptionRef from "effect/SubscriptionRef";import { AsyncResult, Atom } from "effect/unstable/reactivity"; import { createAtomCommandScheduler, createEnvironmentRpcCommand, createEnvironmentRpcQueryAtomFamily, createEnvironmentRpcSubscriptionAtomFamily,} from "./runtime.ts";import type { EnvironmentRegistry } from "../connection/registry.ts";import { EnvironmentSupervisor } from "../connection/supervisor.ts";import { safeErrorLogAttributes } from "../errors/safeLog.ts";import { EnvironmentCacheStore } from "../platform/persistence.ts";import { subscribe, type EnvironmentRpcInput } from "../rpc/client.ts";import { followStreamInEnvironment } from "./runtime.ts"; export interface ServerConfigProjection { readonly config: ServerConfig; readonly latestEvent: ServerConfigStreamEvent; readonly source: "cache" | "live";} export function applyServerConfigProjection( current: Option.Option<ServerConfigProjection>, event: ServerConfigStreamEvent,): Option.Option<ServerConfigProjection> { switch (event.type) { case "snapshot": return Option.some({ config: event.config, latestEvent: event, source: "live", }); case "keybindingsUpdated": return Option.map(current, (projection) => ({ config: { ...projection.config, keybindings: event.payload.keybindings, issues: event.payload.issues, }, latestEvent: event, source: "live", })); case "providerStatuses": return Option.map(current, (projection) => ({ config: { ...projection.config, providers: event.payload.providers, }, latestEvent: event, source: "live", })); case "settingsUpdated": return Option.map(current, (projection) => ({ config: { ...projection.config, settings: event.payload.settings, }, latestEvent: event, source: "live", })); }} export function projectServerConfig( current: Option.Option<ServerConfigProjection>, event: ServerConfigStreamEvent,): readonly [Option.Option<ServerConfigProjection>, ReadonlyArray<ServerConfigProjection>] { const next = applyServerConfigProjection(current, event); return [next, Option.toArray(next)];} const cachedConfigSnapshotEvent = (config: ServerConfig): ServerConfigStreamEvent => ({ version: 1, type: "snapshot", config,}); /** * Keeps a complete server configuration available during reconnects. Server * config carries the provider/model catalogue used by task creation, so it is * useful—and safe—to retain after a transport session ends. */export const makeEnvironmentServerConfigState = Effect.fn("EnvironmentServerConfigState.make")( function* () { const supervisor = yield* EnvironmentSupervisor; const cache = yield* EnvironmentCacheStore; const environmentId = supervisor.target.environmentId; const cachedConfig = yield* cache.loadServerConfig(environmentId).pipe( Effect.catch((error) => Effect.logWarning("Could not load cached server configuration.").pipe( Effect.annotateLogs({ environmentId, ...safeErrorLogAttributes(error), }), Effect.as(Option.none<ServerConfig>()), ), ), ); const state = yield* SubscriptionRef.make<Option.Option<ServerConfigProjection>>( Option.map(cachedConfig, (config) => ({ config, latestEvent: cachedConfigSnapshotEvent(config), source: "cache" as const, })), ); const persistence = yield* Queue.sliding<ServerConfig>(1); const pendingPersistence = yield* Ref.make<Option.Option<ServerConfig>>(Option.none()); const persist = Effect.fn("EnvironmentServerConfigState.persist")(function* ( config: ServerConfig, ) { return yield* cache.saveServerConfig(environmentId, config).pipe( Effect.as(true), Effect.catch((error) => Effect.logWarning("Could not persist cached server configuration.").pipe( Effect.annotateLogs({ environmentId, ...safeErrorLogAttributes(error), }), Effect.as(false), ), ), ); }); const persistPending = Effect.fn("EnvironmentServerConfigState.persistPending")(function* ( config: ServerConfig, ) { if (!(yield* persist(config))) { return; } yield* Ref.update(pendingPersistence, (pending) => Option.isSome(pending) && pending.value === config ? Option.none() : pending, ); }); yield* Stream.fromQueue(persistence).pipe( Stream.debounce("500 millis"), Stream.runForEach(persistPending), Effect.forkScoped, ); yield* subscribe(WS_METHODS.subscribeServerConfig, {}).pipe( Stream.runForEach((event) => Effect.gen(function* () { const next = applyServerConfigProjection(yield* SubscriptionRef.get(state), event); if (Option.isNone(next)) { return; } yield* Ref.set(pendingPersistence, Option.some(next.value.config)); yield* SubscriptionRef.set(state, next); yield* Queue.offer(persistence, next.value.config); }), ), Effect.forkScoped, ); yield* Effect.addFinalizer(() => Ref.get(pendingPersistence).pipe( Effect.flatMap( Option.match({ onNone: () => Effect.void, onSome: (config) => persist(config).pipe(Effect.asVoid), }), ), ), ); return state; },); export function serverConfigStateChanges(environmentId: EnvironmentId) { return followStreamInEnvironment( environmentId, Stream.unwrap( makeEnvironmentServerConfigState().pipe( Effect.map((state) => SubscriptionRef.changes(state).pipe( Stream.filterMap((projection) => Option.match(projection, { onNone: () => Result.failVoid, onSome: (value) => Result.succeed(value), }), ), ), ), ), ), );} export function projectServerWelcome( current: Option.Option<ServerLifecycleWelcomePayload>, event: { readonly type: "welcome" | "ready"; readonly payload: unknown; },): readonly [ Option.Option<ServerLifecycleWelcomePayload>, ReadonlyArray<ServerLifecycleWelcomePayload>,] { if (event.type !== "welcome") { return [current, []]; } const welcome = event.payload as ServerLifecycleWelcomePayload; return [Option.some(welcome), [welcome]];} export function resolveServerConfigValue( projection: ServerConfigProjection | null, initialConfig: ServerConfig | null,): ServerConfig | null { if (projection?.source === "live") return projection.config; return initialConfig ?? projection?.config ?? null;} export function createServerEnvironmentAtoms<R, E>( runtime: Atom.AtomRuntime<EnvironmentRegistry | EnvironmentCacheStore | R, E>, options: { readonly initialConfigValueAtom: ( environmentId: EnvironmentId, ) => Atom.Atom<ServerConfig | null>; },) { const configScheduler = createAtomCommandScheduler(); const configConcurrency = { mode: "serial" as const, key: ({ environmentId }: { readonly environmentId: string }) => environmentId, }; const configProjectionFamily = Atom.family((environmentId: EnvironmentId) => runtime .atom(serverConfigStateChanges(environmentId)) .pipe( Atom.setIdleTTL(5 * 60_000), Atom.withLabel(`environment-data:server:config-projection:${environmentId}`), ), ); const configProjection = (target: { readonly environmentId: EnvironmentId; readonly input: EnvironmentRpcInput<typeof WS_METHODS.subscribeServerConfig>; }) => configProjectionFamily(target.environmentId); const emptyConfigAtom = Atom.make<ServerConfig | null>(null).pipe( Atom.withLabel("environment-data:server:config:empty"), ); const configValueAtom = Atom.family((environmentId: EnvironmentId | null) => { if (environmentId === null) { return emptyConfigAtom; } return Atom.make((get): ServerConfig | null => { const projection = Option.getOrNull( AsyncResult.value(get(configProjection({ environmentId, input: {} }))), ); return resolveServerConfigValue( projection, get(options.initialConfigValueAtom(environmentId)), ); }).pipe(Atom.withLabel(`environment-data:server:config:${environmentId}`)); }); const settingsValueAtom = Atom.family((environmentId: EnvironmentId) => Atom.make((get) => get(configValueAtom(environmentId))?.settings ?? null).pipe( Atom.withLabel(`environment-data:server:settings:${environmentId}`), ), ); const providersValueAtom = Atom.family((environmentId: EnvironmentId) => Atom.make((get) => get(configValueAtom(environmentId))?.providers ?? null).pipe( Atom.withLabel(`environment-data:server:providers:${environmentId}`), ), ); return { configValueAtom, settingsValueAtom, providersValueAtom, traceDiagnostics: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:server:trace-diagnostics", tag: WS_METHODS.serverGetTraceDiagnostics, }), processDiagnostics: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:server:process-diagnostics", tag: WS_METHODS.serverGetProcessDiagnostics, }), processResourceHistory: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:server:process-resource-history", tag: WS_METHODS.serverGetProcessResourceHistory, }), configProjection, welcome: createEnvironmentRpcSubscriptionAtomFamily(runtime, { label: "environment-data:server:welcome", tag: WS_METHODS.subscribeServerLifecycle, transform: (stream) => stream.pipe( Stream.mapAccum(Option.none<ServerLifecycleWelcomePayload>, projectServerWelcome), ), }), refreshProviders: createEnvironmentRpcCommand(runtime, { label: "environment-data:server:refresh-providers", tag: WS_METHODS.serverRefreshProviders, concurrency: { mode: "singleFlight", key: ({ environmentId }) => environmentId, }, }), updateProvider: createEnvironmentRpcCommand(runtime, { label: "environment-data:server:update-provider", tag: WS_METHODS.serverUpdateProvider, scheduler: configScheduler, concurrency: configConcurrency, }), updateServer: createEnvironmentRpcCommand(runtime, { label: "environment-data:server:update-server", tag: WS_METHODS.serverUpdateServer, scheduler: configScheduler, concurrency: configConcurrency, }), upsertKeybinding: createEnvironmentRpcCommand(runtime, { label: "environment-data:server:upsert-keybinding", tag: WS_METHODS.serverUpsertKeybinding, scheduler: configScheduler, concurrency: configConcurrency, }), removeKeybinding: createEnvironmentRpcCommand(runtime, { label: "environment-data:server:remove-keybinding", tag: WS_METHODS.serverRemoveKeybinding, scheduler: configScheduler, concurrency: configConcurrency, }), updateSettings: createEnvironmentRpcCommand(runtime, { label: "environment-data:server:update-settings", tag: WS_METHODS.serverUpdateSettings, scheduler: configScheduler, concurrency: configConcurrency, }), signalProcess: createEnvironmentRpcCommand(runtime, { label: "environment-data:server:signal-process", tag: WS_METHODS.serverSignalProcess, }), };}
import {
type EnvironmentId,
type ServerConfig,
type ServerConfigStreamEvent,
type ServerLifecycleWelcomePayload,
WS_METHODS,
} from "@t3tools/contracts";
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 Result from "effect/Result";
import * as Stream from "effect/Stream";
import * as SubscriptionRef from "effect/SubscriptionRef";
import { AsyncResult, Atom } from "effect/unstable/reactivity";
import {
createAtomCommandScheduler,
createEnvironmentRpcCommand,
createEnvironmentRpcQueryAtomFamily,
createEnvironmentRpcSubscriptionAtomFamily,
} from "./runtime.ts";
import type { EnvironmentRegistry } from "../connection/registry.ts";
import { EnvironmentSupervisor } from "../connection/supervisor.ts";
import { safeErrorLogAttributes } from "../errors/safeLog.ts";
import { EnvironmentCacheStore } from "../platform/persistence.ts";
import { subscribe, type EnvironmentRpcInput } from "../rpc/client.ts";
import { followStreamInEnvironment } from "./runtime.ts";
export interface ServerConfigProjection {
readonly config: ServerConfig;
readonly latestEvent: ServerConfigStreamEvent;
readonly source: "cache" | "live";
}
export function applyServerConfigProjection(
current: Option.Option<ServerConfigProjection>,
event: ServerConfigStreamEvent,
): Option.Option<ServerConfigProjection> {
switch (event.type) {
case "snapshot":
return Option.some({
config: event.config,
latestEvent: event,
source: "live",
});
case "keybindingsUpdated":
return Option.map(current, (projection) => ({
config: {
...projection.config,
keybindings: event.payload.keybindings,
issues: event.payload.issues,
},
latestEvent: event,
source: "live",
}));
case "providerStatuses":
return Option.map(current, (projection) => ({
config: {
...projection.config,
providers: event.payload.providers,
},
latestEvent: event,
source: "live",
}));
case "settingsUpdated":
return Option.map(current, (projection) => ({
config: {
...projection.config,
settings: event.payload.settings,
},
latestEvent: event,
source: "live",
}));
}
}
export function projectServerConfig(
current: Option.Option<ServerConfigProjection>,
event: ServerConfigStreamEvent,
): readonly [Option.Option<ServerConfigProjection>, ReadonlyArray<ServerConfigProjection>] {
const next = applyServerConfigProjection(current, event);
return [next, Option.toArray(next)];
}
const cachedConfigSnapshotEvent = (config: ServerConfig): ServerConfigStreamEvent => ({
version: 1,
type: "snapshot",
config,
});
/**
* Keeps a complete server configuration available during reconnects. Server
* config carries the provider/model catalogue used by task creation, so it is
* useful—and safe—to retain after a transport session ends.
*/
export const makeEnvironmentServerConfigState = Effect.fn("EnvironmentServerConfigState.make")(
function* () {
const supervisor = yield* EnvironmentSupervisor;
const cache = yield* EnvironmentCacheStore;
const environmentId = supervisor.target.environmentId;
const cachedConfig = yield* cache.loadServerConfig(environmentId).pipe(
Effect.catch((error) =>
Effect.logWarning("Could not load cached server configuration.").pipe(
Effect.annotateLogs({
environmentId,
...safeErrorLogAttributes(error),
}),
Effect.as(Option.none<ServerConfig>()),
),
),
);
const state = yield* SubscriptionRef.make<Option.Option<ServerConfigProjection>>(
Option.map(cachedConfig, (config) => ({
config,
latestEvent: cachedConfigSnapshotEvent(config),
source: "cache" as const,
})),
);
const persistence = yield* Queue.sliding<ServerConfig>(1);
const pendingPersistence = yield* Ref.make<Option.Option<ServerConfig>>(Option.none());
const persist = Effect.fn("EnvironmentServerConfigState.persist")(function* (
config: ServerConfig,
) {
return yield* cache.saveServerConfig(environmentId, config).pipe(
Effect.as(true),
Effect.catch((error) =>
Effect.logWarning("Could not persist cached server configuration.").pipe(
Effect.annotateLogs({
environmentId,
...safeErrorLogAttributes(error),
}),
Effect.as(false),
),
),
);
});
const persistPending = Effect.fn("EnvironmentServerConfigState.persistPending")(function* (
config: ServerConfig,
) {
if (!(yield* persist(config))) {
return;
}
yield* Ref.update(pendingPersistence, (pending) =>
Option.isSome(pending) && pending.value === config ? Option.none() : pending,
);
});
yield* Stream.fromQueue(persistence).pipe(
Stream.debounce("500 millis"),
Stream.runForEach(persistPending),
Effect.forkScoped,
);
yield* subscribe(WS_METHODS.subscribeServerConfig, {}).pipe(
Stream.runForEach((event) =>
Effect.gen(function* () {
const next = applyServerConfigProjection(yield* SubscriptionRef.get(state), event);
if (Option.isNone(next)) {
return;
}
yield* Ref.set(pendingPersistence, Option.some(next.value.config));
yield* SubscriptionRef.set(state, next);
yield* Queue.offer(persistence, next.value.config);
}),
),
Effect.forkScoped,
);
yield* Effect.addFinalizer(() =>
Ref.get(pendingPersistence).pipe(
Effect.flatMap(
Option.match({
onNone: () => Effect.void,
onSome: (config) => persist(config).pipe(Effect.asVoid),
}),
),
),
);
return state;
},
);
export function serverConfigStateChanges(environmentId: EnvironmentId) {
return followStreamInEnvironment(
environmentId,
Stream.unwrap(
makeEnvironmentServerConfigState().pipe(
Effect.map((state) =>
SubscriptionRef.changes(state).pipe(
Stream.filterMap((projection) =>
Option.match(projection, {
onNone: () => Result.failVoid,
onSome: (value) => Result.succeed(value),
}),
),
),
),
),
),
);
}
export function projectServerWelcome(
current: Option.Option<ServerLifecycleWelcomePayload>,
event: {
readonly type: "welcome" | "ready";
readonly payload: unknown;
},
): readonly [
Option.Option<ServerLifecycleWelcomePayload>,
ReadonlyArray<ServerLifecycleWelcomePayload>,
] {
if (event.type !== "welcome") {
return [current, []];
}
const welcome = event.payload as ServerLifecycleWelcomePayload;
return [Option.some(welcome), [welcome]];
}
export function resolveServerConfigValue(
projection: ServerConfigProjection | null,
initialConfig: ServerConfig | null,
): ServerConfig | null {
if (projection?.source === "live") return projection.config;
return initialConfig ?? projection?.config ?? null;
}
export function createServerEnvironmentAtoms<R, E>(
runtime: Atom.AtomRuntime<EnvironmentRegistry | EnvironmentCacheStore | R, E>,
options: {
readonly initialConfigValueAtom: (
environmentId: EnvironmentId,
) => Atom.Atom<ServerConfig | null>;
},
) {
const configScheduler = createAtomCommandScheduler();
const configConcurrency = {
mode: "serial" as const,
key: ({ environmentId }: { readonly environmentId: string }) => environmentId,
};
const configProjectionFamily = Atom.family((environmentId: EnvironmentId) =>
runtime
.atom(serverConfigStateChanges(environmentId))
.pipe(
Atom.setIdleTTL(5 * 60_000),
Atom.withLabel(`environment-data:server:config-projection:${environmentId}`),
),
);
const configProjection = (target: {
readonly environmentId: EnvironmentId;
readonly input: EnvironmentRpcInput<typeof WS_METHODS.subscribeServerConfig>;
}) => configProjectionFamily(target.environmentId);
const emptyConfigAtom = Atom.make<ServerConfig | null>(null).pipe(
Atom.withLabel("environment-data:server:config:empty"),
);
const configValueAtom = Atom.family((environmentId: EnvironmentId | null) => {
if (environmentId === null) {
return emptyConfigAtom;
}
return Atom.make((get): ServerConfig | null => {
const projection = Option.getOrNull(
AsyncResult.value(get(configProjection({ environmentId, input: {} }))),
);
return resolveServerConfigValue(
projection,
get(options.initialConfigValueAtom(environmentId)),
);
}).pipe(Atom.withLabel(`environment-data:server:config:${environmentId}`));
});
const settingsValueAtom = Atom.family((environmentId: EnvironmentId) =>
Atom.make((get) => get(configValueAtom(environmentId))?.settings ?? null).pipe(
Atom.withLabel(`environment-data:server:settings:${environmentId}`),
),
);
const providersValueAtom = Atom.family((environmentId: EnvironmentId) =>
Atom.make((get) => get(configValueAtom(environmentId))?.providers ?? null).pipe(
Atom.withLabel(`environment-data:server:providers:${environmentId}`),
),
);
return {
configValueAtom,
settingsValueAtom,
providersValueAtom,
traceDiagnostics: createEnvironmentRpcQueryAtomFamily(runtime, {
label: "environment-data:server:trace-diagnostics",
tag: WS_METHODS.serverGetTraceDiagnostics,
}),
processDiagnostics: createEnvironmentRpcQueryAtomFamily(runtime, {
label: "environment-data:server:process-diagnostics",
tag: WS_METHODS.serverGetProcessDiagnostics,
}),
processResourceHistory: createEnvironmentRpcQueryAtomFamily(runtime, {
label: "environment-data:server:process-resource-history",
tag: WS_METHODS.serverGetProcessResourceHistory,
}),
configProjection,
welcome: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
label: "environment-data:server:welcome",
tag: WS_METHODS.subscribeServerLifecycle,
transform: (stream) =>
stream.pipe(
Stream.mapAccum(Option.none<ServerLifecycleWelcomePayload>, projectServerWelcome),
),
}),
refreshProviders: createEnvironmentRpcCommand(runtime, {
label: "environment-data:server:refresh-providers",
tag: WS_METHODS.serverRefreshProviders,
concurrency: {
mode: "singleFlight",
key: ({ environmentId }) => environmentId,
},
}),
updateProvider: createEnvironmentRpcCommand(runtime, {
label: "environment-data:server:update-provider",
tag: WS_METHODS.serverUpdateProvider,
scheduler: configScheduler,
concurrency: configConcurrency,
}),
updateServer: createEnvironmentRpcCommand(runtime, {
label: "environment-data:server:update-server",
tag: WS_METHODS.serverUpdateServer,
scheduler: configScheduler,
concurrency: configConcurrency,
}),
upsertKeybinding: createEnvironmentRpcCommand(runtime, {
label: "environment-data:server:upsert-keybinding",
tag: WS_METHODS.serverUpsertKeybinding,
scheduler: configScheduler,
concurrency: configConcurrency,
}),
removeKeybinding: createEnvironmentRpcCommand(runtime, {
label: "environment-data:server:remove-keybinding",
tag: WS_METHODS.serverRemoveKeybinding,
scheduler: configScheduler,
concurrency: configConcurrency,
}),
updateSettings: createEnvironmentRpcCommand(runtime, {
label: "environment-data:server:update-settings",
tag: WS_METHODS.serverUpdateSettings,
scheduler: configScheduler,
concurrency: configConcurrency,
}),
signalProcess: createEnvironmentRpcCommand(runtime, {
label: "environment-data:server:signal-process",
tag: WS_METHODS.serverSignalProcess,
}),
};
}