From 74593119559a4ee790f500459bdae1a9e3747217 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 28 Jun 2026 16:08:01 +0000 Subject: [PATCH] Make diagnostics parsing more idiomatic Effect Co-authored-by: Julius Marminge --- .../diagnostics/ProcessDiagnostics.test.ts | 80 +++++++- .../src/diagnostics/ProcessDiagnostics.ts | 98 ++++++---- .../src/diagnostics/TraceDiagnostics.ts | 185 ++++++++++-------- 3 files changed, 247 insertions(+), 116 deletions(-) diff --git a/apps/server/src/diagnostics/ProcessDiagnostics.test.ts b/apps/server/src/diagnostics/ProcessDiagnostics.test.ts index 7d16a11c829..b422e15f285 100644 --- a/apps/server/src/diagnostics/ProcessDiagnostics.test.ts +++ b/apps/server/src/diagnostics/ProcessDiagnostics.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from "@effect/vitest"; +import { assert, describe, expect, it } from "@effect/vitest"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; @@ -220,6 +220,84 @@ describe("ProcessDiagnostics", () => { }), ); + it.effect("queries Windows processes through schema-decoded PowerShell JSON", () => + Effect.gen(function* () { + const commands: Array<{ readonly command: string; readonly args: ReadonlyArray }> = + []; + const spawnerLayer = Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make((command) => { + const childProcess = command as unknown as { + readonly command: string; + readonly args: ReadonlyArray; + }; + commands.push({ command: childProcess.command, args: childProcess.args }); + return Effect.succeed( + mockHandle({ + stdout: JSON.stringify([ + { + ProcessId: process.pid, + ParentProcessId: 1, + Name: "node.exe", + CommandLine: "t3 server", + Status: "Running", + WorkingSetSize: 1024, + PercentProcessorTime: 0, + }, + { + ProcessId: 4242, + ParentProcessId: process.pid, + Name: "agent.exe", + CommandLine: "agent.exe --work", + Status: "Running", + WorkingSetSize: 4096.4, + PercentProcessorTime: 12.25, + }, + { + ProcessId: "invalid", + ParentProcessId: process.pid, + Name: "ignored.exe", + }, + ]), + }), + ); + }), + ); + const layer = ProcessDiagnostics.layer.pipe(Layer.provide(spawnerLayer)); + + const diagnostics = yield* Effect.service(ProcessDiagnostics.ProcessDiagnostics).pipe( + Effect.flatMap((pd) => pd.read), + Effect.provide(layer), + Effect.provideService(HostProcessPlatform, "win32"), + ); + + assert.deepStrictEqual( + diagnostics.processes.map((process) => ({ + pid: process.pid, + ppid: process.ppid, + command: process.command, + rssBytes: process.rssBytes, + cpuPercent: process.cpuPercent, + })), + [ + { + pid: 4242, + ppid: process.pid, + command: "agent.exe --work", + rssBytes: 4096, + cpuPercent: 12.25, + }, + ], + ); + assert.equal(commands[0]?.command, "powershell.exe"); + assert.deepStrictEqual(commands[0]?.args.slice(0, 3), [ + "-NoProfile", + "-NonInteractive", + "-Command", + ]); + }), + ); + it.effect("keeps bounded command diagnostics when the process query exits unsuccessfully", () => Effect.gen(function* () { const spawnerLayer = Layer.succeed( diff --git a/apps/server/src/diagnostics/ProcessDiagnostics.ts b/apps/server/src/diagnostics/ProcessDiagnostics.ts index b39d560a228..e02b93a73f7 100644 --- a/apps/server/src/diagnostics/ProcessDiagnostics.ts +++ b/apps/server/src/diagnostics/ProcessDiagnostics.ts @@ -32,6 +32,19 @@ 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; +const WindowsProcessRecord = Schema.Struct({ + ProcessId: Schema.Number.check(Schema.isFinite()), + ParentProcessId: Schema.Number.check(Schema.isFinite()), + CommandLine: Schema.optional(Schema.Unknown), + Name: Schema.optional(Schema.Unknown), + Status: Schema.optional(Schema.Unknown), + WorkingSetSize: Schema.optional(Schema.Unknown), + PercentProcessorTime: Schema.optional(Schema.Unknown), +}); +const WindowsProcessDocument = Schema.fromJsonString(Schema.Unknown); +const decodeWindowsProcessDocument = Schema.decodeUnknownOption(WindowsProcessDocument); +const decodeWindowsProcessRecord = Schema.decodeUnknownOption(WindowsProcessRecord); + export class ProcessDiagnostics extends Context.Service< ProcessDiagnostics, { @@ -136,6 +149,14 @@ function parseNumber(value: string): number | null { return Number.isFinite(parsed) ? parsed : null; } +function finiteNumber(value: unknown): Option.Option { + return typeof value === "number" && Number.isFinite(value) ? Option.some(value) : Option.none(); +} + +function nonBlankString(value: unknown): Option.Option { + return typeof value === "string" && value.trim().length > 0 ? Option.some(value) : Option.none(); +} + export function parsePosixProcessRows(output: string): ReadonlyArray { const rows: ProcessRow[] = []; const rowPattern = @@ -201,51 +222,58 @@ export function parsePosixProcessRows(output: string): ReadonlyArray return rows; } -function normalizeWindowsProcessRow(value: unknown): ProcessRow | null { - if (typeof value !== "object" || value === null) return null; - const record = value as Record; - const pid = typeof record.ProcessId === "number" ? record.ProcessId : null; - const ppid = typeof record.ParentProcessId === "number" ? record.ParentProcessId : null; - const commandLine = - typeof record.CommandLine === "string" && record.CommandLine.trim().length > 0 - ? record.CommandLine - : typeof record.Name === "string" - ? record.Name - : null; - const workingSet = - typeof record.WorkingSetSize === "number" && Number.isFinite(record.WorkingSetSize) - ? Math.max(0, Math.round(record.WorkingSetSize)) - : 0; - const cpuPercent = - typeof record.PercentProcessorTime === "number" && Number.isFinite(record.PercentProcessorTime) - ? Math.max(0, record.PercentProcessorTime) - : 0; - - if (!pid || pid <= 0 || ppid === null || ppid < 0 || !commandLine) return null; - return { +function normalizeWindowsProcessRow(value: unknown): Option.Option { + const recordOption = decodeWindowsProcessRecord(value); + if (Option.isNone(recordOption)) return Option.none(); + const record = recordOption.value; + const pid = record.ProcessId; + const ppid = record.ParentProcessId; + const commandLineOption = nonBlankString(record.CommandLine).pipe( + Option.orElse(() => nonBlankString(record.Name)), + ); + + if ( + !Number.isInteger(pid) || + pid <= 0 || + !Number.isInteger(ppid) || + ppid < 0 || + Option.isNone(commandLineOption) + ) { + return Option.none(); + } + + const workingSet = Option.getOrElse( + finiteNumber(record.WorkingSetSize).pipe(Option.map((bytes) => Math.max(0, Math.round(bytes)))), + () => 0, + ); + const cpuPercent = Option.getOrElse( + finiteNumber(record.PercentProcessorTime).pipe(Option.map((percent) => Math.max(0, percent))), + () => 0, + ); + + return Option.some({ pid, ppid, pgid: null, - status: typeof record.Status === "string" && record.Status.length > 0 ? record.Status : "Live", + status: Option.getOrElse(nonBlankString(record.Status), () => "Live"), cpuPercent, rssBytes: workingSet, elapsed: "", - command: commandLine, - }; + command: commandLineOption.value, + }); } function parseWindowsProcessRows(output: string): ReadonlyArray { if (output.trim().length === 0) return []; - try { - const parsed = JSON.parse(output) as unknown; - const records = Array.isArray(parsed) ? parsed : [parsed]; - return records.flatMap((record) => { - const row = normalizeWindowsProcessRow(record); - return row ? [row] : []; - }); - } catch { - return []; - } + const document = decodeWindowsProcessDocument(output); + if (Option.isNone(document)) return []; + const records = Array.isArray(document.value) ? document.value : [document.value]; + return records.flatMap((record) => + Option.match(normalizeWindowsProcessRow(record), { + onNone: () => [], + onSome: (row) => [row], + }), + ); } export function buildDescendantEntries( diff --git a/apps/server/src/diagnostics/TraceDiagnostics.ts b/apps/server/src/diagnostics/TraceDiagnostics.ts index d54e033380c..ffcc6200ee7 100644 --- a/apps/server/src/diagnostics/TraceDiagnostics.ts +++ b/apps/server/src/diagnostics/TraceDiagnostics.ts @@ -17,22 +17,31 @@ import * as PlatformError from "effect/PlatformError"; import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; -interface TraceRecordLike { - readonly name?: unknown; - readonly traceId?: unknown; - readonly spanId?: unknown; - readonly startTimeUnixNano?: unknown; - readonly endTimeUnixNano?: unknown; - readonly durationMs?: unknown; - readonly exit?: unknown; - readonly events?: unknown; -} +const TraceRecordLine = Schema.Struct({ + name: Schema.String, + traceId: Schema.String, + spanId: Schema.String, + startTimeUnixNano: Schema.optional(Schema.Unknown), + endTimeUnixNano: Schema.String, + durationMs: Schema.Number.check(Schema.isFinite()), + exit: Schema.optional(Schema.Unknown), + events: Schema.optional(Schema.Array(Schema.Unknown)), +}); +const decodeTraceRecordLine = Schema.decodeUnknownOption(Schema.fromJsonString(TraceRecordLine)); -interface TraceEventLike { - readonly name?: unknown; - readonly timeUnixNano?: unknown; - readonly attributes?: unknown; -} +const TraceExit = Schema.Struct({ + _tag: Schema.String, + cause: Schema.optional(Schema.Unknown), +}); +const decodeTraceExit = Schema.decodeUnknownOption(TraceExit); + +const TraceEvent = Schema.Struct({ + name: Schema.optional(Schema.Unknown), + timeUnixNano: Schema.optional(Schema.Unknown), + attributes: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)), +}); +type TraceEvent = typeof TraceEvent.Type; +const decodeTraceEvent = Schema.decodeUnknownOption(TraceEvent); export interface TraceDiagnosticsOptions { readonly traceFilePath: string; @@ -90,47 +99,38 @@ function toRotatedTracePaths(traceFilePath: string, maxFiles: number): ReadonlyA return [...backups, traceFilePath]; } -function isRecordObject(value: unknown): value is TraceRecordLike { - return typeof value === "object" && value !== null; +function nonBlankString(value: string): Option.Option { + return value.trim().length > 0 ? Option.some(value) : Option.none(); } -function toStringValue(value: unknown): string | null { - return typeof value === "string" && value.trim().length > 0 ? value : null; -} - -function toNumberValue(value: unknown): number | null { - return typeof value === "number" && Number.isFinite(value) ? value : null; -} - -function unixNanoToDateTime(value: unknown): DateTime.Utc | null { - const text = toStringValue(value); - if (!text) return null; +function unixNanoToDateTime(value: unknown): Option.Option { + if (typeof value !== "string" || value.trim().length === 0) return Option.none(); try { - const millis = Number(BigInt(text) / 1_000_000n); - return Option.getOrNull(DateTime.make(millis)); + const millis = Number(BigInt(value) / 1_000_000n); + return DateTime.make(millis); } catch { - return null; + return Option.none(); } } -function readExitTag(exit: unknown): string | null { - if (!isRecordObject(exit) || !("_tag" in exit)) return null; - return toStringValue(exit._tag); +function readExitTag(exit: unknown): Option.Option { + return decodeTraceExit(exit).pipe(Option.flatMap((decoded) => nonBlankString(decoded._tag))); } function readExitCause(exit: unknown): string { - if (!isRecordObject(exit) || !("cause" in exit)) return "Failure"; - return toStringValue(exit.cause)?.trim() ?? "Failure"; -} - -function isTraceEvent(value: unknown): value is TraceEventLike { - return typeof value === "object" && value !== null; + return Option.getOrElse( + decodeTraceExit(exit).pipe( + Option.flatMap((decoded) => Option.fromNullishOr(decoded.cause)), + Option.filter((cause): cause is string => typeof cause === "string"), + Option.map((cause) => cause.trim()), + Option.filter((cause) => cause.length > 0), + ), + () => "Failure", + ); } -function readEventAttributes(event: TraceEventLike): Readonly> { - return typeof event.attributes === "object" && event.attributes !== null - ? (event.attributes as Readonly>) - : {}; +function readEventAttributes(event: TraceEvent): Readonly> { + return event.attributes ?? {}; } function makeEmptyDiagnostics(input: { @@ -211,8 +211,8 @@ export function aggregateTraceDiagnostics( let failureCount = 0; let interruptionCount = 0; let slowSpanCount = 0; - let firstSpanAt: DateTime.Utc | null = null; - let lastSpanAt: DateTime.Utc | null = null; + let firstSpanAt = Option.none(); + let lastSpanAt = Option.none(); const spansByName = new Map< string, @@ -229,40 +229,52 @@ export function aggregateTraceDiagnostics( for (const line of lines) { if (line.trim().length === 0) continue; - let parsed: unknown; - try { - parsed = JSON.parse(line); - } catch { - parseErrorCount += 1; - continue; - } - - if (!isRecordObject(parsed)) { + const parsedOption = decodeTraceRecordLine(line); + if (Option.isNone(parsedOption)) { parseErrorCount += 1; continue; } + const parsed = parsedOption.value; - const name = toStringValue(parsed.name); - const traceId = toStringValue(parsed.traceId); - const spanId = toStringValue(parsed.spanId); - const durationMs = toNumberValue(parsed.durationMs); - const endedAt = unixNanoToDateTime(parsed.endTimeUnixNano); + const nameOption = nonBlankString(parsed.name); + const traceIdOption = nonBlankString(parsed.traceId); + const spanIdOption = nonBlankString(parsed.spanId); + const endedAtOption = unixNanoToDateTime(parsed.endTimeUnixNano); const startedAt = unixNanoToDateTime(parsed.startTimeUnixNano); - - if (!name || !traceId || !spanId || durationMs === null || !endedAt) { + if ( + Option.isNone(nameOption) || + Option.isNone(traceIdOption) || + Option.isNone(spanIdOption) || + Option.isNone(endedAtOption) + ) { parseErrorCount += 1; continue; } + const name = nameOption.value; + const traceId = traceIdOption.value; + const spanId = spanIdOption.value; + const endedAt = endedAtOption.value; + const durationMs = parsed.durationMs; recordCount += 1; - firstSpanAt = - startedAt && (firstSpanAt === null || DateTime.isLessThan(startedAt, firstSpanAt)) - ? startedAt - : firstSpanAt; - lastSpanAt = - lastSpanAt === null || DateTime.isGreaterThan(endedAt, lastSpanAt) ? endedAt : lastSpanAt; - - const exitTag = readExitTag(parsed.exit); + firstSpanAt = Option.match(startedAt, { + onNone: () => firstSpanAt, + onSome: (spanStartedAt) => + Option.match(firstSpanAt, { + onNone: () => Option.some(spanStartedAt), + onSome: (currentFirstSpanAt) => + DateTime.isLessThan(spanStartedAt, currentFirstSpanAt) + ? Option.some(spanStartedAt) + : firstSpanAt, + }), + }); + lastSpanAt = Option.match(lastSpanAt, { + onNone: () => Option.some(endedAt), + onSome: (currentLastSpanAt) => + DateTime.isGreaterThan(endedAt, currentLastSpanAt) ? Option.some(endedAt) : lastSpanAt, + }); + + const exitTag = Option.getOrNull(readExitTag(parsed.exit)); const isFailure = exitTag === "Failure"; const isInterrupted = exitTag === "Interrupted"; if (isFailure) failureCount += 1; @@ -303,12 +315,18 @@ export function aggregateTraceDiagnostics( }); } - if (Array.isArray(parsed.events)) { - for (const rawEvent of parsed.events) { - if (!isTraceEvent(rawEvent)) continue; - const attributes = readEventAttributes(rawEvent); - const level = toStringValue(attributes["effect.logLevel"]); - if (!level) continue; + if (parsed.events) { + for (const encodedEvent of parsed.events) { + const eventOption = decodeTraceEvent(encodedEvent); + if (Option.isNone(eventOption)) continue; + const event = eventOption.value; + const attributes = readEventAttributes(event); + const levelOption = + typeof attributes["effect.logLevel"] === "string" + ? nonBlankString(attributes["effect.logLevel"]) + : Option.none(); + if (Option.isNone(levelOption)) continue; + const level = levelOption.value; logLevelCounts[level] = (logLevelCounts[level] ?? 0) + 1; const normalizedLevel = level.toLowerCase(); @@ -321,8 +339,15 @@ export function aggregateTraceDiagnostics( continue; } - const seenAt = unixNanoToDateTime(rawEvent.timeUnixNano) ?? endedAt; - const message = toStringValue(rawEvent.name)?.trim() ?? "Log event"; + const seenAt = Option.getOrElse(unixNanoToDateTime(event.timeUnixNano), () => endedAt); + const message = Option.getOrElse( + Option.fromNullishOr(event.name).pipe( + Option.filter((name): name is string => typeof name === "string"), + Option.map((name) => name.trim()), + Option.filter((name) => name.length > 0), + ), + () => "Log event", + ); latestWarningAndErrorLogs.push({ spanName: name, level, @@ -354,8 +379,8 @@ export function aggregateTraceDiagnostics( readAt, recordCount, parseErrorCount, - firstSpanAt: Option.fromNullishOr(firstSpanAt), - lastSpanAt: Option.fromNullishOr(lastSpanAt), + firstSpanAt, + lastSpanAt, failureCount, interruptionCount, slowSpanThresholdMs,