diff --git a/docs/process-capture.md b/docs/process-capture.md index 52f0ab4d29..7659988b4b 100644 --- a/docs/process-capture.md +++ b/docs/process-capture.md @@ -311,6 +311,15 @@ that derived observation. A corrected raw-text projection that exceeds `limits.output_bytes` returns a typed capacity failure with collected facts in partial observations. No additional raw-screen copy or replay is retained. +Before failure cleanup renders retained observations, REA cancels scheduled +interactions, waits for sampling to stop, and stops the owned process group. +The terminal finalizer then freezes PTY input, retains normalized and rendered +facts, and releases the renderer. Cleanup retry reuses those completed facts. +All new captures commit +`terminal_observation_version: "deferred-rendered-journal-v1"`, including when +PID and port normalization are disabled. Historical Evidence remains readable; +recapture both sides with the same observation version before comparison. + When path normalization is enabled, aliases apply to selected path identities at clear text delimiters. A selected executable path does not alias a longer sibling filename such as `tool-helper`; selected parent directories still diff --git a/src/domain/process/processCapture.comparison.test.ts b/src/domain/process/processCapture.comparison.test.ts index 7c7cbd590b..1ffb5d2f12 100644 --- a/src/domain/process/processCapture.comparison.test.ts +++ b/src/domain/process/processCapture.comparison.test.ts @@ -1,4 +1,11 @@ import { expect, it } from "vitest"; +import { + parseProcessScenario, + processScenarioCommitment, + processComparisonContract, + digestProcessCommitment, +} from "./processScenario.js"; +import { parseProcessCapture } from "./processCaptureParsing.js"; import { deriveProcessComparisonStatus, @@ -322,3 +329,50 @@ it.each([ }); }, ); + +it("refuses mixed terminal observation versions even with PID and port rules disabled", () => { + const base = emptyCapture(); + const scenario = parseProcessScenario({ + executable: "/usr/bin/true", + normalization: { pids: false, ports: false }, + }); + const committed = processScenarioCommitment( + scenario, + base.manifest.selected_executable_sha256 ?? undefined, + ); + const contract = processComparisonContract(scenario); + expect(contract).not.toHaveProperty("pid_normalization_version"); + expect(contract).not.toHaveProperty("port_normalization_version"); + expect(contract.terminal_observation_version).toBe( + "deferred-rendered-journal-v1", + ); + const capture = parseProcessCapture({ + ...base, + normalization: scenario.normalization, + manifest: { + ...base.manifest, + scenario: committed, + full_scenario_sha256: digestProcessCommitment(committed), + normalization_sha256: digestProcessCommitment(scenario.normalization), + comparison_contract: contract, + comparison_contract_sha256: digestProcessCommitment(contract), + }, + }); + const legacyContract = { ...contract }; + delete legacyContract.terminal_observation_version; + const legacy = parseProcessCapture({ + ...capture, + manifest: { + ...capture.manifest, + comparison_contract: legacyContract, + comparison_contract_sha256: digestProcessCommitment(legacyContract), + }, + }); + expect(() => compareProcessCaptures(capture, legacy)).toThrow( + expect.objectContaining({ + issues: [ + expect.objectContaining({ expected: ["terminal_observation_version"] }), + ], + }), + ); +}); diff --git a/src/domain/process/processScenario.test.ts b/src/domain/process/processScenario.test.ts index 734c7d5522..6c32e5b3b4 100644 --- a/src/domain/process/processScenario.test.ts +++ b/src/domain/process/processScenario.test.ts @@ -148,8 +148,12 @@ it("keeps scenario identity without finalization while versioning corrected port ).toBe("b50005d26269bd1db4a49c79e05bca67c3a18d837b45cd1c8ff591c01eb7623a"); const comparison = processComparisonContract(scenario); expect(comparison.port_normalization_version).toBe("contextual-endpoints-v3"); - const { pid_normalization_version: pidVersion, ...rootOnlyComparison } = - comparison; + const { + pid_normalization_version: pidVersion, + terminal_observation_version: terminalVersion, + ...rootOnlyComparison + } = comparison; + expect(terminalVersion).toBe("deferred-rendered-journal-v1"); expect(pidVersion).toBe("sampled-identifiers-v2"); expect( digestProcessCommitment({ diff --git a/src/domain/process/processScenario.ts b/src/domain/process/processScenario.ts index bc372a583e..11af5318ca 100644 --- a/src/domain/process/processScenario.ts +++ b/src/domain/process/processScenario.ts @@ -163,6 +163,7 @@ export const processComparisonContract = ( : {}), limits: scenario.limits, normalization: scenario.normalization, + terminal_observation_version: "deferred-rendered-journal-v1", ...(scenario.normalization.pids ? { pid_normalization_version: "sampled-identifiers-v2" } : {}), diff --git a/src/process/capture/ProcessCaptureLifecycle.ts b/src/process/capture/ProcessCaptureLifecycle.ts index 0e96e0af22..9b09adbbd3 100644 --- a/src/process/capture/ProcessCaptureLifecycle.ts +++ b/src/process/capture/ProcessCaptureLifecycle.ts @@ -1131,7 +1131,7 @@ const releaseCapturedProcessGroup = async ( export interface ProcessCaptureResourceCleanupOptions { readonly timers: ReadonlySet; readonly terminal: Pick | undefined; - readonly renderer: TerminalRenderer | undefined; + readonly renderer: Pick | undefined; readonly runId: string; readonly temporaryRoot: string; readonly captureBaseline?: ProcessOwnershipBaseline; @@ -1207,8 +1207,8 @@ class ProcessCaptureResourceOwner { async close(): Promise { for (const timer of this.options.timers) timer.cancel(); - await this.#releaseRenderer(); await this.#releaseOwnedProcessGroup(); + await this.#releaseRenderer(); await this.#releaseTemporaryRoot(); return this.report; } @@ -1448,7 +1448,9 @@ export const captureTerminalFrames = (options: { let observedFrames = 0; let normalizedOmittedFrames = 0; let firstIdentifiers: readonly number[] | undefined; - options.terminal.onData((data) => { + let sealed = false; + const subscription = options.terminal.onData((data) => { + if (sealed) return; options.onOutput(); const bytes = Buffer.byteLength(data); observedBytes += bytes; @@ -1480,6 +1482,10 @@ export const captureTerminalFrames = (options: { options.recordEvent("frames", sequence); }); return () => { + if (!sealed) { + sealed = true; + subscription.dispose(); + } const identifiers = options.identifiers(); if ( options.scenario.normalization.pids && diff --git a/src/process/capture/ProcessHarness.ts b/src/process/capture/ProcessHarness.ts index e287b3d0c3..f839781551 100644 --- a/src/process/capture/ProcessHarness.ts +++ b/src/process/capture/ProcessHarness.ts @@ -5,8 +5,10 @@ import { parseProcessCapture } from "../../domain/process/processCaptureParsing. import { err, ok, type Result } from "../../domain/result.js"; import { AnalysisCapabilityUnavailableError } from "../../domain/analysisErrorCore.js"; import { AnalysisError } from "../../domain/analysisErrorBase.js"; -import { projectAnalysisError } from "../../domain/analysisErrorProjection.js"; -import { createProcessTerminalFinalizer } from "./ProcessTerminalObservation.js"; +import { + createProcessTerminalFinalizer, + retainTerminalObservationFailure, +} from "./ProcessTerminalObservation.js"; import { analysisErrorWithCleanupFailure } from "../../domain/analysisErrorCleanup.js"; import { describeProcessCaptureExecutionFailure, @@ -90,7 +92,9 @@ interface StartedCaptureRuntime { readonly startedAt: Date; readonly executableIdentity: ReturnType; readonly lastOutput: () => number; - readonly completeTerminal: ReturnType; + readonly terminalObservation: ReturnType< + typeof createProcessTerminalFinalizer + >; readonly stopSampler: ReturnType; /** One ownership ledger shared by observations and final cleanup for this run. */ readonly ownershipEvidence: ProcessOwnershipEvidenceOwner; @@ -172,6 +176,9 @@ const cleanupFailedStartup = async (options: { readonly timers: Set; readonly terminal: IPty | undefined; readonly renderer: TerminalRenderer | undefined; + readonly terminalObservation: + | ReturnType + | undefined; readonly runId: string; readonly captureBaseline: ProcessOwnershipBaseline; readonly ownershipEvidence: ProcessOwnershipEvidenceOwner; @@ -189,23 +196,10 @@ const cleanupFailedStartup = async (options: { options.observations.process_samples.value, ).filter((groupId) => groupId !== options.terminal?.pid) : undefined; - if (options.observations !== undefined && options.renderer !== undefined) { - try { - options.observations.rendered_frames = { - state: "available", - value: await options.renderer.frames(), - }; - } catch (cause: unknown) { - options.observations.rendered_frames = { - state: "unavailable", - reason: `Rendered terminal frame collection failed: ${cause instanceof Error ? cause.message : "unknown failure"}`, - }; - } - } const cleanupOptions: ProcessCaptureResourceCleanupOptions = { timers: options.timers, terminal: options.terminal, - renderer: options.renderer, + renderer: options.terminalObservation ?? options.renderer, runId: options.runId, captureBaseline: options.captureBaseline, ownershipEvidence: options.ownershipEvidence, @@ -217,11 +211,17 @@ const cleanupFailedStartup = async (options: { options.resourceScope === undefined ? await releaseProcessResources(cleanupOptions) : await options.resourceScope.release(cleanupOptions); + let cause = options.cause; + if (options.terminalObservation !== undefined) { + const terminal = await options.terminalObservation.complete(); + if (!terminal.ok) + cause = retainTerminalObservationFailure(cause, terminal.error); + } const cleanupFailure = cleanupReportFailure(cleanupOutcome.report); if (cleanupFailure !== undefined) { resolveProcessResult( undefined, - options.cause, + cause, cleanupOutcome, options.observations, { @@ -232,7 +232,7 @@ const cleanupFailedStartup = async (options: { }, ); } - throw options.cause; + throw cause; }; const createTerminalRenderer = ( @@ -280,6 +280,10 @@ const startCaptureRuntime = async ( const { ownershipEvidence } = options; let renderer: TerminalRenderer | undefined; let terminal: IPty | undefined; + let terminalObservation: + | ReturnType + | undefined; + let stopSampler: ReturnType | undefined; try { const { spawn } = await import("@lydell/node-pty"); const selectedExecutable = await observeSelectedExecutable( @@ -333,13 +337,18 @@ const startCaptureRuntime = async ( normalizeText, identifiers, }); + terminalObservation = createProcessTerminalFinalizer( + renderer, + terminalRetention, + options.observationBuffer, + ); scheduleScenarioInteractions({ ...options, getTerminal: () => terminal, renderer, started, }); - const stopSampler = startProcessSampler({ + stopSampler = startProcessSampler({ rootPid: terminal.pid, runId: options.runId, started, @@ -355,11 +364,7 @@ const startCaptureRuntime = async ( startedAt, executableIdentity, lastOutput: () => lastOutput, - completeTerminal: createProcessTerminalFinalizer( - renderer, - terminalRetention, - options.observationBuffer, - ), + terminalObservation, stopSampler, signalRoot: retainRootSignaller( terminal.pid, @@ -369,11 +374,14 @@ const startCaptureRuntime = async ( ), }; } catch (cause: unknown) { + for (const timer of options.timers) timer.cancel(); + await stopSampler?.(); return cleanupFailedStartup({ cause, timers: options.timers, terminal, renderer, + terminalObservation, runId: options.runId, captureBaseline: options.ownershipBaseline, ownershipEvidence, @@ -411,27 +419,9 @@ const finishProcessRun = async (options: { readonly signal?: AbortSignal; }): Promise => { options.progress.closeLive(); + for (const timer of options.timers) timer.cancel(); await options.stopSampler(); let { capture, executionFailure } = options; - if (options.runtime !== undefined && capture === undefined) { - const terminal = await options.runtime.completeTerminal(); - if (!terminal.ok) { - if (executionFailure === undefined) executionFailure = terminal.error; - else { - const primary = - executionFailure instanceof AnalysisError - ? executionFailure - : new ProcessCaptureError("process capture failed", { - cause: executionFailure, - }); - if (primary !== terminal.error) - primary.retainSecondaryFailure(terminal.error, { - terminal_observation_error: projectAnalysisError(terminal.error), - }); - executionFailure = primary; - } - } - } const sampledProcessGroupIds = options.runtime === undefined ? undefined @@ -442,7 +432,7 @@ const finishProcessRun = async (options: { const cleanupOptions: ProcessCaptureResourceCleanupOptions = { timers: options.timers, terminal: options.runtime?.terminal, - renderer: options.runtime?.renderer, + renderer: options.runtime?.terminalObservation, runId: options.runId, captureBaseline: options.captureBaseline, ownershipEvidence: options.ownershipEvidence, @@ -454,6 +444,14 @@ const finishProcessRun = async (options: { options.resourceScope === undefined ? await releaseProcessResources(cleanupOptions) : await options.resourceScope.release(cleanupOptions); + if (options.runtime !== undefined && capture === undefined) { + const terminal = await options.runtime.terminalObservation.complete(); + if (!terminal.ok) + executionFailure = retainTerminalObservationFailure( + executionFailure, + terminal.error, + ); + } const cleanup = cleanupOutcome.report; if (executionFailure === undefined && options.signal?.aborted) executionFailure = processCaptureCancelled(options.signal.reason); @@ -511,6 +509,7 @@ const finishProcessRun = async (options: { }; const completeCapture = async (options: { + readonly timers: ReadonlySet; readonly scenario: ProcessScenario; readonly hostPlatform: NodeJS.Platform; readonly runtime: StartedCaptureRuntime; @@ -564,6 +563,7 @@ const completeCapture = async (options: { state: "available", value: settlement, }; + for (const timer of options.timers) timer.cancel(); const sampling = await runtime.stopSampler(); const samplingPartial = sampling.partial; await settleProcessCaptureJournal(options.eventJournal); @@ -608,7 +608,7 @@ const completeCapture = async (options: { }, }; options.recordEvent("filesystem_checkpoints", 1); - const terminal = await runtime.completeTerminal(); + const terminal = await runtime.terminalObservation.complete(); if (!terminal.ok) throw terminal.error; const { renderedFrames, retention } = terminal.value; const checkpoints: UnverifiedProcessCapture["filesystem_checkpoints"] = [ @@ -831,6 +831,7 @@ const runProcessScenario = async ( value: captureExitObservation(exit), }; capture = await completeCapture({ + timers, scenario, hostPlatform, runtime, diff --git a/src/process/capture/ProcessTerminalObservation.test.ts b/src/process/capture/ProcessTerminalObservation.test.ts index 5e3652f7dd..b15f6e7529 100644 --- a/src/process/capture/ProcessTerminalObservation.test.ts +++ b/src/process/capture/ProcessTerminalObservation.test.ts @@ -1,13 +1,106 @@ -import { expect, it, vi } from "vitest"; +import { expect, it, vi, onTestFinished } from "vitest"; +import type { IPty } from "@lydell/node-pty"; +import type { + ProcessSample, + TerminalFrame, +} from "../../domain/process/processCapture.js"; +import { createProcessCaptureJournal } from "./ProcessCaptureJournal.js"; +import { + createBoundedProcessTextNormalizer, + createProcessIdentifierLookup, +} from "./ProcessNormalization.js"; +import { scheduleProcessDelay } from "./ProcessTimer.js"; +import { TerminalRenderer } from "./TerminalRenderer.js"; +import { + ProcessCaptureError, + processCaptureCancelled, +} from "./ProcessCaptureError.js"; import { AnalysisResourceConstraintError } from "../../domain/analysisErrorCore.js"; import { projectAnalysisError } from "../../domain/analysisErrorProjection.js"; import { parseProcessScenario } from "../../domain/process/processScenario.js"; import { createProcessCaptureObservationBuffer, resolveProcessResult, + releaseProcessResources, + captureTerminalFrames, } from "./ProcessCaptureLifecycle.js"; import { createProcessTerminalFinalizer } from "./ProcessTerminalObservation.js"; +const createTerminalCleanupFixture = () => { + const scenario = parseProcessScenario({ + executable: "unused", + terminal: { columns: 80, rows: 4 }, + normalization: { paths: false, ports: false }, + }); + const frames: TerminalFrame[] = []; + const samples: ProcessSample[] = []; + const journal = createProcessCaptureJournal(); + const observations = createProcessCaptureObservationBuffer({ + frames, + samples, + interactions: [], + eventJournal: journal.entries, + before: undefined, + }); + const identifiers = createProcessIdentifierLookup(samples, () => 100); + const normalize = createBoundedProcessTextNormalizer( + scenario, + "/tmp", + identifiers, + ); + const renderer = new TerminalRenderer({ + ...scenario.terminal, + maxBytes: scenario.limits.output_bytes, + normalize, + recordEvent: journal.recordEvent, + }); + const rendererState = { released: false }; + onTestFinished(async () => { + if (!rendererState.released) await renderer.dispose(); + }); + const events: string[] = []; + let onData: ((data: string) => void) | undefined; + const terminal = { + pid: 100, + onData: (callback: (data: string) => void) => { + onData = callback; + return { + dispose: () => { + events.push("input sealed"); + }, + }; + }, + } as unknown as IPty; + const retention = captureTerminalFrames({ + terminal, + scenario, + frames, + started: 0, + onOutput: () => undefined, + renderer, + normalizeText: normalize, + identifiers, + recordEvent: journal.recordEvent, + }); + const owner = createProcessTerminalFinalizer( + renderer, + retention, + observations, + ); + return { + scenario, + frames, + samples, + journal, + observations, + renderer, + rendererState, + events, + owner, + emitOutput: (data: string) => onData?.(data), + }; +}; + it("retains completed terminal facts and the typed capacity failure through finalization", async () => { const failure = new AnalysisResourceConstraintError( "capture_process", @@ -39,14 +132,23 @@ it("retains completed terminal facts and the typed capacity failure through fina serialized_state: "", }, ]; - const renderer = { frames: vi.fn(async () => rendered) }; + const closeFailure = new Error("renderer close failed once"); + const renderer = { + frames: vi.fn(async () => rendered), + dispose: vi + .fn<() => Promise>() + .mockRejectedValueOnce(closeFailure) + .mockResolvedValue(undefined), + }; const finalize = createProcessTerminalFinalizer( renderer, readRetention, observations, ); - const result = await finalize(); - expect(await finalize()).toBe(result); + const result = await finalize.complete(); + await expect(finalize.dispose()).rejects.toBe(closeFailure); + await finalize.dispose(); + expect(await finalize.complete()).toBe(result); expect(readRetention).toHaveBeenCalledOnce(); expect(renderer.frames).toHaveBeenCalledOnce(); if (result.ok) throw new Error("Expected terminal capacity refusal"); @@ -88,3 +190,122 @@ it("retains completed terminal facts and the typed capacity failure through fina }, }); }); + +it.each(["cancelled", "failed"] as const)( + "stops pending interactions and the target before sealing %s terminal facts", + async (outcome) => { + const { + scenario, + frames, + samples, + journal, + observations, + renderer, + rendererState, + events, + owner, + emitOutput, + } = createTerminalCleanupFixture(); + let pendingInteraction: (() => void) | undefined; + const timer = scheduleProcessDelay( + 100, + () => { + events.push("unexpected interaction"); + renderer.resize(40, 4, 2); + }, + (callback) => { + pendingInteraction = callback; + return { + cancel: () => { + events.push("timer cancelled"); + }, + }; + }, + ); + emitOutput("root 100\r\n"); + const cleanup = await releaseProcessResources({ + timers: new Set([timer]), + terminal: { pid: 100 }, + renderer: owner, + runId: "fixture-run", + temporaryRoot: "/fixture/temporary-root", + host: { + platform: "linux", + cleanupProcessGroup: async () => { + events.push("target stopped"); + pendingInteraction?.(); + emitOutput("child 200\r\n"); + samples.push({ + at_ms: 1, + pid: 200, + parent_pid: 100, + process_group_id: 100, + session_id: null, + command: "child", + }); + journal.recordEvent("process_samples", 0); + return { cleaned: true, signaled: true }; + }, + removeTemporaryRoot: async () => { + events.push("temporary root removed"); + }, + }, + }); + rendererState.released = + cleanup.report.terminal_renderer.state === "cleaned"; + expect(events).toEqual([ + "timer cancelled", + "target stopped", + "input sealed", + "temporary root removed", + ]); + const completed = await owner.complete(); + if (!completed.ok) throw completed.error; + emitOutput("late callback after sealing"); + expect(await owner.complete()).toBe(completed); + expect(completed.value.retention.raw_terminal).toMatchObject({ + observed_frames: 2, + retained_frames: 2, + observed_bytes: 21, + retained_bytes: 21, + }); + expect(frames.map(({ data }) => data).join("")).toBe( + "root \r\nchild \r\n", + ); + expect(frames.map(({ raw_data, data }) => raw_data ?? data).join("")).toBe( + "root 100\r\nchild 200\r\n", + ); + expect(completed.value.renderedFrames.at(-1)?.lines).toEqual( + expect.arrayContaining(["root ", "child "]), + ); + const reason = new Error("original capture failure"); + const primary = + outcome === "cancelled" ? processCaptureCancelled(reason) : reason; + let failure: unknown; + try { + resolveProcessResult(undefined, primary, cleanup, observations, { + scenario, + rootPid: 100, + }); + } catch (cause: unknown) { + failure = cause; + } + if (!(failure instanceof ProcessCaptureError)) + throw new Error("Expected a typed capture failure"); + expect(failure.cause).toBe(reason); + expect(projectAnalysisError(failure)).toMatchObject({ + code: outcome === "cancelled" ? "cancelled" : "process_capture_failed", + details: { + partial_observation: { + observations: { + frames: { state: "available", value: frames }, + rendered_frames: { + state: "available", + value: completed.value.renderedFrames, + }, + }, + }, + }, + }); + }, +); diff --git a/src/process/capture/ProcessTerminalObservation.ts b/src/process/capture/ProcessTerminalObservation.ts index 6d639ad484..db55f770ff 100644 --- a/src/process/capture/ProcessTerminalObservation.ts +++ b/src/process/capture/ProcessTerminalObservation.ts @@ -22,17 +22,46 @@ const terminalFailure = (cause: unknown): AnalysisError => { cause }, ); +/** Keep a primary capture failure while retaining the finalizer's typed diagnostic. */ +export const retainTerminalObservationFailure = ( + primary: unknown, + failure: AnalysisError, +): AnalysisError => { + if (primary === undefined) return failure; + const error = + primary instanceof AnalysisError + ? primary + : new ProcessCaptureError("process capture failed", { cause: primary }); + if (error !== failure) + error.retainSecondaryFailure(failure, { + terminal_observation_error: projectAnalysisError(failure), + }); + return error; +}; + /** Finalize retained chunks once after sampling, for both completion and error cleanup. */ export const createProcessTerminalFinalizer = ( - renderer: Pick, + renderer: Pick, retention: () => TerminalCaptureRetention, observations: ProcessCaptureObservationBuffer, -): (() => Promise>) => { +): { + readonly complete: () => Promise< + Result + >; + readonly dispose: () => Promise; +} => { let completed: | Promise> | undefined; - return () => + const complete = () => (completed ??= observeTerminal(renderer, retention, observations)); + return { + complete, + dispose: async () => { + await complete(); + await renderer.dispose(); + }, + }; }; const observeTerminal = async ( diff --git a/tests/boundary/process/processCaptureLifecycle.test.ts b/tests/boundary/process/processCaptureLifecycle.test.ts index 68694d876b..10c92357c1 100644 --- a/tests/boundary/process/processCaptureLifecycle.test.ts +++ b/tests/boundary/process/processCaptureLifecycle.test.ts @@ -552,7 +552,18 @@ itWithCaptureCapability( file.path.endsWith(":escape"), ); expect(escaped?.symlink_target).toBe("/etc/passwd"); - expect(processSourceTruncated(capture, "filesystem")).toBe(false); + expect(processSourceTruncated(capture, "filesystem")).toBe(true); + for (const coverage of [ + capture.truncation_details.filesystem_before, + capture.truncation_details.filesystem_after, + ]) { + expect(coverage.enumeration_reasons).toContain("symlink_not_followed"); + expect(coverage.enumeration_omissions).toContainEqual({ + path: escaped?.path, + reason: "symlink_not_followed", + system_code: null, + }); + } expect(JSON.stringify(capture.files_after)).not.toContain(root); expect( capture.files_after.some((file) => file.path.includes("passwd")), diff --git a/tests/boundary/process/processDescendantPidAdapters.test.ts b/tests/boundary/process/processDescendantPidAdapters.test.ts new file mode 100644 index 0000000000..8629e1d3ab --- /dev/null +++ b/tests/boundary/process/processDescendantPidAdapters.test.ts @@ -0,0 +1,167 @@ +import { execFile } from "node:child_process"; +import { readFile, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { promisify } from "node:util"; +import { Client, InMemoryTransport } from "@modelcontextprotocol/client"; +import { expect, onTestFinished } from "vitest"; +import { z } from "zod"; +import { parseEvidence } from "../../../src/domain/evidence.js"; +import { parseProcessCapture } from "../../../src/domain/process/processCaptureParsing.js"; +import { compareProcessCaptures } from "../../../src/domain/process/processComparison.js"; +import { digestProcessCommitment } from "../../../src/domain/process/processScenario.js"; +import { createServer } from "../../../src/server/createServer.js"; +import { createTestBinarySession } from "../../fixtures/binarySession.js"; +import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js"; +import { itWithCaptureCapability } from "./processCaptureCapability.js"; + +const exec = promisify(execFile); +const fixture = fileURLToPath( + new URL("../../fixtures/processDescendantIdentifiers.mjs", import.meta.url), +); +const identitySchema = z.strictObject({ + root_pid: z.int().positive(), + child_pid: z.int().positive(), +}); + +itWithCaptureCapability.each([ + { adapter: "cli", pids: true }, + { adapter: "mcp", pids: true }, + { adapter: "cli", pids: false }, + { adapter: "mcp", pids: false }, +] as const)( + "normalizes sampled child output and preserves original bytes with pids=$pids through $adapter", + async ({ adapter, pids }) => { + const root = await createTestTempDirectory("rea-descendant-pid-adapter-"); + const identityPath = join(root, "identity.json"); + const scenarioPath = join(root, "scenario.json"); + const scenario = { + executable: process.execPath, + arguments: [fixture, identityPath], + working_directory: root, + timeout_ms: 5_000, + idle_timeout_ms: 5_000, + settle_ms: 100, + normalization: { + pids, + ports: false, + paths: false, + time_bucket_ms: 10_000, + }, + }; + await writeFile(scenarioPath, JSON.stringify(scenario)); + let client: Client | undefined; + if (adapter === "mcp") { + const session = createTestBinarySession(() => { + throw new Error("Process capture must not launch a binary provider"); + }); + const server = createServer({ kind: "session", session }); + client = new Client({ + name: "descendant-pid-normalization", + version: "1", + }); + const connected = client; + onTestFinished(async () => { + await connected.close(); + await server.close(); + await session.close(); + }); + const [clientTransport, serverTransport] = + InMemoryTransport.createLinkedPair(); + await server.connect(serverTransport); + await client.connect(clientTransport); + } + const captures = []; + for (let run = 0; run < (pids ? 2 : 1); run++) { + let evidence; + if (client !== undefined) { + const response = await client.callTool({ + name: "capture_process_scenario", + arguments: scenario, + }); + expect(response.isError, JSON.stringify(response)).not.toBe(true); + evidence = parseEvidence(response.structuredContent); + await client.ping(); + } else { + const { stdout } = await exec( + process.execPath, + [ + "scripts/rea.mjs", + "capture-process", + scenarioPath, + "--format", + "json", + ], + { timeout: 20_000 }, + ); + evidence = parseEvidence(JSON.parse(stdout)); + } + const capture = parseProcessCapture(evidence.normalized_result); + const identity = identitySchema.parse( + JSON.parse(await readFile(identityPath, "utf8")), + ); + const original = `root pid ${identity.root_pid}\r\nchild pid ${identity.child_pid}\r\ncount 9007199254740991\r\n`; + const expected = pids + ? "root pid \r\nchild pid \r\ncount 9007199254740991\r\n" + : original; + const childSamples = capture.process_samples.filter(({ command }) => + command.endsWith("processDescendantIdentifiers.mjs child"), + ); + expect( + childSamples.length, + "the child must have been positively sampled", + ).toBeGreaterThan(0); + expect( + childSamples.some(({ pid, parent_pid }) => + pids + ? pid !== 1 && parent_pid === 1 + : pid === identity.child_pid && parent_pid === identity.root_pid, + ), + ).toBe(true); + expect(capture.frames.map(({ data }) => data).join("")).toBe(expected); + expect( + capture.frames.map(({ raw_data, data }) => raw_data ?? data).join(""), + ).toBe(original); + expect(capture.rendered_frames.at(-1)?.lines).toEqual( + expect.arrayContaining(expected.trimEnd().split("\r\n")), + ); + expect(capture.rendered_frames.at(-1)?.serialized_state).toContain( + pids ? "child pid " : `child pid ${identity.child_pid}`, + ); + expect(capture.truncation_details.raw_terminal).toMatchObject({ + observed_bytes: Buffer.byteLength(original), + retained_bytes: Buffer.byteLength(original), + }); + expect(capture.manifest.comparison_contract).toHaveProperty( + "terminal_observation_version", + "deferred-rendered-journal-v1", + ); + expect(capture.exit).toMatchObject({ code: 0, reason: "exited" }); + captures.push(capture); + } + const [first, second] = captures; + if (first === undefined) throw new Error("Capture was not retained"); + if (second !== undefined) + expect(compareProcessCaptures(first, second).terminal).toBe("unchanged"); + const legacyContract = { ...first.manifest.comparison_contract }; + delete legacyContract.terminal_observation_version; + const legacy = parseProcessCapture({ + ...first, + manifest: { + ...first.manifest, + comparison_contract: legacyContract, + comparison_contract_sha256: digestProcessCommitment(legacyContract), + }, + }); + expect(() => compareProcessCaptures(legacy, first)).toThrow( + expect.objectContaining({ + issues: [ + expect.objectContaining({ + expected: ["terminal_observation_version"], + }), + ], + }), + ); + }, + 30_000, +); diff --git a/tests/fixtures/processDescendantIdentifiers.mjs b/tests/fixtures/processDescendantIdentifiers.mjs new file mode 100644 index 0000000000..1bf16e5356 --- /dev/null +++ b/tests/fixtures/processDescendantIdentifiers.mjs @@ -0,0 +1,29 @@ +import { spawn } from "node:child_process"; +import { writeFileSync } from "node:fs"; +import { fileURLToPath } from "node:url"; + +if (process.argv[2] === "child") { + setTimeout(() => {}, 1_000); +} else { + const child = spawn( + process.execPath, + [fileURLToPath(import.meta.url), "child"], + { + stdio: "inherit", + }, + ); + child.on("error", (cause) => { + throw cause; + }); + if (child.pid === undefined) throw new Error("Child did not launch"); + writeFileSync( + process.argv[2], + JSON.stringify({ root_pid: process.pid, child_pid: child.pid }), + ); + process.stdout.write( + `root pid ${process.pid}\nchild pid ${child.pid}\ncount 9007199254740991\n`, + ); + child.on("exit", (code) => { + process.exitCode = code ?? 1; + }); +}