Commit ccf8331600f323939a796169dfe2598a4c1b59c4 | the-knife

author	Julius Marminge <julius0216@outlook.com> 2026-06-19 21:29:25-07:00
committer	GitHub <noreply@github.com> 2026-06-19 21:29:25-07:00
log	ccf8331600f323939a796169dfe2598a4c1b59c4
tree	2c6d22cad818c99435435ec9f584e2daa700ac77
parent	ffae12f329c4de9680efd1757b488a3b5bbe3cd2
signature	Signed by PGP key B5690EEEBB952194

[codex] Align diagnostics and telemetry Effect services (#3189)

Co-authored-by: codex <codex@users.noreply.github.com>

20 files changed, 718 insertions(+), 556 deletions(-)

apps
- server
  - src
    diagnostics
    ProcessDiagnostics.ts
    ProcessResourceMonitor.test.ts
    ProcessResourceMonitor.ts
    TraceDiagnostics.ts
    http.ts
    observability
    BrowserTraceCollector.ts
    Layers
    Observability.ts
    Services
    BrowserTraceCollector.ts
    provider
    Layers
    ProviderService.test.ts
    ProviderService.ts
    server.test.ts
    server.ts
    serverRuntimeStartup.test.ts
    serverRuntimeStartup.ts
    telemetry
    AnalyticsService.test.ts
    AnalyticsService.ts
    Identify.ts
    Layers
    AnalyticsService.test.ts
    AnalyticsService.ts
    Services
    AnalyticsService.ts

apps/server/src/diagnostics/ProcessDiagnostics.ts+103-40

...	...	@@ -12,7 +12,8 @@ import * as Effect from "effect/Effect";
12	12	import * as Layer from "effect/Layer";
13	13	import * as Option from "effect/Option";
14	14	import * as Schema from "effect/Schema";
15		import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
	15	import * as ChildProcess from "effect/unstable/process/ChildProcess";
	16	import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner";
16	17
17	18	import { collectUint8StreamText } from "../stream/collectUint8StreamText.ts";
18	19
...	...	@@ -31,35 +32,80 @@ const PROCESS_QUERY_TIMEOUT_MS = 1_000;
31	32	const POSIX_PROCESS_QUERY_COMMAND = "pid=,ppid=,pgid=,stat=,pcpu=,rss=,etime=,command=";
32	33	const PROCESS_QUERY_MAX_OUTPUT_BYTES = 2 * 1024 * 1024;
33	34
34		export interface ProcessDiagnosticsShape {
35		readonly read: Effect.Effect<ServerProcessDiagnosticsResult>;
36		readonly signal: (input: {
37		readonly pid: number;
38		readonly signal: ServerProcessSignal;
39		}) => Effect.Effect<ServerSignalProcessResult>;
40		}
41
42	35	export class ProcessDiagnostics extends Context.Service<
43	36	ProcessDiagnostics,
44		ProcessDiagnosticsShape
	37	{
	38	readonly read: Effect.Effect<ServerProcessDiagnosticsResult>;
	39	readonly signal: (input: {
	40	readonly pid: number;
	41	readonly signal: ServerProcessSignal;
	42	}) => Effect.Effect<ServerSignalProcessResult>;
	43	}
45	44	>()("t3/diagnostics/ProcessDiagnostics") {}
46	45
47		class ProcessDiagnosticsError extends Schema.TaggedErrorClass<ProcessDiagnosticsError>()(
48		"ProcessDiagnosticsError",
	46	class ProcessDiagnosticsQueryTimeoutError extends Schema.TaggedErrorClass<ProcessDiagnosticsQueryTimeoutError>()(
	47	"ProcessDiagnosticsQueryTimeoutError",
	48	{ command: Schema.String },
	49	) {
	50	override get message(): string {
	51	return `Process diagnostics query '${this.command}' timed out.`;
	52	}
	53	}
	54
	55	class ProcessDiagnosticsQueryFailedError extends Schema.TaggedErrorClass<ProcessDiagnosticsQueryFailedError>()(
	56	"ProcessDiagnosticsQueryFailedError",
49	57	{
50		message: Schema.String,
	58	command: Schema.String,
	59	stderr: Schema.optional(Schema.String),
51	60	cause: Schema.optional(Schema.Defect()),
52	61	},
53		) {}
54		const isProcessDiagnosticsError = Schema.is(ProcessDiagnosticsError);
	62	) {
	63	override get message(): string {
	64	return this.stderr?.trim() \|\| `Failed to query process diagnostics with '${this.command}'.`;
	65	}
	66	}
55	67
56		function toProcessDiagnosticsError(message: string, cause?: unknown): ProcessDiagnosticsError {
57		return new ProcessDiagnosticsError({
58		message,
59		...(cause === undefined ? {} : { cause }),
60		});
	68	class ProcessDiagnosticsServerProcessSignalError extends Schema.TaggedErrorClass<ProcessDiagnosticsServerProcessSignalError>()(
	69	"ProcessDiagnosticsServerProcessSignalError",
	70	{ pid: Schema.Number },
	71	) {
	72	override get message(): string {
	73	return "Refusing to signal the T3 server process.";
	74	}
	75	}
	76
	77	class ProcessDiagnosticsNotDescendantError extends Schema.TaggedErrorClass<ProcessDiagnosticsNotDescendantError>()(
	78	"ProcessDiagnosticsNotDescendantError",
	79	{ pid: Schema.Number },
	80	) {
	81	override get message(): string {
	82	return `Process ${this.pid} is not a live descendant of the T3 server.`;
	83	}
61	84	}
62	85
	86	class ProcessDiagnosticsSignalFailedError extends Schema.TaggedErrorClass<ProcessDiagnosticsSignalFailedError>()(
	87	"ProcessDiagnosticsSignalFailedError",
	88	{
	89	pid: Schema.Number,
	90	signal: Schema.String,
	91	cause: Schema.Defect(),
	92	},
	93	) {
	94	override get message(): string {
	95	return `Failed to signal process ${this.pid} with ${this.signal}.`;
	96	}
	97	}
	98
	99	const ProcessDiagnosticsError = Schema.Union([
	100	ProcessDiagnosticsQueryTimeoutError,
	101	ProcessDiagnosticsQueryFailedError,
	102	ProcessDiagnosticsServerProcessSignalError,
	103	ProcessDiagnosticsNotDescendantError,
	104	ProcessDiagnosticsSignalFailedError,
	105	]);
	106	type ProcessDiagnosticsError = typeof ProcessDiagnosticsError.Type;
	107	const isProcessDiagnosticsError = Schema.is(ProcessDiagnosticsError);
	108
63	109	function parsePositiveInt(value: string): number \| null {
64	110	const parsed = Number.parseInt(value, 10);
65	111	return Number.isInteger(parsed) && parsed > 0 ? parsed : null;
...	...	@@ -272,11 +318,7 @@ interface ProcessOutput {
272	318	}
273	319
274	320	const runProcess = Effect.fn("runProcess")(
275		function* (input: {
276		readonly command: string;
277		readonly args: ReadonlyArray<string>;
278		readonly errorMessage: string;
279		}) {
	321	function* (input: { readonly command: string; readonly args: ReadonlyArray<string> }) {
280	322	const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
281	323	// `ps` and `powershell.exe` are real executables; spawning through cmd.exe
282	324	// shell mode would re-tokenize the PowerShell `-Command` payload (which
...	...	@@ -315,14 +357,22 @@ const runProcess = Effect.fn("runProcess")(
315	357	Effect.timeoutOption(Duration.millis(PROCESS_QUERY_TIMEOUT_MS)),
316	358	Effect.flatMap((result) =>
317	359	Option.match(result, {
318		onNone: () => Effect.fail(toProcessDiagnosticsError(`${input.errorMessage} timed out.`)),
	360	onNone: () =>
	361	Effect.fail(
	362	new ProcessDiagnosticsQueryTimeoutError({
	363	command: input.command,
	364	}),
	365	),
319	366	onSome: Effect.succeed,
320	367	}),
321	368	),
322	369	Effect.mapError((cause) =>
323	370	isProcessDiagnosticsError(cause)
324	371	? cause
325		: toProcessDiagnosticsError(input.errorMessage, cause),
	372	: new ProcessDiagnosticsQueryFailedError({
	373	command: input.command,
	374	cause,
	375	}),
326	376	),
327	377	),
328	378	);
...	...	@@ -335,11 +385,15 @@ function readPosixProcessRows(): Effect.Effect<
335	385	return runProcess({
336	386	command: "ps",
337	387	args: ["-axo", POSIX_PROCESS_QUERY_COMMAND],
338		errorMessage: "Failed to query process diagnostics.",
339	388	}).pipe(
340	389	Effect.flatMap((result) =>
341	390	result.exitCode !== 0
342		? Effect.fail(toProcessDiagnosticsError(result.stderr.trim() \|\| "ps failed."))
	391	? Effect.fail(
	392	new ProcessDiagnosticsQueryFailedError({
	393	command: "ps",
	394	stderr: result.stderr.trim() \|\| "ps failed.",
	395	}),
	396	)
343	397	: Effect.succeed(parsePosixProcessRows(result.stdout)),
344	398	),
345	399	);
...	...	@@ -361,12 +415,14 @@ function readWindowsProcessRows(): Effect.Effect<
361	415	return runProcess({
362	416	command: "powershell.exe",
363	417	args: ["-NoProfile", "-NonInteractive", "-Command", command],
364		errorMessage: "Failed to query process diagnostics.",
365	418	}).pipe(
366	419	Effect.flatMap((result) =>
367	420	result.exitCode !== 0
368	421	? Effect.fail(
369		toProcessDiagnosticsError(result.stderr.trim() \|\| "PowerShell process query failed."),
	422	new ProcessDiagnosticsQueryFailedError({
	423	command: "powershell.exe",
	424	stderr: result.stderr.trim() \|\| "PowerShell process query failed.",
	425	}),
370	426	)
371	427	: Effect.succeed(parseWindowsProcessRows(result.stdout)),
372	428	),
...	...	@@ -390,7 +446,11 @@ function assertDescendantPid(
390	446	pid: number,
391	447	): Effect.Effect<void, ProcessDiagnosticsError, ChildProcessSpawner.ChildProcessSpawner> {
392	448	if (pid === process.pid) {
393		return Effect.fail(toProcessDiagnosticsError("Refusing to signal the T3 server process."));
	449	return Effect.fail(
	450	new ProcessDiagnosticsServerProcessSignalError({
	451	pid,
	452	}),
	453	);
394	454	}
395	455
396	456	return readProcessRows.pipe(
...	...	@@ -402,16 +462,18 @@ function assertDescendantPid(
402	462	return descendant
403	463	? Effect.void
404	464	: Effect.fail(
405		toProcessDiagnosticsError(`Process ${pid} is not a live descendant of the T3 server.`),
	465	new ProcessDiagnosticsNotDescendantError({
	466	pid,
	467	}),
406	468	);
407	469	}),
408	470	);
409	471	}
410	472
411		export const make = Effect.fn("makeProcessDiagnostics")(function* () {
	473	export const make = Effect.gen(function* () {
412	474	const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
413	475
414		const read: ProcessDiagnosticsShape["read"] = Effect.gen(function* () {
	476	const read: ProcessDiagnostics["Service"]["read"] = Effect.gen(function* () {
415	477	const readAt = yield* DateTime.now;
416	478	const rows = yield* readProcessRows.pipe(
417	479	Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
...	...	@@ -427,7 +489,7 @@ export const make = Effect.fn("makeProcessDiagnostics")(function* () {
427	489	),
428	490	);
429	491
430		const signal: ProcessDiagnosticsShape["signal"] = Effect.fn("ProcessDiagnostics.signal")(
	492	const signal: ProcessDiagnostics["Service"]["signal"] = Effect.fn("ProcessDiagnostics.signal")(
431	493	function* (input) {
432	494	return yield* assertDescendantPid(input.pid).pipe(
433	495	Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
...	...	@@ -443,10 +505,11 @@ export const make = Effect.fn("makeProcessDiagnostics")(function* () {
443	505	};
444	506	},
445	507	catch: (cause) =>
446		toProcessDiagnosticsError(
447		`Failed to signal process ${input.pid} with ${input.signal}.`,
	508	new ProcessDiagnosticsSignalFailedError({
	509	pid: input.pid,
	510	signal: input.signal,
448	511	cause,
449		),
	512	}),
450	513	}),
451	514	),
452	515	Effect.catch((error: ProcessDiagnosticsError) =>
...	...	@@ -464,4 +527,4 @@ export const make = Effect.fn("makeProcessDiagnostics")(function* () {
464	527	return ProcessDiagnostics.of({ read, signal });
465	528	});
466	529
467		export const layer = Layer.effect(ProcessDiagnostics, make());
	530	export const layer = Layer.effect(ProcessDiagnostics, make);

apps/server/src/diagnostics/ProcessResourceMonitor.test.ts+10-13

...	...	@@ -3,16 +3,13 @@ import * as DateTime from "effect/DateTime";
3	3	import * as Effect from "effect/Effect";
4	4	import * as Option from "effect/Option";
5	5
6		import {
7		aggregateProcessResourceHistory,
8		collectMonitoredSamples,
9		} from "./ProcessResourceMonitor.ts";
	6	import * as ProcessResourceMonitor from "./ProcessResourceMonitor.ts";
10	7
11	8	describe("ProcessResourceMonitor", () => {
12	9	it.effect("samples the server root process and descendants", () =>
13	10	Effect.sync(() => {
14	11	const sampledAt = DateTime.makeUnsafe("2026-05-05T10:00:00.000Z");
15		const samples = collectMonitoredSamples({
	12	const samples = ProcessResourceMonitor.collectMonitoredSamples({
16	13	serverPid: 100,
17	14	sampledAt,
18	15	sampledAtMs: DateTime.toEpochMillis(sampledAt),
...	...	@@ -72,7 +69,7 @@ describe("ProcessResourceMonitor", () => {
72	69	const firstAt = DateTime.makeUnsafe("2026-05-05T10:00:00.000Z");
73	70	const secondAt = DateTime.makeUnsafe("2026-05-05T10:00:05.000Z");
74	71	const samples = [
75		...collectMonitoredSamples({
	72	...ProcessResourceMonitor.collectMonitoredSamples({
76	73	serverPid: 100,
77	74	sampledAt: firstAt,
78	75	sampledAtMs: DateTime.toEpochMillis(firstAt),
...	...	@@ -89,7 +86,7 @@ describe("ProcessResourceMonitor", () => {
89	86	},
90	87	],
91	88	}),
92		...collectMonitoredSamples({
	89	...ProcessResourceMonitor.collectMonitoredSamples({
93	90	serverPid: 100,
94	91	sampledAt: secondAt,
95	92	sampledAtMs: DateTime.toEpochMillis(secondAt),
...	...	@@ -108,7 +105,7 @@ describe("ProcessResourceMonitor", () => {
108	105	}),
109	106	];
110	107
111		const result = aggregateProcessResourceHistory({
	108	const result = ProcessResourceMonitor.aggregateProcessResourceHistory({
112	109	samples,
113	110	readAt: secondAt,
114	111	readAtMs: DateTime.toEpochMillis(secondAt),
...	...	@@ -132,7 +129,7 @@ describe("ProcessResourceMonitor", () => {
132	129	const firstAt = DateTime.makeUnsafe("2026-05-05T10:00:00.400Z");
133	130	const secondAt = DateTime.makeUnsafe("2026-05-05T10:00:05.900Z");
134	131	const samples = [
135		...collectMonitoredSamples({
	132	...ProcessResourceMonitor.collectMonitoredSamples({
136	133	serverPid: 100,
137	134	sampledAt: firstAt,
138	135	sampledAtMs: DateTime.toEpochMillis(firstAt),
...	...	@@ -149,7 +146,7 @@ describe("ProcessResourceMonitor", () => {
149	146	},
150	147	],
151	148	}),
152		...collectMonitoredSamples({
	149	...ProcessResourceMonitor.collectMonitoredSamples({
153	150	serverPid: 100,
154	151	sampledAt: secondAt,
155	152	sampledAtMs: DateTime.toEpochMillis(secondAt),
...	...	@@ -168,7 +165,7 @@ describe("ProcessResourceMonitor", () => {
168	165	}),
169	166	];
170	167
171		const result = aggregateProcessResourceHistory({
	168	const result = ProcessResourceMonitor.aggregateProcessResourceHistory({
172	169	samples,
173	170	readAt: secondAt,
174	171	readAtMs: DateTime.toEpochMillis(secondAt),
...	...	@@ -187,7 +184,7 @@ describe("ProcessResourceMonitor", () => {
187	184	it.effect("returns all process summaries in the selected window", () =>
188	185	Effect.sync(() => {
189	186	const sampledAt = DateTime.makeUnsafe("2026-05-05T10:00:00.000Z");
190		const samples = collectMonitoredSamples({
	187	const samples = ProcessResourceMonitor.collectMonitoredSamples({
191	188	serverPid: 100,
192	189	sampledAt,
193	190	sampledAtMs: DateTime.toEpochMillis(sampledAt),
...	...	@@ -215,7 +212,7 @@ describe("ProcessResourceMonitor", () => {
215	212	],
216	213	});
217	214
218		const result = aggregateProcessResourceHistory({
	215	const result = ProcessResourceMonitor.aggregateProcessResourceHistory({
219	216	samples,
220	217	readAt: sampledAt,
221	218	readAtMs: DateTime.toEpochMillis(sampledAt),

apps/server/src/diagnostics/ProcessResourceMonitor.ts+21-23

...	...	@@ -10,14 +10,9 @@ import * as Effect from "effect/Effect";
10	10	import * as Layer from "effect/Layer";
11	11	import * as Option from "effect/Option";
12	12	import * as Ref from "effect/Ref";
13		import { ChildProcessSpawner } from "effect/unstable/process";
	13	import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner";
14	14
15		import {
16		buildDescendantEntries,
17		isDiagnosticsQueryProcess,
18		type ProcessRow,
19		readProcessRows,
20		} from "./ProcessDiagnostics.ts";
	15	import * as ProcessDiagnostics from "./ProcessDiagnostics.ts";
21	16
22	17	const SAMPLE_INTERVAL_MS = 5_000;
23	18	const RETENTION_MS = 60 * 60_000;
...	...	@@ -41,38 +36,41 @@ interface MonitorState {
41	36	readonly lastError: string \| null;
42	37	}
43	38
44		export interface ProcessResourceMonitorShape {
45		readonly readHistory: (
46		input: ServerProcessResourceHistoryInput,
47		) => Effect.Effect<ServerProcessResourceHistoryResult>;
48		}
49
50	39	export class ProcessResourceMonitor extends Context.Service<
51	40	ProcessResourceMonitor,
52		ProcessResourceMonitorShape
	41	{
	42	readonly readHistory: (
	43	input: ServerProcessResourceHistoryInput,
	44	) => Effect.Effect<ServerProcessResourceHistoryResult>;
	45	}
53	46	>()("t3/diagnostics/ProcessResourceMonitor") {}
54	47
55	48	function dateTimeFromMillis(ms: number): DateTime.Utc {
56	49	return DateTime.makeUnsafe(ms);
57	50	}
58	51
59		function sampleKey(row: Pick<ProcessRow, "pid" \| "command">): string {
	52	function sampleKey(row: Pick<ProcessDiagnostics.ProcessRow, "pid" \| "command">): string {
60	53	return `${row.pid}:${row.command}`;
61	54	}
62	55
63		function findServerRootRow(rows: ReadonlyArray<ProcessRow>, serverPid: number): ProcessRow \| null {
	56	function findServerRootRow(
	57	rows: ReadonlyArray<ProcessDiagnostics.ProcessRow>,
	58	serverPid: number,
	59	): ProcessDiagnostics.ProcessRow \| null {
64	60	return rows.find((row) => row.pid === serverPid) ?? null;
65	61	}
66	62
67	63	export function collectMonitoredSamples(input: {
68		readonly rows: ReadonlyArray<ProcessRow>;
	64	readonly rows: ReadonlyArray<ProcessDiagnostics.ProcessRow>;
69	65	readonly serverPid: number;
70	66	readonly sampledAt: DateTime.Utc;
71	67	readonly sampledAtMs: number;
72	68	}): ReadonlyArray<ProcessResourceSample> {
73		const rows = input.rows.filter((row) => !isDiagnosticsQueryProcess(row, input.serverPid));
	69	const rows = input.rows.filter(
	70	(row) => !ProcessDiagnostics.isDiagnosticsQueryProcess(row, input.serverPid),
	71	);
74	72	const root = findServerRootRow(rows, input.serverPid);
75		const descendants = buildDescendantEntries(rows, input.serverPid);
	73	const descendants = ProcessDiagnostics.buildDescendantEntries(rows, input.serverPid);
76	74	const samples: ProcessResourceSample[] = [];
77	75
78	76	if (root) {
...	...	@@ -245,14 +243,14 @@ export function aggregateProcessResourceHistory(input: {
245	243	};
246	244	}
247	245
248		export const make = Effect.fn("makeProcessResourceMonitor")(function* () {
	246	export const make = Effect.gen(function* () {
249	247	const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
250	248	const state = yield* Ref.make<MonitorState>({ samples: [], lastError: null });
251	249
252	250	const sampleOnce = Effect.gen(function* () {
253	251	const sampledAt = yield* DateTime.now;
254	252	const sampledAtMs = DateTime.toEpochMillis(sampledAt);
255		const rows = yield* readProcessRows.pipe(
	253	const rows = yield* ProcessDiagnostics.readProcessRows.pipe(
256	254	Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
257	255	);
258	256	const samples = collectMonitoredSamples({
...	...	@@ -278,7 +276,7 @@ export const make = Effect.fn("makeProcessResourceMonitor")(function* () {
278	276	Effect.forkScoped,
279	277	);
280	278
281		const readHistory: ProcessResourceMonitorShape["readHistory"] = (input) =>
	279	const readHistory: ProcessResourceMonitor["Service"]["readHistory"] = (input) =>
282	280	Effect.gen(function* () {
283	281	const readAt = yield* DateTime.now;
284	282	const readAtMs = DateTime.toEpochMillis(readAt);
...	...	@@ -296,4 +294,4 @@ export const make = Effect.fn("makeProcessResourceMonitor")(function* () {
296	294	return ProcessResourceMonitor.of({ readHistory });
297	295	});
298	296
299		export const layer = Layer.effect(ProcessResourceMonitor, make());
	297	export const layer = Layer.effect(ProcessResourceMonitor, make);

apps/server/src/diagnostics/TraceDiagnostics.ts+11-10

...	...	@@ -39,13 +39,14 @@ export interface TraceDiagnosticsOptions {
39	39	readonly readAt?: DateTime.Utc;
40	40	}
41	41
42		export interface TraceDiagnosticsShape {
43		readonly read: (options: TraceDiagnosticsOptions) => Effect.Effect<ServerTraceDiagnosticsResult>;
44		}
45
46		export class TraceDiagnostics extends Context.Service<TraceDiagnostics, TraceDiagnosticsShape>()(
47		"t3/diagnostics/TraceDiagnostics",
48		) {}
	42	export class TraceDiagnostics extends Context.Service<
	43	TraceDiagnostics,
	44	{
	45	readonly read: (
	46	options: TraceDiagnosticsOptions,
	47	) => Effect.Effect<ServerTraceDiagnosticsResult>;
	48	}
	49	>()("t3/diagnostics/TraceDiagnostics") {}
49	50
50	51	interface TraceDiagnosticsInput {
51	52	readonly traceFilePath: string;
...	...	@@ -395,10 +396,10 @@ function readTraceFile(
395	396	);
396	397	}
397	398
398		export const make = Effect.fn("makeTraceDiagnostics")(function* () {
	399	export const make = Effect.gen(function* () {
399	400	const fileSystem = yield* FileSystem.FileSystem;
400	401
401		const read: TraceDiagnosticsShape["read"] = Effect.fn("TraceDiagnostics.read")(
	402	const read: TraceDiagnostics["Service"]["read"] = Effect.fn("TraceDiagnostics.read")(
402	403	function* (options) {
403	404	const readAt = options.readAt ?? (yield* DateTime.now);
404	405	const slowSpanThresholdMs = options.slowSpanThresholdMs ?? DEFAULT_SLOW_SPAN_THRESHOLD_MS;
...	...	@@ -449,7 +450,7 @@ export const make = Effect.fn("makeTraceDiagnostics")(function* () {
449	450	return TraceDiagnostics.of({ read });
450	451	});
451	452
452		export const layer = Layer.effect(TraceDiagnostics, make());
	453	export const layer = Layer.effect(TraceDiagnostics, make);
453	454
454	455	export function readTraceDiagnostics(
455	456	options: TraceDiagnosticsOptions,

apps/server/src/http.ts+10-9

...	...	@@ -24,13 +24,13 @@ import {
24	24	import * as HttpApiBuilder from "effect/unstable/httpapi/HttpApiBuilder";
25	25	import { OtlpTracer } from "effect/unstable/observability";
26	26
27		import { resolveStaticDir, ServerConfig } from "./config.ts";
	27	import * as ServerConfig from "./config.ts";
28	28	import {
29	29	ASSET_ROUTE_PREFIX,
30	30	FALLBACK_PROJECT_FAVICON_SVG,
31	31	resolveAsset,
32	32	} from "./assets/AssetAccess.ts";
33		import { BrowserTraceCollector } from "./observability/Services/BrowserTraceCollector.ts";
	33	import * as BrowserTraceCollector from "./observability/BrowserTraceCollector.ts";
34	34	import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts";
35	35	import { traceRelayRequest } from "./cloud/traceRelayRequest.ts";
36	36	import {
...	...	@@ -39,7 +39,7 @@ import {
39	39	failEnvironmentAuthInvalid,
40	40	failEnvironmentInternal,
41	41	} from "./auth/http.ts";
42		import { ServerEnvironment } from "./environment/Services/ServerEnvironment.ts";
	42	import * as ServerEnvironment from "./environment/Services/ServerEnvironment.ts";
43	43	import { browserApiCorsAllowedHeaders, browserApiCorsAllowedMethods } from "./httpCors.ts";
44	44
45	45	const OTLP_TRACES_PROXY_PATH = "/api/observability/v1/traces";
...	...	@@ -47,7 +47,7 @@ const LOOPBACK_HOSTNAMES = new Set(["127.0.0.1", "::1", "localhost"]);
47	47
48	48	export const browserApiCorsLayer = Layer.unwrap(
49	49	Effect.gen(function* () {
50		const config = yield* ServerConfig;
	50	const config = yield* ServerConfig.ServerConfig;
51	51	const devOrigin = config.devUrl?.origin;
52	52	return HttpRouter.cors({
53	53	...(devOrigin ? { allowedOrigins: [devOrigin], credentials: true } : {}),
...	...	@@ -95,7 +95,7 @@ export const serverEnvironmentHttpApiLayer = HttpApiBuilder.group(
95	95	EnvironmentHttpApi,
96	96	"metadata",
97	97	Effect.fnUntraced(function* (handlers) {
98		const serverEnvironment = yield* ServerEnvironment;
	98	const serverEnvironment = yield* ServerEnvironment.ServerEnvironment;
99	99	return handlers.handle(
100	100	"descriptor",
101	101	Effect.fn("environment.metadata.descriptor")(function* (args) {
...	...	@@ -117,9 +117,9 @@ export const otlpTracesProxyRouteLayer = HttpRouter.add(
117	117	Effect.gen(function* () {
118	118	yield* authenticateRawRouteWithScope(AuthOrchestrationOperateScope);
119	119	const request = yield* HttpServerRequest.HttpServerRequest;
120		const config = yield* ServerConfig;
	120	const config = yield* ServerConfig.ServerConfig;
121	121	const otlpTracesUrl = config.otlpTracesUrl;
122		const browserTraceCollector = yield* BrowserTraceCollector;
	122	const browserTraceCollector = yield* BrowserTraceCollector.BrowserTraceCollector;
123	123	const httpClient = yield* HttpClient.HttpClient;
124	124	const bodyJson = cast<unknown, OtlpTracer.TraceData>(yield* request.json);
125	125
...	...	@@ -223,14 +223,15 @@ export const staticAndDevRouteLayer = HttpRouter.add(
223	223	return HttpServerResponse.text("Bad Request", { status: 400 });
224	224	}
225	225
226		const config = yield* ServerConfig;
	226	const config = yield* ServerConfig.ServerConfig;
227	227	if (config.devUrl && isLoopbackHostname(url.value.hostname)) {
228	228	return HttpServerResponse.redirect(resolveDevRedirectUrl(config.devUrl, url.value), {
229	229	status: 302,
230	230	});
231	231	}
232	232
233		const staticDir = config.staticDir ?? (config.devUrl ? yield* resolveStaticDir() : undefined);
	233	const staticDir =
	234	config.staticDir ?? (config.devUrl ? yield* ServerConfig.resolveStaticDir() : undefined);
234	235	if (!staticDir) {
235	236	return HttpServerResponse.text("No static directory configured and no dev URL set.", {
236	237	status: 503,

apps/server/src/observability/BrowserTraceCollector.ts created+23

...	...	@@ -0,0 +1,23 @@
	1	import type { TraceRecord, TraceSink } from "@t3tools/shared/observability";
	2	import * as Context from "effect/Context";
	3	import * as Effect from "effect/Effect";
	4	import * as Layer from "effect/Layer";
	5
	6	export class BrowserTraceCollector extends Context.Service<
	7	BrowserTraceCollector,
	8	{
	9	readonly record: (records: ReadonlyArray<TraceRecord>) => Effect.Effect<void>;
	10	}
	11	>()("t3/observability/BrowserTraceCollector") {}
	12
	13	export const make = (sink: TraceSink): BrowserTraceCollector["Service"] =>
	14	BrowserTraceCollector.of({
	15	record: (records) =>
	16	Effect.sync(() => {
	17	for (const record of records) {
	18	sink.push(record);
	19	}
	20	}),
	21	});
	22
	23	export const layer = (sink: TraceSink) => Layer.succeed(BrowserTraceCollector, make(sink));

apps/server/src/observability/Layers/Observability.ts+7-12

...	...	@@ -4,17 +4,19 @@ import * as Effect from "effect/Effect";
4	4	import * as Layer from "effect/Layer";
5	5	import * as References from "effect/References";
6	6	import * as Tracer from "effect/Tracer";
7		import { OtlpMetrics, OtlpSerialization, OtlpTracer } from "effect/unstable/observability";
	7	import * as OtlpMetrics from "effect/unstable/observability/OtlpMetrics";
	8	import * as OtlpSerialization from "effect/unstable/observability/OtlpSerialization";
	9	import * as OtlpTracer from "effect/unstable/observability/OtlpTracer";
8	10
9		import { ServerConfig } from "../../config.ts";
	11	import * as ServerConfig from "../../config.ts";
10	12	import { ServerLoggerLive } from "../../serverLogger.ts";
11		import { BrowserTraceCollector } from "../Services/BrowserTraceCollector.ts";
	13	import * as BrowserTraceCollector from "../BrowserTraceCollector.ts";
12	14
13	15	const otlpSerializationLayer = OtlpSerialization.layerJson;
14	16
15	17	export const ObservabilityLive = Layer.unwrap(
16	18	Effect.gen(function* () {
17		const config = yield* ServerConfig;
	19	const config = yield* ServerConfig.ServerConfig;
18	20
19	21	const traceReferencesLayer = Layer.mergeAll(
20	22	Layer.succeed(Tracer.MinimumTraceLevel, config.traceMinLevel),
...	...	@@ -56,14 +58,7 @@ export const ObservabilityLive = Layer.unwrap(
56	58
57	59	return Layer.mergeAll(
58	60	Layer.succeed(Tracer.Tracer, tracer),
59		Layer.succeed(BrowserTraceCollector, {
60		record: (records) =>
61		Effect.sync(() => {
62		for (const record of records) {
63		sink.push(record);
64		}
65		}),
66		}),
	61	BrowserTraceCollector.layer(sink),
67	62	);
68	63	}),
69	64	).pipe(Layer.provideMerge(otlpSerializationLayer));

apps/server/src/observability/Services/BrowserTraceCollector.ts deleted-12

...	...	@@ -1,12 +0,0 @@
1		import type { TraceRecord } from "@t3tools/shared/observability";
2		import * as Context from "effect/Context";
3		import type * as Effect from "effect/Effect";
4
5		export interface BrowserTraceCollectorShape {
6		readonly record: (records: ReadonlyArray<TraceRecord>) => Effect.Effect<void>;
7		}
8
9		export class BrowserTraceCollector extends Context.Service<
10		BrowserTraceCollector,
11		BrowserTraceCollectorShape
12		>()("t3/observability/Services/BrowserTraceCollector") {}

apps/server/src/provider/Layers/ProviderService.test.ts+164-75

...	...	@@ -43,14 +43,11 @@ import {
43	43	type ProviderAdapterError,
44	44	} from "../Errors.ts";
45	45	import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts";
46		import {
47		ProviderAdapterRegistry,
48		type ProviderAdapterRegistryShape,
49		} from "../Services/ProviderAdapterRegistry.ts";
50		import { ProviderService } from "../Services/ProviderService.ts";
51		import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts";
	46	import * as ProviderAdapterRegistry from "../Services/ProviderAdapterRegistry.ts";
	47	import * as ProviderService from "../Services/ProviderService.ts";
	48	import * as ProviderSessionDirectory from "../Services/ProviderSessionDirectory.ts";
52	49	import { makeProviderServiceLive } from "./ProviderService.ts";
53		import { NoOpProviderEventLoggers, ProviderEventLoggers } from "./ProviderEventLoggers.ts";
	50	import * as ProviderEventLoggers from "./ProviderEventLoggers.ts";
54	51	import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts";
55	52	import * as NodeServices from "@effect/platform-node/NodeServices";
56	53	import * as ProviderSessionRuntime from "../../persistence/ProviderSessionRuntime.ts";
...	...	@@ -58,11 +55,11 @@ import {
58	55	makeSqlitePersistenceLive,
59	56	SqlitePersistenceMemory,
60	57	} from "../../persistence/Layers/Sqlite.ts";
61		import { ServerSettingsService } from "../../serverSettings.ts";
62		import { AnalyticsService } from "../../telemetry/Services/AnalyticsService.ts";
	58	import * as ServerSettings from "../../serverSettings.ts";
	59	import * as AnalyticsService from "../../telemetry/AnalyticsService.ts";
63	60	import { makeAdapterRegistryMock } from "../testUtils/providerAdapterRegistryMock.ts";
64	61
65		const defaultServerSettingsLayer = ServerSettingsService.layerTest();
	62	const defaultServerSettingsLayer = ServerSettings.ServerSettingsService.layerTest();
66	63
67	64	const asRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.make(value);
68	65	const asEventId = (value: string): EventId => EventId.make(value);
...	...	@@ -280,7 +277,10 @@ function makeProviderServiceLayer() {
280	277	[ProviderDriverKind.make("cursor")]: cursor.adapter,
281	278	});
282	279
283		const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry);
	280	const providerAdapterLayer = Layer.succeed(
	281	ProviderAdapterRegistry.ProviderAdapterRegistry,
	282	registry,
	283	);
284	284	const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe(
285	285	Layer.provide(SqlitePersistenceMemory),
286	286	);
...	...	@@ -293,7 +293,12 @@ function makeProviderServiceLayer() {
293	293	Layer.provide(directoryLayer),
294	294	Layer.provide(defaultServerSettingsLayer),
295	295	Layer.provideMerge(AnalyticsService.layerTest),
296		Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)),
	296	Layer.provide(
	297	Layer.succeed(
	298	ProviderEventLoggers.ProviderEventLoggers,
	299	ProviderEventLoggers.NoOpProviderEventLoggers,
	300	),
	301	),
297	302	),
298	303	directoryLayer,
299	304
...	...	@@ -325,7 +330,10 @@ it.effect("ProviderServiceLive catches stopAll failures during shutdown", () =>
325	330	const registry = makeAdapterRegistryMock({
326	331	[CODEX_DRIVER]: codex.adapter,
327	332	});
328		const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry);
	333	const providerAdapterLayer = Layer.succeed(
	334	ProviderAdapterRegistry.ProviderAdapterRegistry,
	335	registry,
	336	);
329	337	const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe(
330	338	Layer.provide(SqlitePersistenceMemory),
331	339	);
...	...	@@ -336,7 +344,12 @@ it.effect("ProviderServiceLive catches stopAll failures during shutdown", () =>
336	344	Layer.provide(directoryLayer),
337	345	Layer.provide(defaultServerSettingsLayer),
338	346	Layer.provideMerge(AnalyticsService.layerTest),
339		Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)),
	347	Layer.provide(
	348	Layer.succeed(
	349	ProviderEventLoggers.ProviderEventLoggers,
	350	ProviderEventLoggers.NoOpProviderEventLoggers,
	351	),
	352	),
340	353	),
341	354	directoryLayer,
342	355	runtimeRepositoryLayer,
...	...	@@ -345,7 +358,7 @@ it.effect("ProviderServiceLive catches stopAll failures during shutdown", () =>
345	358	const scope = yield* Scope.make();
346	359	const runtimeServices = yield* Layer.build(providerLayer).pipe(Scope.provide(scope));
347	360
348		yield* ProviderService.pipe(Effect.provide(runtimeServices));
	361	yield* ProviderService.ProviderService.pipe(Effect.provide(runtimeServices));
349	362	const closeExit = yield* Scope.close(scope, Exit.void).pipe(Effect.exit);
350	363
351	364	assert.equal(Exit.isSuccess(closeExit), true);
...	...	@@ -361,7 +374,7 @@ it.effect("ProviderServiceLive rejects new sessions for disabled providers", ()
361	374	[CODEX_DRIVER]: codex.adapter,
362	375	[CLAUDE_AGENT_DRIVER]: claude.adapter,
363	376	});
364		const registry: ProviderAdapterRegistryShape = {
	377	const registry: ProviderAdapterRegistry.ProviderAdapterRegistry["Service"] = {
365	378	...registryBase,
366	379	getInstanceInfo: (instanceId) =>
367	380	instanceId === claudeAgentInstanceId
...	...	@@ -377,7 +390,10 @@ it.effect("ProviderServiceLive rejects new sessions for disabled providers", ()
377	390	})
378	391	: registryBase.getInstanceInfo(instanceId),
379	392	};
380		const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry);
	393	const providerAdapterLayer = Layer.succeed(
	394	ProviderAdapterRegistry.ProviderAdapterRegistry,
	395	registry,
	396	);
381	397	const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe(
382	398	Layer.provide(SqlitePersistenceMemory),
383	399	);
...	...	@@ -387,12 +403,17 @@ it.effect("ProviderServiceLive rejects new sessions for disabled providers", ()
387	403	Layer.provide(directoryLayer),
388	404	Layer.provide(defaultServerSettingsLayer),
389	405	Layer.provide(AnalyticsService.layerTest),
390		Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)),
	406	Layer.provide(
	407	Layer.succeed(
	408	ProviderEventLoggers.ProviderEventLoggers,
	409	ProviderEventLoggers.NoOpProviderEventLoggers,
	410	),
	411	),
391	412	);
392	413
393	414	const failure = yield* Effect.flip(
394	415	Effect.gen(function* () {
395		const provider = yield* ProviderService;
	416	const provider = yield* ProviderService.ProviderService;
396	417	return yield* provider.startSession(asThreadId("thread-disabled"), {
397	418	provider: ProviderDriverKind.make("claudeAgent"),
398	419	providerInstanceId: claudeAgentInstanceId,
...	...	@@ -419,7 +440,7 @@ it.effect(
419	440	new ProviderUnsupportedError({
420	441	provider: driverKind,
421	442	});
422		const registry: ProviderAdapterRegistryShape = {
	443	const registry: ProviderAdapterRegistry.ProviderAdapterRegistry["Service"] = {
423	444	getByInstance: (requestedInstanceId) =>
424	445	requestedInstanceId === instanceId
425	446	? Effect.succeed(codex.adapter)
...	...	@@ -444,8 +465,11 @@ it.effect(
444	465	PubSub.subscribe(pubsub),
445	466	),
446	467	};
447		const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry);
448		const serverSettingsLayer = ServerSettingsService.layerTest({
	468	const providerAdapterLayer = Layer.succeed(
	469	ProviderAdapterRegistry.ProviderAdapterRegistry,
	470	registry,
	471	);
	472	const serverSettingsLayer = ServerSettings.ServerSettingsService.layerTest({
449	473	providers: {
450	474	codex: {
451	475	enabled: false,
...	...	@@ -463,11 +487,16 @@ it.effect(
463	487	Layer.provide(directoryLayer),
464	488	Layer.provide(serverSettingsLayer),
465	489	Layer.provide(AnalyticsService.layerTest),
466		Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)),
	490	Layer.provide(
	491	Layer.succeed(
	492	ProviderEventLoggers.ProviderEventLoggers,
	493	ProviderEventLoggers.NoOpProviderEventLoggers,
	494	),
	495	),
467	496	);
468	497
469	498	const session = yield* Effect.gen(function* () {
470		const provider = yield* ProviderService;
	499	const provider = yield* ProviderService.ProviderService;
471	500	return yield* provider.startSession(asThreadId("thread-enabled-custom"), {
472	501	provider: driverKind,
473	502	providerInstanceId: instanceId,
...	...	@@ -490,7 +519,7 @@ it.effect("ProviderServiceLive rejects new sessions for disabled custom instance
490	519	new ProviderUnsupportedError({
491	520	provider: ProviderDriverKind.make("codex"),
492	521	});
493		const registry: ProviderAdapterRegistryShape = {
	522	const registry: ProviderAdapterRegistry.ProviderAdapterRegistry["Service"] = {
494	523	getByInstance: (requestedInstanceId) =>
495	524	requestedInstanceId === instanceId
496	525	? Effect.succeed(codex.adapter)
...	...	@@ -515,7 +544,10 @@ it.effect("ProviderServiceLive rejects new sessions for disabled custom instance
515	544	PubSub.subscribe(pubsub),
516	545	),
517	546	};
518		const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry);
	547	const providerAdapterLayer = Layer.succeed(
	548	ProviderAdapterRegistry.ProviderAdapterRegistry,
	549	registry,
	550	);
519	551	const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe(
520	552	Layer.provide(SqlitePersistenceMemory),
521	553	);
...	...	@@ -525,12 +557,17 @@ it.effect("ProviderServiceLive rejects new sessions for disabled custom instance
525	557	Layer.provide(directoryLayer),
526	558	Layer.provide(defaultServerSettingsLayer),
527	559	Layer.provide(AnalyticsService.layerTest),
528		Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)),
	560	Layer.provide(
	561	Layer.succeed(
	562	ProviderEventLoggers.ProviderEventLoggers,
	563	ProviderEventLoggers.NoOpProviderEventLoggers,
	564	),
	565	),
529	566	);
530	567
531	568	const failure = yield* Effect.flip(
532	569	Effect.gen(function* () {
533		const provider = yield* ProviderService;
	570	const provider = yield* ProviderService.ProviderService;
534	571	return yield* provider.startSession(asThreadId("thread-disabled-instance"), {
535	572	provider: ProviderDriverKind.make("codex"),
536	573	providerInstanceId: instanceId,
...	...	@@ -571,15 +608,20 @@ it.effect("ProviderServiceLive writes canonical events to the emitting thread se
571	608	close: () => Effect.void,
572	609	},
573	610	}).pipe(
574		Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)),
	611	Layer.provide(Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, registry)),
575	612	Layer.provide(directoryLayer),
576	613	Layer.provide(defaultServerSettingsLayer),
577	614	Layer.provide(AnalyticsService.layerTest),
578		Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)),
	615	Layer.provide(
	616	Layer.succeed(
	617	ProviderEventLoggers.ProviderEventLoggers,
	618	ProviderEventLoggers.NoOpProviderEventLoggers,
	619	),
	620	),
579	621	);
580	622
581	623	yield* Effect.gen(function* () {
582		yield* ProviderService;
	624	yield* ProviderService.ProviderService;
583	625	yield* advanceTestClock(10);
584	626	codex.emit({
585	627	eventId: asEventId("evt-canonical-thread-segment"),
...	...	@@ -617,7 +659,7 @@ it.effect("ProviderServiceLive keeps persisted resumable sessions on startup", (
617	659	const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer));
618	660
619	661	yield* Effect.gen(function* () {
620		const directory = yield* ProviderSessionDirectory;
	662	const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;
621	663	yield* directory.upsert({
622	664	provider: ProviderDriverKind.make("codex"),
623	665	providerInstanceId: codexInstanceId,
...	...	@@ -626,17 +668,22 @@ it.effect("ProviderServiceLive keeps persisted resumable sessions on startup", (
626	668	}).pipe(Effect.provide(directoryLayer));
627	669
628	670	const providerLayer = makeProviderServiceLive().pipe(
629		Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)),
	671	Layer.provide(Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, registry)),
630	672	Layer.provide(directoryLayer),
631	673	Layer.provide(defaultServerSettingsLayer),
632	674	Layer.provide(AnalyticsService.layerTest),
633		Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)),
	675	Layer.provide(
	676	Layer.succeed(
	677	ProviderEventLoggers.ProviderEventLoggers,
	678	ProviderEventLoggers.NoOpProviderEventLoggers,
	679	),
	680	),
634	681	);
635	682
636		yield* ProviderService.pipe(Effect.provide(providerLayer));
	683	yield* ProviderService.ProviderService.pipe(Effect.provide(providerLayer));
637	684
638	685	const persistedProvider = yield* Effect.gen(function* () {
639		const directory = yield* ProviderSessionDirectory;
	686	const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;
640	687	return yield* directory.getProvider(asThreadId("thread-stale"));
641	688	}).pipe(Effect.provide(directoryLayer));
642	689	assert.equal(persistedProvider, "codex");
...	...	@@ -683,11 +730,18 @@ it.effect(
683	730	Layer.provide(runtimeRepositoryLayer),
684	731	);
685	732	const firstProviderLayer = makeProviderServiceLive().pipe(
686		Layer.provide(Layer.succeed(ProviderAdapterRegistry, firstRegistry)),
	733	Layer.provide(
	734	Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, firstRegistry),
	735	),
687	736	Layer.provide(firstDirectoryLayer),
688	737	Layer.provide(defaultServerSettingsLayer),
689	738	Layer.provide(AnalyticsService.layerTest),
690		Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)),
	739	Layer.provide(
	740	Layer.succeed(
	741	ProviderEventLoggers.ProviderEventLoggers,
	742	ProviderEventLoggers.NoOpProviderEventLoggers,
	743	),
	744	),
691	745	);
692	746	const updatedResumeCursor = {
693	747	threadId: asThreadId("thread-1"),
...	...	@@ -697,7 +751,7 @@ it.effect(
697	751	};
698	752
699	753	const startedSession = yield* Effect.gen(function* () {
700		const provider = yield* ProviderService;
	754	const provider = yield* ProviderService.ProviderService;
701	755	const threadId = asThreadId("thread-1");
702	756	const session = yield* provider.startSession(threadId, {
703	757	provider: ProviderDriverKind.make("codex"),
...	...	@@ -735,18 +789,25 @@ it.effect(
735	789	Layer.provide(runtimeRepositoryLayer),
736	790	);
737	791	const secondProviderLayer = makeProviderServiceLive().pipe(
738		Layer.provide(Layer.succeed(ProviderAdapterRegistry, secondRegistry)),
	792	Layer.provide(
	793	Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, secondRegistry),
	794	),
739	795	Layer.provide(secondDirectoryLayer),
740	796	Layer.provide(defaultServerSettingsLayer),
741	797	Layer.provide(AnalyticsService.layerTest),
742		Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)),
	798	Layer.provide(
	799	Layer.succeed(
	800	ProviderEventLoggers.ProviderEventLoggers,
	801	ProviderEventLoggers.NoOpProviderEventLoggers,
	802	),
	803	),
743	804	);
744	805
745	806	secondCodex.startSession.mockClear();
746	807	secondCodex.rollbackThread.mockClear();
747	808
748	809	yield* Effect.gen(function* () {
749		const provider = yield* ProviderService;
	810	const provider = yield* ProviderService.ProviderService;
750	811	yield* provider.rollbackConversation({
751	812	threadId: startedSession.threadId,
752	813	numTurns: 1,
...	...	@@ -780,7 +841,7 @@ it.effect(
780	841	routing.layer("ProviderServiceLive routing", (it) => {
781	842	it.effect("routes provider operations and rollback conversation", () =>
782	843	Effect.gen(function* () {
783		const provider = yield* ProviderService;
	844	const provider = yield* ProviderService.ProviderService;
784	845
785	846	const session = yield* provider.startSession(asThreadId("thread-1"), {
786	847	provider: ProviderDriverKind.make("codex"),
...	...	@@ -866,7 +927,7 @@ routing.layer("ProviderServiceLive routing", (it) => {
866	927
867	928	it.effect("recovers stale persisted sessions for rollback by resuming thread identity", () =>
868	929	Effect.gen(function* () {
869		const provider = yield* ProviderService;
	930	const provider = yield* ProviderService.ProviderService;
870	931
871	932	const initial = yield* provider.startSession(asThreadId("thread-1"), {
872	933	provider: ProviderDriverKind.make("codex"),
...	...	@@ -907,7 +968,7 @@ routing.layer("ProviderServiceLive routing", (it) => {
907	968
908	969	it.effect("preserves the persisted binding when stopping a session", () =>
909	970	Effect.gen(function* () {
910		const provider = yield* ProviderService;
	971	const provider = yield* ProviderService.ProviderService;
911	972	const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
912	973
913	974	const initial = yield* provider.startSession(asThreadId("thread-reap-preserve"), {
...	...	@@ -959,7 +1020,7 @@ routing.layer("ProviderServiceLive routing", (it) => {
959	1020
960	1021	it.effect("routes explicit claudeAgent provider session starts to the claude adapter", () =>
961	1022	Effect.gen(function* () {
962		const provider = yield* ProviderService;
	1023	const provider = yield* ProviderService.ProviderService;
963	1024
964	1025	const session = yield* provider.startSession(asThreadId("thread-claude"), {
965	1026	provider: ProviderDriverKind.make("claudeAgent"),
...	...	@@ -988,8 +1049,8 @@ routing.layer("ProviderServiceLive routing", (it) => {
988	1049
989	1050	it.effect("dies when an active session conflicts with its persisted binding", () =>
990	1051	Effect.gen(function* () {
991		const provider = yield* ProviderService;
992		const directory = yield* ProviderSessionDirectory;
	1052	const provider = yield* ProviderService.ProviderService;
	1053	const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;
993	1054	const threadId = asThreadId("thread-binding-mismatch");
994	1055
995	1056	yield* provider.startSession(threadId, {
...	...	@@ -1019,7 +1080,7 @@ routing.layer("ProviderServiceLive routing", (it) => {
1019	1080
1020	1081	it.effect("stops stale sessions in other providers after a successful replacement start", () =>
1021	1082	Effect.gen(function* () {
1022		const provider = yield* ProviderService;
	1083	const provider = yield* ProviderService.ProviderService;
1023	1084	const threadId = asThreadId("thread-provider-replacement");
1024	1085
1025	1086	const codexSession = yield* provider.startSession(threadId, {
...	...	@@ -1058,7 +1119,7 @@ routing.layer("ProviderServiceLive routing", (it) => {
1058	1119
1059	1120	it.effect("recovers stale sessions for sendTurn using persisted cwd", () =>
1060	1121	Effect.gen(function* () {
1061		const provider = yield* ProviderService;
	1122	const provider = yield* ProviderService.ProviderService;
1062	1123
1063	1124	const initial = yield* provider.startSession(asThreadId("thread-1"), {
1064	1125	provider: ProviderDriverKind.make("codex"),
...	...	@@ -1099,7 +1160,7 @@ routing.layer("ProviderServiceLive routing", (it) => {
1099	1160
1100	1161	it.effect("recovers stale claudeAgent sessions for sendTurn using persisted cwd", () =>
1101	1162	Effect.gen(function* () {
1102		const provider = yield* ProviderService;
	1163	const provider = yield* ProviderService.ProviderService;
1103	1164
1104	1165	const initial = yield* provider.startSession(asThreadId("thread-claude-send-turn"), {
1105	1166	provider: ProviderDriverKind.make("claudeAgent"),
...	...	@@ -1152,7 +1213,7 @@ routing.layer("ProviderServiceLive routing", (it) => {
1152	1213
1153	1214	it.effect("lists no sessions after adapter runtime clears", () =>
1154	1215	Effect.gen(function* () {
1155		const provider = yield* ProviderService;
	1216	const provider = yield* ProviderService.ProviderService;
1156	1217
1157	1218	yield* provider.startSession(asThreadId("thread-1"), {
1158	1219	provider: ProviderDriverKind.make("codex"),
...	...	@@ -1177,7 +1238,7 @@ routing.layer("ProviderServiceLive routing", (it) => {
1177	1238
1178	1239	it.effect("persists runtime status transitions in provider_session_runtime", () =>
1179	1240	Effect.gen(function* () {
1180		const provider = yield* ProviderService;
	1241	const provider = yield* ProviderService.ProviderService;
1181	1242	const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
1182	1243
1183	1244	const threadId = asThreadId("thread-runtime-status");
...	...	@@ -1237,15 +1298,22 @@ routing.layer("ProviderServiceLive routing", (it) => {
1237	1298	Layer.provide(runtimeRepositoryLayer),
1238	1299	);
1239	1300	const firstProviderLayer = makeProviderServiceLive().pipe(
1240		Layer.provide(Layer.succeed(ProviderAdapterRegistry, firstRegistry)),
	1301	Layer.provide(
	1302	Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, firstRegistry),
	1303	),
1241	1304	Layer.provide(firstDirectoryLayer),
1242	1305	Layer.provide(defaultServerSettingsLayer),
1243	1306	Layer.provide(AnalyticsService.layerTest),
1244		Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)),
	1307	Layer.provide(
	1308	Layer.succeed(
	1309	ProviderEventLoggers.ProviderEventLoggers,
	1310	ProviderEventLoggers.NoOpProviderEventLoggers,
	1311	),
	1312	),
1245	1313	);
1246	1314
1247	1315	const initial = yield* Effect.gen(function* () {
1248		const provider = yield* ProviderService;
	1316	const provider = yield* ProviderService.ProviderService;
1249	1317	return yield* provider.startSession(asThreadId("thread-claude-start"), {
1250	1318	provider: ProviderDriverKind.make("claudeAgent"),
1251	1319	providerInstanceId: claudeAgentInstanceId,
...	...	@@ -1256,7 +1324,7 @@ routing.layer("ProviderServiceLive routing", (it) => {
1256	1324	}).pipe(Effect.provide(firstProviderLayer));
1257	1325
1258	1326	yield* Effect.gen(function* () {
1259		const provider = yield* ProviderService;
	1327	const provider = yield* ProviderService.ProviderService;
1260	1328	yield* provider.listSessions();
1261	1329	}).pipe(Effect.provide(firstProviderLayer));
1262	1330
...	...	@@ -1268,17 +1336,24 @@ routing.layer("ProviderServiceLive routing", (it) => {
1268	1336	Layer.provide(runtimeRepositoryLayer),
1269	1337	);
1270	1338	const secondProviderLayer = makeProviderServiceLive().pipe(
1271		Layer.provide(Layer.succeed(ProviderAdapterRegistry, secondRegistry)),
	1339	Layer.provide(
	1340	Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, secondRegistry),
	1341	),
1272	1342	Layer.provide(secondDirectoryLayer),
1273	1343	Layer.provide(defaultServerSettingsLayer),
1274	1344	Layer.provide(AnalyticsService.layerTest),
1275		Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)),
	1345	Layer.provide(
	1346	Layer.succeed(
	1347	ProviderEventLoggers.ProviderEventLoggers,
	1348	ProviderEventLoggers.NoOpProviderEventLoggers,
	1349	),
	1350	),
1276	1351	);
1277	1352
1278	1353	secondClaude.startSession.mockClear();
1279	1354
1280	1355	yield* Effect.gen(function* () {
1281		const provider = yield* ProviderService;
	1356	const provider = yield* ProviderService.ProviderService;
1282	1357	yield* provider.startSession(initial.threadId, {
1283	1358	provider: ProviderDriverKind.make("claudeAgent"),
1284	1359	providerInstanceId: claudeAgentInstanceId,
...	...	@@ -1327,15 +1402,22 @@ routing.layer("ProviderServiceLive routing", (it) => {
1327	1402	Layer.provide(runtimeRepositoryLayer),
1328	1403	);
1329	1404	const firstProviderLayer = makeProviderServiceLive().pipe(
1330		Layer.provide(Layer.succeed(ProviderAdapterRegistry, firstRegistry)),
	1405	Layer.provide(
	1406	Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, firstRegistry),
	1407	),
1331	1408	Layer.provide(firstDirectoryLayer),
1332	1409	Layer.provide(defaultServerSettingsLayer),
1333	1410	Layer.provide(AnalyticsService.layerTest),
1334		Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)),
	1411	Layer.provide(
	1412	Layer.succeed(
	1413	ProviderEventLoggers.ProviderEventLoggers,
	1414	ProviderEventLoggers.NoOpProviderEventLoggers,
	1415	),
	1416	),
1335	1417	);
1336	1418
1337	1419	const initial = yield* Effect.gen(function* () {
1338		const provider = yield* ProviderService;
	1420	const provider = yield* ProviderService.ProviderService;
1339	1421	return yield* provider.startSession(asThreadId("thread-claude-cwd"), {
1340	1422	provider: ProviderDriverKind.make("claudeAgent"),
1341	1423	providerInstanceId: claudeAgentInstanceId,
...	...	@@ -1353,17 +1435,24 @@ routing.layer("ProviderServiceLive routing", (it) => {
1353	1435	Layer.provide(runtimeRepositoryLayer),
1354	1436	);
1355	1437	const secondProviderLayer = makeProviderServiceLive().pipe(
1356		Layer.provide(Layer.succeed(ProviderAdapterRegistry, secondRegistry)),
	1438	Layer.provide(
	1439	Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, secondRegistry),
	1440	),
1357	1441	Layer.provide(secondDirectoryLayer),
1358	1442	Layer.provide(defaultServerSettingsLayer),
1359	1443	Layer.provide(AnalyticsService.layerTest),
1360		Layer.provide(Layer.succeed(ProviderEventLoggers, NoOpProviderEventLoggers)),
	1444	Layer.provide(
	1445	Layer.succeed(
	1446	ProviderEventLoggers.ProviderEventLoggers,
	1447	ProviderEventLoggers.NoOpProviderEventLoggers,
	1448	),
	1449	),
1361	1450	);
1362	1451
1363	1452	secondClaude.startSession.mockClear();
1364	1453
1365	1454	yield* Effect.gen(function* () {
1366		const provider = yield* ProviderService;
	1455	const provider = yield* ProviderService.ProviderService;
1367	1456	yield* provider.startSession(initial.threadId, {
1368	1457	provider: ProviderDriverKind.make("claudeAgent"),
1369	1458	providerInstanceId: claudeAgentInstanceId,
...	...	@@ -1397,7 +1486,7 @@ const fanout = makeProviderServiceLayer();
1397	1486	fanout.layer("ProviderServiceLive fanout", (it) => {
1398	1487	it.effect("fans out adapter turn completion events", () =>
1399	1488	Effect.gen(function* () {
1400		const provider = yield* ProviderService;
	1489	const provider = yield* ProviderService.ProviderService;
1401	1490	const session = yield* provider.startSession(asThreadId("thread-1"), {
1402	1491	provider: ProviderDriverKind.make("codex"),
1403	1492	providerInstanceId: codexInstanceId,
...	...	@@ -1443,7 +1532,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => {
1443	1532
1444	1533	it.effect("fans out canonical runtime events in emission order", () =>
1445	1534	Effect.gen(function* () {
1446		const provider = yield* ProviderService;
	1535	const provider = yield* ProviderService.ProviderService;
1447	1536	const session = yield* provider.startSession(asThreadId("thread-seq"), {
1448	1537	provider: ProviderDriverKind.make("codex"),
1449	1538	providerInstanceId: codexInstanceId,
...	...	@@ -1499,7 +1588,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => {
1499	1588
1500	1589	it.effect("keeps subscriber delivery ordered and isolates failing subscribers", () =>
1501	1590	Effect.gen(function* () {
1502		const provider = yield* ProviderService;
	1591	const provider = yield* ProviderService.ProviderService;
1503	1592	const session = yield* provider.startSession(asThreadId("thread-1"), {
1504	1593	provider: ProviderDriverKind.make("codex"),
1505	1594	providerInstanceId: codexInstanceId,
...	...	@@ -1571,7 +1660,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => {
1571	1660
1572	1661	it.effect("records provider metrics with the routed provider label", () =>
1573	1662	Effect.gen(function* () {
1574		const provider = yield* ProviderService;
	1663	const provider = yield* ProviderService.ProviderService;
1575	1664
1576	1665	const session = yield* provider.startSession(asThreadId("thread-metrics"), {
1577	1666	provider: ProviderDriverKind.make("claudeAgent"),
...	...	@@ -1649,7 +1738,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => {
1649	1738	"records sendTurn metrics with the resolved provider when modelSelection is omitted",
1650	1739	() =>
1651	1740	Effect.gen(function* () {
1652		const provider = yield* ProviderService;
	1741	const provider = yield* ProviderService.ProviderService;
1653	1742
1654	1743	const session = yield* provider.startSession(asThreadId("thread-send-metrics"), {
1655	1744	provider: ProviderDriverKind.make("claudeAgent"),
...	...	@@ -1690,7 +1779,7 @@ const validation = makeProviderServiceLayer();
1690	1779	validation.layer("ProviderServiceLive validation", (it) => {
1691	1780	it.effect("rejects session starts without an explicit provider instance id", () =>
1692	1781	Effect.gen(function* () {
1693		const provider = yield* ProviderService;
	1782	const provider = yield* ProviderService.ProviderService;
1694	1783
1695	1784	validation.codex.startSession.mockClear();
1696	1785	const failure = yield* Effect.flip(
...	...	@@ -1709,7 +1798,7 @@ validation.layer("ProviderServiceLive validation", (it) => {
1709	1798
1710	1799	it.effect("rejects mismatched provider kind and provider instance id", () =>
1711	1800	Effect.gen(function* () {
1712		const provider = yield* ProviderService;
	1801	const provider = yield* ProviderService.ProviderService;
1713	1802
1714	1803	validation.codex.startSession.mockClear();
1715	1804	validation.claude.startSession.mockClear();
...	...	@@ -1734,7 +1823,7 @@ validation.layer("ProviderServiceLive validation", (it) => {
1734	1823
1735	1824	it.effect("returns ProviderValidationError for invalid input payloads", () =>
1736	1825	Effect.gen(function* () {
1737		const provider = yield* ProviderService;
	1826	const provider = yield* ProviderService.ProviderService;
1738	1827
1739	1828	const failure = yield* Effect.result(
1740	1829	provider.startSession(asThreadId("thread-validation"), {
...	...	@@ -1759,7 +1848,7 @@ validation.layer("ProviderServiceLive validation", (it) => {
1759	1848
1760	1849	it.effect("accepts startSession when adapter has not emitted provider thread id yet", () =>
1761	1850	Effect.gen(function* () {
1762		const provider = yield* ProviderService;
	1851	const provider = yield* ProviderService.ProviderService;
1763	1852	const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
1764	1853
1765	1854	validation.codex.startSession.mockImplementationOnce((input: ProviderSessionStartInput) =>

apps/server/src/provider/Layers/ProviderService.ts+44-32

...	...	@@ -47,15 +47,12 @@ import {
47	47	} from "../../observability/Metrics.ts";
48	48	import { type ProviderAdapterError, ProviderValidationError } from "../Errors.ts";
49	49	import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts";
50		import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts";
51		import { ProviderService, type ProviderServiceShape } from "../Services/ProviderService.ts";
52		import {
53		ProviderSessionDirectory,
54		type ProviderRuntimeBinding,
55		} from "../Services/ProviderSessionDirectory.ts";
	50	import * as ProviderAdapterRegistry from "../Services/ProviderAdapterRegistry.ts";
	51	import * as ProviderService from "../Services/ProviderService.ts";
	52	import * as ProviderSessionDirectory from "../Services/ProviderSessionDirectory.ts";
56	53	import { type EventNdjsonLogger } from "./EventNdjsonLogger.ts";
57		import { ProviderEventLoggers } from "./ProviderEventLoggers.ts";
58		import { AnalyticsService } from "../../telemetry/Services/AnalyticsService.ts";
	54	import * as ProviderEventLoggers from "./ProviderEventLoggers.ts";
	55	import * as AnalyticsService from "../../telemetry/AnalyticsService.ts";
59	56	import * as McpProviderSession from "../../mcp/McpProviderSession.ts";
60	57	import * as McpSessionRegistry from "../../mcp/McpSessionRegistry.ts";
61	58	const isModelSelection = Schema.is(ModelSelection);
...	...	@@ -69,6 +66,9 @@ export interface ProviderServiceLiveOptions {
69	66	readonly canonicalEventLogger?: EventNdjsonLogger;
70	67	}
71	68
	69	type ProviderServiceMethod<Name extends keyof ProviderService.ProviderService["Service"]> =
	70	ProviderService.ProviderService["Service"][Name];
	71
72	72	const ProviderRollbackConversationInput = Schema.Struct({
73	73	threadId: ThreadId,
74	74	numTurns: NonNegativeInt,
...	...	@@ -141,7 +141,7 @@ function toRuntimePayloadFromSession(
141	141	}
142	142
143	143	function readPersistedModelSelection(
144		runtimePayload: ProviderRuntimeBinding["runtimePayload"],
	144	runtimePayload: ProviderSessionDirectory.ProviderRuntimeBinding["runtimePayload"],
145	145	): ModelSelection \| undefined {
146	146	if (!runtimePayload \|\| typeof runtimePayload !== "object" \|\| Array.isArray(runtimePayload)) {
147	147	return undefined;
...	...	@@ -151,7 +151,7 @@ function readPersistedModelSelection(
151	151	}
152	152
153	153	function readPersistedCwd(
154		runtimePayload: ProviderRuntimeBinding["runtimePayload"],
	154	runtimePayload: ProviderSessionDirectory.ProviderRuntimeBinding["runtimePayload"],
155	155	): string \| undefined {
156	156	if (!runtimePayload \|\| typeof runtimePayload !== "object" \|\| Array.isArray(runtimePayload)) {
157	157	return undefined;
...	...	@@ -202,16 +202,16 @@ const correlateRuntimeEventWithInstance = (
202	202	const makeProviderService = Effect.fn("makeProviderService")(function* (
203	203	options?: ProviderServiceLiveOptions,
204	204	) {
205		const analytics = yield* Effect.service(AnalyticsService);
206		const eventLoggers = yield* ProviderEventLoggers;
	205	const analytics = yield* Effect.service(AnalyticsService.AnalyticsService);
	206	const eventLoggers = yield* ProviderEventLoggers.ProviderEventLoggers;
207	207	// Options-provided logger wins (test overrides); otherwise we take whatever
208	208	// the `ProviderEventLoggers` tag exposes — `undefined` means "no canonical
209	209	// log writer is attached", which downstream code already handles as a
210	210	// no-op.
211	211	const canonicalEventLogger = options?.canonicalEventLogger ?? eventLoggers.canonical;
212	212
213		const registry = yield* ProviderAdapterRegistry;
214		const directory = yield* ProviderSessionDirectory;
	213	const registry = yield* ProviderAdapterRegistry.ProviderAdapterRegistry;
	214	const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;
215	215	const runtimeEventPubSub = yield* PubSub.unbounded<ProviderRuntimeEvent>();
216	216	const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
217	217	const prepareMcpSession = (threadId: ThreadId, providerInstanceId: ProviderInstanceId) =>
...	...	@@ -353,7 +353,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
353	353	).pipe(Effect.forkScoped);
354	354
355	355	const recoverSessionForThread = Effect.fn("recoverSessionForThread")(function* (input: {
356		readonly binding: ProviderRuntimeBinding;
	356	readonly binding: ProviderSessionDirectory.ProviderRuntimeBinding;
357	357	readonly operation: string;
358	358	}) {
359	359	const bindingInstanceId = yield* requireBindingInstanceId(input.operation, input.binding);
...	...	@@ -519,7 +519,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
519	519	);
520	520	});
521	521
522		const startSession: ProviderServiceShape["startSession"] = Effect.fn("startSession")(
	522	const startSession: ProviderServiceMethod<"startSession"> = Effect.fn("startSession")(
523	523	function* (threadId, rawInput) {
524	524	const parsed = yield* decodeInputOrValidationError({
525	525	operation: "ProviderService.startSession",
...	...	@@ -642,7 +642,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
642	642	},
643	643	);
644	644
645		const sendTurn: ProviderServiceShape["sendTurn"] = Effect.fn("sendTurn")(function* (rawInput) {
	645	const sendTurn: ProviderServiceMethod<"sendTurn"> = Effect.fn("sendTurn")(function* (rawInput) {
646	646	const parsed = yield* decodeInputOrValidationError({
647	647	operation: "ProviderService.sendTurn",
648	648	schema: ProviderSendTurnInput,
...	...	@@ -717,7 +717,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
717	717	);
718	718	});
719	719
720		const interruptTurn: ProviderServiceShape["interruptTurn"] = Effect.fn("interruptTurn")(
	720	const interruptTurn: ProviderServiceMethod<"interruptTurn"> = Effect.fn("interruptTurn")(
721	721	function* (rawInput) {
722	722	const input = yield* decodeInputOrValidationError({
723	723	operation: "ProviderService.interruptTurn",
...	...	@@ -754,7 +754,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
754	754	},
755	755	);
756	756
757		const respondToRequest: ProviderServiceShape["respondToRequest"] = Effect.fn("respondToRequest")(
	757	const respondToRequest: ProviderServiceMethod<"respondToRequest"> = Effect.fn("respondToRequest")(
758	758	function* (rawInput) {
759	759	const input = yield* decodeInputOrValidationError({
760	760	operation: "ProviderService.respondToRequest",
...	...	@@ -792,7 +792,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
792	792	},
793	793	);
794	794
795		const respondToUserInput: ProviderServiceShape["respondToUserInput"] = Effect.fn(
	795	const respondToUserInput: ProviderServiceMethod<"respondToUserInput"> = Effect.fn(
796	796	"respondToUserInput",
797	797	)(function* (rawInput) {
798	798	const input = yield* decodeInputOrValidationError({
...	...	@@ -826,7 +826,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
826	826	);
827	827	});
828	828
829		const stopSession: ProviderServiceShape["stopSession"] = Effect.fn("stopSession")(
	829	const stopSession: ProviderServiceMethod<"stopSession"> = Effect.fn("stopSession")(
830	830	function* (rawInput) {
831	831	const input = yield* decodeInputOrValidationError({
832	832	operation: "ProviderService.stopSession",
...	...	@@ -874,7 +874,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
874	874	},
875	875	);
876	876
877		const listSessions: ProviderServiceShape["listSessions"] = Effect.fn("listSessions")(
	877	const listSessions: ProviderServiceMethod<"listSessions"> = Effect.fn("listSessions")(
878	878	function* () {
879	879	const currentAdapters = yield* getAdapterEntries;
880	880	const sessionsByProvider = yield* Effect.forEach(currentAdapters, ([instanceId, adapter]) =>
...	...	@@ -895,13 +895,22 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
895	895	(threadId) =>
896	896	directory
897	897	.getBinding(threadId)
898		.pipe(Effect.orElseSucceed(() => Option.none<ProviderRuntimeBinding>())),
	898	.pipe(
	899	Effect.orElseSucceed(() =>
	900	Option.none<ProviderSessionDirectory.ProviderRuntimeBinding>(),
	901	),
	902	),
899	903	{ concurrency: "unbounded" },
900	904	),
901	905	),
902		Effect.orElseSucceed(() => [] as Array<Option.Option<ProviderRuntimeBinding>>),
	906	Effect.orElseSucceed(
	907	() => [] as Array<Option.Option<ProviderSessionDirectory.ProviderRuntimeBinding>>,
	908	),
903	909	);
904		const bindingsByThreadId = new Map<ThreadId, ProviderRuntimeBinding>();
	910	const bindingsByThreadId = new Map<
	911	ThreadId,
	912	ProviderSessionDirectory.ProviderRuntimeBinding
	913	>();
905	914	for (const bindingOption of persistedBindings) {
906	915	const binding = Option.getOrUndefined(bindingOption);
907	916	if (binding) {
...	...	@@ -952,13 +961,13 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
952	961	},
953	962	);
954	963
955		const getCapabilities: ProviderServiceShape["getCapabilities"] = (instanceId) =>
	964	const getCapabilities: ProviderServiceMethod<"getCapabilities"> = (instanceId) =>
956	965	registry.getByInstance(instanceId).pipe(Effect.map((adapter) => adapter.capabilities));
957	966
958		const getInstanceInfo: ProviderServiceShape["getInstanceInfo"] = (instanceId) =>
	967	const getInstanceInfo: ProviderServiceMethod<"getInstanceInfo"> = (instanceId) =>
959	968	registry.getInstanceInfo(instanceId);
960	969
961		const rollbackConversation: ProviderServiceShape["rollbackConversation"] = Effect.fn(
	970	const rollbackConversation: ProviderServiceMethod<"rollbackConversation"> = Effect.fn(
962	971	"rollbackConversation",
963	972	)(function* (rawInput) {
964	973	const input = yield* decodeInputOrValidationError({
...	...	@@ -1071,14 +1080,17 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
1071	1080	// Each access creates a fresh PubSub subscription so that multiple
1072	1081	// consumers (ProviderRuntimeIngestion, CheckpointReactor, etc.) each
1073	1082	// independently receive all runtime events.
1074		get streamEvents(): ProviderServiceShape["streamEvents"] {
	1083	get streamEvents(): ProviderServiceMethod<"streamEvents"> {
1075	1084	return Stream.fromPubSub(runtimeEventPubSub);
1076	1085	},
1077		} satisfies ProviderServiceShape;
	1086	} satisfies ProviderService.ProviderService["Service"];
1078	1087	});
1079	1088
1080		export const ProviderServiceLive = Layer.effect(ProviderService, makeProviderService());
	1089	export const ProviderServiceLive = Layer.effect(
	1090	ProviderService.ProviderService,
	1091	makeProviderService(),
	1092	);
1081	1093
1082	1094	export function makeProviderServiceLive(options?: ProviderServiceLiveOptions) {
1083		return Layer.effect(ProviderService, makeProviderService(options));
	1095	return Layer.effect(ProviderService.ProviderService, makeProviderService(options));
1084	1096	}

apps/server/src/server.test.ts+1-1

...	...	@@ -88,7 +88,7 @@ import * as ServerSettings from "./serverSettings.ts";
88	88	import * as TerminalManager from "./terminal/Services/Manager.ts";
89	89	import * as PreviewManager from "./preview/Manager.ts";
90	90	import * as PortScanner from "./preview/PortScanner.ts";
91		import * as BrowserTraceCollector from "./observability/Services/BrowserTraceCollector.ts";
	91	import * as BrowserTraceCollector from "./observability/BrowserTraceCollector.ts";
92	92	import { ProjectFaviconResolverLive } from "./project/Layers/ProjectFaviconResolver.ts";
93	93	import * as ProjectSetupScriptRunner from "./project/Services/ProjectSetupScriptRunner.ts";
94	94	import * as RepositoryIdentityResolver from "./project/Services/RepositoryIdentityResolver.ts";

apps/server/src/server.ts+2-2

...	...	@@ -17,7 +17,7 @@ import { websocketRpcRouteLayer } from "./ws.ts";
17	17	import * as ExternalLauncher from "./process/externalLauncher.ts";
18	18	import { layerConfig as SqlitePersistenceLayerLive } from "./persistence/Layers/Sqlite.ts";
19	19	import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts";
20		import { AnalyticsServiceLayerLive } from "./telemetry/Layers/AnalyticsService.ts";
	20	import * as AnalyticsService from "./telemetry/AnalyticsService.ts";
21	21	import { ProviderSessionDirectoryLive } from "./provider/Layers/ProviderSessionDirectory.ts";
22	22	import * as ProviderSessionRuntime from "./persistence/ProviderSessionRuntime.ts";
23	23	import { ProviderAdapterRegistryLive } from "./provider/Layers/ProviderAdapterRegistry.ts";
...	...	@@ -332,7 +332,7 @@ const RuntimeDependenciesLive = RuntimeCoreDependenciesLive.pipe(
332	332	Layer.provideMerge(ProcessDiagnostics.layer),
333	333	Layer.provideMerge(ProcessResourceMonitor.layer),
334	334	Layer.provideMerge(TraceDiagnostics.layer),
335		Layer.provideMerge(AnalyticsServiceLayerLive),
	335	Layer.provideMerge(AnalyticsService.layer),
336	336	Layer.provideMerge(ExternalLauncher.layer),
337	337	Layer.provideMerge(ServerLifecycleEvents.layer),
338	338	Layer.provide(NetService.layer),

apps/server/src/serverRuntimeStartup.test.ts+1-1

...	...	@@ -13,7 +13,7 @@ import * as Stream from "effect/Stream";
13	13	import * as ServerConfig from "./config.ts";
14	14	import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts";
15	15	import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts";
16		import * as AnalyticsService from "./telemetry/Services/AnalyticsService.ts";
	16	import * as AnalyticsService from "./telemetry/AnalyticsService.ts";
17	17	import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts";
18	18
19	19	it("uses the canonical Codex default for auto-bootstrapped model selection", () => {

apps/server/src/serverRuntimeStartup.ts+1-1

...	...	@@ -31,7 +31,7 @@ import * as OrchestrationReactor from "./orchestration/Services/OrchestrationRea
31	31	import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts";
32	32	import * as ServerSettings from "./serverSettings.ts";
33	33	import * as ServerEnvironment from "./environment/Services/ServerEnvironment.ts";
34		import * as AnalyticsService from "./telemetry/Services/AnalyticsService.ts";
	34	import * as AnalyticsService from "./telemetry/AnalyticsService.ts";
35	35	import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts";
36	36	import * as ProviderSessionReaper from "./provider/Services/ProviderSessionReaper.ts";
37	37	import {

apps/server/src/telemetry/AnalyticsService.test.ts created+120

...	...	@@ -0,0 +1,120 @@
	1	import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";
	2	import * as NodeServices from "@effect/platform-node/NodeServices";
	3	import { assert, it } from "@effect/vitest";
	4	import * as ConfigProvider from "effect/ConfigProvider";
	5	import * as Effect from "effect/Effect";
	6	import * as Layer from "effect/Layer";
	7	import * as HttpServer from "effect/unstable/http/HttpServer";
	8	import * as HttpServerRequest from "effect/unstable/http/HttpServerRequest";
	9	import * as HttpServerResponse from "effect/unstable/http/HttpServerResponse";
	10
	11	import * as ServerConfig from "../config.ts";
	12	import { getTelemetryIdentifier } from "./Identify.ts";
	13	import * as AnalyticsService from "./AnalyticsService.ts";
	14
	15	interface RecordedBatchRequest {
	16	readonly path: string;
	17	readonly body: {
	18	readonly batch?: ReadonlyArray<{
	19	readonly event?: string;
	20	readonly properties?: {
	21	readonly index?: number;
	22	readonly clientType?: string;
	23	};
	24	}>;
	25	} \| null;
	26	}
	27
	28	interface RecordedBatchBody {
	29	readonly batch: ReadonlyArray<{
	30	readonly event?: string;
	31	readonly properties?: {
	32	readonly index?: number;
	33	readonly clientType?: string;
	34	};
	35	}>;
	36	}
	37
	38	it.layer(NodeServices.layer)("AnalyticsService test", (it) => {
	39	it.effect("flush drains all buffered events across multiple batches", () =>
	40	Effect.gen(function* () {
	41	const capturedRequests: Array<RecordedBatchRequest> = [];
	42	const serverConfigLayer = ServerConfig.ServerConfig.layerTest(process.cwd(), {
	43	prefix: "t3-telemetry-base-",
	44	});
	45
	46	const telemetryLayer = AnalyticsService.layer.pipe(Layer.provideMerge(serverConfigLayer));
	47	const configLayer = ConfigProvider.layer(
	48	ConfigProvider.fromUnknown({
	49	T3CODE_TELEMETRY_ENABLED: true,
	50	T3CODE_POSTHOG_KEY: "phc_test_key",
	51	T3CODE_POSTHOG_HOST: "",
	52	T3CODE_TELEMETRY_FLUSH_BATCH_SIZE: 20,
	53	}),
	54	);
	55	const batchServerLayer = HttpServer.serve(
	56	Effect.gen(function* () {
	57	const request = yield* HttpServerRequest.HttpServerRequest;
	58	if (request.method !== "POST") {
	59	return HttpServerResponse.empty({ status: 404 });
	60	}
	61
	62	const payload = yield* request.json.pipe(
	63	Effect.map((body) => body as RecordedBatchRequest["body"]),
	64	Effect.orElseSucceed(() => null),
	65	);
	66
	67	capturedRequests.push({ path: request.url, body: payload });
	68
	69	return HttpServerResponse.jsonUnsafe({});
	70	}),
	71	);
	72	const runtimeLayer = telemetryLayer.pipe(
	73	Layer.provide(configLayer),
	74	Layer.provideMerge(NodeHttpServer.layerTest),
	75	);
	76
	77	yield* Effect.gen(function* () {
	78	yield* Layer.launch(batchServerLayer).pipe(Effect.forkScoped);
	79	const telemetryIdentifier = yield* getTelemetryIdentifier;
	80	assert.equal(telemetryIdentifier !== null, true);
	81	const analytics = yield* AnalyticsService.AnalyticsService;
	82
	83	for (let index = 0; index < 45; index += 1) {
	84	yield* analytics.record("test.flush.drain", { index });
	85	}
	86
	87	yield* analytics.flush;
	88	}).pipe(Effect.provide(runtimeLayer));
	89
	90	const batchRequests = capturedRequests.filter(
	91	(request): request is RecordedBatchRequest & { readonly body: RecordedBatchBody } =>
	92	Array.isArray(request.body?.batch),
	93	);
	94	assert.equal(batchRequests.length, 3);
	95	assert.equal(
	96	batchRequests.every((request) => request.path === "/batch/" \|\| request.path === "/batch"),
	97	true,
	98	);
	99	const deliveredIndexes = batchRequests.flatMap((request) =>
	100	request.body.batch
	101	.filter((event) => event.event === "test.flush.drain")
	102	.map((event) => event.properties?.index)
	103	.filter((index): index is number => typeof index === "number"),
	104	);
	105
	106	const sorted = deliveredIndexes.toSorted((a, b) => a - b);
	107	assert.equal(sorted.length, 45);
	108	assert.deepEqual(
	109	sorted,
	110	Array.from({ length: 45 }, (_, index) => index),
	111	);
	112	assert.equal(
	113	batchRequests.every((request) =>
	114	request.body.batch.every((event) => event.properties?.clientType === "cli-web-client"),
	115	),
	116	true,
	117	);
	118	}),
	119	);
	120	});

apps/server/src/telemetry/AnalyticsService.ts created+186

...	...	@@ -0,0 +1,186 @@
	1	/**
	2	* Anonymous PostHog telemetry service.
	3	*
	4	* Persists an installation-scoped anonymous identifier, buffers events in
	5	* memory, and flushes batches over Effect's HTTP client.
	6	*
	7	* @module AnalyticsService
	8	*/
	9	import { HostProcessArchitecture, HostProcessPlatform } from "@t3tools/shared/hostProcess";
	10	import * as Config from "effect/Config";
	11	import * as Context from "effect/Context";
	12	import * as DateTime from "effect/DateTime";
	13	import * as Effect from "effect/Effect";
	14	import * as Layer from "effect/Layer";
	15	import * as Option from "effect/Option";
	16	import * as Ref from "effect/Ref";
	17	import * as HttpClient from "effect/unstable/http/HttpClient";
	18	import * as HttpClientRequest from "effect/unstable/http/HttpClientRequest";
	19	import * as HttpClientResponse from "effect/unstable/http/HttpClientResponse";
	20
	21	import packageJson from "../../package.json" with { type: "json" };
	22	import * as ServerConfig from "../config.ts";
	23	import { getTelemetryIdentifier } from "./Identify.ts";
	24
	25	interface BufferedAnalyticsEvent {
	26	readonly event: string;
	27	readonly properties?: Readonly<Record<string, unknown>>;
	28	readonly capturedAt: string;
	29	}
	30
	31	const TelemetryEnvConfig = Config.all({
	32	posthogKey: Config.string("T3CODE_POSTHOG_KEY").pipe(
	33	Config.withDefault("phc_XOWci4oZP4VvLiEyrFqkFjP4CZn55mjYYBMREK5Wd6m"),
	34	),
	35	posthogHost: Config.string("T3CODE_POSTHOG_HOST").pipe(
	36	Config.withDefault("https://us.i.posthog.com"),
	37	),
	38	enabled: Config.boolean("T3CODE_TELEMETRY_ENABLED").pipe(Config.withDefault(true)),
	39	flushBatchSize: Config.number("T3CODE_TELEMETRY_FLUSH_BATCH_SIZE").pipe(Config.withDefault(20)),
	40	maxBufferedEvents: Config.number("T3CODE_TELEMETRY_MAX_BUFFERED_EVENTS").pipe(
	41	Config.withDefault(1_000),
	42	),
	43	wslDistroName: Config.string("WSL_DISTRO_NAME").pipe(Config.option),
	44	});
	45
	46	export class AnalyticsService extends Context.Service<
	47	AnalyticsService,
	48	{
	49	/** Record an anonymous event for best-effort buffered delivery. */
	50	readonly record: (
	51	event: string,
	52	properties?: Readonly<Record<string, unknown>>,
	53	) => Effect.Effect<void>;
	54
	55	/** Flush all currently queued telemetry events. */
	56	readonly flush: Effect.Effect<void>;
	57	}
	58	>()("t3/telemetry/AnalyticsService") {
	59	/** No-op layer for callers that intentionally disable telemetry. */
	60	static readonly layerTest = Layer.succeed(
	61	AnalyticsService,
	62	AnalyticsService.of({
	63	record: () => Effect.void,
	64	flush: Effect.void,
	65	}),
	66	);
	67	}
	68
	69	export const make = Effect.gen(function* () {
	70	const telemetryConfig = yield* TelemetryEnvConfig;
	71	const httpClient = yield* HttpClient.HttpClient;
	72	const serverConfig = yield* ServerConfig.ServerConfig;
	73	const identifier = yield* getTelemetryIdentifier;
	74	const bufferRef = yield* Ref.make<ReadonlyArray<BufferedAnalyticsEvent>>([]);
	75	const clientType = serverConfig.mode === "desktop" ? "desktop-app" : "cli-web-client";
	76	const hostPlatform = yield* HostProcessPlatform;
	77	const hostArchitecture = yield* HostProcessArchitecture;
	78
	79	const enqueueBufferedEvent = (event: string, properties?: Readonly<Record<string, unknown>>) =>
	80	Effect.flatMap(DateTime.now, (now) =>
	81	Ref.modify(bufferRef, (current) => {
	82	const appended = [
	83	...current,
	84	{
	85	event,
	86	...(properties ? { properties } : {}),
	87	capturedAt: DateTime.formatIso(now),
	88	} satisfies BufferedAnalyticsEvent,
	89	];
	90
	91	const next =
	92	appended.length > telemetryConfig.maxBufferedEvents
	93	? appended.slice(appended.length - telemetryConfig.maxBufferedEvents)
	94	: appended;
	95
	96	return [
	97	{
	98	size: next.length,
	99	dropped: next.length !== appended.length,
	100	} as const,
	101	next,
	102	] as const;
	103	}),
	104	);
	105
	106	const sendBatch = Effect.fn("AnalyticsService.sendBatch")(function* (
	107	events: ReadonlyArray<BufferedAnalyticsEvent>,
	108	) {
	109	if (!telemetryConfig.enabled \|\| !identifier) return;
	110
	111	const payload = {
	112	api_key: telemetryConfig.posthogKey,
	113	batch: events.map((event) => ({
	114	event: event.event,
	115	distinct_id: identifier,
	116	properties: {
	117	...event.properties,
	118	$process_person_profile: false,
	119	platform: hostPlatform,
	120	wsl: Option.getOrUndefined(telemetryConfig.wslDistroName),
	121	arch: hostArchitecture,
	122	t3CodeVersion: packageJson.version,
	123	clientType,
	124	},
	125	timestamp: event.capturedAt,
	126	})),
	127	};
	128
	129	yield* HttpClientRequest.post(`${telemetryConfig.posthogHost}/batch/`).pipe(
	130	HttpClientRequest.bodyJson(payload),
	131	Effect.flatMap(httpClient.execute),
	132	Effect.flatMap(HttpClientResponse.filterStatusOk),
	133	);
	134	});
	135
	136	const flush: AnalyticsService["Service"]["flush"] = Effect.gen(function* () {
	137	while (true) {
	138	const batch = yield* Ref.modify(bufferRef, (current) => {
	139	if (current.length === 0) {
	140	return [[] as ReadonlyArray<BufferedAnalyticsEvent>, current] as const;
	141	}
	142	const nextBatch = current.slice(0, telemetryConfig.flushBatchSize);
	143	const remaining = current.slice(nextBatch.length);
	144	return [nextBatch, remaining] as const;
	145	});
	146
	147	if (batch.length === 0) {
	148	return;
	149	}
	150
	151	yield* sendBatch(batch).pipe(
	152	Effect.catch((error) =>
	153	Ref.update(bufferRef, (current) => [...batch, ...current]).pipe(
	154	Effect.flatMap(() => Effect.fail(error)),
	155	),
	156	),
	157	);
	158	}
	159	}).pipe(Effect.catch((cause) => Effect.logError("Failed to flush telemetry", { cause })));
	160
	161	const record: AnalyticsService["Service"]["record"] = Effect.fn("AnalyticsService.record")(
	162	function* (event, properties) {
	163	if (!telemetryConfig.enabled \|\| !identifier) return;
	164
	165	const enqueueResult = yield* enqueueBufferedEvent(event, properties);
	166	if (enqueueResult.dropped) {
	167	yield* Effect.logDebug("analytics buffer full; dropping oldest event", {
	168	size: enqueueResult.size,
	169	event,
	170	});
	171	}
	172	},
	173	);
	174
	175	yield* Effect.forever(Effect.sleep(1000).pipe(Effect.flatMap(() => flush)), {
	176	disableYield: true,
	177	}).pipe(Effect.forkScoped);
	178
	179	yield* Effect.addFinalizer(() => flush);
	180
	181	return AnalyticsService.of({ record, flush });
	182	});
	183
	184	export const layer = Layer.effect(AnalyticsService, make);
	185
	186	export const layerTest = AnalyticsService.layerTest;

apps/server/src/telemetry/Identify.ts+12-6

...	...	@@ -6,7 +6,7 @@ import * as FileSystem from "effect/FileSystem";
6	6	import * as Path from "effect/Path";
7	7	import * as Schema from "effect/Schema";
8	8
9		import { ServerConfig } from "../config.ts";
	9	import * as ServerConfig from "../config.ts";
10	10
11	11	const CodexAuthJsonSchema = Schema.Struct({
12	12	tokens: Schema.Struct({
...	...	@@ -19,9 +19,14 @@ const ClaudeJsonSchema = Schema.Struct({
19	19	});
20	20
21	21	class IdentifyUserError extends Schema.TaggedErrorClass<IdentifyUserError>()("IdentifyUserError", {
22		message: Schema.String,
23		cause: Schema.optional(Schema.Defect()),
24		}) {}
	22	operation: Schema.Literal("hash_identifier"),
	23	algorithm: Schema.Literal("SHA-256"),
	24	cause: Schema.Defect(),
	25	}) {
	26	override get message(): string {
	27	return `Failed to hash telemetry identifier with ${this.algorithm}.`;
	28	}
	29	}
25	30
26	31	const hash = (value: string) =>
27	32	Crypto.Crypto.pipe(
...	...	@@ -30,7 +35,8 @@ const hash = (value: string) =>
30	35	Effect.mapError(
31	36	(cause) =>
32	37	new IdentifyUserError({
33		message: "Failed to hash identifier",
	38	operation: "hash_identifier",
	39	algorithm: "SHA-256",
34	40	cause,
35	41	}),
36	42	),
...	...	@@ -64,7 +70,7 @@ const getClaudeUserId = Effect.gen(function* () {
64	70
65	71	const upsertAnonymousId = Effect.gen(function* () {
66	72	const fileSystem = yield* FileSystem.FileSystem;
67		const { anonymousIdPath } = yield* ServerConfig;
	73	const { anonymousIdPath } = yield* ServerConfig.ServerConfig;
68	74
69	75	const anonymousId = yield* fileSystem.readFileString(anonymousIdPath).pipe(
70	76	Effect.catch(() =>

apps/server/src/telemetry/Layers/AnalyticsService.test.ts deleted-121

...	...	@@ -1,121 +0,0 @@
1		import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";
2		import * as NodeServices from "@effect/platform-node/NodeServices";
3		import { assert, it } from "@effect/vitest";
4		import * as ConfigProvider from "effect/ConfigProvider";
5		import * as Effect from "effect/Effect";
6		import * as Layer from "effect/Layer";
7		import * as HttpServer from "effect/unstable/http/HttpServer";
8		import * as HttpServerRequest from "effect/unstable/http/HttpServerRequest";
9		import * as HttpServerResponse from "effect/unstable/http/HttpServerResponse";
10
11		import { ServerConfig } from "../../config.ts";
12		import { getTelemetryIdentifier } from "../Identify.ts";
13		import { AnalyticsService } from "../Services/AnalyticsService.ts";
14		import { AnalyticsServiceLayerLive } from "./AnalyticsService.ts";
15
16		interface RecordedBatchRequest {
17		readonly path: string;
18		readonly body: {
19		readonly batch?: ReadonlyArray<{
20		readonly event?: string;
21		readonly properties?: {
22		readonly index?: number;
23		readonly clientType?: string;
24		};
25		}>;
26		} \| null;
27		}
28
29		interface RecordedBatchBody {
30		readonly batch: ReadonlyArray<{
31		readonly event?: string;
32		readonly properties?: {
33		readonly index?: number;
34		readonly clientType?: string;
35		};
36		}>;
37		}
38
39		it.layer(NodeServices.layer)("AnalyticsService test", (it) => {
40		it.effect("flush drains all buffered events across multiple batches", () =>
41		Effect.gen(function* () {
42		const capturedRequests: Array<RecordedBatchRequest> = [];
43		const serverConfigLayer = ServerConfig.layerTest(process.cwd(), {
44		prefix: "t3-telemetry-base-",
45		});
46
47		const telemetryLayer = AnalyticsServiceLayerLive.pipe(Layer.provideMerge(serverConfigLayer));
48		const configLayer = ConfigProvider.layer(
49		ConfigProvider.fromUnknown({
50		T3CODE_TELEMETRY_ENABLED: true,
51		T3CODE_POSTHOG_KEY: "phc_test_key",
52		T3CODE_POSTHOG_HOST: "",
53		T3CODE_TELEMETRY_FLUSH_BATCH_SIZE: 20,
54		}),
55		);
56		const batchServerLayer = HttpServer.serve(
57		Effect.gen(function* () {
58		const request = yield* HttpServerRequest.HttpServerRequest;
59		if (request.method !== "POST") {
60		return HttpServerResponse.empty({ status: 404 });
61		}
62
63		const payload = yield* request.json.pipe(
64		Effect.map((body) => body as RecordedBatchRequest["body"]),
65		Effect.orElseSucceed(() => null),
66		);
67
68		capturedRequests.push({ path: request.url, body: payload });
69
70		return HttpServerResponse.jsonUnsafe({});
71		}),
72		);
73		const runtimeLayer = telemetryLayer.pipe(
74		Layer.provide(configLayer),
75		Layer.provideMerge(NodeHttpServer.layerTest),
76		);
77
78		yield* Effect.gen(function* () {
79		yield* Layer.launch(batchServerLayer).pipe(Effect.forkScoped);
80		const telemetryIdentifier = yield* getTelemetryIdentifier;
81		assert.equal(telemetryIdentifier !== null, true);
82		const analytics = yield* AnalyticsService;
83
84		for (let index = 0; index < 45; index += 1) {
85		yield* analytics.record("test.flush.drain", { index });
86		}
87
88		yield* analytics.flush;
89		}).pipe(Effect.provide(runtimeLayer));
90
91		const batchRequests = capturedRequests.filter(
92		(request): request is RecordedBatchRequest & { readonly body: RecordedBatchBody } =>
93		Array.isArray(request.body?.batch),
94		);
95		assert.equal(batchRequests.length, 3);
96		assert.equal(
97		batchRequests.every((request) => request.path === "/batch/" \|\| request.path === "/batch"),
98		true,
99		);
100		const deliveredIndexes = batchRequests.flatMap((request) =>
101		request.body.batch
102		.filter((event) => event.event === "test.flush.drain")
103		.map((event) => event.properties?.index)
104		.filter((index): index is number => typeof index === "number"),
105		);
106
107		const sorted = deliveredIndexes.toSorted((a, b) => a - b);
108		assert.equal(sorted.length, 45);
109		assert.deepEqual(
110		sorted,
111		Array.from({ length: 45 }, (_, index) => index),
112		);
113		assert.equal(
114		batchRequests.every((request) =>
115		request.body.batch.every((event) => event.properties?.clientType === "cli-web-client"),
116		),
117		true,
118		);
119		}),
120		);
121		});

apps/server/src/telemetry/Layers/AnalyticsService.ts deleted-163

...	...	@@ -1,163 +0,0 @@
1		/**
2		* AnalyticsServiceLive - Anonymous PostHog telemetry layer.
3		*
4		* Persists a random installation-scoped anonymous id to state dir, buffers
5		* events in memory, and flushes batches to PostHog over Effect HttpClient.
6		*
7		* @module AnalyticsServiceLive
8		*/
9
10		import { HostProcessArchitecture, HostProcessPlatform } from "@t3tools/shared/hostProcess";
11		import * as Config from "effect/Config";
12		import * as DateTime from "effect/DateTime";
13		import * as Effect from "effect/Effect";
14		import * as Layer from "effect/Layer";
15		import * as Option from "effect/Option";
16		import * as Ref from "effect/Ref";
17		import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http";
18
19		import { ServerConfig } from "../../config.ts";
20		import { AnalyticsService, type AnalyticsServiceShape } from "../Services/AnalyticsService.ts";
21		import { getTelemetryIdentifier } from "../Identify.ts";
22		import packageJson from "../../../package.json" with { type: "json" };
23
24		interface BufferedAnalyticsEvent {
25		readonly event: string;
26		readonly properties?: Readonly<Record<string, unknown>>;
27		readonly capturedAt: string;
28		}
29
30		const TelemetryEnvConfig = Config.all({
31		posthogKey: Config.string("T3CODE_POSTHOG_KEY").pipe(
32		Config.withDefault("phc_XOWci4oZP4VvLiEyrFqkFjP4CZn55mjYYBMREK5Wd6m"),
33		),
34		posthogHost: Config.string("T3CODE_POSTHOG_HOST").pipe(
35		Config.withDefault("https://us.i.posthog.com"),
36		),
37		enabled: Config.boolean("T3CODE_TELEMETRY_ENABLED").pipe(Config.withDefault(true)),
38		flushBatchSize: Config.number("T3CODE_TELEMETRY_FLUSH_BATCH_SIZE").pipe(Config.withDefault(20)),
39		maxBufferedEvents: Config.number("T3CODE_TELEMETRY_MAX_BUFFERED_EVENTS").pipe(
40		Config.withDefault(1_000),
41		),
42		wslDistroName: Config.string("WSL_DISTRO_NAME").pipe(Config.option),
43		});
44
45		const makeAnalyticsService = Effect.gen(function* () {
46		const telemetryConfig = yield* TelemetryEnvConfig;
47		const httpClient = yield* HttpClient.HttpClient;
48		const serverConfig = yield* ServerConfig;
49		const identifier = yield* getTelemetryIdentifier;
50		const bufferRef = yield* Ref.make<ReadonlyArray<BufferedAnalyticsEvent>>([]);
51		const clientType = serverConfig.mode === "desktop" ? "desktop-app" : "cli-web-client";
52		const hostPlatform = yield* HostProcessPlatform;
53		const hostArchitecture = yield* HostProcessArchitecture;
54
55		const enqueueBufferedEvent = (event: string, properties?: Readonly<Record<string, unknown>>) =>
56		Effect.flatMap(DateTime.now, (now) =>
57		Ref.modify(bufferRef, (current) => {
58		const appended = [
59		...current,
60		{
61		event,
62		...(properties ? { properties } : {}),
63		capturedAt: DateTime.formatIso(now),
64		} satisfies BufferedAnalyticsEvent,
65		];
66
67		const next =
68		appended.length > telemetryConfig.maxBufferedEvents
69		? appended.slice(appended.length - telemetryConfig.maxBufferedEvents)
70		: appended;
71
72		return [
73		{
74		size: next.length,
75		dropped: next.length !== appended.length,
76		} as const,
77		next,
78		] as const;
79		}),
80		);
81
82		const sendBatch = Effect.fn("sendBatch")(function* (
83		events: ReadonlyArray<BufferedAnalyticsEvent>,
84		) {
85		if (!telemetryConfig.enabled \|\| !identifier) return;
86
87		const payload = {
88		api_key: telemetryConfig.posthogKey,
89		batch: events.map((event) => ({
90		event: event.event,
91		distinct_id: identifier,
92		properties: {
93		...event.properties,
94		$process_person_profile: false,
95		platform: hostPlatform,
96		wsl: Option.getOrUndefined(telemetryConfig.wslDistroName),
97		arch: hostArchitecture,
98		t3CodeVersion: packageJson.version,
99		clientType,
100		},
101		timestamp: event.capturedAt,
102		})),
103		};
104
105		yield* HttpClientRequest.post(`${telemetryConfig.posthogHost}/batch/`).pipe(
106		HttpClientRequest.bodyJson(payload),
107		Effect.flatMap(httpClient.execute),
108		Effect.flatMap(HttpClientResponse.filterStatusOk),
109		);
110		});
111
112		const flush: AnalyticsServiceShape["flush"] = Effect.gen(function* () {
113		while (true) {
114		const batch = yield* Ref.modify(bufferRef, (current) => {
115		if (current.length === 0) {
116		return [[] as ReadonlyArray<BufferedAnalyticsEvent>, current] as const;
117		}
118		const nextBatch = current.slice(0, telemetryConfig.flushBatchSize);
119		const remaining = current.slice(nextBatch.length);
120		return [nextBatch, remaining] as const;
121		});
122
123		if (batch.length === 0) {
124		return;
125		}
126
127		yield* sendBatch(batch).pipe(
128		Effect.catch((error) =>
129		Ref.update(bufferRef, (current) => [...batch, ...current]).pipe(
130		Effect.flatMap(() => Effect.fail(error)),
131		),
132		),
133		);
134		}
135		}).pipe(Effect.catch((cause) => Effect.logError("Failed to flush telemetry", { cause })));
136
137		const record: AnalyticsServiceShape["record"] = Effect.fn("record")(
138		function* (event, properties) {
139		if (!telemetryConfig.enabled \|\| !identifier) return;
140
141		const enqueueResult = yield* enqueueBufferedEvent(event, properties);
142		if (enqueueResult.dropped) {
143		yield* Effect.logDebug("analytics buffer full; dropping oldest event", {
144		size: enqueueResult.size,
145		event,
146		});
147		}
148		},
149		);
150
151		yield* Effect.forever(Effect.sleep(1000).pipe(Effect.flatMap(() => flush)), {
152		disableYield: true,
153		}).pipe(Effect.forkScoped);
154
155		yield* Effect.addFinalizer(() => flush);
156
157		return {
158		record,
159		flush,
160		} satisfies AnalyticsServiceShape;
161		});
162
163		export const AnalyticsServiceLayerLive = Layer.effect(AnalyticsService, makeAnalyticsService);

apps/server/src/telemetry/Services/AnalyticsService.ts+2-35

...	...	@@ -1,35 +1,2 @@
1		/**
2		* AnalyticsService - Anonymous telemetry capture contract.
3		*
4		* Provides a best-effort event API for runtime telemetry and a strict
5		* `captureImmediate` method for call sites that need explicit error handling.
6		*
7		* @module AnalyticsService
8		*/
9		import * as Effect from "effect/Effect";
10		import * as Layer from "effect/Layer";
11		import * as Context from "effect/Context";
12
13		export interface AnalyticsServiceShape {
14		/**
15		* Capture an event immediately; returns typed failure when capture fails.
16		*/
17		readonly record: (
18		event: string,
19		properties?: Readonly<Record<string, unknown>>,
20		) => Effect.Effect<void, never>;
21
22		/**
23		* Flush queued telemetry.
24		*/
25		readonly flush: Effect.Effect<void, never>;
26		}
27
28		export class AnalyticsService extends Context.Service<AnalyticsService, AnalyticsServiceShape>()(
29		"t3/telemetry/Services/AnalyticsService",
30		) {
31		static readonly layerTest = Layer.succeed(AnalyticsService, {
32		record: () => Effect.void,
33		flush: Effect.void,
34		});
35		}
	1	// Compatibility shim for the intentionally excluded orchestration harness.
	2	export { AnalyticsService } from "../AnalyticsService.ts";