diff --git a/apps/server/src/diagnostics/ProcessDiagnostics.ts b/apps/server/src/diagnostics/ProcessDiagnostics.ts index f5f746134f2..40e7f347be1 100644 --- a/apps/server/src/diagnostics/ProcessDiagnostics.ts +++ b/apps/server/src/diagnostics/ProcessDiagnostics.ts @@ -12,7 +12,8 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; -import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; +import * as ChildProcess from "effect/unstable/process/ChildProcess"; +import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner"; import { collectUint8StreamText } from "../stream/collectUint8StreamText.ts"; @@ -31,35 +32,80 @@ const PROCESS_QUERY_TIMEOUT_MS = 1_000; const POSIX_PROCESS_QUERY_COMMAND = "pid=,ppid=,pgid=,stat=,pcpu=,rss=,etime=,command="; const PROCESS_QUERY_MAX_OUTPUT_BYTES = 2 * 1024 * 1024; -export interface ProcessDiagnosticsShape { - readonly read: Effect.Effect; - readonly signal: (input: { - readonly pid: number; - readonly signal: ServerProcessSignal; - }) => Effect.Effect; -} - export class ProcessDiagnostics extends Context.Service< ProcessDiagnostics, - ProcessDiagnosticsShape + { + readonly read: Effect.Effect; + readonly signal: (input: { + readonly pid: number; + readonly signal: ServerProcessSignal; + }) => Effect.Effect; + } >()("t3/diagnostics/ProcessDiagnostics") {} -class ProcessDiagnosticsError extends Schema.TaggedErrorClass()( - "ProcessDiagnosticsError", +class ProcessDiagnosticsQueryTimeoutError extends Schema.TaggedErrorClass()( + "ProcessDiagnosticsQueryTimeoutError", + { command: Schema.String }, +) { + override get message(): string { + return `Process diagnostics query '${this.command}' timed out.`; + } +} + +class ProcessDiagnosticsQueryFailedError extends Schema.TaggedErrorClass()( + "ProcessDiagnosticsQueryFailedError", { - message: Schema.String, + command: Schema.String, + stderr: Schema.optional(Schema.String), cause: Schema.optional(Schema.Defect()), }, -) {} -const isProcessDiagnosticsError = Schema.is(ProcessDiagnosticsError); +) { + override get message(): string { + return this.stderr?.trim() || `Failed to query process diagnostics with '${this.command}'.`; + } +} -function toProcessDiagnosticsError(message: string, cause?: unknown): ProcessDiagnosticsError { - return new ProcessDiagnosticsError({ - message, - ...(cause === undefined ? {} : { cause }), - }); +class ProcessDiagnosticsServerProcessSignalError extends Schema.TaggedErrorClass()( + "ProcessDiagnosticsServerProcessSignalError", + { pid: Schema.Number }, +) { + override get message(): string { + return "Refusing to signal the T3 server process."; + } +} + +class ProcessDiagnosticsNotDescendantError extends Schema.TaggedErrorClass()( + "ProcessDiagnosticsNotDescendantError", + { pid: Schema.Number }, +) { + override get message(): string { + return `Process ${this.pid} is not a live descendant of the T3 server.`; + } } +class ProcessDiagnosticsSignalFailedError extends Schema.TaggedErrorClass()( + "ProcessDiagnosticsSignalFailedError", + { + pid: Schema.Number, + signal: Schema.String, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return `Failed to signal process ${this.pid} with ${this.signal}.`; + } +} + +const ProcessDiagnosticsError = Schema.Union([ + ProcessDiagnosticsQueryTimeoutError, + ProcessDiagnosticsQueryFailedError, + ProcessDiagnosticsServerProcessSignalError, + ProcessDiagnosticsNotDescendantError, + ProcessDiagnosticsSignalFailedError, +]); +type ProcessDiagnosticsError = typeof ProcessDiagnosticsError.Type; +const isProcessDiagnosticsError = Schema.is(ProcessDiagnosticsError); + function parsePositiveInt(value: string): number | null { const parsed = Number.parseInt(value, 10); return Number.isInteger(parsed) && parsed > 0 ? parsed : null; @@ -272,11 +318,7 @@ interface ProcessOutput { } const runProcess = Effect.fn("runProcess")( - function* (input: { - readonly command: string; - readonly args: ReadonlyArray; - readonly errorMessage: string; - }) { + function* (input: { readonly command: string; readonly args: ReadonlyArray }) { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; // `ps` and `powershell.exe` are real executables; spawning through cmd.exe // shell mode would re-tokenize the PowerShell `-Command` payload (which @@ -315,14 +357,22 @@ const runProcess = Effect.fn("runProcess")( Effect.timeoutOption(Duration.millis(PROCESS_QUERY_TIMEOUT_MS)), Effect.flatMap((result) => Option.match(result, { - onNone: () => Effect.fail(toProcessDiagnosticsError(`${input.errorMessage} timed out.`)), + onNone: () => + Effect.fail( + new ProcessDiagnosticsQueryTimeoutError({ + command: input.command, + }), + ), onSome: Effect.succeed, }), ), Effect.mapError((cause) => isProcessDiagnosticsError(cause) ? cause - : toProcessDiagnosticsError(input.errorMessage, cause), + : new ProcessDiagnosticsQueryFailedError({ + command: input.command, + cause, + }), ), ), ); @@ -335,11 +385,15 @@ function readPosixProcessRows(): Effect.Effect< return runProcess({ command: "ps", args: ["-axo", POSIX_PROCESS_QUERY_COMMAND], - errorMessage: "Failed to query process diagnostics.", }).pipe( Effect.flatMap((result) => result.exitCode !== 0 - ? Effect.fail(toProcessDiagnosticsError(result.stderr.trim() || "ps failed.")) + ? Effect.fail( + new ProcessDiagnosticsQueryFailedError({ + command: "ps", + stderr: result.stderr.trim() || "ps failed.", + }), + ) : Effect.succeed(parsePosixProcessRows(result.stdout)), ), ); @@ -361,12 +415,14 @@ function readWindowsProcessRows(): Effect.Effect< return runProcess({ command: "powershell.exe", args: ["-NoProfile", "-NonInteractive", "-Command", command], - errorMessage: "Failed to query process diagnostics.", }).pipe( Effect.flatMap((result) => result.exitCode !== 0 ? Effect.fail( - toProcessDiagnosticsError(result.stderr.trim() || "PowerShell process query failed."), + new ProcessDiagnosticsQueryFailedError({ + command: "powershell.exe", + stderr: result.stderr.trim() || "PowerShell process query failed.", + }), ) : Effect.succeed(parseWindowsProcessRows(result.stdout)), ), @@ -390,7 +446,11 @@ function assertDescendantPid( pid: number, ): Effect.Effect { if (pid === process.pid) { - return Effect.fail(toProcessDiagnosticsError("Refusing to signal the T3 server process.")); + return Effect.fail( + new ProcessDiagnosticsServerProcessSignalError({ + pid, + }), + ); } return readProcessRows.pipe( @@ -402,16 +462,18 @@ function assertDescendantPid( return descendant ? Effect.void : Effect.fail( - toProcessDiagnosticsError(`Process ${pid} is not a live descendant of the T3 server.`), + new ProcessDiagnosticsNotDescendantError({ + pid, + }), ); }), ); } -export const make = Effect.fn("makeProcessDiagnostics")(function* () { +export const make = Effect.gen(function* () { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const read: ProcessDiagnosticsShape["read"] = Effect.gen(function* () { + const read: ProcessDiagnostics["Service"]["read"] = Effect.gen(function* () { const readAt = yield* DateTime.now; const rows = yield* readProcessRows.pipe( Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), @@ -427,7 +489,7 @@ export const make = Effect.fn("makeProcessDiagnostics")(function* () { ), ); - const signal: ProcessDiagnosticsShape["signal"] = Effect.fn("ProcessDiagnostics.signal")( + const signal: ProcessDiagnostics["Service"]["signal"] = Effect.fn("ProcessDiagnostics.signal")( function* (input) { return yield* assertDescendantPid(input.pid).pipe( Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), @@ -443,10 +505,11 @@ export const make = Effect.fn("makeProcessDiagnostics")(function* () { }; }, catch: (cause) => - toProcessDiagnosticsError( - `Failed to signal process ${input.pid} with ${input.signal}.`, + new ProcessDiagnosticsSignalFailedError({ + pid: input.pid, + signal: input.signal, cause, - ), + }), }), ), Effect.catch((error: ProcessDiagnosticsError) => @@ -464,4 +527,4 @@ export const make = Effect.fn("makeProcessDiagnostics")(function* () { return ProcessDiagnostics.of({ read, signal }); }); -export const layer = Layer.effect(ProcessDiagnostics, make()); +export const layer = Layer.effect(ProcessDiagnostics, make); diff --git a/apps/server/src/diagnostics/ProcessResourceMonitor.test.ts b/apps/server/src/diagnostics/ProcessResourceMonitor.test.ts index 11d12c012db..49a9676ab11 100644 --- a/apps/server/src/diagnostics/ProcessResourceMonitor.test.ts +++ b/apps/server/src/diagnostics/ProcessResourceMonitor.test.ts @@ -3,16 +3,13 @@ import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; -import { - aggregateProcessResourceHistory, - collectMonitoredSamples, -} from "./ProcessResourceMonitor.ts"; +import * as ProcessResourceMonitor from "./ProcessResourceMonitor.ts"; describe("ProcessResourceMonitor", () => { it.effect("samples the server root process and descendants", () => Effect.sync(() => { const sampledAt = DateTime.makeUnsafe("2026-05-05T10:00:00.000Z"); - const samples = collectMonitoredSamples({ + const samples = ProcessResourceMonitor.collectMonitoredSamples({ serverPid: 100, sampledAt, sampledAtMs: DateTime.toEpochMillis(sampledAt), @@ -72,7 +69,7 @@ describe("ProcessResourceMonitor", () => { const firstAt = DateTime.makeUnsafe("2026-05-05T10:00:00.000Z"); const secondAt = DateTime.makeUnsafe("2026-05-05T10:00:05.000Z"); const samples = [ - ...collectMonitoredSamples({ + ...ProcessResourceMonitor.collectMonitoredSamples({ serverPid: 100, sampledAt: firstAt, sampledAtMs: DateTime.toEpochMillis(firstAt), @@ -89,7 +86,7 @@ describe("ProcessResourceMonitor", () => { }, ], }), - ...collectMonitoredSamples({ + ...ProcessResourceMonitor.collectMonitoredSamples({ serverPid: 100, sampledAt: secondAt, sampledAtMs: DateTime.toEpochMillis(secondAt), @@ -108,7 +105,7 @@ describe("ProcessResourceMonitor", () => { }), ]; - const result = aggregateProcessResourceHistory({ + const result = ProcessResourceMonitor.aggregateProcessResourceHistory({ samples, readAt: secondAt, readAtMs: DateTime.toEpochMillis(secondAt), @@ -132,7 +129,7 @@ describe("ProcessResourceMonitor", () => { const firstAt = DateTime.makeUnsafe("2026-05-05T10:00:00.400Z"); const secondAt = DateTime.makeUnsafe("2026-05-05T10:00:05.900Z"); const samples = [ - ...collectMonitoredSamples({ + ...ProcessResourceMonitor.collectMonitoredSamples({ serverPid: 100, sampledAt: firstAt, sampledAtMs: DateTime.toEpochMillis(firstAt), @@ -149,7 +146,7 @@ describe("ProcessResourceMonitor", () => { }, ], }), - ...collectMonitoredSamples({ + ...ProcessResourceMonitor.collectMonitoredSamples({ serverPid: 100, sampledAt: secondAt, sampledAtMs: DateTime.toEpochMillis(secondAt), @@ -168,7 +165,7 @@ describe("ProcessResourceMonitor", () => { }), ]; - const result = aggregateProcessResourceHistory({ + const result = ProcessResourceMonitor.aggregateProcessResourceHistory({ samples, readAt: secondAt, readAtMs: DateTime.toEpochMillis(secondAt), @@ -187,7 +184,7 @@ describe("ProcessResourceMonitor", () => { it.effect("returns all process summaries in the selected window", () => Effect.sync(() => { const sampledAt = DateTime.makeUnsafe("2026-05-05T10:00:00.000Z"); - const samples = collectMonitoredSamples({ + const samples = ProcessResourceMonitor.collectMonitoredSamples({ serverPid: 100, sampledAt, sampledAtMs: DateTime.toEpochMillis(sampledAt), @@ -215,7 +212,7 @@ describe("ProcessResourceMonitor", () => { ], }); - const result = aggregateProcessResourceHistory({ + const result = ProcessResourceMonitor.aggregateProcessResourceHistory({ samples, readAt: sampledAt, readAtMs: DateTime.toEpochMillis(sampledAt), diff --git a/apps/server/src/diagnostics/ProcessResourceMonitor.ts b/apps/server/src/diagnostics/ProcessResourceMonitor.ts index efeeb66256d..b6e71dd2423 100644 --- a/apps/server/src/diagnostics/ProcessResourceMonitor.ts +++ b/apps/server/src/diagnostics/ProcessResourceMonitor.ts @@ -10,14 +10,9 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; -import { ChildProcessSpawner } from "effect/unstable/process"; +import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner"; -import { - buildDescendantEntries, - isDiagnosticsQueryProcess, - type ProcessRow, - readProcessRows, -} from "./ProcessDiagnostics.ts"; +import * as ProcessDiagnostics from "./ProcessDiagnostics.ts"; const SAMPLE_INTERVAL_MS = 5_000; const RETENTION_MS = 60 * 60_000; @@ -41,38 +36,41 @@ interface MonitorState { readonly lastError: string | null; } -export interface ProcessResourceMonitorShape { - readonly readHistory: ( - input: ServerProcessResourceHistoryInput, - ) => Effect.Effect; -} - export class ProcessResourceMonitor extends Context.Service< ProcessResourceMonitor, - ProcessResourceMonitorShape + { + readonly readHistory: ( + input: ServerProcessResourceHistoryInput, + ) => Effect.Effect; + } >()("t3/diagnostics/ProcessResourceMonitor") {} function dateTimeFromMillis(ms: number): DateTime.Utc { return DateTime.makeUnsafe(ms); } -function sampleKey(row: Pick): string { +function sampleKey(row: Pick): string { return `${row.pid}:${row.command}`; } -function findServerRootRow(rows: ReadonlyArray, serverPid: number): ProcessRow | null { +function findServerRootRow( + rows: ReadonlyArray, + serverPid: number, +): ProcessDiagnostics.ProcessRow | null { return rows.find((row) => row.pid === serverPid) ?? null; } export function collectMonitoredSamples(input: { - readonly rows: ReadonlyArray; + readonly rows: ReadonlyArray; readonly serverPid: number; readonly sampledAt: DateTime.Utc; readonly sampledAtMs: number; }): ReadonlyArray { - const rows = input.rows.filter((row) => !isDiagnosticsQueryProcess(row, input.serverPid)); + const rows = input.rows.filter( + (row) => !ProcessDiagnostics.isDiagnosticsQueryProcess(row, input.serverPid), + ); const root = findServerRootRow(rows, input.serverPid); - const descendants = buildDescendantEntries(rows, input.serverPid); + const descendants = ProcessDiagnostics.buildDescendantEntries(rows, input.serverPid); const samples: ProcessResourceSample[] = []; if (root) { @@ -245,14 +243,14 @@ export function aggregateProcessResourceHistory(input: { }; } -export const make = Effect.fn("makeProcessResourceMonitor")(function* () { +export const make = Effect.gen(function* () { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const state = yield* Ref.make({ samples: [], lastError: null }); const sampleOnce = Effect.gen(function* () { const sampledAt = yield* DateTime.now; const sampledAtMs = DateTime.toEpochMillis(sampledAt); - const rows = yield* readProcessRows.pipe( + const rows = yield* ProcessDiagnostics.readProcessRows.pipe( Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), ); const samples = collectMonitoredSamples({ @@ -278,7 +276,7 @@ export const make = Effect.fn("makeProcessResourceMonitor")(function* () { Effect.forkScoped, ); - const readHistory: ProcessResourceMonitorShape["readHistory"] = (input) => + const readHistory: ProcessResourceMonitor["Service"]["readHistory"] = (input) => Effect.gen(function* () { const readAt = yield* DateTime.now; const readAtMs = DateTime.toEpochMillis(readAt); @@ -296,4 +294,4 @@ export const make = Effect.fn("makeProcessResourceMonitor")(function* () { return ProcessResourceMonitor.of({ readHistory }); }); -export const layer = Layer.effect(ProcessResourceMonitor, make()); +export const layer = Layer.effect(ProcessResourceMonitor, make); diff --git a/apps/server/src/diagnostics/TraceDiagnostics.ts b/apps/server/src/diagnostics/TraceDiagnostics.ts index ff63410b9bc..d396f4e4ee9 100644 --- a/apps/server/src/diagnostics/TraceDiagnostics.ts +++ b/apps/server/src/diagnostics/TraceDiagnostics.ts @@ -39,13 +39,14 @@ export interface TraceDiagnosticsOptions { readonly readAt?: DateTime.Utc; } -export interface TraceDiagnosticsShape { - readonly read: (options: TraceDiagnosticsOptions) => Effect.Effect; -} - -export class TraceDiagnostics extends Context.Service()( - "t3/diagnostics/TraceDiagnostics", -) {} +export class TraceDiagnostics extends Context.Service< + TraceDiagnostics, + { + readonly read: ( + options: TraceDiagnosticsOptions, + ) => Effect.Effect; + } +>()("t3/diagnostics/TraceDiagnostics") {} interface TraceDiagnosticsInput { readonly traceFilePath: string; @@ -395,10 +396,10 @@ function readTraceFile( ); } -export const make = Effect.fn("makeTraceDiagnostics")(function* () { +export const make = Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; - const read: TraceDiagnosticsShape["read"] = Effect.fn("TraceDiagnostics.read")( + const read: TraceDiagnostics["Service"]["read"] = Effect.fn("TraceDiagnostics.read")( function* (options) { const readAt = options.readAt ?? (yield* DateTime.now); const slowSpanThresholdMs = options.slowSpanThresholdMs ?? DEFAULT_SLOW_SPAN_THRESHOLD_MS; @@ -449,7 +450,7 @@ export const make = Effect.fn("makeTraceDiagnostics")(function* () { return TraceDiagnostics.of({ read }); }); -export const layer = Layer.effect(TraceDiagnostics, make()); +export const layer = Layer.effect(TraceDiagnostics, make); export function readTraceDiagnostics( options: TraceDiagnosticsOptions, diff --git a/apps/server/src/http.ts b/apps/server/src/http.ts index 37baff432fe..032fd501b01 100644 --- a/apps/server/src/http.ts +++ b/apps/server/src/http.ts @@ -24,13 +24,13 @@ import { import * as HttpApiBuilder from "effect/unstable/httpapi/HttpApiBuilder"; import { OtlpTracer } from "effect/unstable/observability"; -import { resolveStaticDir, ServerConfig } from "./config.ts"; +import * as ServerConfig from "./config.ts"; import { ASSET_ROUTE_PREFIX, FALLBACK_PROJECT_FAVICON_SVG, resolveAsset, } from "./assets/AssetAccess.ts"; -import { BrowserTraceCollector } from "./observability/Services/BrowserTraceCollector.ts"; +import * as BrowserTraceCollector from "./observability/BrowserTraceCollector.ts"; import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; import { traceRelayRequest } from "./cloud/traceRelayRequest.ts"; import { @@ -39,7 +39,7 @@ import { failEnvironmentAuthInvalid, failEnvironmentInternal, } from "./auth/http.ts"; -import { ServerEnvironment } from "./environment/Services/ServerEnvironment.ts"; +import * as ServerEnvironment from "./environment/Services/ServerEnvironment.ts"; import { browserApiCorsAllowedHeaders, browserApiCorsAllowedMethods } from "./httpCors.ts"; const OTLP_TRACES_PROXY_PATH = "/api/observability/v1/traces"; @@ -47,7 +47,7 @@ const LOOPBACK_HOSTNAMES = new Set(["127.0.0.1", "::1", "localhost"]); export const browserApiCorsLayer = Layer.unwrap( Effect.gen(function* () { - const config = yield* ServerConfig; + const config = yield* ServerConfig.ServerConfig; const devOrigin = config.devUrl?.origin; return HttpRouter.cors({ ...(devOrigin ? { allowedOrigins: [devOrigin], credentials: true } : {}), @@ -95,7 +95,7 @@ export const serverEnvironmentHttpApiLayer = HttpApiBuilder.group( EnvironmentHttpApi, "metadata", Effect.fnUntraced(function* (handlers) { - const serverEnvironment = yield* ServerEnvironment; + const serverEnvironment = yield* ServerEnvironment.ServerEnvironment; return handlers.handle( "descriptor", Effect.fn("environment.metadata.descriptor")(function* (args) { @@ -117,9 +117,9 @@ export const otlpTracesProxyRouteLayer = HttpRouter.add( Effect.gen(function* () { yield* authenticateRawRouteWithScope(AuthOrchestrationOperateScope); const request = yield* HttpServerRequest.HttpServerRequest; - const config = yield* ServerConfig; + const config = yield* ServerConfig.ServerConfig; const otlpTracesUrl = config.otlpTracesUrl; - const browserTraceCollector = yield* BrowserTraceCollector; + const browserTraceCollector = yield* BrowserTraceCollector.BrowserTraceCollector; const httpClient = yield* HttpClient.HttpClient; const bodyJson = cast(yield* request.json); @@ -223,14 +223,15 @@ export const staticAndDevRouteLayer = HttpRouter.add( return HttpServerResponse.text("Bad Request", { status: 400 }); } - const config = yield* ServerConfig; + const config = yield* ServerConfig.ServerConfig; if (config.devUrl && isLoopbackHostname(url.value.hostname)) { return HttpServerResponse.redirect(resolveDevRedirectUrl(config.devUrl, url.value), { status: 302, }); } - const staticDir = config.staticDir ?? (config.devUrl ? yield* resolveStaticDir() : undefined); + const staticDir = + config.staticDir ?? (config.devUrl ? yield* ServerConfig.resolveStaticDir() : undefined); if (!staticDir) { return HttpServerResponse.text("No static directory configured and no dev URL set.", { status: 503, diff --git a/apps/server/src/observability/BrowserTraceCollector.ts b/apps/server/src/observability/BrowserTraceCollector.ts new file mode 100644 index 00000000000..300a50fe330 --- /dev/null +++ b/apps/server/src/observability/BrowserTraceCollector.ts @@ -0,0 +1,23 @@ +import type { TraceRecord, TraceSink } from "@t3tools/shared/observability"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +export class BrowserTraceCollector extends Context.Service< + BrowserTraceCollector, + { + readonly record: (records: ReadonlyArray) => Effect.Effect; + } +>()("t3/observability/BrowserTraceCollector") {} + +export const make = (sink: TraceSink): BrowserTraceCollector["Service"] => + BrowserTraceCollector.of({ + record: (records) => + Effect.sync(() => { + for (const record of records) { + sink.push(record); + } + }), + }); + +export const layer = (sink: TraceSink) => Layer.succeed(BrowserTraceCollector, make(sink)); diff --git a/apps/server/src/observability/Layers/Observability.ts b/apps/server/src/observability/Layers/Observability.ts index 95263866d80..11463cc1d85 100644 --- a/apps/server/src/observability/Layers/Observability.ts +++ b/apps/server/src/observability/Layers/Observability.ts @@ -4,17 +4,19 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as References from "effect/References"; import * as Tracer from "effect/Tracer"; -import { OtlpMetrics, OtlpSerialization, OtlpTracer } from "effect/unstable/observability"; +import * as OtlpMetrics from "effect/unstable/observability/OtlpMetrics"; +import * as OtlpSerialization from "effect/unstable/observability/OtlpSerialization"; +import * as OtlpTracer from "effect/unstable/observability/OtlpTracer"; -import { ServerConfig } from "../../config.ts"; +import * as ServerConfig from "../../config.ts"; import { ServerLoggerLive } from "../../serverLogger.ts"; -import { BrowserTraceCollector } from "../Services/BrowserTraceCollector.ts"; +import * as BrowserTraceCollector from "../BrowserTraceCollector.ts"; const otlpSerializationLayer = OtlpSerialization.layerJson; export const ObservabilityLive = Layer.unwrap( Effect.gen(function* () { - const config = yield* ServerConfig; + const config = yield* ServerConfig.ServerConfig; const traceReferencesLayer = Layer.mergeAll( Layer.succeed(Tracer.MinimumTraceLevel, config.traceMinLevel), @@ -56,14 +58,7 @@ export const ObservabilityLive = Layer.unwrap( return Layer.mergeAll( Layer.succeed(Tracer.Tracer, tracer), - Layer.succeed(BrowserTraceCollector, { - record: (records) => - Effect.sync(() => { - for (const record of records) { - sink.push(record); - } - }), - }), + BrowserTraceCollector.layer(sink), ); }), ).pipe(Layer.provideMerge(otlpSerializationLayer)); diff --git a/apps/server/src/observability/Services/BrowserTraceCollector.ts b/apps/server/src/observability/Services/BrowserTraceCollector.ts deleted file mode 100644 index b704804c963..00000000000 --- a/apps/server/src/observability/Services/BrowserTraceCollector.ts +++ /dev/null @@ -1,12 +0,0 @@ -import type { TraceRecord } from "@t3tools/shared/observability"; -import * as Context from "effect/Context"; -import type * as Effect from "effect/Effect"; - -export interface BrowserTraceCollectorShape { - readonly record: (records: ReadonlyArray) => Effect.Effect; -} - -export class BrowserTraceCollector extends Context.Service< - BrowserTraceCollector, - BrowserTraceCollectorShape ->()("t3/observability/Services/BrowserTraceCollector") {} diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 8581a11213b..fbb8acfb9e6 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -43,14 +43,11 @@ import { type ProviderAdapterError, } from "../Errors.ts"; import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts"; -import { - ProviderAdapterRegistry, - type ProviderAdapterRegistryShape, -} from "../Services/ProviderAdapterRegistry.ts"; -import { ProviderService } from "../Services/ProviderService.ts"; -import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; +import * as ProviderAdapterRegistry from "../Services/ProviderAdapterRegistry.ts"; +import * as ProviderService from "../Services/ProviderService.ts"; +import * as ProviderSessionDirectory from "../Services/ProviderSessionDirectory.ts"; import { makeProviderServiceLive } from "./ProviderService.ts"; -import { NoOpProviderEventLoggers, ProviderEventLoggers } from "./ProviderEventLoggers.ts"; +import * as ProviderEventLoggers from "./ProviderEventLoggers.ts"; import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts"; import * as NodeServices from "@effect/platform-node/NodeServices"; import * as ProviderSessionRuntime from "../../persistence/ProviderSessionRuntime.ts"; @@ -58,11 +55,11 @@ import { makeSqlitePersistenceLive, SqlitePersistenceMemory, } from "../../persistence/Layers/Sqlite.ts"; -import { ServerSettingsService } from "../../serverSettings.ts"; -import { AnalyticsService } from "../../telemetry/Services/AnalyticsService.ts"; +import * as ServerSettings from "../../serverSettings.ts"; +import * as AnalyticsService from "../../telemetry/AnalyticsService.ts"; import { makeAdapterRegistryMock } from "../testUtils/providerAdapterRegistryMock.ts"; -const defaultServerSettingsLayer = ServerSettingsService.layerTest(); +const defaultServerSettingsLayer = ServerSettings.ServerSettingsService.layerTest(); const asRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.make(value); const asEventId = (value: string): EventId => EventId.make(value); @@ -280,7 +277,10 @@ function makeProviderServiceLayer() { [ProviderDriverKind.make("cursor")]: cursor.adapter, }); - const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry); + const providerAdapterLayer = Layer.succeed( + ProviderAdapterRegistry.ProviderAdapterRegistry, + registry, + ); const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(SqlitePersistenceMemory), ); @@ -293,7 +293,12 @@ function makeProviderServiceLayer() { Layer.provide(directoryLayer), Layer.provide(defaultServerSettingsLayer), Layer.provideMerge(AnalyticsService.layerTest), - Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), ), directoryLayer, @@ -325,7 +330,10 @@ it.effect("ProviderServiceLive catches stopAll failures during shutdown", () => const registry = makeAdapterRegistryMock({ [CODEX_DRIVER]: codex.adapter, }); - const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry); + const providerAdapterLayer = Layer.succeed( + ProviderAdapterRegistry.ProviderAdapterRegistry, + registry, + ); const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(SqlitePersistenceMemory), ); @@ -336,7 +344,12 @@ it.effect("ProviderServiceLive catches stopAll failures during shutdown", () => Layer.provide(directoryLayer), Layer.provide(defaultServerSettingsLayer), Layer.provideMerge(AnalyticsService.layerTest), - Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), ), directoryLayer, runtimeRepositoryLayer, @@ -345,7 +358,7 @@ it.effect("ProviderServiceLive catches stopAll failures during shutdown", () => const scope = yield* Scope.make(); const runtimeServices = yield* Layer.build(providerLayer).pipe(Scope.provide(scope)); - yield* ProviderService.pipe(Effect.provide(runtimeServices)); + yield* ProviderService.ProviderService.pipe(Effect.provide(runtimeServices)); const closeExit = yield* Scope.close(scope, Exit.void).pipe(Effect.exit); assert.equal(Exit.isSuccess(closeExit), true); @@ -361,7 +374,7 @@ it.effect("ProviderServiceLive rejects new sessions for disabled providers", () [CODEX_DRIVER]: codex.adapter, [CLAUDE_AGENT_DRIVER]: claude.adapter, }); - const registry: ProviderAdapterRegistryShape = { + const registry: ProviderAdapterRegistry.ProviderAdapterRegistry["Service"] = { ...registryBase, getInstanceInfo: (instanceId) => instanceId === claudeAgentInstanceId @@ -377,7 +390,10 @@ it.effect("ProviderServiceLive rejects new sessions for disabled providers", () }) : registryBase.getInstanceInfo(instanceId), }; - const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry); + const providerAdapterLayer = Layer.succeed( + ProviderAdapterRegistry.ProviderAdapterRegistry, + registry, + ); const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(SqlitePersistenceMemory), ); @@ -387,12 +403,17 @@ it.effect("ProviderServiceLive rejects new sessions for disabled providers", () Layer.provide(directoryLayer), Layer.provide(defaultServerSettingsLayer), Layer.provide(AnalyticsService.layerTest), - Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), ); const failure = yield* Effect.flip( Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; return yield* provider.startSession(asThreadId("thread-disabled"), { provider: ProviderDriverKind.make("claudeAgent"), providerInstanceId: claudeAgentInstanceId, @@ -419,7 +440,7 @@ it.effect( new ProviderUnsupportedError({ provider: driverKind, }); - const registry: ProviderAdapterRegistryShape = { + const registry: ProviderAdapterRegistry.ProviderAdapterRegistry["Service"] = { getByInstance: (requestedInstanceId) => requestedInstanceId === instanceId ? Effect.succeed(codex.adapter) @@ -444,8 +465,11 @@ it.effect( PubSub.subscribe(pubsub), ), }; - const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry); - const serverSettingsLayer = ServerSettingsService.layerTest({ + const providerAdapterLayer = Layer.succeed( + ProviderAdapterRegistry.ProviderAdapterRegistry, + registry, + ); + const serverSettingsLayer = ServerSettings.ServerSettingsService.layerTest({ providers: { codex: { enabled: false, @@ -463,11 +487,16 @@ it.effect( Layer.provide(directoryLayer), Layer.provide(serverSettingsLayer), Layer.provide(AnalyticsService.layerTest), - Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), ); const session = yield* Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; return yield* provider.startSession(asThreadId("thread-enabled-custom"), { provider: driverKind, providerInstanceId: instanceId, @@ -490,7 +519,7 @@ it.effect("ProviderServiceLive rejects new sessions for disabled custom instance new ProviderUnsupportedError({ provider: ProviderDriverKind.make("codex"), }); - const registry: ProviderAdapterRegistryShape = { + const registry: ProviderAdapterRegistry.ProviderAdapterRegistry["Service"] = { getByInstance: (requestedInstanceId) => requestedInstanceId === instanceId ? Effect.succeed(codex.adapter) @@ -515,7 +544,10 @@ it.effect("ProviderServiceLive rejects new sessions for disabled custom instance PubSub.subscribe(pubsub), ), }; - const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry); + const providerAdapterLayer = Layer.succeed( + ProviderAdapterRegistry.ProviderAdapterRegistry, + registry, + ); const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(SqlitePersistenceMemory), ); @@ -525,12 +557,17 @@ it.effect("ProviderServiceLive rejects new sessions for disabled custom instance Layer.provide(directoryLayer), Layer.provide(defaultServerSettingsLayer), Layer.provide(AnalyticsService.layerTest), - Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), ); const failure = yield* Effect.flip( Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; return yield* provider.startSession(asThreadId("thread-disabled-instance"), { provider: ProviderDriverKind.make("codex"), providerInstanceId: instanceId, @@ -571,15 +608,20 @@ it.effect("ProviderServiceLive writes canonical events to the emitting thread se close: () => Effect.void, }, }).pipe( - Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)), + Layer.provide(Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, registry)), Layer.provide(directoryLayer), Layer.provide(defaultServerSettingsLayer), Layer.provide(AnalyticsService.layerTest), - Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), ); yield* Effect.gen(function* () { - yield* ProviderService; + yield* ProviderService.ProviderService; yield* advanceTestClock(10); codex.emit({ eventId: asEventId("evt-canonical-thread-segment"), @@ -617,7 +659,7 @@ it.effect("ProviderServiceLive keeps persisted resumable sessions on startup", ( const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)); yield* Effect.gen(function* () { - const directory = yield* ProviderSessionDirectory; + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; yield* directory.upsert({ provider: ProviderDriverKind.make("codex"), providerInstanceId: codexInstanceId, @@ -626,17 +668,22 @@ it.effect("ProviderServiceLive keeps persisted resumable sessions on startup", ( }).pipe(Effect.provide(directoryLayer)); const providerLayer = makeProviderServiceLive().pipe( - Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)), + Layer.provide(Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, registry)), Layer.provide(directoryLayer), Layer.provide(defaultServerSettingsLayer), Layer.provide(AnalyticsService.layerTest), - Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), ); - yield* ProviderService.pipe(Effect.provide(providerLayer)); + yield* ProviderService.ProviderService.pipe(Effect.provide(providerLayer)); const persistedProvider = yield* Effect.gen(function* () { - const directory = yield* ProviderSessionDirectory; + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; return yield* directory.getProvider(asThreadId("thread-stale")); }).pipe(Effect.provide(directoryLayer)); assert.equal(persistedProvider, "codex"); @@ -683,11 +730,18 @@ it.effect( Layer.provide(runtimeRepositoryLayer), ); const firstProviderLayer = makeProviderServiceLive().pipe( - Layer.provide(Layer.succeed(ProviderAdapterRegistry, firstRegistry)), + Layer.provide( + Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, firstRegistry), + ), Layer.provide(firstDirectoryLayer), Layer.provide(defaultServerSettingsLayer), Layer.provide(AnalyticsService.layerTest), - Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), ); const updatedResumeCursor = { threadId: asThreadId("thread-1"), @@ -697,7 +751,7 @@ it.effect( }; const startedSession = yield* Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; const threadId = asThreadId("thread-1"); const session = yield* provider.startSession(threadId, { provider: ProviderDriverKind.make("codex"), @@ -735,18 +789,25 @@ it.effect( Layer.provide(runtimeRepositoryLayer), ); const secondProviderLayer = makeProviderServiceLive().pipe( - Layer.provide(Layer.succeed(ProviderAdapterRegistry, secondRegistry)), + Layer.provide( + Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, secondRegistry), + ), Layer.provide(secondDirectoryLayer), Layer.provide(defaultServerSettingsLayer), Layer.provide(AnalyticsService.layerTest), - Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), ); secondCodex.startSession.mockClear(); secondCodex.rollbackThread.mockClear(); yield* Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; yield* provider.rollbackConversation({ threadId: startedSession.threadId, numTurns: 1, @@ -780,7 +841,7 @@ it.effect( routing.layer("ProviderServiceLive routing", (it) => { it.effect("routes provider operations and rollback conversation", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; const session = yield* provider.startSession(asThreadId("thread-1"), { provider: ProviderDriverKind.make("codex"), @@ -866,7 +927,7 @@ routing.layer("ProviderServiceLive routing", (it) => { it.effect("recovers stale persisted sessions for rollback by resuming thread identity", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; const initial = yield* provider.startSession(asThreadId("thread-1"), { provider: ProviderDriverKind.make("codex"), @@ -907,7 +968,7 @@ routing.layer("ProviderServiceLive routing", (it) => { it.effect("preserves the persisted binding when stopping a session", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; const initial = yield* provider.startSession(asThreadId("thread-reap-preserve"), { @@ -959,7 +1020,7 @@ routing.layer("ProviderServiceLive routing", (it) => { it.effect("routes explicit claudeAgent provider session starts to the claude adapter", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; const session = yield* provider.startSession(asThreadId("thread-claude"), { provider: ProviderDriverKind.make("claudeAgent"), @@ -988,8 +1049,8 @@ routing.layer("ProviderServiceLive routing", (it) => { it.effect("dies when an active session conflicts with its persisted binding", () => Effect.gen(function* () { - const provider = yield* ProviderService; - const directory = yield* ProviderSessionDirectory; + const provider = yield* ProviderService.ProviderService; + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; const threadId = asThreadId("thread-binding-mismatch"); yield* provider.startSession(threadId, { @@ -1019,7 +1080,7 @@ routing.layer("ProviderServiceLive routing", (it) => { it.effect("stops stale sessions in other providers after a successful replacement start", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; const threadId = asThreadId("thread-provider-replacement"); const codexSession = yield* provider.startSession(threadId, { @@ -1058,7 +1119,7 @@ routing.layer("ProviderServiceLive routing", (it) => { it.effect("recovers stale sessions for sendTurn using persisted cwd", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; const initial = yield* provider.startSession(asThreadId("thread-1"), { provider: ProviderDriverKind.make("codex"), @@ -1099,7 +1160,7 @@ routing.layer("ProviderServiceLive routing", (it) => { it.effect("recovers stale claudeAgent sessions for sendTurn using persisted cwd", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; const initial = yield* provider.startSession(asThreadId("thread-claude-send-turn"), { provider: ProviderDriverKind.make("claudeAgent"), @@ -1152,7 +1213,7 @@ routing.layer("ProviderServiceLive routing", (it) => { it.effect("lists no sessions after adapter runtime clears", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; yield* provider.startSession(asThreadId("thread-1"), { provider: ProviderDriverKind.make("codex"), @@ -1177,7 +1238,7 @@ routing.layer("ProviderServiceLive routing", (it) => { it.effect("persists runtime status transitions in provider_session_runtime", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; const threadId = asThreadId("thread-runtime-status"); @@ -1237,15 +1298,22 @@ routing.layer("ProviderServiceLive routing", (it) => { Layer.provide(runtimeRepositoryLayer), ); const firstProviderLayer = makeProviderServiceLive().pipe( - Layer.provide(Layer.succeed(ProviderAdapterRegistry, firstRegistry)), + Layer.provide( + Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, firstRegistry), + ), Layer.provide(firstDirectoryLayer), Layer.provide(defaultServerSettingsLayer), Layer.provide(AnalyticsService.layerTest), - Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), ); const initial = yield* Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; return yield* provider.startSession(asThreadId("thread-claude-start"), { provider: ProviderDriverKind.make("claudeAgent"), providerInstanceId: claudeAgentInstanceId, @@ -1256,7 +1324,7 @@ routing.layer("ProviderServiceLive routing", (it) => { }).pipe(Effect.provide(firstProviderLayer)); yield* Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; yield* provider.listSessions(); }).pipe(Effect.provide(firstProviderLayer)); @@ -1268,17 +1336,24 @@ routing.layer("ProviderServiceLive routing", (it) => { Layer.provide(runtimeRepositoryLayer), ); const secondProviderLayer = makeProviderServiceLive().pipe( - Layer.provide(Layer.succeed(ProviderAdapterRegistry, secondRegistry)), + Layer.provide( + Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, secondRegistry), + ), Layer.provide(secondDirectoryLayer), Layer.provide(defaultServerSettingsLayer), Layer.provide(AnalyticsService.layerTest), - Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), ); secondClaude.startSession.mockClear(); yield* Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; yield* provider.startSession(initial.threadId, { provider: ProviderDriverKind.make("claudeAgent"), providerInstanceId: claudeAgentInstanceId, @@ -1327,15 +1402,22 @@ routing.layer("ProviderServiceLive routing", (it) => { Layer.provide(runtimeRepositoryLayer), ); const firstProviderLayer = makeProviderServiceLive().pipe( - Layer.provide(Layer.succeed(ProviderAdapterRegistry, firstRegistry)), + Layer.provide( + Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, firstRegistry), + ), Layer.provide(firstDirectoryLayer), Layer.provide(defaultServerSettingsLayer), Layer.provide(AnalyticsService.layerTest), - Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), ); const initial = yield* Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; return yield* provider.startSession(asThreadId("thread-claude-cwd"), { provider: ProviderDriverKind.make("claudeAgent"), providerInstanceId: claudeAgentInstanceId, @@ -1353,17 +1435,24 @@ routing.layer("ProviderServiceLive routing", (it) => { Layer.provide(runtimeRepositoryLayer), ); const secondProviderLayer = makeProviderServiceLive().pipe( - Layer.provide(Layer.succeed(ProviderAdapterRegistry, secondRegistry)), + Layer.provide( + Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, secondRegistry), + ), Layer.provide(secondDirectoryLayer), Layer.provide(defaultServerSettingsLayer), Layer.provide(AnalyticsService.layerTest), - Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), ); secondClaude.startSession.mockClear(); yield* Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; yield* provider.startSession(initial.threadId, { provider: ProviderDriverKind.make("claudeAgent"), providerInstanceId: claudeAgentInstanceId, @@ -1397,7 +1486,7 @@ const fanout = makeProviderServiceLayer(); fanout.layer("ProviderServiceLive fanout", (it) => { it.effect("fans out adapter turn completion events", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; const session = yield* provider.startSession(asThreadId("thread-1"), { provider: ProviderDriverKind.make("codex"), providerInstanceId: codexInstanceId, @@ -1443,7 +1532,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { it.effect("fans out canonical runtime events in emission order", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; const session = yield* provider.startSession(asThreadId("thread-seq"), { provider: ProviderDriverKind.make("codex"), providerInstanceId: codexInstanceId, @@ -1499,7 +1588,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { it.effect("keeps subscriber delivery ordered and isolates failing subscribers", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; const session = yield* provider.startSession(asThreadId("thread-1"), { provider: ProviderDriverKind.make("codex"), providerInstanceId: codexInstanceId, @@ -1571,7 +1660,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { it.effect("records provider metrics with the routed provider label", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; const session = yield* provider.startSession(asThreadId("thread-metrics"), { provider: ProviderDriverKind.make("claudeAgent"), @@ -1649,7 +1738,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { "records sendTurn metrics with the resolved provider when modelSelection is omitted", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; const session = yield* provider.startSession(asThreadId("thread-send-metrics"), { provider: ProviderDriverKind.make("claudeAgent"), @@ -1690,7 +1779,7 @@ const validation = makeProviderServiceLayer(); validation.layer("ProviderServiceLive validation", (it) => { it.effect("rejects session starts without an explicit provider instance id", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; validation.codex.startSession.mockClear(); const failure = yield* Effect.flip( @@ -1709,7 +1798,7 @@ validation.layer("ProviderServiceLive validation", (it) => { it.effect("rejects mismatched provider kind and provider instance id", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; validation.codex.startSession.mockClear(); validation.claude.startSession.mockClear(); @@ -1734,7 +1823,7 @@ validation.layer("ProviderServiceLive validation", (it) => { it.effect("returns ProviderValidationError for invalid input payloads", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; const failure = yield* Effect.result( provider.startSession(asThreadId("thread-validation"), { @@ -1759,7 +1848,7 @@ validation.layer("ProviderServiceLive validation", (it) => { it.effect("accepts startSession when adapter has not emitted provider thread id yet", () => Effect.gen(function* () { - const provider = yield* ProviderService; + const provider = yield* ProviderService.ProviderService; const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; validation.codex.startSession.mockImplementationOnce((input: ProviderSessionStartInput) => diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index ecb1dd2dbd3..c15d50eed62 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -47,15 +47,12 @@ import { } from "../../observability/Metrics.ts"; import { type ProviderAdapterError, ProviderValidationError } from "../Errors.ts"; import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts"; -import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts"; -import { ProviderService, type ProviderServiceShape } from "../Services/ProviderService.ts"; -import { - ProviderSessionDirectory, - type ProviderRuntimeBinding, -} from "../Services/ProviderSessionDirectory.ts"; +import * as ProviderAdapterRegistry from "../Services/ProviderAdapterRegistry.ts"; +import * as ProviderService from "../Services/ProviderService.ts"; +import * as ProviderSessionDirectory from "../Services/ProviderSessionDirectory.ts"; import { type EventNdjsonLogger } from "./EventNdjsonLogger.ts"; -import { ProviderEventLoggers } from "./ProviderEventLoggers.ts"; -import { AnalyticsService } from "../../telemetry/Services/AnalyticsService.ts"; +import * as ProviderEventLoggers from "./ProviderEventLoggers.ts"; +import * as AnalyticsService from "../../telemetry/AnalyticsService.ts"; import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; import * as McpSessionRegistry from "../../mcp/McpSessionRegistry.ts"; const isModelSelection = Schema.is(ModelSelection); @@ -69,6 +66,9 @@ export interface ProviderServiceLiveOptions { readonly canonicalEventLogger?: EventNdjsonLogger; } +type ProviderServiceMethod = + ProviderService.ProviderService["Service"][Name]; + const ProviderRollbackConversationInput = Schema.Struct({ threadId: ThreadId, numTurns: NonNegativeInt, @@ -141,7 +141,7 @@ function toRuntimePayloadFromSession( } function readPersistedModelSelection( - runtimePayload: ProviderRuntimeBinding["runtimePayload"], + runtimePayload: ProviderSessionDirectory.ProviderRuntimeBinding["runtimePayload"], ): ModelSelection | undefined { if (!runtimePayload || typeof runtimePayload !== "object" || Array.isArray(runtimePayload)) { return undefined; @@ -151,7 +151,7 @@ function readPersistedModelSelection( } function readPersistedCwd( - runtimePayload: ProviderRuntimeBinding["runtimePayload"], + runtimePayload: ProviderSessionDirectory.ProviderRuntimeBinding["runtimePayload"], ): string | undefined { if (!runtimePayload || typeof runtimePayload !== "object" || Array.isArray(runtimePayload)) { return undefined; @@ -202,16 +202,16 @@ const correlateRuntimeEventWithInstance = ( const makeProviderService = Effect.fn("makeProviderService")(function* ( options?: ProviderServiceLiveOptions, ) { - const analytics = yield* Effect.service(AnalyticsService); - const eventLoggers = yield* ProviderEventLoggers; + const analytics = yield* Effect.service(AnalyticsService.AnalyticsService); + const eventLoggers = yield* ProviderEventLoggers.ProviderEventLoggers; // Options-provided logger wins (test overrides); otherwise we take whatever // the `ProviderEventLoggers` tag exposes — `undefined` means "no canonical // log writer is attached", which downstream code already handles as a // no-op. const canonicalEventLogger = options?.canonicalEventLogger ?? eventLoggers.canonical; - const registry = yield* ProviderAdapterRegistry; - const directory = yield* ProviderSessionDirectory; + const registry = yield* ProviderAdapterRegistry.ProviderAdapterRegistry; + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; const runtimeEventPubSub = yield* PubSub.unbounded(); const nowIso = Effect.map(DateTime.now, DateTime.formatIso); const prepareMcpSession = (threadId: ThreadId, providerInstanceId: ProviderInstanceId) => @@ -353,7 +353,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ).pipe(Effect.forkScoped); const recoverSessionForThread = Effect.fn("recoverSessionForThread")(function* (input: { - readonly binding: ProviderRuntimeBinding; + readonly binding: ProviderSessionDirectory.ProviderRuntimeBinding; readonly operation: string; }) { const bindingInstanceId = yield* requireBindingInstanceId(input.operation, input.binding); @@ -519,7 +519,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ); }); - const startSession: ProviderServiceShape["startSession"] = Effect.fn("startSession")( + const startSession: ProviderServiceMethod<"startSession"> = Effect.fn("startSession")( function* (threadId, rawInput) { const parsed = yield* decodeInputOrValidationError({ operation: "ProviderService.startSession", @@ -642,7 +642,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( }, ); - const sendTurn: ProviderServiceShape["sendTurn"] = Effect.fn("sendTurn")(function* (rawInput) { + const sendTurn: ProviderServiceMethod<"sendTurn"> = Effect.fn("sendTurn")(function* (rawInput) { const parsed = yield* decodeInputOrValidationError({ operation: "ProviderService.sendTurn", schema: ProviderSendTurnInput, @@ -717,7 +717,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ); }); - const interruptTurn: ProviderServiceShape["interruptTurn"] = Effect.fn("interruptTurn")( + const interruptTurn: ProviderServiceMethod<"interruptTurn"> = Effect.fn("interruptTurn")( function* (rawInput) { const input = yield* decodeInputOrValidationError({ operation: "ProviderService.interruptTurn", @@ -754,7 +754,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( }, ); - const respondToRequest: ProviderServiceShape["respondToRequest"] = Effect.fn("respondToRequest")( + const respondToRequest: ProviderServiceMethod<"respondToRequest"> = Effect.fn("respondToRequest")( function* (rawInput) { const input = yield* decodeInputOrValidationError({ operation: "ProviderService.respondToRequest", @@ -792,7 +792,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( }, ); - const respondToUserInput: ProviderServiceShape["respondToUserInput"] = Effect.fn( + const respondToUserInput: ProviderServiceMethod<"respondToUserInput"> = Effect.fn( "respondToUserInput", )(function* (rawInput) { const input = yield* decodeInputOrValidationError({ @@ -826,7 +826,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ); }); - const stopSession: ProviderServiceShape["stopSession"] = Effect.fn("stopSession")( + const stopSession: ProviderServiceMethod<"stopSession"> = Effect.fn("stopSession")( function* (rawInput) { const input = yield* decodeInputOrValidationError({ operation: "ProviderService.stopSession", @@ -874,7 +874,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( }, ); - const listSessions: ProviderServiceShape["listSessions"] = Effect.fn("listSessions")( + const listSessions: ProviderServiceMethod<"listSessions"> = Effect.fn("listSessions")( function* () { const currentAdapters = yield* getAdapterEntries; const sessionsByProvider = yield* Effect.forEach(currentAdapters, ([instanceId, adapter]) => @@ -895,13 +895,22 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( (threadId) => directory .getBinding(threadId) - .pipe(Effect.orElseSucceed(() => Option.none())), + .pipe( + Effect.orElseSucceed(() => + Option.none(), + ), + ), { concurrency: "unbounded" }, ), ), - Effect.orElseSucceed(() => [] as Array>), + Effect.orElseSucceed( + () => [] as Array>, + ), ); - const bindingsByThreadId = new Map(); + const bindingsByThreadId = new Map< + ThreadId, + ProviderSessionDirectory.ProviderRuntimeBinding + >(); for (const bindingOption of persistedBindings) { const binding = Option.getOrUndefined(bindingOption); if (binding) { @@ -952,13 +961,13 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( }, ); - const getCapabilities: ProviderServiceShape["getCapabilities"] = (instanceId) => + const getCapabilities: ProviderServiceMethod<"getCapabilities"> = (instanceId) => registry.getByInstance(instanceId).pipe(Effect.map((adapter) => adapter.capabilities)); - const getInstanceInfo: ProviderServiceShape["getInstanceInfo"] = (instanceId) => + const getInstanceInfo: ProviderServiceMethod<"getInstanceInfo"> = (instanceId) => registry.getInstanceInfo(instanceId); - const rollbackConversation: ProviderServiceShape["rollbackConversation"] = Effect.fn( + const rollbackConversation: ProviderServiceMethod<"rollbackConversation"> = Effect.fn( "rollbackConversation", )(function* (rawInput) { const input = yield* decodeInputOrValidationError({ @@ -1071,14 +1080,17 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( // Each access creates a fresh PubSub subscription so that multiple // consumers (ProviderRuntimeIngestion, CheckpointReactor, etc.) each // independently receive all runtime events. - get streamEvents(): ProviderServiceShape["streamEvents"] { + get streamEvents(): ProviderServiceMethod<"streamEvents"> { return Stream.fromPubSub(runtimeEventPubSub); }, - } satisfies ProviderServiceShape; + } satisfies ProviderService.ProviderService["Service"]; }); -export const ProviderServiceLive = Layer.effect(ProviderService, makeProviderService()); +export const ProviderServiceLive = Layer.effect( + ProviderService.ProviderService, + makeProviderService(), +); export function makeProviderServiceLive(options?: ProviderServiceLiveOptions) { - return Layer.effect(ProviderService, makeProviderService(options)); + return Layer.effect(ProviderService.ProviderService, makeProviderService(options)); } diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index bf4a77743a2..4b92b34cc62 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -88,7 +88,7 @@ import * as ServerSettings from "./serverSettings.ts"; import * as TerminalManager from "./terminal/Services/Manager.ts"; import * as PreviewManager from "./preview/Manager.ts"; import * as PortScanner from "./preview/PortScanner.ts"; -import * as BrowserTraceCollector from "./observability/Services/BrowserTraceCollector.ts"; +import * as BrowserTraceCollector from "./observability/BrowserTraceCollector.ts"; import { ProjectFaviconResolverLive } from "./project/Layers/ProjectFaviconResolver.ts"; import * as ProjectSetupScriptRunner from "./project/Services/ProjectSetupScriptRunner.ts"; import * as RepositoryIdentityResolver from "./project/Services/RepositoryIdentityResolver.ts"; diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index f1e900c0b5a..f3cbe764264 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -17,7 +17,7 @@ import { websocketRpcRouteLayer } from "./ws.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; import { layerConfig as SqlitePersistenceLayerLive } from "./persistence/Layers/Sqlite.ts"; import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; -import { AnalyticsServiceLayerLive } from "./telemetry/Layers/AnalyticsService.ts"; +import * as AnalyticsService from "./telemetry/AnalyticsService.ts"; import { ProviderSessionDirectoryLive } from "./provider/Layers/ProviderSessionDirectory.ts"; import * as ProviderSessionRuntime from "./persistence/ProviderSessionRuntime.ts"; import { ProviderAdapterRegistryLive } from "./provider/Layers/ProviderAdapterRegistry.ts"; @@ -332,7 +332,7 @@ const RuntimeDependenciesLive = RuntimeCoreDependenciesLive.pipe( Layer.provideMerge(ProcessDiagnostics.layer), Layer.provideMerge(ProcessResourceMonitor.layer), Layer.provideMerge(TraceDiagnostics.layer), - Layer.provideMerge(AnalyticsServiceLayerLive), + Layer.provideMerge(AnalyticsService.layer), Layer.provideMerge(ExternalLauncher.layer), Layer.provideMerge(ServerLifecycleEvents.layer), Layer.provide(NetService.layer), diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts index a11beba794d..2109f4c5458 100644 --- a/apps/server/src/serverRuntimeStartup.test.ts +++ b/apps/server/src/serverRuntimeStartup.test.ts @@ -13,7 +13,7 @@ import * as Stream from "effect/Stream"; import * as ServerConfig from "./config.ts"; import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts"; -import * as AnalyticsService from "./telemetry/Services/AnalyticsService.ts"; +import * as AnalyticsService from "./telemetry/AnalyticsService.ts"; import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts"; it("uses the canonical Codex default for auto-bootstrapped model selection", () => { diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index dab6143e11c..35ac5a06fc9 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -31,7 +31,7 @@ import * as OrchestrationReactor from "./orchestration/Services/OrchestrationRea import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; import * as ServerSettings from "./serverSettings.ts"; import * as ServerEnvironment from "./environment/Services/ServerEnvironment.ts"; -import * as AnalyticsService from "./telemetry/Services/AnalyticsService.ts"; +import * as AnalyticsService from "./telemetry/AnalyticsService.ts"; import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; import * as ProviderSessionReaper from "./provider/Services/ProviderSessionReaper.ts"; import { diff --git a/apps/server/src/telemetry/Layers/AnalyticsService.test.ts b/apps/server/src/telemetry/AnalyticsService.test.ts similarity index 89% rename from apps/server/src/telemetry/Layers/AnalyticsService.test.ts rename to apps/server/src/telemetry/AnalyticsService.test.ts index 5aa47406d9b..d69bab32feb 100644 --- a/apps/server/src/telemetry/Layers/AnalyticsService.test.ts +++ b/apps/server/src/telemetry/AnalyticsService.test.ts @@ -8,10 +8,9 @@ import * as HttpServer from "effect/unstable/http/HttpServer"; import * as HttpServerRequest from "effect/unstable/http/HttpServerRequest"; import * as HttpServerResponse from "effect/unstable/http/HttpServerResponse"; -import { ServerConfig } from "../../config.ts"; -import { getTelemetryIdentifier } from "../Identify.ts"; -import { AnalyticsService } from "../Services/AnalyticsService.ts"; -import { AnalyticsServiceLayerLive } from "./AnalyticsService.ts"; +import * as ServerConfig from "../config.ts"; +import { getTelemetryIdentifier } from "./Identify.ts"; +import * as AnalyticsService from "./AnalyticsService.ts"; interface RecordedBatchRequest { readonly path: string; @@ -40,11 +39,11 @@ it.layer(NodeServices.layer)("AnalyticsService test", (it) => { it.effect("flush drains all buffered events across multiple batches", () => Effect.gen(function* () { const capturedRequests: Array = []; - const serverConfigLayer = ServerConfig.layerTest(process.cwd(), { + const serverConfigLayer = ServerConfig.ServerConfig.layerTest(process.cwd(), { prefix: "t3-telemetry-base-", }); - const telemetryLayer = AnalyticsServiceLayerLive.pipe(Layer.provideMerge(serverConfigLayer)); + const telemetryLayer = AnalyticsService.layer.pipe(Layer.provideMerge(serverConfigLayer)); const configLayer = ConfigProvider.layer( ConfigProvider.fromUnknown({ T3CODE_TELEMETRY_ENABLED: true, @@ -79,7 +78,7 @@ it.layer(NodeServices.layer)("AnalyticsService test", (it) => { yield* Layer.launch(batchServerLayer).pipe(Effect.forkScoped); const telemetryIdentifier = yield* getTelemetryIdentifier; assert.equal(telemetryIdentifier !== null, true); - const analytics = yield* AnalyticsService; + const analytics = yield* AnalyticsService.AnalyticsService; for (let index = 0; index < 45; index += 1) { yield* analytics.record("test.flush.drain", { index }); diff --git a/apps/server/src/telemetry/Layers/AnalyticsService.ts b/apps/server/src/telemetry/AnalyticsService.ts similarity index 72% rename from apps/server/src/telemetry/Layers/AnalyticsService.ts rename to apps/server/src/telemetry/AnalyticsService.ts index 0d51d7c66b1..5fdc7bdeb19 100644 --- a/apps/server/src/telemetry/Layers/AnalyticsService.ts +++ b/apps/server/src/telemetry/AnalyticsService.ts @@ -1,25 +1,26 @@ /** - * AnalyticsServiceLive - Anonymous PostHog telemetry layer. + * Anonymous PostHog telemetry service. * - * Persists a random installation-scoped anonymous id to state dir, buffers - * events in memory, and flushes batches to PostHog over Effect HttpClient. + * Persists an installation-scoped anonymous identifier, buffers events in + * memory, and flushes batches over Effect's HTTP client. * - * @module AnalyticsServiceLive + * @module AnalyticsService */ - import { HostProcessArchitecture, HostProcessPlatform } from "@t3tools/shared/hostProcess"; import * as Config from "effect/Config"; +import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; -import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"; +import * as HttpClient from "effect/unstable/http/HttpClient"; +import * as HttpClientRequest from "effect/unstable/http/HttpClientRequest"; +import * as HttpClientResponse from "effect/unstable/http/HttpClientResponse"; -import { ServerConfig } from "../../config.ts"; -import { AnalyticsService, type AnalyticsServiceShape } from "../Services/AnalyticsService.ts"; -import { getTelemetryIdentifier } from "../Identify.ts"; -import packageJson from "../../../package.json" with { type: "json" }; +import packageJson from "../../package.json" with { type: "json" }; +import * as ServerConfig from "../config.ts"; +import { getTelemetryIdentifier } from "./Identify.ts"; interface BufferedAnalyticsEvent { readonly event: string; @@ -42,10 +43,33 @@ const TelemetryEnvConfig = Config.all({ wslDistroName: Config.string("WSL_DISTRO_NAME").pipe(Config.option), }); -const makeAnalyticsService = Effect.gen(function* () { +export class AnalyticsService extends Context.Service< + AnalyticsService, + { + /** Record an anonymous event for best-effort buffered delivery. */ + readonly record: ( + event: string, + properties?: Readonly>, + ) => Effect.Effect; + + /** Flush all currently queued telemetry events. */ + readonly flush: Effect.Effect; + } +>()("t3/telemetry/AnalyticsService") { + /** No-op layer for callers that intentionally disable telemetry. */ + static readonly layerTest = Layer.succeed( + AnalyticsService, + AnalyticsService.of({ + record: () => Effect.void, + flush: Effect.void, + }), + ); +} + +export const make = Effect.gen(function* () { const telemetryConfig = yield* TelemetryEnvConfig; const httpClient = yield* HttpClient.HttpClient; - const serverConfig = yield* ServerConfig; + const serverConfig = yield* ServerConfig.ServerConfig; const identifier = yield* getTelemetryIdentifier; const bufferRef = yield* Ref.make>([]); const clientType = serverConfig.mode === "desktop" ? "desktop-app" : "cli-web-client"; @@ -79,7 +103,7 @@ const makeAnalyticsService = Effect.gen(function* () { }), ); - const sendBatch = Effect.fn("sendBatch")(function* ( + const sendBatch = Effect.fn("AnalyticsService.sendBatch")(function* ( events: ReadonlyArray, ) { if (!telemetryConfig.enabled || !identifier) return; @@ -109,7 +133,7 @@ const makeAnalyticsService = Effect.gen(function* () { ); }); - const flush: AnalyticsServiceShape["flush"] = Effect.gen(function* () { + const flush: AnalyticsService["Service"]["flush"] = Effect.gen(function* () { while (true) { const batch = yield* Ref.modify(bufferRef, (current) => { if (current.length === 0) { @@ -134,7 +158,7 @@ const makeAnalyticsService = Effect.gen(function* () { } }).pipe(Effect.catch((cause) => Effect.logError("Failed to flush telemetry", { cause }))); - const record: AnalyticsServiceShape["record"] = Effect.fn("record")( + const record: AnalyticsService["Service"]["record"] = Effect.fn("AnalyticsService.record")( function* (event, properties) { if (!telemetryConfig.enabled || !identifier) return; @@ -154,10 +178,9 @@ const makeAnalyticsService = Effect.gen(function* () { yield* Effect.addFinalizer(() => flush); - return { - record, - flush, - } satisfies AnalyticsServiceShape; + return AnalyticsService.of({ record, flush }); }); -export const AnalyticsServiceLayerLive = Layer.effect(AnalyticsService, makeAnalyticsService); +export const layer = Layer.effect(AnalyticsService, make); + +export const layerTest = AnalyticsService.layerTest; diff --git a/apps/server/src/telemetry/Identify.ts b/apps/server/src/telemetry/Identify.ts index 364273a9e1d..f7458bcd8c8 100644 --- a/apps/server/src/telemetry/Identify.ts +++ b/apps/server/src/telemetry/Identify.ts @@ -6,7 +6,7 @@ import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; -import { ServerConfig } from "../config.ts"; +import * as ServerConfig from "../config.ts"; const CodexAuthJsonSchema = Schema.Struct({ tokens: Schema.Struct({ @@ -19,9 +19,14 @@ const ClaudeJsonSchema = Schema.Struct({ }); class IdentifyUserError extends Schema.TaggedErrorClass()("IdentifyUserError", { - message: Schema.String, - cause: Schema.optional(Schema.Defect()), -}) {} + operation: Schema.Literal("hash_identifier"), + algorithm: Schema.Literal("SHA-256"), + cause: Schema.Defect(), +}) { + override get message(): string { + return `Failed to hash telemetry identifier with ${this.algorithm}.`; + } +} const hash = (value: string) => Crypto.Crypto.pipe( @@ -30,7 +35,8 @@ const hash = (value: string) => Effect.mapError( (cause) => new IdentifyUserError({ - message: "Failed to hash identifier", + operation: "hash_identifier", + algorithm: "SHA-256", cause, }), ), @@ -64,7 +70,7 @@ const getClaudeUserId = Effect.gen(function* () { const upsertAnonymousId = Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; - const { anonymousIdPath } = yield* ServerConfig; + const { anonymousIdPath } = yield* ServerConfig.ServerConfig; const anonymousId = yield* fileSystem.readFileString(anonymousIdPath).pipe( Effect.catch(() => diff --git a/apps/server/src/telemetry/Services/AnalyticsService.ts b/apps/server/src/telemetry/Services/AnalyticsService.ts index a2717c790dc..879a1de7cdb 100644 --- a/apps/server/src/telemetry/Services/AnalyticsService.ts +++ b/apps/server/src/telemetry/Services/AnalyticsService.ts @@ -1,35 +1,2 @@ -/** - * AnalyticsService - Anonymous telemetry capture contract. - * - * Provides a best-effort event API for runtime telemetry and a strict - * `captureImmediate` method for call sites that need explicit error handling. - * - * @module AnalyticsService - */ -import * as Effect from "effect/Effect"; -import * as Layer from "effect/Layer"; -import * as Context from "effect/Context"; - -export interface AnalyticsServiceShape { - /** - * Capture an event immediately; returns typed failure when capture fails. - */ - readonly record: ( - event: string, - properties?: Readonly>, - ) => Effect.Effect; - - /** - * Flush queued telemetry. - */ - readonly flush: Effect.Effect; -} - -export class AnalyticsService extends Context.Service()( - "t3/telemetry/Services/AnalyticsService", -) { - static readonly layerTest = Layer.succeed(AnalyticsService, { - record: () => Effect.void, - flush: Effect.void, - }); -} +// Compatibility shim for the intentionally excluded orchestration harness. +export { AnalyticsService } from "../AnalyticsService.ts";