Skip to content

[codex] Align diagnostics and telemetry Effect services #3189

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
juliusmarminge merged 1 commit into from
Jun 20, 2026
Merged

[codex] Align diagnostics and telemetry Effect services #3189

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
143 changes: 103 additions & 40 deletions apps/server/src/diagnostics/ProcessDiagnostics.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,8 @@ import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Schema from "effect/Schema";
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
import * as ChildProcess from "effect/unstable/process/ChildProcess";
import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner";

import { collectUint8StreamText } from "../stream/collectUint8StreamText.ts";

Expand All @@ -31,35 +32,80 @@ const PROCESS_QUERY_TIMEOUT_MS = 1_000;
const POSIX_PROCESS_QUERY_COMMAND = "pid=,ppid=,pgid=,stat=,pcpu=,rss=,etime=,command=";
const PROCESS_QUERY_MAX_OUTPUT_BYTES = 2 * 1024 * 1024;

export interface ProcessDiagnosticsShape {
readonly read: Effect.Effect<ServerProcessDiagnosticsResult>;
readonly signal: (input: {
readonly pid: number;
readonly signal: ServerProcessSignal;
}) => Effect.Effect<ServerSignalProcessResult>;
}

export class ProcessDiagnostics extends Context.Service<
ProcessDiagnostics,
ProcessDiagnosticsShape
{
readonly read: Effect.Effect<ServerProcessDiagnosticsResult>;
readonly signal: (input: {
readonly pid: number;
readonly signal: ServerProcessSignal;
}) => Effect.Effect<ServerSignalProcessResult>;
}
>()("t3/diagnostics/ProcessDiagnostics") {}

class ProcessDiagnosticsError extends Schema.TaggedErrorClass<ProcessDiagnosticsError>()(
"ProcessDiagnosticsError",
class ProcessDiagnosticsQueryTimeoutError extends Schema.TaggedErrorClass<ProcessDiagnosticsQueryTimeoutError>()(
"ProcessDiagnosticsQueryTimeoutError",
{ command: Schema.String },
) {
override get message(): string {
return `Process diagnostics query '${this.command}' timed out.`;
}
}

class ProcessDiagnosticsQueryFailedError extends Schema.TaggedErrorClass<ProcessDiagnosticsQueryFailedError>()(
"ProcessDiagnosticsQueryFailedError",
{
message: Schema.String,
command: Schema.String,
stderr: Schema.optional(Schema.String),
cause: Schema.optional(Schema.Defect()),
},
) {}
const isProcessDiagnosticsError = Schema.is(ProcessDiagnosticsError);
) {
override get message(): string {
return this.stderr?.trim() || `Failed to query process diagnostics with '${this.command}'.`;
}
}

function toProcessDiagnosticsError(message: string, cause?: unknown): ProcessDiagnosticsError {
return new ProcessDiagnosticsError({
message,
...(cause === undefined ? {} : { cause }),
});
class ProcessDiagnosticsServerProcessSignalError extends Schema.TaggedErrorClass<ProcessDiagnosticsServerProcessSignalError>()(
"ProcessDiagnosticsServerProcessSignalError",
{ pid: Schema.Number },
) {
override get message(): string {
return "Refusing to signal the T3 server process.";
}
}

class ProcessDiagnosticsNotDescendantError extends Schema.TaggedErrorClass<ProcessDiagnosticsNotDescendantError>()(
"ProcessDiagnosticsNotDescendantError",
{ pid: Schema.Number },
) {
override get message(): string {
return `Process ${this.pid} is not a live descendant of the T3 server.`;
}
}

class ProcessDiagnosticsSignalFailedError extends Schema.TaggedErrorClass<ProcessDiagnosticsSignalFailedError>()(
"ProcessDiagnosticsSignalFailedError",
{
pid: Schema.Number,
signal: Schema.String,
cause: Schema.Defect(),
},
) {
override get message(): string {
return `Failed to signal process ${this.pid} with ${this.signal}.`;
}
}

const ProcessDiagnosticsError = Schema.Union([
ProcessDiagnosticsQueryTimeoutError,
ProcessDiagnosticsQueryFailedError,
ProcessDiagnosticsServerProcessSignalError,
ProcessDiagnosticsNotDescendantError,
ProcessDiagnosticsSignalFailedError,
]);
type ProcessDiagnosticsError = typeof ProcessDiagnosticsError.Type;
const isProcessDiagnosticsError = Schema.is(ProcessDiagnosticsError);

function parsePositiveInt(value: string): number | null {
const parsed = Number.parseInt(value, 10);
return Number.isInteger(parsed) && parsed > 0 ? parsed : null;
Expand Down Expand Up @@ -272,11 +318,7 @@ interface ProcessOutput {
}

const runProcess = Effect.fn("runProcess")(
function* (input: {
readonly command: string;
readonly args: ReadonlyArray<string>;
readonly errorMessage: string;
}) {
function* (input: { readonly command: string; readonly args: ReadonlyArray<string> }) {
const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
// `ps` and `powershell.exe` are real executables; spawning through cmd.exe
// shell mode would re-tokenize the PowerShell `-Command` payload (which
Expand Down Expand Up @@ -315,14 +357,22 @@ const runProcess = Effect.fn("runProcess")(
Effect.timeoutOption(Duration.millis(PROCESS_QUERY_TIMEOUT_MS)),
Effect.flatMap((result) =>
Option.match(result, {
onNone: () => Effect.fail(toProcessDiagnosticsError(`${input.errorMessage} timed out.`)),
onNone: () =>
Effect.fail(
new ProcessDiagnosticsQueryTimeoutError({
command: input.command,
}),
),
onSome: Effect.succeed,
}),
),
Effect.mapError((cause) =>
isProcessDiagnosticsError(cause)
? cause
: toProcessDiagnosticsError(input.errorMessage, cause),
: new ProcessDiagnosticsQueryFailedError({
command: input.command,
cause,
}),
),
),
);
Expand All @@ -335,11 +385,15 @@ function readPosixProcessRows(): Effect.Effect<
return runProcess({
command: "ps",
args: ["-axo", POSIX_PROCESS_QUERY_COMMAND],
errorMessage: "Failed to query process diagnostics.",
}).pipe(
Effect.flatMap((result) =>
result.exitCode !== 0
? Effect.fail(toProcessDiagnosticsError(result.stderr.trim() || "ps failed."))
? Effect.fail(
new ProcessDiagnosticsQueryFailedError({
command: "ps",
stderr: result.stderr.trim() || "ps failed.",
}),
)
: Effect.succeed(parsePosixProcessRows(result.stdout)),
),
);
Expand All @@ -361,12 +415,14 @@ function readWindowsProcessRows(): Effect.Effect<
return runProcess({
command: "powershell.exe",
args: ["-NoProfile", "-NonInteractive", "-Command", command],
errorMessage: "Failed to query process diagnostics.",
}).pipe(
Effect.flatMap((result) =>
result.exitCode !== 0
? Effect.fail(
toProcessDiagnosticsError(result.stderr.trim() || "PowerShell process query failed."),
new ProcessDiagnosticsQueryFailedError({
command: "powershell.exe",
stderr: result.stderr.trim() || "PowerShell process query failed.",
}),
)
: Effect.succeed(parseWindowsProcessRows(result.stdout)),
),
Expand All @@ -390,7 +446,11 @@ function assertDescendantPid(
pid: number,
): Effect.Effect<void, ProcessDiagnosticsError, ChildProcessSpawner.ChildProcessSpawner> {
if (pid === process.pid) {
return Effect.fail(toProcessDiagnosticsError("Refusing to signal the T3 server process."));
return Effect.fail(
new ProcessDiagnosticsServerProcessSignalError({
pid,
}),
);
}

return readProcessRows.pipe(
Expand All @@ -402,16 +462,18 @@ function assertDescendantPid(
return descendant
? Effect.void
: Effect.fail(
toProcessDiagnosticsError(`Process ${pid} is not a live descendant of the T3 server.`),
new ProcessDiagnosticsNotDescendantError({
pid,
}),
);
}),
);
}

export const make = Effect.fn("makeProcessDiagnostics")(function* () {
export const make = Effect.gen(function* () {
const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;

const read: ProcessDiagnosticsShape["read"] = Effect.gen(function* () {
const read: ProcessDiagnostics["Service"]["read"] = Effect.gen(function* () {
const readAt = yield* DateTime.now;
const rows = yield* readProcessRows.pipe(
Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
Expand All @@ -427,7 +489,7 @@ export const make = Effect.fn("makeProcessDiagnostics")(function* () {
),
);

const signal: ProcessDiagnosticsShape["signal"] = Effect.fn("ProcessDiagnostics.signal")(
const signal: ProcessDiagnostics["Service"]["signal"] = Effect.fn("ProcessDiagnostics.signal")(
function* (input) {
return yield* assertDescendantPid(input.pid).pipe(
Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
Expand All @@ -443,10 +505,11 @@ export const make = Effect.fn("makeProcessDiagnostics")(function* () {
};
},
catch: (cause) =>
toProcessDiagnosticsError(
`Failed to signal process ${input.pid} with ${input.signal}.`,
new ProcessDiagnosticsSignalFailedError({
pid: input.pid,
signal: input.signal,
cause,
),
}),
}),
),
Effect.catch((error: ProcessDiagnosticsError) =>
Expand All @@ -464,4 +527,4 @@ export const make = Effect.fn("makeProcessDiagnostics")(function* () {
return ProcessDiagnostics.of({ read, signal });
});

export const layer = Layer.effect(ProcessDiagnostics, make());
export const layer = Layer.effect(ProcessDiagnostics, make);
23 changes: 10 additions & 13 deletions apps/server/src/diagnostics/ProcessResourceMonitor.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,16 +3,13 @@ import * as DateTime from "effect/DateTime";
import * as Effect from "effect/Effect";
import * as Option from "effect/Option";

import {
aggregateProcessResourceHistory,
collectMonitoredSamples,
} from "./ProcessResourceMonitor.ts";
import * as ProcessResourceMonitor from "./ProcessResourceMonitor.ts";

describe("ProcessResourceMonitor", () => {
it.effect("samples the server root process and descendants", () =>
Effect.sync(() => {
const sampledAt = DateTime.makeUnsafe("2026-05-05T10:00:00.000Z");
const samples = collectMonitoredSamples({
const samples = ProcessResourceMonitor.collectMonitoredSamples({
serverPid: 100,
sampledAt,
sampledAtMs: DateTime.toEpochMillis(sampledAt),
Expand Down Expand Up @@ -72,7 +69,7 @@ describe("ProcessResourceMonitor", () => {
const firstAt = DateTime.makeUnsafe("2026-05-05T10:00:00.000Z");
const secondAt = DateTime.makeUnsafe("2026-05-05T10:00:05.000Z");
const samples = [
...collectMonitoredSamples({
...ProcessResourceMonitor.collectMonitoredSamples({
serverPid: 100,
sampledAt: firstAt,
sampledAtMs: DateTime.toEpochMillis(firstAt),
Expand All @@ -89,7 +86,7 @@ describe("ProcessResourceMonitor", () => {
},
],
}),
...collectMonitoredSamples({
...ProcessResourceMonitor.collectMonitoredSamples({
serverPid: 100,
sampledAt: secondAt,
sampledAtMs: DateTime.toEpochMillis(secondAt),
Expand All @@ -108,7 +105,7 @@ describe("ProcessResourceMonitor", () => {
}),
];

const result = aggregateProcessResourceHistory({
const result = ProcessResourceMonitor.aggregateProcessResourceHistory({
samples,
readAt: secondAt,
readAtMs: DateTime.toEpochMillis(secondAt),
Expand All @@ -132,7 +129,7 @@ describe("ProcessResourceMonitor", () => {
const firstAt = DateTime.makeUnsafe("2026-05-05T10:00:00.400Z");
const secondAt = DateTime.makeUnsafe("2026-05-05T10:00:05.900Z");
const samples = [
...collectMonitoredSamples({
...ProcessResourceMonitor.collectMonitoredSamples({
serverPid: 100,
sampledAt: firstAt,
sampledAtMs: DateTime.toEpochMillis(firstAt),
Expand All @@ -149,7 +146,7 @@ describe("ProcessResourceMonitor", () => {
},
],
}),
...collectMonitoredSamples({
...ProcessResourceMonitor.collectMonitoredSamples({
serverPid: 100,
sampledAt: secondAt,
sampledAtMs: DateTime.toEpochMillis(secondAt),
Expand All @@ -168,7 +165,7 @@ describe("ProcessResourceMonitor", () => {
}),
];

const result = aggregateProcessResourceHistory({
const result = ProcessResourceMonitor.aggregateProcessResourceHistory({
samples,
readAt: secondAt,
readAtMs: DateTime.toEpochMillis(secondAt),
Expand All @@ -187,7 +184,7 @@ describe("ProcessResourceMonitor", () => {
it.effect("returns all process summaries in the selected window", () =>
Effect.sync(() => {
const sampledAt = DateTime.makeUnsafe("2026-05-05T10:00:00.000Z");
const samples = collectMonitoredSamples({
const samples = ProcessResourceMonitor.collectMonitoredSamples({
serverPid: 100,
sampledAt,
sampledAtMs: DateTime.toEpochMillis(sampledAt),
Expand Down Expand Up @@ -215,7 +212,7 @@ describe("ProcessResourceMonitor", () => {
],
});

const result = aggregateProcessResourceHistory({
const result = ProcessResourceMonitor.aggregateProcessResourceHistory({
samples,
readAt: sampledAt,
readAtMs: DateTime.toEpochMillis(sampledAt),
Expand Down
Loading
Loading