Make process 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,7 +42,7 @@ describe("ProcessDiagnostics", () => {
         ].join("\n"),
       );
 
-      expect(rows).toEqual([
+      assert.deepEqual(rows, [
         {
           pid: 10,
           ppid: 1,
@@ -126,15 +126,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]);
     }),
   );
 
@@ -177,7 +183,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 +219,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 +253,7 @@ describe("ProcessDiagnostics", () => {
         Effect.flip,
       );
 
-      expect(error).toMatchObject({
+      assert.deepInclude(error, {
         _tag: "ProcessDiagnosticsQueryFailedError",
         command: "ps",
         argCount: 2,
@@ -252,7 +264,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,12 +293,93 @@ describe("ProcessDiagnostics", () => {
         Effect.provide(layer),
       );
 
-      expect(result).toEqual({
+      assert.deepEqual(result, {
         pid: 4242,
         signal: "SIGINT",
         signaled: false,
         message: Option.some("Process 4242 is not a live descendant of the T3 server."),
       });
     }),
   );
+
+  it.effect("parses Windows JSON process rows through Schema", () =>
+    Effect.gen(function* () {
+      const commands: Array<{ readonly command: string; readonly args: ReadonlyArray<string> }> =
+        [];
+      const spawnerLayer = Layer.succeed(
+        ChildProcessSpawner.ChildProcessSpawner,
+        ChildProcessSpawner.make((command) => {
+          const childProcess = command as unknown as {
+            readonly command: string;
+            readonly args: ReadonlyArray<string>;
+          };
+          commands.push({ command: childProcess.command, args: childProcess.args });
+          return Effect.succeed(
+            mockHandle({
+              stdout: JSON.stringify([
+                {
+                  ProcessId: 4242,
+                  ParentProcessId: process.pid,
+                  Name: "agent.exe",
+                  CommandLine: "agent.exe --flag",
+                  Status: "Running",
+                  WorkingSetSize: 4096,
+                  PercentProcessorTime: 12.5,
+                },
+                {
+                  ProcessId: 4243,
+                  ParentProcessId: 4242,
+                  Name: "fallback.exe",
+                  CommandLine: null,
+                  Status: null,
+                  WorkingSetSize: null,
+                  PercentProcessorTime: null,
+                },
+                {
+                  ProcessId: "not-a-number",
+                  ParentProcessId: process.pid,
+                  Name: "ignored.exe",
+                  CommandLine: "ignored.exe",
+                },
+              ]),
+            }),
+          );
+        }),
+      );
+
+      const rows = yield* ProcessDiagnostics.readProcessRows.pipe(
+        Effect.provide(spawnerLayer),
+        Effect.provideService(HostProcessPlatform, "win32"),
+      );
+
+      assert.deepEqual(rows, [
+        {
+          pid: 4242,
+          ppid: process.pid,
+          pgid: null,
+          status: "Running",
+          cpuPercent: 12.5,
+          rssBytes: 4096,
+          elapsed: "",
+          command: "agent.exe --flag",
+        },
+        {
+          pid: 4243,
+          ppid: 4242,
+          pgid: null,
+          status: "Live",
+          cpuPercent: 0,
+          rssBytes: 0,
+          elapsed: "",
+          command: "fallback.exe",
+        },
+      ]);
+      assert.equal(commands[0]?.command, "powershell.exe");
+      assert.deepEqual(commands[0]?.args.slice(0, 3), [
+        "-NoProfile",
+        "-NonInteractive",
+        "-Command",
+      ]);
+    }),
+  );
 });

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,
+  ParentProcessId: Schema.Number,
+  CommandLine: Schema.optional(Schema.NullOr(Schema.String)),
+  Name: Schema.optional(Schema.NullOr(Schema.String)),
+  Status: Schema.optional(Schema.NullOr(Schema.String)),
+  WorkingSetSize: Schema.optional(Schema.NullOr(Schema.Number)),
+  PercentProcessorTime: Schema.optional(Schema.NullOr(Schema.Number)),
+});
+type WindowsProcessRecord = typeof WindowsProcessRecord.Type;
+const decodeWindowsProcessJson = Schema.decodeOption(Schema.UnknownFromJsonString);
+const decodeWindowsProcessRecord = Schema.decodeUnknownOption(WindowsProcessRecord);
+
 export class ProcessDiagnostics extends Context.Service<
   ProcessDiagnostics,
   {
@@ -121,19 +134,47 @@ const ProcessDiagnosticsError = Schema.Union([
 type ProcessDiagnosticsError = typeof ProcessDiagnosticsError.Type;
 const isProcessDiagnosticsError = Schema.is(ProcessDiagnosticsError);
 
-function parsePositiveInt(value: string): number | null {
+function parsePositiveInt(value: string): Option.Option<number> {
+  const parsed = Number.parseInt(value, 10);
+  return Number.isInteger(parsed) && parsed > 0 ? Option.some(parsed) : Option.none();
+}
+
+function parseNonNegativeInt(value: string): Option.Option<number> {
   const parsed = Number.parseInt(value, 10);
-  return Number.isInteger(parsed) && parsed > 0 ? parsed : null;
+  return Number.isInteger(parsed) && parsed >= 0 ? Option.some(parsed) : Option.none();
 }
 
-function parseNonNegativeInt(value: string): number | null {
+function parseInteger(value: string): Option.Option<number> {
   const parsed = Number.parseInt(value, 10);
-  return Number.isInteger(parsed) && parsed >= 0 ? parsed : null;
+  return Number.isInteger(parsed) ? Option.some(parsed) : Option.none();
 }
 
-function parseNumber(value: string): number | null {
+function parseNumber(value: string): Option.Option<number> {
   const parsed = Number.parseFloat(value);
-  return Number.isFinite(parsed) ? parsed : null;
+  return Number.isFinite(parsed) ? Option.some(parsed) : Option.none();
+}
+
+function positiveInteger(value: number): Option.Option<number> {
+  return Number.isInteger(value) && value > 0 ? Option.some(value) : Option.none();
+}
+
+function nonNegativeInteger(value: number): Option.Option<number> {
+  return Number.isInteger(value) && value >= 0 ? Option.some(value) : Option.none();
+}
+
+function trimNonEmpty(value: string | null | undefined): Option.Option<string> {
+  return Option.fromNullishOr(value).pipe(
+    Option.map((text) => text.trim()),
+    Option.filter((text) => text.length > 0),
+  );
+}
+
+function finiteNonNegativeNumberOrZero(value: number | null | undefined): number {
+  return Option.fromNullishOr(value).pipe(
+    Option.filter(Number.isFinite),
+    Option.map((number) => Math.max(0, number)),
+    Option.getOrElse(() => 0),
+  );
 }
 
 export function parsePosixProcessRows(output: string): ReadonlyArray<ProcessRow> {
@@ -168,31 +209,24 @@ export function parsePosixProcessRows(output: string): ReadonlyArray<ProcessRow>
       continue;
     }
 
-    const pid = parsePositiveInt(pidText);
-    const ppid = parseNonNegativeInt(ppidText);
-    const pgid = Number.parseInt(pgidText, 10);
-    const cpuPercent = parseNumber(cpuText);
-    const rssKiB = parseNonNegativeInt(rssText);
-    if (
-      pid === null ||
-      ppid === null ||
-      !Number.isInteger(pgid) ||
-      cpuPercent === null ||
-      rssKiB === null ||
-      !status ||
-      !elapsed ||
-      !command
-    ) {
+    const parsed = Option.all({
+      pid: parsePositiveInt(pidText),
+      ppid: parseNonNegativeInt(ppidText),
+      pgid: parseInteger(pgidText),
+      cpuPercent: parseNumber(cpuText),
+      rssKiB: parseNonNegativeInt(rssText),
+    });
+    if (Option.isNone(parsed) || !status || !elapsed || !command) {
       continue;
     }
 
     rows.push({
-      pid,
-      ppid,
-      pgid,
+      pid: parsed.value.pid,
+      ppid: parsed.value.ppid,
+      pgid: parsed.value.pgid,
       status,
-      cpuPercent,
-      rssBytes: rssKiB * 1024,
+      cpuPercent: parsed.value.cpuPercent,
+      rssBytes: parsed.value.rssKiB * 1024,
       elapsed,
       command,
     });
@@ -201,51 +235,38 @@ 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,
-    elapsed: "",
-    command: commandLine,
-  };
+function normalizeWindowsProcessRow(value: unknown): Option.Option<ProcessRow> {
+  return Option.gen(function* () {
+    const record: WindowsProcessRecord = yield* decodeWindowsProcessRecord(value);
+    const pid = yield* positiveInteger(record.ProcessId);
+    const ppid = yield* nonNegativeInteger(record.ParentProcessId);
+    const command = yield* Option.firstSomeOf([
+      trimNonEmpty(record.CommandLine),
+      trimNonEmpty(record.Name),
+    ]);
+    const status = Option.getOrElse(trimNonEmpty(record.Status), () => "Live");
+
+    return {
+      pid,
+      ppid,
+      pgid: null,
+      status,
+      cpuPercent: finiteNonNegativeNumberOrZero(record.PercentProcessorTime),
+      rssBytes: Math.round(finiteNonNegativeNumberOrZero(record.WorkingSetSize)),
+      elapsed: "",
+      command,
+    };
+  });
 }
 
 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 {
-    return [];
-  }
+  const parsed = decodeWindowsProcessJson(output);
+  if (Option.isNone(parsed)) return [];
+  const records = Array.isArray(parsed.value) ? parsed.value : [parsed.value];
+  return records.flatMap((record) => {
+    const row = normalizeWindowsProcessRow(record);
+    return Option.isSome(row) ? [row.value] : [];
+  });
 }
 
 export function buildDescendantEntries(