Make diagnostics parsing more idiomatic Effect

apps/server/src/diagnostics/ProcessDiagnostics.test.ts

@@ -1,4 +1,4 @@
-import { describe, expect, it } from "@effect/vitest";
+import { assert, describe, it } from "@effect/vitest";
 import * as DateTime from "effect/DateTime";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
@@ -42,11 +42,11 @@ describe("ProcessDiagnostics", () => {
         ].join("\n"),
       );
 
-      expect(rows).toEqual([
+      assert.deepEqual(rows, [
         {
           pid: 10,
           ppid: 1,
-          pgid: 10,
+          pgid: Option.some(10),
           status: "Ss",
           cpuPercent: 0,
           rssBytes: 1024 * 1024,
@@ -56,7 +56,7 @@ describe("ProcessDiagnostics", () => {
         {
           pid: 11,
           ppid: 10,
-          pgid: 10,
+          pgid: Option.some(10),
           status: "S+",
           cpuPercent: 12.5,
           rssBytes: 20480 * 1024,
@@ -67,6 +67,43 @@ describe("ProcessDiagnostics", () => {
     }),
   );
 
+  it.effect("parses Windows process JSON through Schema and skips invalid rows", () =>
+    Effect.sync(() => {
+      const rows = ProcessDiagnostics.parseWindowsProcessRows(
+        `[
+          {
+            "ProcessId": 10,
+            "ParentProcessId": 1,
+            "CommandLine": "",
+            "Name": "node.exe",
+            "Status": "",
+            "WorkingSetSize": 1024.4,
+            "PercentProcessorTime": 12.5
+          },
+          {
+            "ProcessId": "not-a-number",
+            "ParentProcessId": 1,
+            "Name": "bad.exe"
+          }
+        ]`,
+      );
+
+      assert.deepEqual(rows, [
+        {
+          pid: 10,
+          ppid: 1,
+          pgid: Option.none(),
+          status: "Live",
+          cpuPercent: 12.5,
+          rssBytes: 1024,
+          elapsed: "",
+          command: "node.exe",
+        },
+      ]);
+      assert.deepEqual(ProcessDiagnostics.parseWindowsProcessRows("not-json"), []);
+    }),
+  );
+
   it.effect("aggregates only descendants of the server process", () =>
     Effect.sync(() => {
       const diagnostics = ProcessDiagnostics.aggregateProcessDiagnostics({
@@ -76,7 +113,7 @@ describe("ProcessDiagnostics", () => {
           {
             pid: 100,
             ppid: 1,
-            pgid: 100,
+            pgid: Option.some(100),
             status: "S",
             cpuPercent: 0,
             rssBytes: 1_000,
@@ -86,7 +123,7 @@ describe("ProcessDiagnostics", () => {
           {
             pid: 101,
             ppid: 100,
-            pgid: 100,
+            pgid: Option.some(100),
             status: "S",
             cpuPercent: 1.5,
             rssBytes: 2_000,
@@ -96,7 +133,7 @@ describe("ProcessDiagnostics", () => {
           {
             pid: 102,
             ppid: 101,
-            pgid: 100,
+            pgid: Option.some(100),
             status: "R",
             cpuPercent: 3.25,
             rssBytes: 4_000,
@@ -106,7 +143,7 @@ describe("ProcessDiagnostics", () => {
           {
             pid: 200,
             ppid: 1,
-            pgid: 200,
+            pgid: Option.some(200),
             status: "S",
             cpuPercent: 99,
             rssBytes: 8_000,
@@ -116,7 +153,7 @@ describe("ProcessDiagnostics", () => {
           {
             pid: 201,
             ppid: 100,
-            pgid: 100,
+            pgid: Option.some(100),
             status: "R",
             cpuPercent: 9,
             rssBytes: 9_000,
@@ -126,15 +163,21 @@ describe("ProcessDiagnostics", () => {
         ],
       });
 
-      expect(diagnostics.serverPid).toBe(100);
-      expect(DateTime.formatIso(diagnostics.readAt)).toBe("2026-05-05T10:00:00.000Z");
-      expect(diagnostics.processCount).toBe(2);
-      expect(diagnostics.totalRssBytes).toBe(6_000);
-      expect(diagnostics.totalCpuPercent).toBe(4.75);
-      expect(diagnostics.processes.map((process) => process.pid)).toEqual([101, 102]);
-      expect(diagnostics.processes.map((process) => process.depth)).toEqual([0, 1]);
-      expect(Option.getOrNull(diagnostics.processes[0]!.pgid)).toBe(100);
-      expect(diagnostics.processes[0]?.childPids).toEqual([102]);
+      assert.equal(diagnostics.serverPid, 100);
+      assert.equal(DateTime.formatIso(diagnostics.readAt), "2026-05-05T10:00:00.000Z");
+      assert.equal(diagnostics.processCount, 2);
+      assert.equal(diagnostics.totalRssBytes, 6_000);
+      assert.equal(diagnostics.totalCpuPercent, 4.75);
+      assert.deepEqual(
+        diagnostics.processes.map((process) => process.pid),
+        [101, 102],
+      );
+      assert.deepEqual(
+        diagnostics.processes.map((process) => process.depth),
+        [0, 1],
+      );
+      assert.equal(Option.getOrNull(diagnostics.processes[0]!.pgid), 100);
+      assert.deepEqual(diagnostics.processes[0]?.childPids, [102]);
     }),
   );
 
@@ -147,7 +190,7 @@ describe("ProcessDiagnostics", () => {
           {
             pid: 101,
             ppid: 100,
-            pgid: 100,
+            pgid: Option.some(100),
             status: "S",
             cpuPercent: 0,
             rssBytes: 100,
@@ -157,7 +200,7 @@ describe("ProcessDiagnostics", () => {
           {
             pid: 103,
             ppid: 101,
-            pgid: 100,
+            pgid: Option.some(100),
             status: "S",
             cpuPercent: 0,
             rssBytes: 100,
@@ -167,7 +210,7 @@ describe("ProcessDiagnostics", () => {
           {
             pid: 102,
             ppid: 101,
-            pgid: 100,
+            pgid: Option.some(100),
             status: "S",
             cpuPercent: 0,
             rssBytes: 100,
@@ -177,7 +220,10 @@ describe("ProcessDiagnostics", () => {
         ],
       });
 
-      expect(diagnostics.processes.map((process) => process.pid)).toEqual([101, 102, 103]);
+      assert.deepEqual(
+        diagnostics.processes.map((process) => process.pid),
+        [101, 102, 103],
+      );
     }),
   );
 
@@ -210,8 +256,11 @@ describe("ProcessDiagnostics", () => {
         Effect.provide(layer),
       );
 
-      expect(diagnostics.processes.map((process) => process.pid)).toEqual([4242]);
-      expect(commands).toEqual([
+      assert.deepEqual(
+        diagnostics.processes.map((process) => process.pid),
+        [4242],
+      );
+      assert.deepEqual(commands, [
         {
           command: "ps",
           args: ["-axo", "pid=,ppid=,pgid=,stat=,pcpu=,rss=,etime=,command="],
@@ -241,7 +290,7 @@ describe("ProcessDiagnostics", () => {
         Effect.flip,
       );
 
-      expect(error).toMatchObject({
+      assert.deepInclude(error, {
         _tag: "ProcessDiagnosticsQueryFailedError",
         command: "ps",
         argCount: 2,
@@ -252,7 +301,8 @@ describe("ProcessDiagnostics", () => {
         stdoutTruncated: false,
         stderrTruncated: false,
       });
-      expect(error.message).toBe(
+      assert.equal(
+        error.message,
         `Process diagnostics query 'ps' failed with exit code 17 in '${process.cwd()}'.`,
       );
     }),
@@ -280,7 +330,7 @@ describe("ProcessDiagnostics", () => {
         Effect.provide(layer),
       );
 
-      expect(result).toEqual({
+      assert.deepEqual(result, {
         pid: 4242,
         signal: "SIGINT",
         signaled: false,

apps/server/src/diagnostics/ProcessDiagnostics.ts

@@ -11,16 +11,19 @@ import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
+import * as Result from "effect/Result";
 import * as Schema from "effect/Schema";
 import * as ChildProcess from "effect/unstable/process/ChildProcess";
 import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner";
 
+import { decodeJsonResult } from "@t3tools/shared/schemaJson";
+
 import { collectUint8StreamText } from "../stream/collectUint8StreamText.ts";
 
 export interface ProcessRow {
   readonly pid: number;
   readonly ppid: number;
-  readonly pgid: number | null;
+  readonly pgid: Option.Option<number>;
   readonly status: string;
   readonly cpuPercent: number;
   readonly rssBytes: number;
@@ -31,6 +34,20 @@ export interface ProcessRow {
 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 NullableString = Schema.Union([Schema.String, Schema.Null]);
+const NullableNumber = Schema.Union([Schema.Number, Schema.Null]);
+const WindowsProcessRecord = Schema.Struct({
+  ProcessId: Schema.Number,
+  ParentProcessId: Schema.Number,
+  CommandLine: Schema.optional(NullableString),
+  Name: Schema.optional(NullableString),
+  Status: Schema.optional(NullableString),
+  WorkingSetSize: Schema.optional(NullableNumber),
+  PercentProcessorTime: Schema.optional(NullableNumber),
+});
+type WindowsProcessRecord = typeof WindowsProcessRecord.Type;
+const decodeWindowsProcessJson = decodeJsonResult(Schema.Unknown);
+const decodeWindowsProcessRecord = Schema.decodeUnknownOption(WindowsProcessRecord);
 
 export class ProcessDiagnostics extends Context.Service<
   ProcessDiagnostics,
@@ -136,6 +153,14 @@ function parseNumber(value: string): number | null {
   return Number.isFinite(parsed) ? parsed : null;
 }
 
+function toNonEmptyStringOption(value: string | null | undefined): Option.Option<string> {
+  return typeof value === "string" && value.trim().length > 0 ? Option.some(value) : Option.none();
+}
+
+function toNonNegativeNumber(value: number | null | undefined): number {
+  return typeof value === "number" && Number.isFinite(value) ? Math.max(0, value) : 0;
+}
+
 export function parsePosixProcessRows(output: string): ReadonlyArray<ProcessRow> {
   const rows: ProcessRow[] = [];
   const rowPattern =
@@ -189,7 +214,7 @@ export function parsePosixProcessRows(output: string): ReadonlyArray<ProcessRow>
     rows.push({
       pid,
       ppid,
-      pgid,
+      pgid: Option.some(pgid),
       status,
       cpuPercent,
       rssBytes: rssKiB * 1024,
@@ -201,51 +226,46 @@ export function parsePosixProcessRows(output: string): ReadonlyArray<ProcessRow>
   return rows;
 }
 
-function normalizeWindowsProcessRow(value: unknown): ProcessRow | null {
-  if (typeof value !== "object" || value === null) return null;
-  const record = value as Record<string, unknown>;
-  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 {
-    pid,
-    ppid,
-    pgid: null,
-    status: typeof record.Status === "string" && record.Status.length > 0 ? record.Status : "Live",
-    cpuPercent,
-    rssBytes: workingSet,
+function normalizeWindowsProcessRow(record: WindowsProcessRecord): Option.Option<ProcessRow> {
+  const commandLine = Option.firstSomeOf([
+    toNonEmptyStringOption(record.CommandLine),
+    toNonEmptyStringOption(record.Name),
+  ]);
+  if (
+    !Number.isInteger(record.ProcessId) ||
+    record.ProcessId <= 0 ||
+    !Number.isInteger(record.ParentProcessId) ||
+    record.ParentProcessId < 0 ||
+    Option.isNone(commandLine)
+  ) {
+    return Option.none();
+  }
+
+  return Option.some({
+    pid: record.ProcessId,
+    ppid: record.ParentProcessId,
+    pgid: Option.none(),
+    status: Option.getOrElse(toNonEmptyStringOption(record.Status), () => "Live"),
+    cpuPercent: toNonNegativeNumber(record.PercentProcessorTime),
+    rssBytes: Math.round(toNonNegativeNumber(record.WorkingSetSize)),
     elapsed: "",
-    command: commandLine,
-  };
+    command: commandLine.value,
+  });
 }
 
-function parseWindowsProcessRows(output: string): ReadonlyArray<ProcessRow> {
+export function parseWindowsProcessRows(output: string): ReadonlyArray<ProcessRow> {
   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 {
+  const parsed = decodeWindowsProcessJson(output);
+  if (Result.isFailure(parsed)) {
     return [];
   }
+  const records = Array.isArray(parsed.success) ? parsed.success : [parsed.success];
+  return records.flatMap((rawRecord) => {
+    const decodedRecord = decodeWindowsProcessRecord(rawRecord);
+    if (Option.isNone(decodedRecord)) return [];
+    const row = normalizeWindowsProcessRow(decodedRecord.value);
+    return Option.isSome(row) ? [row.value] : [];
+  });
 }
 
 export function buildDescendantEntries(
@@ -276,7 +296,7 @@ export function buildDescendantEntries(
     entries.push({
       pid: item.row.pid,
       ppid: item.row.ppid,
-      pgid: Option.fromNullishOr(item.row.pgid),
+      pgid: item.row.pgid,
       status: item.row.status,
       cpuPercent: item.row.cpuPercent,
       rssBytes: item.row.rssBytes,