fix(process): honor disabled PID normalization in samples

src/application/ProcessNormalization.test.ts

@@ -0,0 +1,167 @@
+import { expect, it } from "vitest";
+
+import { normalizeProcessSamples } from "./ProcessNormalization.js";
+import {
+  parseProcessScenario,
+  type ProcessSample,
+} from "../domain/processCapture.js";
+
+const rootPid = 41_010;
+const samples = Object.freeze(
+  [
+    {
+      at_ms: 17,
+      pid: rootPid,
+      parent_pid: 40_000,
+      process_group_id: rootPid,
+      session_id: rootPid,
+      command: " root 41010 ",
+    },
+    {
+      at_ms: 28,
+      pid: 42_020,
+      parent_pid: rootPid,
+      process_group_id: 43_030,
+      session_id: 44_040,
+      command: "child 42020 parent 41010 group 43030 session 44040",
+    },
+    {
+      at_ms: 39,
+      pid: 45_050,
+      parent_pid: 42_020,
+      process_group_id: 43_030,
+      session_id: 44_040,
+      command: "peer 45050 unrelated 145050",
+    },
+    {
+      at_ms: 41,
+      pid: 46_060,
+      parent_pid: 0,
+      process_group_id: null,
+      session_id: 0,
+      command: "unknown 46060",
+    },
+    {
+      at_ms: 52,
+      pid: 47_070,
+      parent_pid: 0,
+      process_group_id: 0,
+      session_id: null,
+      command: "unavailable 47070",
+    },
+  ].map((sample) =>
+    Object.freeze({ ...sample, runtime_metadata: "must not be serialized" }),
+  ),
+);
+
+it.each([false, true, undefined])(
+  "projects only declared sample fields without mutating inputs when pids is %s",
+  (pids) => {
+    const scenario = parseProcessScenario({
+      executable: "/usr/bin/node",
+      ...(pids === undefined ? {} : { normalization: { pids } }),
+    });
+    const before = JSON.stringify(samples);
+    const normalized = normalizeProcessSamples(samples, scenario, rootPid);
+    const identities =
+      pids === false
+        ? [
+            [rootPid, 40_000, rootPid, rootPid],
+            [42_020, rootPid, 43_030, 44_040],
+            [45_050, 42_020, 43_030, 44_040],
+            [46_060, 0, null, 0],
+            [47_070, 0, 0, null],
+          ]
+        : [
+            [1, 2, 1, 1],
+            [3, 1, 4, 5],
+            [6, 3, 4, 5],
+            [7, 0, null, 0],
+            [8, 0, 0, null],
+          ];
+    const commands =
+      pids === false
+        ? samples.map(({ command }) => command.trim())
+        : [
+            "root <pid>",
+            "child <pid> parent <pid> group <pid> session <pid>",
+            "peer <pid> unrelated 145050",
+            "unknown <pid>",
+            "unavailable <pid>",
+          ];
+
+    expect(scenario.normalization.pids).toBe(pids ?? true);
+    expect(normalized).toEqual(
+      identities.map(
+        ([pid, parent_pid, process_group_id, session_id], index) => ({
+          at_ms: (index + 1) * 10,
+          pid,
+          parent_pid,
+          process_group_id,
+          session_id,
+          command: commands[index],
+        }),
+      ),
+    );
+    expect(normalized).not.toBe(samples);
+    for (const [index, sample] of normalized.entries()) {
+      expect(sample).not.toBe(samples[index]);
+      expect(Object.keys(sample).sort()).toEqual(
+        [
+          "at_ms",
+          "pid",
+          "parent_pid",
+          "process_group_id",
+          "session_id",
+          "command",
+        ].sort(),
+      );
+    }
+    expect(JSON.stringify(samples)).toBe(before);
+  },
+);
+
+for (const pids of [false, true]) {
+  for (const paths of [false, true]) {
+    it.each([false, true])(
+      `keeps time, patterns, paths=${String(paths)}, and ports=%s independent of pids=${String(pids)}`,
+      (ports) => {
+        const sample: ProcessSample = Object.freeze({
+          at_ms: 29,
+          pid: 42_020,
+          parent_pid: rootPid,
+          process_group_id: 43_030,
+          session_id: 44_040,
+          command:
+            " /workspace/program child 42020 parent 41010 group 43030 session 44040 listen:8080 marker ",
+        });
+        const scenario = parseProcessScenario({
+          executable: "/usr/bin/node",
+          working_directory: "/workspace",
+          normalization: {
+            pids,
+            paths,
+            ports,
+            time_bucket_ms: 25,
+            patterns: [{ pattern: "marker", replacement: "selected" }],
+          },
+        });
+        const normalized = normalizeProcessSamples([sample], scenario, rootPid);
+        const directory = paths ? "<working-directory>" : "/workspace";
+        const identifiers = pids
+          ? "child <pid> parent <pid> group <pid> session <pid>"
+          : "child 42020 parent 41010 group 43030 session 44040";
+        expect(normalized).toEqual([
+          {
+            at_ms: 25,
+            pid: pids ? 2 : 42_020,
+            parent_pid: pids ? 1 : rootPid,
+            process_group_id: pids ? 3 : 43_030,
+            session_id: pids ? 4 : 44_040,
+            command: `${directory}/program ${identifiers} listen:${ports ? "<port>" : "8080"} selected`,
+          },
+        ]);
+      },
+    );
+  }
+}

src/application/ProcessNormalization.ts

@@ -39,6 +39,7 @@ export const normalizeProcessText = (
   return normalized;
 };
 
+/** Project sampled observations under the caller-selected normalization rules. */
 export const normalizeProcessSamples = (
   samples: readonly ProcessSample[],
   scenario: ProcessScenario,
@@ -74,14 +75,20 @@ export const normalizeProcessSamples = (
       sample.at_ms,
       scenario.normalization.time_bucket_ms,
     ),
-    pid: mapping.get(sample.pid) ?? 1,
-    parent_pid: mapping.get(sample.parent_pid) ?? 0,
+    pid: scenario.normalization.pids
+      ? (mapping.get(sample.pid) ?? 1)
+      : sample.pid,
+    parent_pid: scenario.normalization.pids
+      ? (mapping.get(sample.parent_pid) ?? 0)
+      : sample.parent_pid,
     process_group_id:
-      sample.process_group_id === null
-        ? null
+      !scenario.normalization.pids || sample.process_group_id === null
+        ? sample.process_group_id
         : (mapping.get(sample.process_group_id) ?? 0),
     session_id:
-      sample.session_id === null ? null : (mapping.get(sample.session_id) ?? 0),
+      !scenario.normalization.pids || sample.session_id === null
+        ? sample.session_id
+        : (mapping.get(sample.session_id) ?? 0),
     command: normalizeCommand(sample.command),
   }));
 };

tests/boundary/process/processCaptureNormalization.test.ts

@@ -0,0 +1,93 @@
+import { readFile } from "node:fs/promises";
+import { join } from "node:path";
+import { fileURLToPath } from "node:url";
+import { expect, it } from "vitest";
+import { z } from "zod";
+
+import { captureProcessScenario } from "../../../src/application/ProcessHarness.js";
+import { createProcessCaptureEvidence } from "../../../src/application/ProcessEvidence.js";
+import {
+  digestProcessCommitment,
+  parseProcessCapture,
+  parseProcessScenario,
+} from "../../../src/domain/processCapture.js";
+import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
+import { itWithCaptureCapability } from "./processCaptureCapability.js";
+
+const processFixture = fileURLToPath(
+  new URL("../../fixtures/processIdentity.mjs", import.meta.url),
+);
+const identitySchema = z.object({
+  pid: z.number().int().positive(),
+  parent_pid: z.number().int().positive(),
+  process_group_id: z.number().int().positive(),
+  session_id: z.number().int().positive(),
+});
+
+const itWithLinuxCaptureCapability =
+  process.platform === "linux" ? itWithCaptureCapability : it.skip;
+
+itWithLinuxCaptureCapability.each([false, true])(
+  "honors pids=%s in owned Linux capture samples and committed Evidence",
+  async (pids) => {
+    const root = await createTestTempDirectory("rea-process-identifiers-");
+    const identityPath = join(root, "identity.json");
+    const scenario = parseProcessScenario({
+      executable: process.execPath,
+      arguments: [processFixture, identityPath],
+      working_directory: root,
+      normalization: { pids, ports: false, paths: false },
+      timeout_ms: 10_000,
+      idle_timeout_ms: 10_000,
+    });
+    const result = await captureProcessScenario(scenario);
+    expect(result.ok).toBe(true);
+    if (!result.ok) throw result.error;
+    const capture = result.value;
+    const identity = identitySchema.parse(
+      JSON.parse(await readFile(identityPath, "utf8")),
+    );
+    expect(identity.process_group_id).toBe(identity.pid);
+    expect(identity.session_id).toBe(identity.pid);
+    expect(identity.parent_pid).not.toBe(identity.pid);
+    const fixtureSamples = capture.process_samples.filter(({ command }) =>
+      command.includes(processFixture),
+    );
+    expect(fixtureSamples.length).toBeGreaterThan(0);
+    for (const sample of fixtureSamples)
+      expect(sample).toMatchObject(
+        pids
+          ? { pid: 1, parent_pid: 2, process_group_id: 1, session_id: 1 }
+          : identity,
+      );
+    const output = capture.frames.map(({ data }) => data).join("");
+    if (pids) expect(output).toContain('"pid":<pid>');
+    else expect(output).toContain(JSON.stringify(identity));
+
+    const evidence = createProcessCaptureEvidence(scenario, capture);
+    const serializedCapture = parseProcessCapture(evidence.normalized_result);
+    expect(evidence.parameters).toMatchObject({ normalization: { pids } });
+    expect(serializedCapture.process_samples).toEqual(capture.process_samples);
+    expect(serializedCapture.normalization.pids).toBe(pids);
+    expect(serializedCapture.manifest.scenario).toMatchObject({
+      normalization: { pids },
+    });
+    expect(serializedCapture.manifest.comparison_contract).toMatchObject({
+      normalization: { pids },
+    });
+    expect(serializedCapture.manifest.normalization_sha256).toBe(
+      digestProcessCommitment(scenario.normalization),
+    );
+    expect(capture.exit).toMatchObject({ code: 0, reason: "exited" });
+    expect([null, 0]).toContain(capture.exit.signal);
+    expect(capture.settlement).toMatchObject({
+      state: "quiesced",
+      cleanup_outcome: "not_required",
+    });
+    expect(capture.cleanup).toEqual({
+      owned_process_group: "verified",
+      temporary_root: "removed",
+    });
+  },
+  20_000,
+);

tests/fixtures/processIdentity.mjs

@@ -0,0 +1,19 @@
+import { readFile, writeFile } from "node:fs/promises";
+import { setTimeout } from "node:timers/promises";
+
+// Read the producer's own kernel identity independently of REA's sampler.
+const stat = await readFile("/proc/self/stat", "utf8");
+const fields = stat
+  .slice(stat.lastIndexOf(")") + 2)
+  .trim()
+  .split(/\s+/u);
+const identity = {
+  pid: Number(stat.slice(0, stat.indexOf(" "))),
+  parent_pid: Number(fields[1]),
+  process_group_id: Number(fields[2]),
+  session_id: Number(fields[3]),
+};
+await writeFile(process.argv[2], JSON.stringify(identity));
+process.stdout.write(`${JSON.stringify(identity)}\n`);
+// Remain observable across multiple process-sampling intervals, then exit cleanly.
+await setTimeout(500);