mirror of
https://github.com/morluto/rea.git
synced 2026-11-09 00:07:18 +00:00
fix(application): harden replay and process cleanup
This commit is contained in:
1 parent
c240ae19c6
commit
b5dc7b70ec
16 files changed
+620
-61
No files matched your search
@@ -31,6 +31,11 @@ Use `--expected-revision N` when an external coordinator needs an explicit
|
||||
compare-and-swap guard. Traversal, byte, page, and comparison limits are also
|
||||
available as CLI options.
|
||||
|
||||
After a completed run returns its stable run ID, `--replay-run-id RUN_ID`
|
||||
selects that exact result without reading or rescanning either version input.
|
||||
The workspace still requires read authority. REA rejects a selector that is
|
||||
missing, incomplete, or inconsistent with the requested paths and controls.
|
||||
|
||||
## MCP
|
||||
|
||||
`find_changed_behavior` retains its existing aggregation mode. Its additive
|
||||
@@ -75,6 +80,12 @@ report checkpoints. A process interrupted after a checkpoint can resume from
|
||||
the last complete stage. A completed run is reused when both graph commitments
|
||||
and every bounded option are identical.
|
||||
|
||||
By default, REA rescans the version inputs so changed content creates a new
|
||||
run. For an explicit cache-only replay, add the returned `run_id` as
|
||||
`investigation_run.replay_run_id`. REA first validates the workspace revision,
|
||||
the complete run, its inventory paths and controls, and its linked Evidence;
|
||||
only then may it return without requesting input-read authority.
|
||||
|
||||
The run ID is a SHA-256 commitment to both root and graph digests plus the
|
||||
normalized options. Local paths and timestamps do not affect that identity.
|
||||
Changing either artifact or a budget creates a new run in the same workspace;
|
||||
|
||||
@@ -45,6 +45,7 @@ import {
|
||||
ARTIFACT_COMPARISON_PROVIDER,
|
||||
CHANGED_BEHAVIOR_PROVIDER,
|
||||
} from "./InvestigationProviders.js";
|
||||
import { selectCompletedInvestigationReplay } from "./CrossVersionInvestigationReplay.js";
|
||||
|
||||
export interface CrossVersionInvestigationOutcome {
|
||||
readonly evidence: Evidence;
|
||||
@@ -59,6 +60,7 @@ export interface CrossVersionInvestigationExecution {
|
||||
readonly signal?: AbortSignal;
|
||||
readonly progress?: ProgressReporter;
|
||||
readonly integrityContinueEnabled?: boolean;
|
||||
readonly authorizeInputRead?: () => Promise<Result<null, AnalysisError>>;
|
||||
readonly authorizeWorkspaceWrite?: () => Promise<Result<null, AnalysisError>>;
|
||||
}
|
||||
|
||||
@@ -84,6 +86,23 @@ export const runCrossVersionInvestigation = async (
|
||||
if (!preflight.ok) return preflight;
|
||||
if (isCancelled(execution.signal))
|
||||
return err(new AnalysisCancelledError("find_changed_behavior"));
|
||||
if (parsed.data.replay_run_id !== undefined) {
|
||||
const replay = selectCompletedInvestigationReplay(
|
||||
initial.value,
|
||||
parsed.data,
|
||||
);
|
||||
if (!replay.ok) return replay;
|
||||
return completeOutcome(
|
||||
replay.value.workspace,
|
||||
replay.value.run,
|
||||
true,
|
||||
execution.session,
|
||||
);
|
||||
}
|
||||
if (execution.authorizeInputRead !== undefined) {
|
||||
const authorized = await execution.authorizeInputRead();
|
||||
if (!authorized.ok) return authorized;
|
||||
}
|
||||
await execution.progress?.report({
|
||||
phase: "scan_versions",
|
||||
completed: 0,
|
||||
|
||||
@@ -0,0 +1,189 @@
|
||||
import { changedBehaviorResultSchema } from "../domain/changedBehavior.js";
|
||||
import {
|
||||
EvidenceIntegrityError,
|
||||
InvestigationWorkspaceError,
|
||||
} from "../domain/errors.js";
|
||||
import type {
|
||||
CrossVersionInvestigationInput,
|
||||
InvestigationRun,
|
||||
InvestigationWorkspace,
|
||||
} from "../domain/investigationWorkspace.js";
|
||||
import { err, ok, type Result } from "../domain/result.js";
|
||||
import {
|
||||
ARTIFACT_COMPARISON_PROVIDER,
|
||||
ARTIFACT_GRAPH_PROVIDER,
|
||||
CHANGED_BEHAVIOR_PROVIDER,
|
||||
} from "./InvestigationProviders.js";
|
||||
|
||||
/** Validated workspace state returned by an explicit replay selector. */
|
||||
export interface CompletedInvestigationReplay {
|
||||
readonly workspace: InvestigationWorkspace;
|
||||
readonly run: InvestigationRun;
|
||||
}
|
||||
|
||||
/** Select and verify one caller-identified complete run without input access. */
|
||||
export const selectCompletedInvestigationReplay = (
|
||||
workspace: InvestigationWorkspace | null,
|
||||
input: CrossVersionInvestigationInput,
|
||||
): Result<
|
||||
CompletedInvestigationReplay,
|
||||
InvestigationWorkspaceError | EvidenceIntegrityError
|
||||
> => {
|
||||
const run = workspace?.runs.find(
|
||||
({ run_id: runId }) => runId === input.replay_run_id,
|
||||
);
|
||||
if (workspace === null || run === undefined || run.status !== "complete")
|
||||
return err(new InvestigationWorkspaceError("read", "revision-conflict"));
|
||||
if (!replayRequestMatches(workspace, run, input))
|
||||
return err(new InvestigationWorkspaceError("read", "revision-conflict"));
|
||||
const evidence = validateCompletedReplayEvidence(workspace, run);
|
||||
return evidence.ok ? ok({ workspace, run }) : evidence;
|
||||
};
|
||||
|
||||
const replayRequestMatches = (
|
||||
workspace: InvestigationWorkspace,
|
||||
run: InvestigationRun,
|
||||
input: CrossVersionInvestigationInput,
|
||||
): boolean =>
|
||||
JSON.stringify(run.options) === JSON.stringify(input.options) &&
|
||||
run.integrity_policy === input.integrity_policy &&
|
||||
run.integrity_continue_approved === input.integrity_continue_approved &&
|
||||
run.max_integrity_mismatches === input.max_integrity_mismatches &&
|
||||
inventoryEvidenceMatches(
|
||||
workspace,
|
||||
run.left_inventory_evidence_ids,
|
||||
input.left_path,
|
||||
run.left,
|
||||
) &&
|
||||
inventoryEvidenceMatches(
|
||||
workspace,
|
||||
run.right_inventory_evidence_ids,
|
||||
input.right_path,
|
||||
run.right,
|
||||
);
|
||||
|
||||
const inventoryEvidenceMatches = (
|
||||
workspace: InvestigationWorkspace,
|
||||
evidenceIds: readonly string[],
|
||||
path: string,
|
||||
target: InvestigationRun["left"],
|
||||
): boolean => {
|
||||
const byId = new Map(
|
||||
workspace.bundle.records.map((record) => [record.evidence_id, record]),
|
||||
);
|
||||
return evidenceIds.every((evidenceId) => {
|
||||
const evidence = byId.get(evidenceId);
|
||||
return (
|
||||
evidence?.operation === "inventory_artifact" &&
|
||||
providerMatches(evidence.provider, ARTIFACT_GRAPH_PROVIDER) &&
|
||||
evidence.subject?.local_path === path &&
|
||||
evidence.subject.digest.sha256 === target.root_sha256 &&
|
||||
evidence.subject.format === target.format
|
||||
);
|
||||
});
|
||||
};
|
||||
|
||||
const validateCompletedReplayEvidence = (
|
||||
workspace: InvestigationWorkspace,
|
||||
run: InvestigationRun,
|
||||
): Result<null, EvidenceIntegrityError> => {
|
||||
const comparison = workspace.bundle.records.find(
|
||||
({ evidence_id: evidenceId }) => evidenceId === run.comparison_evidence_id,
|
||||
);
|
||||
const result = workspace.bundle.records.find(
|
||||
({ evidence_id: evidenceId }) => evidenceId === run.result_evidence_id,
|
||||
);
|
||||
const parsedResult = changedBehaviorResultSchema.safeParse(
|
||||
result?.normalized_result,
|
||||
);
|
||||
const summary = parsedResult.success
|
||||
? parsedResult.data.investigation_run
|
||||
: undefined;
|
||||
const parsedResultLinks = parsedResult.success
|
||||
? parsedResult.data.evidence_links
|
||||
: [];
|
||||
const inventoryIds = [
|
||||
...run.left_inventory_evidence_ids,
|
||||
...run.right_inventory_evidence_ids,
|
||||
];
|
||||
const resultLinks =
|
||||
comparison === undefined
|
||||
? []
|
||||
: [...new Set([comparison.evidence_id, ...inventoryIds])].sort(
|
||||
(left, right) => left.localeCompare(right, "en"),
|
||||
);
|
||||
const valid =
|
||||
comparison?.operation === "compare_artifacts" &&
|
||||
comparison.predicate_type === "rea.artifact-comparison/v1" &&
|
||||
providerMatches(comparison.provider, ARTIFACT_COMPARISON_PROVIDER) &&
|
||||
comparison.subject === null &&
|
||||
comparison.confidence === "derived" &&
|
||||
comparison.authority === "analyst-inference" &&
|
||||
JSON.stringify(comparison.evidence_links) ===
|
||||
JSON.stringify(inventoryIds) &&
|
||||
comparisonParametersMatch(comparison.parameters, run) &&
|
||||
result?.operation === "find_changed_behavior" &&
|
||||
result.predicate_type === "rea.changed-behavior/v1" &&
|
||||
providerMatches(result.provider, CHANGED_BEHAVIOR_PROVIDER) &&
|
||||
result.subject === null &&
|
||||
result.confidence === "derived" &&
|
||||
result.authority === "analyst-inference" &&
|
||||
JSON.stringify(result.evidence_links) === JSON.stringify(resultLinks) &&
|
||||
resultParametersMatch(result.parameters, workspace, run) &&
|
||||
summary?.workspace_id === workspace.workspace_id &&
|
||||
summary.run_id === run.run_id &&
|
||||
summary.left_manifest_id === run.left.manifest_id &&
|
||||
summary.right_manifest_id === run.right.manifest_id &&
|
||||
summary.inventory_evidence_count === inventoryIds.length &&
|
||||
summary.comparison_evidence_id === run.comparison_evidence_id &&
|
||||
JSON.stringify(summary.limitations) === JSON.stringify(run.limitations) &&
|
||||
JSON.stringify(parsedResultLinks) === JSON.stringify(resultLinks);
|
||||
return valid
|
||||
? ok(null)
|
||||
: err(
|
||||
new EvidenceIntegrityError(
|
||||
"Completed investigation replay Evidence is inconsistent",
|
||||
),
|
||||
);
|
||||
};
|
||||
|
||||
const comparisonParametersMatch = (
|
||||
parameters: Readonly<Record<string, unknown>>,
|
||||
run: InvestigationRun,
|
||||
): boolean =>
|
||||
Object.keys(parameters).length === 4 &&
|
||||
JSON.stringify(parameters.left_evidence_ids) ===
|
||||
JSON.stringify(run.left_inventory_evidence_ids) &&
|
||||
JSON.stringify(parameters.right_evidence_ids) ===
|
||||
JSON.stringify(run.right_inventory_evidence_ids) &&
|
||||
parameters.offset === 0 &&
|
||||
parameters.limit === run.options.change_limit;
|
||||
|
||||
const resultParametersMatch = (
|
||||
parameters: Readonly<Record<string, unknown>>,
|
||||
workspace: InvestigationWorkspace,
|
||||
run: InvestigationRun,
|
||||
): boolean =>
|
||||
Object.keys(parameters).length === 5 &&
|
||||
parameters.workspace_id === workspace.workspace_id &&
|
||||
parameters.run_id === run.run_id &&
|
||||
JSON.stringify(parameters.comparison_evidence_ids) ===
|
||||
JSON.stringify([run.comparison_evidence_id]) &&
|
||||
parameters.offset === 0 &&
|
||||
parameters.limit === Math.min(run.options.change_limit, 100);
|
||||
|
||||
const providerMatches = (
|
||||
actual: {
|
||||
readonly id: string;
|
||||
readonly name: string;
|
||||
readonly version: string | null;
|
||||
},
|
||||
expected: {
|
||||
readonly id: string;
|
||||
readonly name: string;
|
||||
readonly version: string | null;
|
||||
},
|
||||
): boolean =>
|
||||
actual.id === expected.id &&
|
||||
actual.name === expected.name &&
|
||||
actual.version === expected.version;
|
||||
@@ -19,6 +19,7 @@ interface ProcessGroupMember {
|
||||
readonly pid: number;
|
||||
readonly parentPid: number;
|
||||
readonly processGroupId: number;
|
||||
readonly state: string;
|
||||
readonly command: string;
|
||||
}
|
||||
|
||||
@@ -30,9 +31,19 @@ export interface ProcessOwnershipHost {
|
||||
}
|
||||
|
||||
/** Cleanup succeeds only when the group is absent or every member is owned. */
|
||||
export interface ProcessOwnershipValidationFailure {
|
||||
readonly pid: number;
|
||||
readonly reason: "environment-unreadable" | "run-token-mismatch";
|
||||
}
|
||||
|
||||
/** Cleanup outcome with per-member diagnostics when ownership is uncertain. */
|
||||
export type ProcessCleanupResult =
|
||||
| { readonly cleaned: true; readonly signaled: boolean }
|
||||
| { readonly cleaned: false; readonly reason: string };
|
||||
| {
|
||||
readonly cleaned: false;
|
||||
readonly reason: string;
|
||||
readonly failures?: readonly ProcessOwnershipValidationFailure[];
|
||||
};
|
||||
|
||||
/** Token-verified liveness result used by post-root-exit settlement. */
|
||||
export type ProcessGroupObservation =
|
||||
@@ -59,17 +70,18 @@ const systemHost: ProcessOwnershipHost = {
|
||||
if (process.platform === "win32") return [];
|
||||
const { stdout } = await execFileAsync("ps", [
|
||||
"-axo",
|
||||
"pid=,ppid=,pgid=,command=",
|
||||
"pid=,ppid=,pgid=,stat=,command=",
|
||||
]);
|
||||
return stdout
|
||||
.split("\n")
|
||||
.map((line) => /\s*(\d+)\s+(\d+)\s+(\d+)\s+(.*)/u.exec(line))
|
||||
.map((line) => /\s*(\d+)\s+(\d+)\s+(\d+)\s+(\S+)\s+(.*)/u.exec(line))
|
||||
.filter((match): match is RegExpExecArray => match !== null)
|
||||
.map((match) => ({
|
||||
pid: Number(match[1]),
|
||||
parentPid: Number(match[2]),
|
||||
processGroupId: Number(match[3]),
|
||||
command: match[4] ?? "",
|
||||
state: match[4] ?? "",
|
||||
command: match[5] ?? "",
|
||||
}))
|
||||
.filter((member) => member.processGroupId === processGroupId);
|
||||
},
|
||||
@@ -109,8 +121,11 @@ export const cleanupOwnedProcessGroup = async (
|
||||
} catch {
|
||||
return { cleaned: false, reason: "process group could not be inspected" };
|
||||
}
|
||||
if (members.length === 0) return { cleaned: true, signaled: false };
|
||||
const leader = members.find((member) => member.pid === ownership.leaderPid);
|
||||
const liveMembers = liveProcessGroupMembers(members);
|
||||
if (liveMembers.length === 0) return { cleaned: true, signaled: false };
|
||||
const leader = liveMembers.find(
|
||||
(member) => member.pid === ownership.leaderPid,
|
||||
);
|
||||
if (leader !== undefined) {
|
||||
if (
|
||||
ownership.expectedParentPid !== undefined &&
|
||||
@@ -129,23 +144,24 @@ export const cleanupOwnedProcessGroup = async (
|
||||
reason: "owned launcher command identity did not match",
|
||||
};
|
||||
}
|
||||
for (const member of members) {
|
||||
let environment: Readonly<Record<string, string>>;
|
||||
const failures: ProcessOwnershipValidationFailure[] = [];
|
||||
for (const member of liveMembers) {
|
||||
try {
|
||||
environment = await host.environment(member.pid);
|
||||
const environment = await host.environment(member.pid);
|
||||
if (environment.REA_PROCESS_RUN_ID !== ownership.runId)
|
||||
failures.push({ pid: member.pid, reason: "run-token-mismatch" });
|
||||
} catch {
|
||||
return {
|
||||
cleaned: false,
|
||||
reason: "process ownership could not be revalidated",
|
||||
};
|
||||
}
|
||||
if (environment.REA_PROCESS_RUN_ID !== ownership.runId) {
|
||||
return {
|
||||
cleaned: false,
|
||||
reason: "process group contains an unowned or PID-reused process",
|
||||
};
|
||||
failures.push({ pid: member.pid, reason: "environment-unreadable" });
|
||||
}
|
||||
}
|
||||
if (failures.length > 0)
|
||||
return {
|
||||
cleaned: false,
|
||||
reason: failures.some(({ reason }) => reason === "run-token-mismatch")
|
||||
? "process group contains an unowned or PID-reused process"
|
||||
: "process ownership could not be revalidated",
|
||||
failures,
|
||||
};
|
||||
try {
|
||||
host.signalGroup(ownership.processGroupId, "SIGKILL");
|
||||
} catch (cause: unknown) {
|
||||
@@ -171,8 +187,9 @@ export const observeOwnedProcessGroup = async (
|
||||
reason: "process group could not be inspected",
|
||||
};
|
||||
}
|
||||
if (members.length === 0) return { state: "empty" };
|
||||
for (const member of members) {
|
||||
const liveMembers = liveProcessGroupMembers(members);
|
||||
if (liveMembers.length === 0) return { state: "empty" };
|
||||
for (const member of liveMembers) {
|
||||
try {
|
||||
if (
|
||||
(await host.environment(member.pid)).REA_PROCESS_RUN_ID !==
|
||||
@@ -192,6 +209,11 @@ export const observeOwnedProcessGroup = async (
|
||||
return { state: "alive" };
|
||||
};
|
||||
|
||||
const liveProcessGroupMembers = (
|
||||
members: readonly ProcessGroupMember[],
|
||||
): readonly ProcessGroupMember[] =>
|
||||
members.filter(({ state }) => !state.startsWith("Z"));
|
||||
|
||||
const commandMatches = (actual: string, expected: string): boolean => {
|
||||
const normalizedActual = actual.trim();
|
||||
const normalizedExpected = expected.trim();
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { isSecretLikePath } from "../domain/referenceSourceClassification.js";
|
||||
import {
|
||||
createHistoricalSourceGraph,
|
||||
referenceSourceParseFailureKey,
|
||||
type HistoricalSourceGraph,
|
||||
type HistoricalSourceGraphInput,
|
||||
} from "../domain/referenceSourceGraph.js";
|
||||
@@ -204,18 +205,19 @@ export const importReferenceSource = async (
|
||||
if (isAborted(options.signal)) return err(cancelled());
|
||||
|
||||
const uniqueRelationships = deduplicateRelationships(relationships);
|
||||
const uniqueFailures = [...parseFailures]
|
||||
.sort((left, right) => {
|
||||
const byPath = byCodePoint(left.path, right.path);
|
||||
if (byPath !== 0) return byPath;
|
||||
return byCodePoint(left.parser, right.parser);
|
||||
})
|
||||
.filter((value, index, array) => {
|
||||
if (index === 0) return true;
|
||||
const previous = array[index - 1];
|
||||
if (previous === undefined) return true;
|
||||
return value.path !== previous.path || value.parser !== previous.parser;
|
||||
});
|
||||
const uniqueFailures = [
|
||||
...new Map(
|
||||
parseFailures.map((parseFailure) => [
|
||||
referenceSourceParseFailureKey(parseFailure),
|
||||
parseFailure,
|
||||
]),
|
||||
).values(),
|
||||
].sort((left, right) =>
|
||||
byCodePoint(
|
||||
referenceSourceParseFailureKey(left),
|
||||
referenceSourceParseFailureKey(right),
|
||||
),
|
||||
);
|
||||
|
||||
const sortedEntries = [...entries].sort((left, right) =>
|
||||
byCodePoint(left.path, right.path),
|
||||
|
||||
@@ -18,6 +18,11 @@ const investigationOptionsSchema = z.object({
|
||||
.describe("Approve persistent workspace writes"),
|
||||
workspaceName: z.string().trim().min(1).max(200).default("default"),
|
||||
expectedRevision: z.number().int().min(1).optional(),
|
||||
replayRunId: z
|
||||
.string()
|
||||
.regex(/^run_[a-f0-9]{64}$/u)
|
||||
.optional()
|
||||
.describe("Replay one verified complete workspace run without rescanning"),
|
||||
maxEntries: z.number().int().min(1).max(50_000).default(10_000),
|
||||
maxTotalBytes: z
|
||||
.number()
|
||||
@@ -55,6 +60,7 @@ export const registerInvestigationCommands = (
|
||||
alias: {
|
||||
workspaceName: "workspace-name",
|
||||
expectedRevision: "expected-revision",
|
||||
replayRunId: "replay-run-id",
|
||||
maxEntries: "max-entries",
|
||||
maxTotalBytes: "max-total-bytes",
|
||||
maxEntryBytes: "max-entry-bytes",
|
||||
@@ -99,13 +105,6 @@ const runInvestigation = async (
|
||||
},
|
||||
);
|
||||
if (!workspaceAuthorization.ok) return cliError(workspaceAuthorization.error);
|
||||
const inputAuthorization = await authorizeRootPermission(authority.value, {
|
||||
capability: "investigation_input",
|
||||
roots: [args.leftPath, args.rightPath],
|
||||
access: "read",
|
||||
operation: "investigate_versions",
|
||||
});
|
||||
if (!inputAuthorization.ok) return cliError(inputAuthorization.error);
|
||||
const investigated = await runCrossVersionInvestigation(
|
||||
{
|
||||
approved: true,
|
||||
@@ -114,6 +113,9 @@ const runInvestigation = async (
|
||||
...(options.expectedRevision === undefined
|
||||
? {}
|
||||
: { expected_workspace_revision: options.expectedRevision }),
|
||||
...(options.replayRunId === undefined
|
||||
? {}
|
||||
: { replay_run_id: options.replayRunId }),
|
||||
left_path: args.leftPath,
|
||||
right_path: args.rightPath,
|
||||
integrity_policy: options.integrityPolicy,
|
||||
@@ -134,6 +136,13 @@ const runInvestigation = async (
|
||||
{
|
||||
inputRoots: config.value.investigationInputRoots,
|
||||
integrityContinueEnabled: config.value.artifactIntegrityContinueEnabled,
|
||||
authorizeInputRead: () =>
|
||||
authorizeRootPermission(authority.value, {
|
||||
capability: "investigation_input",
|
||||
roots: [args.leftPath, args.rightPath],
|
||||
access: "read",
|
||||
operation: "investigate_versions",
|
||||
}),
|
||||
authorizeWorkspaceWrite: workspaceAuthorization.value.authorizeWrite,
|
||||
},
|
||||
);
|
||||
|
||||
@@ -59,6 +59,10 @@ export const crossVersionInvestigationInputSchema = z
|
||||
workspace_path: z.string().min(1).max(4_096),
|
||||
workspace_name: z.string().trim().min(1).max(200).default("default"),
|
||||
expected_workspace_revision: z.number().int().min(1).optional(),
|
||||
replay_run_id: z
|
||||
.string()
|
||||
.regex(/^run_[a-f0-9]{64}$/u)
|
||||
.optional(),
|
||||
left_path: z.string().min(1).max(4_096),
|
||||
right_path: z.string().min(1).max(4_096),
|
||||
integrity_policy: z.enum(["fail", "record-and-continue"]).default("fail"),
|
||||
|
||||
@@ -128,6 +128,11 @@ const parseFailureSchema = z.strictObject({
|
||||
reason: boundedTextSchema,
|
||||
});
|
||||
|
||||
/** Stable identity for one distinct reference-source parse failure. */
|
||||
export const referenceSourceParseFailureKey = (
|
||||
failure: z.infer<typeof parseFailureSchema>,
|
||||
): string => JSON.stringify([failure.path, failure.parser, failure.reason]);
|
||||
|
||||
const exclusionSchema = z.strictObject({
|
||||
path: relativePathSchema,
|
||||
reason: z.enum([
|
||||
@@ -400,7 +405,7 @@ function checkGraphInvariants(
|
||||
context,
|
||||
);
|
||||
checkSortedUnique(
|
||||
graph.parse_failures.map(({ path, parser }) => `${path}\u0000${parser}`),
|
||||
graph.parse_failures.map(referenceSourceParseFailureKey),
|
||||
"parse_failures",
|
||||
context,
|
||||
);
|
||||
|
||||
+14
-2
@@ -33,7 +33,7 @@ interface RuntimeDependencies {
|
||||
readonly serve: typeof serveStdio;
|
||||
readonly writeStderr: (text: string) => void;
|
||||
readonly setExitCode: (code: number) => void;
|
||||
readonly registerShutdown: (handler: () => void) => void;
|
||||
readonly registerShutdown: (handler: () => void) => () => void;
|
||||
readonly registerReload?: (handler: () => void) => () => void;
|
||||
}
|
||||
|
||||
@@ -49,6 +49,15 @@ const runtimeDependencies = (): RuntimeDependencies => ({
|
||||
process.once("SIGTERM", handler);
|
||||
process.stdin.once("end", handler);
|
||||
process.stdin.once("close", handler);
|
||||
let registered = true;
|
||||
return () => {
|
||||
if (!registered) return;
|
||||
registered = false;
|
||||
process.off("SIGINT", handler);
|
||||
process.off("SIGTERM", handler);
|
||||
process.stdin.off("end", handler);
|
||||
process.stdin.off("close", handler);
|
||||
};
|
||||
},
|
||||
registerReload: (handler) => {
|
||||
process.on("SIGHUP", handler);
|
||||
@@ -241,9 +250,11 @@ export const run = async (
|
||||
})();
|
||||
}) ?? (() => undefined);
|
||||
|
||||
let unregisterShutdown = (): void => undefined;
|
||||
let shutdownPromise: Promise<void> | undefined;
|
||||
const shutdown = (): Promise<void> => {
|
||||
shutdownPromise ??= (async () => {
|
||||
unregisterShutdown();
|
||||
unregisterReload();
|
||||
await handle.close();
|
||||
await session.close();
|
||||
@@ -259,7 +270,8 @@ export const run = async (
|
||||
});
|
||||
};
|
||||
|
||||
dependencies.registerShutdown(requestShutdown);
|
||||
unregisterShutdown = dependencies.registerShutdown(requestShutdown);
|
||||
if (shutdownPromise !== undefined) unregisterShutdown();
|
||||
return 0;
|
||||
};
|
||||
|
||||
|
||||
@@ -19,7 +19,10 @@ import {
|
||||
} from "../domain/changedBehavior.js";
|
||||
import { createEvidence, type Evidence } from "../domain/evidence.js";
|
||||
import type { EvidenceFilePolicy } from "../domain/evidenceBundle.js";
|
||||
import { EvidenceIntegrityError } from "../domain/errors.js";
|
||||
import {
|
||||
EvidenceIntegrityError,
|
||||
type AnalysisError,
|
||||
} from "../domain/errors.js";
|
||||
import { jsonValueSchema } from "../domain/jsonValue.js";
|
||||
import { err, ok, type Result } from "../domain/result.js";
|
||||
import {
|
||||
@@ -93,14 +96,19 @@ const registerChangedBehavior = (
|
||||
contractOptions(contract),
|
||||
async (input, context) => {
|
||||
const parsed = changedBehaviorInputSchema.parse(input);
|
||||
if (parsed.investigation_run !== undefined) {
|
||||
const investigationRun = parsed.investigation_run;
|
||||
if (investigationRun !== undefined) {
|
||||
const progress = mcpProgressReporter(context);
|
||||
let workspaceAuthorization: DeferredFileWriteAuthorization | undefined;
|
||||
let authorizeInputRead:
|
||||
| (() => Promise<Result<null, AnalysisError>>)
|
||||
| undefined;
|
||||
if (policies.permissionAuthority !== undefined) {
|
||||
const authority = policies.permissionAuthority;
|
||||
const authorizedWorkspace = await authorizeFileReadWithDeferredWrite(
|
||||
policies.permissionAuthority,
|
||||
authority,
|
||||
{
|
||||
path: parsed.investigation_run.workspace_path,
|
||||
path: investigationRun.workspace_path,
|
||||
readCapability: "investigation_workspace_read",
|
||||
writeCapability: "investigation_workspace_write",
|
||||
operation: contract.name,
|
||||
@@ -109,23 +117,16 @@ const registerChangedBehavior = (
|
||||
if (!authorizedWorkspace.ok)
|
||||
return toCallToolResult(authorizedWorkspace, contract);
|
||||
workspaceAuthorization = authorizedWorkspace.value;
|
||||
const authorizedInput = await authorizeRootPermission(
|
||||
policies.permissionAuthority,
|
||||
{
|
||||
authorizeInputRead = () =>
|
||||
authorizeRootPermission(authority, {
|
||||
capability: "investigation_input",
|
||||
roots: [
|
||||
parsed.investigation_run.left_path,
|
||||
parsed.investigation_run.right_path,
|
||||
],
|
||||
roots: [investigationRun.left_path, investigationRun.right_path],
|
||||
access: "read",
|
||||
operation: contract.name,
|
||||
},
|
||||
);
|
||||
if (!authorizedInput.ok)
|
||||
return toCallToolResult(authorizedInput, contract);
|
||||
});
|
||||
}
|
||||
const investigated = await runCrossVersionInvestigation(
|
||||
parsed.investigation_run,
|
||||
investigationRun,
|
||||
policies.evidenceFiles,
|
||||
{
|
||||
...investigationExecution(
|
||||
@@ -141,6 +142,7 @@ const registerChangedBehavior = (
|
||||
authorizeWorkspaceWrite:
|
||||
workspaceAuthorization.authorizeWrite,
|
||||
}),
|
||||
...(authorizeInputRead === undefined ? {} : { authorizeInputRead }),
|
||||
},
|
||||
);
|
||||
if (!investigated.ok) return toCallToolResult(investigated, contract);
|
||||
|
||||
@@ -15,6 +15,7 @@ import {
|
||||
PROCESS_COMPARISON_EVIDENCE,
|
||||
} from "../src/contracts/investigationExamples.js";
|
||||
import { createEvidence, parseEvidence } from "../src/domain/evidence.js";
|
||||
import { changedBehaviorResultSchema } from "../src/domain/changedBehavior.js";
|
||||
import { jsonObjectSchema, jsonValueSchema } from "../src/domain/jsonValue.js";
|
||||
import { createServer } from "../src/server/createServer.js";
|
||||
import { observed } from "./fixtures/analysisExecution.js";
|
||||
@@ -214,6 +215,85 @@ describe("investigation MCP workflows", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("replays a selected complete run after input access is removed", async () => {
|
||||
const directory = await mkdtemp(
|
||||
join(tmpdir(), "rea-investigation-replay-"),
|
||||
);
|
||||
const left = join(directory, "left");
|
||||
const right = join(directory, "right");
|
||||
const workspace = join(directory, "workspace.json");
|
||||
await Promise.all([mkdir(left), mkdir(right)]);
|
||||
await Promise.all([
|
||||
writeFile(join(left, "feature.txt"), "old\n"),
|
||||
writeFile(join(right, "feature.txt"), "new\n"),
|
||||
]);
|
||||
const filePolicy = evidencePolicy(directory);
|
||||
const investigationRun = {
|
||||
approved: true,
|
||||
workspace_path: workspace,
|
||||
workspace_name: "selected-replay",
|
||||
left_path: left,
|
||||
right_path: right,
|
||||
options: { page_size: 500, change_limit: 100 },
|
||||
};
|
||||
const writer = await investigationAuthority(directory, true);
|
||||
const initial = await connected(filePolicy, [directory], writer);
|
||||
let runId: string | undefined;
|
||||
try {
|
||||
const created = await initial.client.callTool({
|
||||
name: "find_changed_behavior",
|
||||
arguments: { investigation_run: investigationRun },
|
||||
});
|
||||
expect(created.isError, JSON.stringify(created)).not.toBe(true);
|
||||
const evidence = parseEvidence(
|
||||
z.object({ result: z.unknown() }).parse(created.structuredContent)
|
||||
.result,
|
||||
);
|
||||
const result = changedBehaviorResultSchema.parse(
|
||||
evidence.normalized_result,
|
||||
);
|
||||
if (result.investigation_run === undefined)
|
||||
throw new Error("missing investigation run summary");
|
||||
runId = result.investigation_run.run_id;
|
||||
} finally {
|
||||
await close(initial.session, initial.server, initial.client);
|
||||
}
|
||||
if (runId === undefined) throw new Error("missing completed run ID");
|
||||
|
||||
await Promise.all([
|
||||
rm(left, { recursive: true, force: true }),
|
||||
rm(right, { recursive: true, force: true }),
|
||||
]);
|
||||
const workspaceOnly = await permissionAuthorityForRoot(
|
||||
directory,
|
||||
[
|
||||
"investigation_workspace_read",
|
||||
"investigation_workspace_write",
|
||||
"investigation_input",
|
||||
],
|
||||
["investigation_workspace_read"],
|
||||
);
|
||||
const replay = await connected(filePolicy, [], workspaceOnly);
|
||||
try {
|
||||
const cached = await replay.client.callTool({
|
||||
name: "find_changed_behavior",
|
||||
arguments: {
|
||||
investigation_run: {
|
||||
...investigationRun,
|
||||
replay_run_id: runId,
|
||||
},
|
||||
},
|
||||
});
|
||||
expect(cached.isError, JSON.stringify(cached)).not.toBe(true);
|
||||
expect(
|
||||
await readInvestigationWorkspace(workspace, filePolicy),
|
||||
).toMatchObject({ ok: true, value: { revision: 3 } });
|
||||
} finally {
|
||||
await close(replay.session, replay.server, replay.client);
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("refuses automatic artifact reads outside operator-approved roots", async () => {
|
||||
const directory = await mkdtemp(join(tmpdir(), "rea-investigation-mcp-"));
|
||||
const approvedInputs = join(directory, "approved-inputs");
|
||||
|
||||
@@ -12,7 +12,7 @@ import { createHash } from "node:crypto";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { createPackageWithOptions } from "@electron/asar";
|
||||
import canonicalize from "canonicalize";
|
||||
|
||||
@@ -31,6 +31,7 @@ import {
|
||||
parseInvestigationWorkspace,
|
||||
serializeInvestigationWorkspace,
|
||||
} from "../src/domain/investigationWorkspace.js";
|
||||
import { ok } from "../src/domain/result.js";
|
||||
|
||||
const digestCanonical = (value: unknown): string => {
|
||||
const encoded = canonicalize(value);
|
||||
@@ -239,6 +240,57 @@ describe("persistent cross-version investigation workspace", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("replays an explicitly selected complete run without input access", async () => {
|
||||
const { left, right, input } = await fixture();
|
||||
if (directory === undefined) throw new Error("missing fixture root");
|
||||
const completed = await runCrossVersionInvestigation(
|
||||
input,
|
||||
policy(directory),
|
||||
{ inputRoots: [directory] },
|
||||
);
|
||||
if (!completed.ok) throw completed.error;
|
||||
const run = completed.value.workspace.runs[0];
|
||||
if (run === undefined) throw new Error("missing completed run");
|
||||
await Promise.all([
|
||||
rm(left, { recursive: true, force: true }),
|
||||
rm(right, { recursive: true, force: true }),
|
||||
]);
|
||||
const authorizeInputRead = vi.fn(() => Promise.resolve(ok(null)));
|
||||
|
||||
await expect(
|
||||
runCrossVersionInvestigation(
|
||||
{ ...input, replay_run_id: run.run_id },
|
||||
policy(directory),
|
||||
{ inputRoots: [], authorizeInputRead },
|
||||
),
|
||||
).resolves.toMatchObject({
|
||||
ok: true,
|
||||
value: {
|
||||
reused: true,
|
||||
evidence: { evidence_id: completed.value.evidence.evidence_id },
|
||||
},
|
||||
});
|
||||
expect(authorizeInputRead).not.toHaveBeenCalled();
|
||||
|
||||
for (const replayInput of [
|
||||
{ ...input, replay_run_id: `run_${"f".repeat(64)}` },
|
||||
{ ...input, right_path: `${right}-other`, replay_run_id: run.run_id },
|
||||
])
|
||||
await expect(
|
||||
runCrossVersionInvestigation(replayInput, policy(directory), {
|
||||
inputRoots: [],
|
||||
authorizeInputRead,
|
||||
}),
|
||||
).resolves.toMatchObject({
|
||||
ok: false,
|
||||
error: {
|
||||
_tag: "InvestigationWorkspaceError",
|
||||
reason: "revision-conflict",
|
||||
},
|
||||
});
|
||||
expect(authorizeInputRead).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("enforces shared JSON depth limits while reading workspaces", async () => {
|
||||
const { path, input } = await fixture();
|
||||
if (directory === undefined) throw new Error("missing fixture root");
|
||||
@@ -300,6 +352,26 @@ describe("persistent cross-version investigation workspace", () => {
|
||||
),
|
||||
).toMatchObject({ ok: true });
|
||||
|
||||
const authorizeInputRead = vi.fn(() => Promise.resolve(ok(null)));
|
||||
await expect(
|
||||
runCrossVersionInvestigation(
|
||||
{
|
||||
...input,
|
||||
workspace_path: resumePath,
|
||||
replay_run_id: partialRun.run_id,
|
||||
},
|
||||
policy(directory),
|
||||
{ inputRoots: [], authorizeInputRead },
|
||||
),
|
||||
).resolves.toMatchObject({
|
||||
ok: false,
|
||||
error: {
|
||||
_tag: "InvestigationWorkspaceError",
|
||||
reason: "revision-conflict",
|
||||
},
|
||||
});
|
||||
expect(authorizeInputRead).not.toHaveBeenCalled();
|
||||
|
||||
const resumed = await runCrossVersionInvestigation(
|
||||
{ ...input, workspace_path: resumePath },
|
||||
policy(directory),
|
||||
|
||||
@@ -18,7 +18,13 @@ const dependencies = (
|
||||
serve,
|
||||
writeStderr: (text) => output.push(text),
|
||||
setExitCode: (code) => exitCodes.push(code),
|
||||
registerShutdown: (handler) => shutdown.push(handler),
|
||||
registerShutdown: (handler) => {
|
||||
shutdown.push(handler);
|
||||
return () => {
|
||||
const index = shutdown.indexOf(handler);
|
||||
if (index >= 0) shutdown.splice(index, 1);
|
||||
};
|
||||
},
|
||||
});
|
||||
|
||||
describe("MCP runtime errors", () => {
|
||||
@@ -52,6 +58,7 @@ describe("MCP runtime errors", () => {
|
||||
shutdown[0]?.();
|
||||
await nextTurn();
|
||||
expect(closeCalls).toBe(1);
|
||||
expect(shutdown).toEqual([]);
|
||||
});
|
||||
|
||||
it("reports transport startup failure without its cause", async () => {
|
||||
@@ -87,6 +94,7 @@ describe("MCP runtime errors", () => {
|
||||
);
|
||||
shutdown[0]?.();
|
||||
await nextTurn();
|
||||
expect(shutdown).toEqual([]);
|
||||
expect(exitCodes).toEqual([1]);
|
||||
expect(output).toEqual([
|
||||
"REA could not close cleanly. End the REA process before starting it again.\n",
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
cleanupOwnedProcessGroup,
|
||||
observeOwnedProcessGroup,
|
||||
parseProcessEnvironment,
|
||||
type ProcessOwnershipHost,
|
||||
} from "../src/application/ProcessOwnership.js";
|
||||
@@ -26,6 +27,7 @@ const host = (
|
||||
pid: Number(pid),
|
||||
parentPid: Number(pid) === 100 ? 1 : 100,
|
||||
processGroupId: 100,
|
||||
state: "S",
|
||||
command: "fixture",
|
||||
})),
|
||||
),
|
||||
@@ -63,10 +65,108 @@ describe("owned process-group cleanup", () => {
|
||||
expect(await cleanupOwnedProcessGroup(ownership, adapter)).toEqual({
|
||||
cleaned: false,
|
||||
reason: "process group contains an unowned or PID-reused process",
|
||||
failures: [{ pid: 101, reason: "run-token-mismatch" }],
|
||||
});
|
||||
expect(signalGroup).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("checks every member and aggregates ownership read failures", async () => {
|
||||
const environment = vi.fn((pid: number) => {
|
||||
if (pid === 100)
|
||||
return Promise.reject(new Error("transient procfs read"));
|
||||
return Promise.resolve(
|
||||
pid === 101
|
||||
? { REA_PROCESS_RUN_ID: "different-run" }
|
||||
: { REA_PROCESS_RUN_ID: "run-token" },
|
||||
);
|
||||
});
|
||||
const signalGroup = vi.fn();
|
||||
const adapter: ProcessOwnershipHost = {
|
||||
listMembers: () =>
|
||||
Promise.resolve(
|
||||
[100, 101, 102].map((pid) => ({
|
||||
pid,
|
||||
parentPid: pid === 100 ? 1 : 100,
|
||||
processGroupId: 100,
|
||||
state: "S",
|
||||
command: "fixture",
|
||||
})),
|
||||
),
|
||||
environment,
|
||||
signalGroup,
|
||||
};
|
||||
|
||||
expect(await cleanupOwnedProcessGroup(ownership, adapter)).toEqual({
|
||||
cleaned: false,
|
||||
reason: "process group contains an unowned or PID-reused process",
|
||||
failures: [
|
||||
{ pid: 100, reason: "environment-unreadable" },
|
||||
{ pid: 101, reason: "run-token-mismatch" },
|
||||
],
|
||||
});
|
||||
expect(environment.mock.calls.map(([pid]) => pid)).toEqual([100, 101, 102]);
|
||||
expect(signalGroup).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("ignores exited zombie members during live ownership checks", async () => {
|
||||
const environment = vi.fn((pid: number) =>
|
||||
Promise.resolve(pid === 101 ? {} : { REA_PROCESS_RUN_ID: "run-token" }),
|
||||
);
|
||||
const signalGroup = vi.fn();
|
||||
const adapter: ProcessOwnershipHost = {
|
||||
listMembers: () =>
|
||||
Promise.resolve([
|
||||
{
|
||||
pid: 100,
|
||||
parentPid: 1,
|
||||
processGroupId: 100,
|
||||
state: "S",
|
||||
command: "fixture",
|
||||
},
|
||||
{
|
||||
pid: 101,
|
||||
parentPid: 100,
|
||||
processGroupId: 100,
|
||||
state: "Z",
|
||||
command: "[node] <defunct>",
|
||||
},
|
||||
]),
|
||||
environment,
|
||||
signalGroup,
|
||||
};
|
||||
|
||||
expect(await cleanupOwnedProcessGroup(ownership, adapter)).toEqual({
|
||||
cleaned: true,
|
||||
signaled: true,
|
||||
});
|
||||
expect(environment).toHaveBeenCalledTimes(1);
|
||||
expect(environment).toHaveBeenCalledWith(100);
|
||||
expect(signalGroup).toHaveBeenCalledWith(100, "SIGKILL");
|
||||
});
|
||||
|
||||
it("observes a zombie-only group as settled", async () => {
|
||||
const environment = vi.fn(() => Promise.resolve({}));
|
||||
const adapter: ProcessOwnershipHost = {
|
||||
listMembers: () =>
|
||||
Promise.resolve([
|
||||
{
|
||||
pid: 101,
|
||||
parentPid: 1,
|
||||
processGroupId: 100,
|
||||
state: "Z+",
|
||||
command: "[node] <defunct>",
|
||||
},
|
||||
]),
|
||||
environment,
|
||||
signalGroup: vi.fn(),
|
||||
};
|
||||
|
||||
expect(await observeOwnedProcessGroup(ownership, adapter)).toEqual({
|
||||
state: "empty",
|
||||
});
|
||||
expect(environment).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("is idempotent when the owned group has already exited", async () => {
|
||||
const { adapter } = host({});
|
||||
expect(await cleanupOwnedProcessGroup(ownership, adapter)).toEqual({
|
||||
|
||||
@@ -192,6 +192,23 @@ describe("historical source graph", () => {
|
||||
).toThrow(/Complete inventory/u);
|
||||
});
|
||||
|
||||
it("keeps distinct failure reasons for the same path and parser", () => {
|
||||
const partial = graphInput();
|
||||
partial.inventory_state = "partial";
|
||||
partial.parse_failures = [
|
||||
{ path: "src/main.ts", parser: "babel", reason: "First failure" },
|
||||
{ path: "src/main.ts", parser: "babel", reason: "Second failure" },
|
||||
];
|
||||
|
||||
expect(createHistoricalSourceGraph(partial).parse_failures).toEqual(
|
||||
partial.parse_failures,
|
||||
);
|
||||
partial.parse_failures.push(partial.parse_failures[1]!);
|
||||
expect(() => createHistoricalSourceGraph(partial)).toThrow(
|
||||
/parse_failures.*unique/iu,
|
||||
);
|
||||
});
|
||||
|
||||
it("rejects strict unknown fields, invalid VCS, dangling internal edges, and false manifests", () => {
|
||||
expect(() =>
|
||||
createHistoricalSourceGraph({ ...graphInput(), surprise: true }),
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { defineConfig } from "vitest/config";
|
||||
|
||||
export default defineConfig({
|
||||
@@ -8,6 +11,10 @@ export default defineConfig({
|
||||
coverage: {
|
||||
enabled: true,
|
||||
provider: "v8",
|
||||
reportsDirectory: join(
|
||||
tmpdir(),
|
||||
`rea-vitest-coverage-${String(process.pid)}`,
|
||||
),
|
||||
include: ["src/**"],
|
||||
thresholds: {
|
||||
statements: 65,
|
||||
|
||||
Reference in new issue
Block a user