Merge branches 'main', 'codex/review-fixes-pr-2403' and 'main' of https://github.com/morluto/rea into HEAD

This commit is contained in:
morluto committed 2026-11-07 23:29:21 +08:00
commit d71022d91c
28 files changed
+1377 -347

No files matched your search

+15
View File
@@ -243,6 +243,20 @@ def _hex(value):
return "0x%x" % value
def _source_location(address):
"""Preserve available line-entry facets without acquiring source files."""
entry = address.GetLineEntry()
if not entry.IsValid():
return None
file_spec = entry.GetFileSpec()
file_path = str(file_spec) if file_spec.IsValid() else None
line = entry.GetLine() or None
column = entry.GetColumn() or None
if not file_path and line is None and column is None:
return None
return {"file_path": file_path or None, "line": line, "column": column}
def _location(address, target):
"""Load and file address, module and symbol of one code address."""
module = address.GetModule()
@@ -256,6 +270,7 @@ def _location(address, target):
"module": module.GetFileSpec().GetFilename() if module.IsValid() else None,
"module_path": str(module.GetFileSpec()) if module.IsValid() else None,
"symbol": symbol.GetName() if symbol.IsValid() else None,
"source_location": _source_location(address),
}
return location
+88 -13
View File
@@ -25,6 +25,45 @@ spec.loader.exec_module(tracer)
class NativeTraceBudgetTests(unittest.TestCase):
def test_source_coordinates_preserve_partial_metadata_without_reading_source(self):
def address(valid=True, file_valid=True, line=23, column=7):
class SourceFileSpec:
def IsValid(self):
return file_valid
def __str__(self):
return "/missing/é space/fixture.pas"
file_spec = SourceFileSpec()
entry = types.SimpleNamespace(
IsValid=lambda: valid,
GetFileSpec=lambda: file_spec,
GetLine=lambda: line,
GetColumn=lambda: column,
)
return types.SimpleNamespace(GetLineEntry=lambda: entry)
self.assertEqual(
tracer._source_location(address()),
{"file_path": "/missing/é space/fixture.pas", "line": 23, "column": 7},
)
self.assertEqual(
tracer._source_location(address(column=0)),
{"file_path": "/missing/é space/fixture.pas", "line": 23, "column": None},
)
self.assertEqual(
tracer._source_location(address(file_valid=False)),
{"file_path": None, "line": 23, "column": 7},
)
self.assertEqual(
tracer._source_location(address(line=0)),
{"file_path": "/missing/é space/fixture.pas", "line": None, "column": 7},
)
self.assertIsNone(tracer._source_location(address(valid=False)))
self.assertIsNone(
tracer._source_location(address(file_valid=False, line=0, column=0))
)
def test_invalid_launch_process_reports_explicit_no_creation(self):
class InvalidProcess:
def IsValid(self):
@@ -390,6 +429,9 @@ class NativeTraceBudgetTests(unittest.TestCase):
return FileSpec()
class Address:
def GetLineEntry(self):
return types.SimpleNamespace(IsValid=lambda: False)
def GetModule(self):
return Module()
@@ -491,7 +533,24 @@ class NativeTraceBudgetTests(unittest.TestCase):
def GetFileSpec(self):
return FileSpec()
class SourceFileSpec:
def IsValid(self):
return True
def __str__(self):
return source_path
class Address:
def GetLineEntry(self):
if source_path is None:
return types.SimpleNamespace(IsValid=lambda: False)
return types.SimpleNamespace(
IsValid=lambda: True,
GetFileSpec=SourceFileSpec,
GetLine=lambda: 15,
GetColumn=lambda: 3,
)
def GetModule(self):
return Module()
@@ -518,19 +577,35 @@ class NativeTraceBudgetTests(unittest.TestCase):
breakpoint_count = 1000
created = [[Breakpoint()] for _ in range(breakpoint_count)]
config = {"breakpoints": [{"kind": "objc-method"}] * breakpoint_count}
report, locations_truncated = tracer._breakpoint_report(
object(), created, config
)
retained_bytes = sum(
len(json.dumps(location).encode("utf-8"))
+ (1 if index > 0 else 0)
for row in report
for index, location in enumerate(row["locations"])
)
self.assertTrue(locations_truncated)
self.assertEqual(sum(row["location_count"] for row in report), 64_000)
self.assertLess(sum(len(row["locations"]) for row in report), 64_000)
self.assertLessEqual(retained_bytes, tracer.MAX_BREAKPOINT_LOCATION_BYTES)
for source_path in (None, "/tmp/" + "源🚀" * 3000 + "/calls.pas"):
with self.subTest(source_path_present=source_path is not None):
report, locations_truncated = tracer._breakpoint_report(
object(), created, config
)
retained_bytes = sum(
len(json.dumps(location).encode("utf-8"))
+ (1 if index > 0 else 0)
for row in report
for index, location in enumerate(row["locations"])
)
self.assertTrue(locations_truncated)
self.assertEqual(
sum(row["location_count"] for row in report), 64_000
)
self.assertLess(
sum(len(row["locations"]) for row in report), 64_000
)
self.assertLessEqual(
retained_bytes, tracer.MAX_BREAKPOINT_LOCATION_BYTES
)
if source_path is not None:
self.assertTrue(
any(
location["source_location"]["file_path"] == source_path
for row in report
for location in row["locations"]
)
)
if __name__ == "__main__":
@@ -1154,6 +1154,25 @@ arguments. The receiver class comes from LLDB's Objective-C runtime reader
without running target code; no expression is evaluated, and REA never
attaches to a process it did not launch.
Resolved breakpoint locations, entries and caller frames also retain
`source_location` from LLDB's line tables when available. Its `file_path`,
one-based `line` and one-based `column` are independently nullable. An invalid
line entry is null; historical captures may omit the field. This is debug metadata
attribution: REA does not read that source path or verify the source file bytes.
Absent coordinates do not imply that the executable has no source code.
Free Pascal programs with retained native symbols and usable DWARF can use this
same bounded call observation on macOS. Select an exact emitted symbol from the
binary rather than assuming a source procedure name. The owned Pascal fixture
uses an explicit C calling convention and a public symbol; its verification
covers call entries, raw integer registers and source coordinates, without
establishing typed Pascal locals or interactive stepping. After building REA,
run the real lane with an existing caller-selected compiler:
```sh
REA_VERIFY_FPC_COMMAND=/absolute/path/to/fpc npm run verify:native:pascal-calls
```
The run ends when the process exits, `max_events` entries are recorded,
`duration_ms` elapses, or the aggregate trace resource budget is exhausted.
Bounded termination reports partial coverage. REA checks process termination
@@ -0,0 +1,39 @@
# Implementation plan
## Boundary and contract
Extend nativeCodeLocationSchema in src/domain/native/nativeCallObservation.ts.
Reuse that shape for events, breakpoint locations, backtraces and retained journal
events. Keep the field optional for older evidence and nullable for current LLDB
observations with unavailable metadata. Each inner facet can be independently
unavailable; avoid dropping a valid line number because a file path is absent.
Read SBAddress.GetLineEntry once in the existing Python _location producer. A
valid SBFileSpec supplies its displayed path without filesystem access. Convert
zero line/column to null. An invalid entry returns null. Existing estimation and
serialization budget checks already traverse the location object recursively.
## Incremental work and validation
1. Define contracts and criteria before code; add failing producer tests.
2. Extend the canonical schema and _location producer. Update existing fake
address objects to implement the actual LLDB line-entry interface.
3. Extend provider/MCP boundary coverage and retained-journal coverage for the
additive field; preserve omission compatibility.
4. Extend the existing real macOS verifier with source-coordinate assertions
against its own Objective-C/C fixture built with debug information.
5. Update the native investigation guide; run Python tests, focused TypeScript
tests, repository checks and the real verifier if host authority permits.
## Effects and compatibility
The existing LLDB operation already owns its breakpoint callbacks and process
lifecycle. This adds a metadata read within that observation. It does not introduce
target execution or a new provider beyond existing behavior. No source file is
acquired from a path in debug information. Consumers of historical evidence retain
omission; consumers of new captures see either a partial/full object or null.
## References
- https://lldb.llvm.org/python_api/lldb.SBAddress.html
- https://lldb.llvm.org/python_api/lldb.SBLineEntry.html
@@ -0,0 +1,36 @@
# Native call source coordinates
Issue: https://github.com/morluto/rea/issues/2382
## Problem and outcome
The existing bounded macOS LLDB call observation retains module, symbol and
addresses, but loses compiler-provided source coordinates. Retain line-entry
metadata at the same producer boundary for breakpoint locations, call entries
and caller frames.
## Requirements
- Add optional nullable source_location to the canonical native code location.
A present object contains nullable file_path, line and column. Positive line
and column values are one-based. Provider zero values mean unavailable.
- Preserve partial facts from a valid line entry. Invalid entries or entries
without any usable facet produce null. Historical captures may omit the field.
- Read only SBAddress line metadata; do not resolve paths, read source contents,
evaluate expressions or change the existing launch, stop or cleanup policy.
- The field establishes LLDB line-table attribution, not verified source bytes.
- Existing trace and breakpoint byte budgets include the new representation.
## Acceptance
Producer tests cover valid, invalid and partial entries, Unicode paths and absent
coordinates. Existing evidence without the field remains accepted. CLI/MCP and
partial retained-journal paths preserve the canonical field. The real native-call
verifier checks source coordinates for its owned debug build, including caller
frames and breakpoint locations, and reports actual host/provider coverage.
## Out of scope
New debugger adapters, interactive stepping, source breakpoints, source-file
association verification, source reads, Pascal/Delphi compatibility claims and
Java/Android/CLR runtime support.
@@ -0,0 +1,14 @@
# Tasks
The checked items below record work completed on an earlier revision. The
current revision's final caller-frame assertions, marker-derived line
expectations, timeout handling, and Unicode budget case remain source-reviewed
only; the full native-call verifier is incomplete.
- [x] Define the single-feature scope, contract and implementation plan.
- [x] Reproduce lost source coordinates with producer regression tests.
- [x] Retain line-entry metadata in the producer and canonical schema.
- [x] Verify boundary/MCP propagation and historical omission.
- [x] Historical: verify real CLI/MCP call coordinates on an owned Free Pascal 3.2.2 program.
- [x] Historical: update documentation; pass Python, focused TypeScript, static and discovery checks.
- [ ] Complete the full native-call verifier after investigating #2397.
@@ -0,0 +1,45 @@
# Historical verification
The records below describe earlier revisions, before the final caller-frame
assertions, marker-derived line expectations, forced compiler timeout termination,
and large Unicode source-location budget coverage. They do not validate the
current revision. Those final changes were reviewed from source only; tests and
automated checks were not run at the user’s explicit request.
Host: macOS ARM64. Compiler: caller-installed Free Pascal 3.2.2, Darwin AArch64.
Debugger: LLDB 2103.0.34.103 from Command Line Tools.
- Python producer/budget/output tests: 10 passed.
- Focused native provider, retained-journal and MCP tests: 18 passed.
- Real Pascal CLI and stdio MCP: three expected calls, integer register values
0/1/2, source path and positive line attribution, successful target exit.
- Repository static checks: six tasks and architecture guards passed.
- Test discovery: all 982 test files discovered exactly once.
The complete native-call verifier did not pass. One run reported exited instead
of event-limit for capped overlapping breakpoints. Another reported unavailable
launch start identity in a later capped scenario. Investigation:
https://github.com/morluto/rea/issues/2397. These results have not yet been isolated
on unchanged upstream. The source-coordinate PR was draft at that historical checkpoint.
The first real native scenario and CLI/MCP source-coordinate comparisons completed
before those broader failures. The Pascal lane verifies bounded native call
observation; it does not establish typed Pascal local values, source breakpoints,
interactive stepping, Delphi compatibility or a GDB provider.
## Unmeasured costs and outstanding execution
Repeated source paths expand event, caller-frame and breakpoint-location payloads.
`nativeCallObservation.test.ts` measures that growth as UTF-8 JSON bytes: three
copies of one location grow by three times the difference between the
`source_location` object and `null`. That byte delta is not a token count.
Equivalent before/after completed workflows have not been measured for agent tokens,
elapsed time, CPU time or peak process-family memory, for either cold or warm runs.
The first compiler probe is only `fpc -iV` with a 30s deadline. Darwin ownership-helper
compilation uses `REA_SWIFTC_COMPILE_TIMEOUT_MS`, default 120s, inside the later
180s `observe-native-calls` run.
The current caller-frame equality checks, marker-derived source lines, forced
compiler timeout termination and Unicode source-path budget case remain unverified
by execution. Earlier Pascal CLI/MCP observations establish only the historical
coverage described above.
+1
View File
@@ -295,6 +295,7 @@
"prepare": "node scripts/prepare.mjs",
"verify:asset-catalog": "npm run build:cached && node scripts/verify/apple/asset-catalog.mjs",
"verify:native-calls": "npm run build:cached && node scripts/verify-native-calls.mjs",
"verify:native:pascal-calls": "npm run build:cached && node scripts/verify-native-pascal-calls.mjs",
"verify:native-ui:permissions": "npm run build:cached && node scripts/verify-native-ui.mjs --allow-permission-denial",
"test:focused": "node scripts/run-development-tests.mjs focused",
"test:sqlite": "npm run test:prepare && npm run test:sqlite:run",
+124 -46
View File
@@ -1,9 +1,8 @@
import assert from "node:assert/strict";
import { execFile } from "node:child_process";
import { promisify } from "node:util";
import { fileURLToPath } from "node:url";
import { Client } from "@modelcontextprotocol/client";
import { StdioClientTransport } from "@modelcontextprotocol/client/stdio";
import { createVerifierCommands } from "./verify-commands.mjs";
import { createOwnedMcpStdioTransport } from "./owned-mcp-stdio.mjs";
import { withVerifierCleanup } from "./verifier-cleanup.mjs";
import { INLINE_EVIDENCE_MCP_ENVIRONMENT } from "./mcp-verification-environment.mjs";
@@ -75,6 +74,23 @@ export async function artifactMcpCall(client, parameters, options = {}) {
return client.callTool(parameters, { ...options, toolDefinition });
}
/** Retain the exact command owner when a verifier cannot finish cleanup. */
class ArtifactVerifierCleanupError extends Error {
constructor(cause, cleanupOwner) {
super("Artifact verifier cleanup is incomplete", { cause });
this.name = "ArtifactVerifierCleanupError";
this.cleanupOwner = cleanupOwner;
}
}
const closeOwnedCommands = async (owner) => {
try {
await owner.close();
} catch (cause) {
throw new ArtifactVerifierCleanupError(cause, owner);
}
};
const environment = () => {
const compileCache =
process.env.NODE_COMPILE_CACHE ??
@@ -109,68 +125,130 @@ export async function artifactCliEvidence(
command,
target,
arguments_ = [],
{ timeoutMs = 60000 } = {},
{ timeoutMs = 60000, signal, commands } = {},
) {
const { stdout } = await promisify(execFile)(
process.execPath,
[
fileURLToPath(new URL("../rea.mjs", import.meta.url)),
command,
target,
...arguments_,
"--json",
],
{ env: environment(), timeout: timeoutMs, maxBuffer: 16 * 1024 * 1024 },
);
signal?.throwIfAborted();
const owner = commands ?? (await createVerifierCommands());
const execute = () =>
owner.exec(
process.execPath,
[
fileURLToPath(new URL("../rea.mjs", import.meta.url)),
command,
target,
...arguments_,
"--json",
],
{
env: environment(),
timeout: timeoutMs,
maxBuffer: 16 * 1024 * 1024,
signal,
},
);
const { stdout } =
commands === undefined
? await withVerifierCleanup(execute, [() => closeOwnedCommands(owner)])
: await execute();
signal?.throwIfAborted();
const evidence = JSON.parse(stdout);
assert.equal(evidence.error, undefined, stdout);
assert.ok(evidence.normalized_result, "CLI omitted normalized result");
return evidence;
}
const withOpenedArtifactMcp = async (target, openTool, verify) => {
const transport = new StdioClientTransport({
const withOpenedArtifactMcp = async (
target,
openTool,
verify,
options = {},
) => {
options.signal?.throwIfAborted();
const owner = options.commands ?? (await createVerifierCommands());
const transport = await createOwnedMcpStdioTransport({
command: process.execPath,
args: [fileURLToPath(new URL("../rea.mjs", import.meta.url)), "mcp"],
env: { ...environment(), ...INLINE_EVIDENCE_MCP_ENVIRONMENT },
stderr: "pipe",
arguments: [fileURLToPath(new URL("../rea.mjs", import.meta.url)), "mcp"],
hostEnvironment: { ...environment(), ...INLINE_EVIDENCE_MCP_ENVIRONMENT },
cwd: process.cwd(),
expectedCommand: null,
sweepTokenOwnedProcesses: true,
signal: options.signal,
});
owner.adopt(transport, "artifact MCP process");
const client = new Client({ name: "artifact-real-e2e", version: "1" });
return withVerifierCleanup(async () => {
await client.connect(transport);
const opened = await artifactMcpCall(client, {
name: openTool,
arguments: { path: target },
});
assert.notEqual(opened.isError, true, JSON.stringify(opened));
if (openTool === "open_bundle_resources")
assert.equal(opened.structuredContent?.result?.caller_path, target);
await verify(client);
}, [
() => {
toolBindings.delete(client);
},
() => client.close(),
() => transport.close(),
]);
let clientClose;
const closeClient = () =>
(clientClose ??= Promise.resolve().then(() => client.close()));
const abortTransport = () => {
// Keep the same promise for finalization; concurrent SDK closes can return
// before the first close has observed the child exit.
void closeClient().catch(() => undefined);
};
options.signal?.addEventListener("abort", abortTransport, { once: true });
try {
const result = await withVerifierCleanup(async () => {
options.signal?.throwIfAborted();
await client.connect(transport);
options.signal?.throwIfAborted();
const opened = await artifactMcpCall(
client,
{
name: openTool,
arguments: { path: target },
},
{ signal: options.signal },
);
assert.notEqual(opened.isError, true, JSON.stringify(opened));
if (openTool === "open_bundle_resources")
assert.equal(opened.structuredContent?.result?.caller_path, target);
await verify(client);
}, [
() => {
toolBindings.delete(client);
},
() => closeClient(),
() => transport.close(),
...(options.commands === undefined
? [() => closeOwnedCommands(owner)]
: []),
]);
options.signal?.throwIfAborted();
return result;
} finally {
options.signal?.removeEventListener("abort", abortTransport);
}
};
/** Run a real stdio MCP subprocess with an explicitly opened binary target. */
export async function withArtifactMcp(target, verify) {
return withOpenedArtifactMcp(target, "open_binary", verify);
export async function withArtifactMcp(target, verify, options = {}) {
return withOpenedArtifactMcp(target, "open_binary", verify, options);
}
/** Run a real stdio MCP subprocess with an explicitly opened bundle resource root. */
export async function withBundleResourcesMcp(target, verify) {
return withOpenedArtifactMcp(target, "open_bundle_resources", verify);
export async function withBundleResourcesMcp(target, verify, options = {}) {
return withOpenedArtifactMcp(
target,
"open_bundle_resources",
verify,
options,
);
}
/** Require a successful structured MCP tool result rather than its text rendering. */
export async function artifactMcpResult(client, name, arguments_ = {}) {
const result = await artifactMcpCall(client, {
name,
arguments: arguments_,
});
export async function artifactMcpResult(
client,
name,
arguments_ = {},
options = {},
) {
const result = await artifactMcpCall(
client,
{
name,
arguments: arguments_,
},
options,
);
assert.notEqual(result.isError, true, JSON.stringify(result));
assert.ok(
result.structuredContent?.normalized_result,
@@ -0,0 +1,5 @@
/** Allocate under a caller-selected parent and return the host's canonical spelling. */
export declare const createCanonicalTemporaryDirectory: (
prefix: string,
parent?: string,
) => Promise<{ readonly path: string; readonly allocatedPath: string }>;
@@ -0,0 +1,17 @@
import { mkdtemp, realpath, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
/** Allocate under a caller-selected parent and return the host's canonical spelling. */
export const createCanonicalTemporaryDirectory = async (
prefix,
parent = tmpdir(),
) => {
const allocatedPath = await mkdtemp(join(parent, prefix));
try {
return { path: await realpath(allocatedPath), allocatedPath };
} catch (cause) {
await rm(allocatedPath, { recursive: true, force: true });
throw cause;
}
};
+24
View File
@@ -0,0 +1,24 @@
import { importVerifierSource } from "./verifier-source-loader.mjs";
let implementation;
/** Create an SDK-compatible transport which retains the exact process owner. */
export async function createOwnedMcpStdioTransport(options) {
implementation ??= importVerifierSource(
new URL("../../src/process/OwnedMcpStdioTransport.ts", import.meta.url)
.href,
import.meta.url,
);
const { OwnedMcpStdioTransport } = await implementation;
return new OwnedMcpStdioTransport(options, {
provider: "artifact-verifier",
label: "Artifact verifier",
// Match the SDK's default maximum buffered stdio message size.
maxFrameBytes: 10 * 1024 * 1024,
// Unlike the JADX adapter, artifact verifiers do not impose a cumulative
// per-connection output limit; the SDK bounds each buffered message.
maxOperationBytes: Infinity,
gracefulInputCloseMs: 2000,
maxDiagnosticBytes: 64 * 1024,
});
}
+18
View File
@@ -61,6 +61,10 @@ const stable = (result) =>
file_address: event.file_address,
selector: event.selector,
receiver_class: event.receiver_class,
source_location: event.source_location,
caller_source_locations: event.backtrace.map(
(frame) => frame.source_location,
),
counter:
event.symbol === GREET
? event.registers[3].value
@@ -96,6 +100,18 @@ const checkScenario = async (binary) => {
});
const greet = result.events.filter(({ symbol }) => symbol === GREET);
const add = result.events.filter(({ symbol }) => symbol === "rea_call_add");
for (const location of [
...result.events,
...result.events.flatMap((event) => event.backtrace),
...result.breakpoints.flatMap((breakpoint) => breakpoint.locations),
]) {
assert.equal(location.source_location?.file_path, SOURCE);
assert.ok(location.source_location.line > 0);
assert.ok(
location.source_location.column === null ||
location.source_location.column > 0,
);
}
assert.equal(greet.length, 3, JSON.stringify(result));
assert.equal(add.length, 3, JSON.stringify(result));
for (const [index, event] of greet.entries()) {
@@ -269,6 +285,8 @@ try {
const binary = join(directory, "ReaCalls");
await exec("/usr/bin/xcrun", [
"clang",
"-g",
"-O0",
"-fobjc-arc",
"-framework",
"Foundation",
+186
View File
@@ -0,0 +1,186 @@
import assert from "node:assert/strict";
import { readFile, rm } from "node:fs/promises";
import { isAbsolute, join } from "node:path";
import { fileURLToPath } from "node:url";
import { createPackageVerifierCommands } from "./lib/verify-package-commands.mjs";
import { withVerifierCleanup } from "./lib/verifier-cleanup.mjs";
import {
artifactCli,
artifactMcpResult,
withArtifactMcp,
} from "./lib/artifact-e2e.mjs";
import { createCanonicalTemporaryDirectory } from "./lib/canonical-temporary-directory.mjs";
const cancellation = new AbortController();
let shutdownSignal;
const signalExitCodes = { SIGINT: 130, SIGTERM: 143 };
const forwardCancellation = (signal) => {
if (shutdownSignal !== undefined) return;
shutdownSignal = signal;
cancellation.abort(new Error(`Verification cancelled by ${signal}`));
};
const signalHandlers = Object.fromEntries(
Object.keys(signalExitCodes).map((signal) => [
signal,
() => forwardCancellation(signal),
]),
);
for (const [signal, handler] of Object.entries(signalHandlers))
process.on(signal, handler);
const compiler = process.env.REA_VERIFY_FPC_COMMAND;
if (process.platform !== "darwin" || !isAbsolute(compiler ?? ""))
throw new Error(
"Requires macOS and an absolute caller-selected REA_VERIFY_FPC_COMMAND; build the runtime first.",
);
const source = fileURLToPath(
new URL("../tests/conformance/native/calls.pas", import.meta.url),
);
const fixtureLines = (await readFile(source, "utf8")).split(/\r?\n/u);
const callerLines = new Set(
["REA_VERIFY_CALLER_LOOP", "REA_VERIFY_CALLER_CALL"].map((marker) => {
const matches = fixtureLines.flatMap((line, index) =>
line.includes(marker) ? [index + 1] : [],
);
assert.equal(matches.length, 1, `fixture has one ${marker} marker`);
return matches[0];
}),
);
const allocatedDirectory =
await createCanonicalTemporaryDirectory("rea-pascal-calls-");
let directory = allocatedDirectory.path;
const commands = await createPackageVerifierCommands({
workspace: {
path: allocatedDirectory.allocatedPath,
close: () =>
rm(allocatedDirectory.allocatedPath, { recursive: true, force: true }),
},
});
const execute = (command, arguments_, { timeout }) =>
commands.exec(command, arguments_, {
timeout,
diagnosticBytes: 1024 * 1024,
signal: cancellation.signal,
});
const input = {
breakpoints: [{ kind: "function", name: "rea_pascal_call" }],
argument_registers: 1,
backtrace_frames: 1,
duration_ms: 30_000,
};
const inspect = (result) => {
assert.equal(result.process.outcome, "exited");
assert.equal(result.process.exit_status, 0);
assert.equal(result.process.terminated, true);
assert.equal(result.process.stdout.text, "pascal 0\npascal 1\npascal 2\n");
assert.equal(result.events.length, 3);
assert.deepEqual(
result.events.map((event) => event.registers[0].value),
["0x0", "0x1", "0x2"],
);
assert.equal(result.breakpoints[0].location_count, 1);
for (const event of result.events) {
assert.equal(event.backtrace.length, 1);
assert.equal(event.backtrace[0].source_location?.file_path, source);
assert.equal(event.backtrace[0].module_path, join(directory, "calls"));
assert.ok(
callerLines.has(event.backtrace[0].source_location.line),
"caller is the Pascal loop/call site",
);
}
for (const location of [
...result.events,
...result.breakpoints[0].locations,
]) {
assert.equal(location.source_location?.file_path, source);
assert.ok(location.source_location.line > 0);
}
};
const stable = (result) =>
result.events.map((event) => ({
file_address: event.file_address,
source_location: event.source_location,
registers: event.registers,
callers: event.backtrace.map((frame) => ({
file_address: frame.file_address,
symbol: frame.symbol,
source_location: frame.source_location,
})),
}));
try {
const result = await withVerifierCleanup(async () => {
const version = (
await execute(compiler, ["-iV"], { timeout: 30_000 })
).stdout.trim();
const architecture = (
await execute(compiler, ["-iTP"], {
timeout: 30_000,
})
).stdout.trim();
await execute(
compiler,
[
"-dDEBUG",
"-gw3",
"-O-",
"-Xs-",
`-FE${directory}`,
`-FU${directory}`,
source,
],
{ timeout: 120_000 },
);
const binary = join(directory, "calls");
const cli = await artifactCli(
"observe-native-calls",
binary,
[JSON.stringify(input)],
{ timeoutMs: 180_000, signal: cancellation.signal, commands },
);
cancellation.signal.throwIfAborted();
inspect(cli);
await withArtifactMcp(
binary,
async (client) => {
const mcp = await artifactMcpResult(
client,
"observe_native_calls",
input,
{ signal: cancellation.signal },
);
cancellation.signal.throwIfAborted();
inspect(mcp);
assert.deepEqual(stable(mcp), stable(cli));
},
{ signal: cancellation.signal, commands },
);
cancellation.signal.throwIfAborted();
return {
ok: true,
mocked: false,
compiler,
version,
architecture,
debugger: cli.debugger,
cli: true,
stdio_mcp: true,
scope:
"bounded native call entries and compiler-provided source coordinates; no typed Pascal locals or interactive stepping",
};
}, [() => commands.close()]);
cancellation.signal.throwIfAborted();
console.log(JSON.stringify(result));
} catch (cause) {
const cancelled =
cause === cancellation.signal.reason ||
(cause?.reason === "cancelled" &&
cause.cancellation?.reason === cancellation.signal.reason &&
cause.cleanupFailure === null);
if (shutdownSignal === undefined || !cancelled) throw cause;
console.error(`Verification cancelled by ${shutdownSignal}.`);
process.exitCode = signalExitCodes[shutdownSignal];
} finally {
for (const signal of Object.keys(signalExitCodes))
process.off(signal, signalHandlers[signal]);
}
+20 -266
View File
@@ -1,275 +1,29 @@
import {
deserializeMessage,
serializeMessage,
type JSONRPCMessage,
type Transport,
} from "@modelcontextprotocol/client";
import { randomUUID } from "node:crypto";
import { finished } from "node:stream/promises";
import type {
OwnedProviderProcessSpawnOptions,
SpawnedOwnedProviderProcess,
} from "../process/ProviderProcess.js";
import type { AnalysisCapturedOutput } from "../domain/analysisErrorBase.js";
import {
ProviderProcessSupervisor,
spawnOwnedProviderProcess,
} from "../process/ProviderProcess.js";
import { cleanupOwnedProcessGroup } from "../process/ProcessOwnership.js";
import { ProviderCleanupError } from "../domain/providerCleanupError.js";
OwnedMcpStdioTransport,
type OwnedMcpLauncher,
type OwnedMcpTransportFailure,
} from "../process/OwnedMcpStdioTransport.js";
import type { OwnedProviderProcessSpawnOptions } from "../process/ProviderProcess.js";
const MAX_FRAME_BYTES = 8 * 1024 * 1024;
const MAX_OPERATION_BYTES = 32 * 1024 * 1024;
/** Injectable acquisition seam; production owns a POSIX group or Windows job. */
export type JadxLauncher = (
options: OwnedProviderProcessSpawnOptions,
) => Promise<SpawnedOwnedProviderProcess>;
/** The first observed failure, recorded before notifying the SDK or beginning cleanup. */
export type JadxTransportFailure =
| { readonly kind: "process"; readonly cause: unknown }
| { readonly kind: "exit" }
| { readonly kind: "protocol"; readonly cause: Error }
| { readonly kind: "interrupted"; readonly reason: unknown };
/** SDK-framed stdio with REA-owned process cleanup rather than an unowned child. */
export class JadxMcpTransport implements Transport {
onclose: Transport["onclose"];
onerror: Transport["onerror"];
onmessage: Transport["onmessage"];
readonly runId = randomUUID();
#spawned: SpawnedOwnedProviderProcess | undefined;
#supervisor: ProviderProcessSupervisor | undefined;
#frameBuffer = Buffer.alloc(0);
#frameLength = 0;
#bytes = 0;
#started = false;
#closed = false;
#closePromise: Promise<void> | undefined;
#starting: Promise<void> | undefined;
#failure: JadxTransportFailure | undefined;
export type JadxLauncher = OwnedMcpLauncher;
export type JadxTransportFailure = OwnedMcpTransportFailure;
/** The JADX provider's bounded stdio protocol transport. */
export class JadxMcpTransport extends OwnedMcpStdioTransport {
constructor(
readonly options: Omit<OwnedProviderProcessSpawnOptions, "runId" | "stdin">,
readonly launcher: JadxLauncher = spawnOwnedProviderProcess,
) {}
/** Acquires exactly one protocol process after the SDK installs callbacks. */
start(): Promise<void> {
if (this.#started || this.#closed)
throw new Error("JADX transport cannot be restarted");
this.#started = true;
this.#starting = this.#acquire().catch((cause: unknown) => {
this.#failure ??= { kind: "process", cause };
throw cause;
});
return this.#starting;
}
async #acquire(): Promise<void> {
const spawned = await this.launcher({
...this.options,
runId: this.runId,
stdin: "pipe",
});
this.#spawned = spawned;
this.#supervisor = new ProviderProcessSupervisor(
options: Omit<OwnedProviderProcessSpawnOptions, "runId" | "stdin">,
launcher?: JadxLauncher,
) {
super(
options,
{
...spawned,
ownsProcessLifetime: true,
cleanup:
spawned.cleanup ??
(() => cleanupOwnedProcessGroup(spawned.ownership)),
provider: "jadx",
label: "JADX",
maxFrameBytes: 8 * 1024 * 1024,
maxOperationBytes: 32 * 1024 * 1024,
notifyOnExit: true,
},
{ captureStdout: false },
launcher,
);
if (
spawned.process.stdin === undefined ||
spawned.process.stdin === null ||
spawned.process.stdout === null
)
throw new Error("JADX launcher did not supply protocol stdin/stdout");
spawned.process.stdout.on("data", this.#read);
spawned.process.stdout.on("error", this.#error);
spawned.process.stderr?.on("data", this.#countStderr);
spawned.process.on("error", this.#error);
spawned.process.once("exit", this.#exit);
spawned.process.stdin.on("error", this.#error);
if (this.#closed) throw new Error("JADX transport closed during startup");
}
/** Writes SDK-serialized messages and observes stream backpressure. */
async send(message: JSONRPCMessage): Promise<void> {
const input = this.#spawned?.process.stdin;
if (this.#closed || input === undefined || input === null)
throw new Error("JADX protocol process is not writable");
await new Promise<void>((resolve, reject) => {
try {
input.write(serializeMessage(message), (cause) => {
if (cause === null || cause === undefined) resolve();
else {
this.#failure ??= { kind: "process", cause };
reject(cause);
}
});
} catch (cause) {
this.#failure ??= { kind: "process", cause };
reject(cause);
}
});
}
/** Verifies owned process exit before declaring the transport closed. */
close(): Promise<void> {
this.#closePromise ??= this.#stop().catch((cause: unknown) => {
this.#closePromise = undefined;
throw cause;
});
return this.#closePromise;
}
/** Keep the interruption reason distinct from errors produced by its teardown. */
interrupt(reason: unknown): Promise<void> {
this.#failure ??= { kind: "interrupted", reason };
return this.close();
}
/** Captured provider stderr without persisting inherited environment values. */
diagnostics(): AnalysisCapturedOutput | undefined {
const owner = this.#supervisor;
if (owner === undefined) return undefined;
const snapshot = owner.snapshot();
return {
stdout: "",
stderr: snapshot.stderr.text,
stderr_bytes: snapshot.stderr.observedBytes,
truncated:
snapshot.diagnosticTruncated === true || !owner.observedOutputClose(),
...(snapshot.exitCode === undefined
? {}
: { exit_code: snapshot.exitCode }),
...(snapshot.signal === undefined ? {} : { signal: snapshot.signal }),
};
}
/** Preserve the failed protocol/resource constraint after the SDK closes requests. */
observedFailure(): JadxTransportFailure | undefined {
return this.#failure;
}
/** Scope both the byte budget and diagnostic retention to the next serialized request. */
beginOperation(): void {
this.#bytes = 0;
this.#supervisor?.resetOutput();
}
async #stop(): Promise<void> {
this.#closed = true;
this.#frameBuffer = Buffer.alloc(0);
this.#frameLength = 0;
// A cancellation may race acquisition. Join it before inspecting ownership.
await this.#starting?.catch(() => undefined);
const process = this.#spawned?.process;
process?.stdout?.off("data", this.#read);
process?.stdout?.off("error", this.#error);
process?.stderr?.off("data", this.#countStderr);
process?.off("error", this.#error);
process?.off("exit", this.#exit);
const input = process?.stdin;
// A job termination can complete an in-flight pipe write asynchronously.
// Observe its errors until the owned Writable has finished closing.
const inputClosed =
input === undefined || input === null
? undefined
: finished(input, { readable: false, cleanup: true }).catch(
() => undefined,
);
const stopped = await this.#supervisor?.stop();
input?.destroy();
await inputClosed;
input?.off("error", this.#error);
this.onclose?.();
if (stopped?.status === "incomplete")
throw new ProviderCleanupError("jadx", [this.runId], {
reason: stopped.reason,
stderr: this.diagnostics()?.stderr ?? "",
});
}
readonly #error = (error: Error): void => {
this.#failure ??= { kind: "process", cause: error };
this.onerror?.(error);
// Reject pending SDK work after joining the same owned teardown on every path.
void this.close().catch((cause: unknown) =>
this.onerror?.(cause instanceof Error ? cause : new Error(String(cause))),
);
};
readonly #exit = (): void => {
this.#failure ??= { kind: "exit" };
this.onclose?.();
};
readonly #countStderr = (chunk: Buffer | string): void => {
this.#bytes += Buffer.byteLength(chunk);
if (this.#bytes > MAX_OPERATION_BYTES)
this.#fail(
new Error(
`JADX operation output exceeds ${MAX_OPERATION_BYTES} bytes; no complete result is available`,
),
);
};
readonly #read = (chunk: Buffer | string): void => {
if (this.#closed) return;
const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
this.#bytes += bytes.length;
if (this.#bytes > MAX_OPERATION_BYTES) {
this.#fail(
new Error(
`JADX operation output exceeds ${MAX_OPERATION_BYTES} bytes; no complete result is available`,
),
);
return;
}
try {
let offset = 0;
while (offset < bytes.length) {
const newline = bytes.indexOf(10, offset);
const end = newline === -1 ? bytes.length : newline;
const fragment = bytes.subarray(offset, end);
const frameBytes = this.#frameLength + fragment.length;
if (frameBytes > MAX_FRAME_BYTES)
throw new Error(`JADX MCP frame exceeds ${MAX_FRAME_BYTES} bytes`);
this.#appendFrame(fragment, frameBytes);
if (newline === -1) return;
const line = new TextDecoder("utf-8", {
fatal: true,
// Preserve a leading BOM so the JSON-RPC parser can reject it.
ignoreBOM: true,
}).decode(this.#frameBuffer.subarray(0, this.#frameLength));
this.#frameBuffer = Buffer.alloc(0);
this.#frameLength = 0;
this.onmessage?.(deserializeMessage(line));
offset = newline + 1;
}
} catch (cause) {
this.#fail(cause instanceof Error ? cause : new Error(String(cause)));
}
};
#appendFrame(fragment: Buffer, frameBytes: number): void {
if (frameBytes > this.#frameBuffer.length) {
const capacity = Math.min(
MAX_FRAME_BYTES,
Math.max(frameBytes, this.#frameBuffer.length * 2, 1),
);
const grown = Buffer.allocUnsafe(capacity);
this.#frameBuffer.copy(grown, 0, 0, this.#frameLength);
this.#frameBuffer = grown;
}
fragment.copy(this.#frameBuffer, this.#frameLength);
this.#frameLength = frameBytes;
}
#fail(error: Error): void {
this.#failure ??= { kind: "protocol", cause: error };
this.#error(error);
}
}
@@ -125,6 +125,18 @@ export const nativeCodeLocationSchema = z.strictObject({
module: z.string().nullable(),
module_path: z.string().nullable(),
symbol: z.string().nullable(),
/** LLDB line-table attribution; does not verify source file contents. */
source_location: z
.strictObject({
file_path: z.string().min(1).nullable(),
line: z.number().int().positive().nullable(),
column: z.number().int().positive().nullable(),
})
.nullable()
.optional()
.describe(
"Compiler or debugger source attribution only; REA does not read or verify the named source-file bytes.",
),
});
/** One observed entry into a requested function or method. */
+1
View File
@@ -82,6 +82,7 @@ export const projectLldbPartialObservation = (input: {
limitations: [
"The LLDB run did not produce a complete native call observation result.",
"The target SHA-256 identifies the selected artifact; LLDB did not verify the bytes loaded into the process.",
"Source paths and lines are LLDB line-table attribution; REA did not read or verify the referenced source bytes.",
"Captured output byte counts are observed lower bounds because LLDB did not report final stream counters.",
...input.journal.limitations,
...(input.journalLimitations ?? []),
+1
View File
@@ -36,6 +36,7 @@ export const NATIVE_CALL_OBSERVATION_LIMITATIONS = [
"Registers are the raw integer argument registers at entry. For Objective-C methods the first two hold the receiver and the selector.",
"Receiver classes come from LLDB's Objective-C runtime reader without running target code; they are null for class methods and for receivers LLDB cannot classify.",
"Calls that never reach a symbol's entry, such as inlined code or objc_direct methods called directly, are not observed.",
"Source locations come from LLDB line-table metadata; REA does not read or verify the named source-file contents.",
"The target is launched directly under LLDB, not through LaunchServices; LLDB disables ASLR for it, so load addresses can equal file addresses.",
"The observation window is checked about once a second, and recording pauses the process at every entry, which slows it.",
] as const;
+303
View File
@@ -0,0 +1,303 @@
import {
deserializeMessage,
serializeMessage,
type JSONRPCMessage,
type Transport,
} from "@modelcontextprotocol/client";
import { randomUUID } from "node:crypto";
import { finished } from "node:stream/promises";
import type {
OwnedProviderProcessSpawnOptions,
SpawnedOwnedProviderProcess,
} from "./ProviderProcess.js";
import type { AnalysisCapturedOutput } from "../domain/analysisErrorBase.js";
import {
ProviderProcessSupervisor,
spawnOwnedProviderProcess,
} from "./ProviderProcess.js";
import { cleanupOwnedProcessGroup } from "./ProcessOwnership.js";
import { ProviderCleanupError } from "../domain/providerCleanupError.js";
/** Injectable acquisition seam; production owns a POSIX group or Windows job. */
export type OwnedMcpLauncher = (
options: OwnedProviderProcessSpawnOptions,
) => Promise<SpawnedOwnedProviderProcess>;
/** The first observed failure, recorded before notifying the SDK or beginning cleanup. */
export type OwnedMcpTransportFailure =
| { readonly kind: "process"; readonly cause: unknown }
| { readonly kind: "exit" }
| { readonly kind: "protocol"; readonly cause: Error }
| { readonly kind: "interrupted"; readonly reason: unknown };
/** SDK-framed stdio with REA-owned process cleanup rather than an unowned child. */
export class OwnedMcpStdioTransport implements Transport {
onclose: Transport["onclose"];
onerror: Transport["onerror"];
onmessage: Transport["onmessage"];
readonly runId = randomUUID();
#spawned: SpawnedOwnedProviderProcess | undefined;
#supervisor: ProviderProcessSupervisor | undefined;
#frameBuffer = Buffer.alloc(0);
#frameLength = 0;
#bytes = 0;
#started = false;
#closed = false;
#closePromise: Promise<void> | undefined;
#starting: Promise<void> | undefined;
#failure: OwnedMcpTransportFailure | undefined;
constructor(
readonly options: Omit<OwnedProviderProcessSpawnOptions, "runId" | "stdin">,
readonly settings: {
readonly provider: string;
readonly label: string;
readonly maxFrameBytes: number;
readonly maxOperationBytes: number;
readonly gracefulInputCloseMs?: number;
readonly maxDiagnosticBytes?: number;
readonly notifyOnExit?: boolean;
},
readonly launcher: OwnedMcpLauncher = spawnOwnedProviderProcess,
) {}
/** Acquires exactly one protocol process after the SDK installs callbacks. */
start(): Promise<void> {
if (this.#started || this.#closed)
throw new Error(`${this.settings.label} transport cannot be restarted`);
this.#started = true;
this.#starting = this.#acquire().catch((cause: unknown) => {
this.#failure ??= { kind: "process", cause };
throw cause;
});
return this.#starting;
}
async #acquire(): Promise<void> {
const spawned = await this.launcher({
...this.options,
runId: this.runId,
stdin: "pipe",
});
this.#spawned = spawned;
this.#supervisor = new ProviderProcessSupervisor(
{
...spawned,
ownsProcessLifetime: true,
cleanup:
spawned.cleanup ??
(() => cleanupOwnedProcessGroup(spawned.ownership)),
},
{
captureStdout: false,
...(this.settings.maxDiagnosticBytes === undefined
? {}
: { maxDiagnosticBytes: this.settings.maxDiagnosticBytes }),
},
);
if (
spawned.process.stdin === undefined ||
spawned.process.stdin === null ||
spawned.process.stdout === null
)
throw new Error(
`${this.settings.label} launcher did not supply protocol stdin/stdout`,
);
spawned.process.stdout.on("data", this.#read);
spawned.process.stdout.on("error", this.#error);
spawned.process.stderr?.on("data", this.#countStderr);
spawned.process.on("error", this.#error);
spawned.process.once("exit", this.#exit);
spawned.process.once("close", this.#outputClosed);
spawned.process.stdin.on("error", this.#error);
if (this.#closed)
throw new Error(`${this.settings.label} transport closed during startup`);
}
/** Writes SDK-serialized messages and observes stream backpressure. */
async send(message: JSONRPCMessage): Promise<void> {
const input = this.#spawned?.process.stdin;
if (this.#closed || input === undefined || input === null)
throw new Error(
`${this.settings.label} protocol process is not writable`,
);
await new Promise<void>((resolve, reject) => {
try {
input.write(serializeMessage(message), (cause) => {
if (cause === null || cause === undefined) resolve();
else {
this.#failure ??= { kind: "process", cause };
reject(cause);
}
});
} catch (cause) {
this.#failure ??= { kind: "process", cause };
reject(cause);
}
});
}
/** Verifies owned process exit before declaring the transport closed. */
close(): Promise<void> {
this.#closePromise ??= this.#stop().catch((cause: unknown) => {
this.#closePromise = undefined;
throw cause;
});
return this.#closePromise;
}
/** Keep the interruption reason distinct from errors produced by its teardown. */
interrupt(reason: unknown): Promise<void> {
this.#failure ??= { kind: "interrupted", reason };
return this.close();
}
/** Captured provider stderr without persisting inherited environment values. */
diagnostics(): AnalysisCapturedOutput | undefined {
const owner = this.#supervisor;
if (owner === undefined) return undefined;
const snapshot = owner.snapshot();
return {
stdout: "",
stderr: snapshot.stderr.text,
stderr_bytes: snapshot.stderr.observedBytes,
truncated:
snapshot.diagnosticTruncated === true || !owner.observedOutputClose(),
...(snapshot.exitCode === undefined
? {}
: { exit_code: snapshot.exitCode }),
...(snapshot.signal === undefined ? {} : { signal: snapshot.signal }),
};
}
/** Preserve the failed protocol/resource constraint after the SDK closes requests. */
observedFailure(): OwnedMcpTransportFailure | undefined {
return this.#failure;
}
/** Scope both the byte budget and diagnostic retention to the next serialized request. */
beginOperation(): void {
this.#bytes = 0;
this.#supervisor?.resetOutput();
}
async #stop(): Promise<void> {
this.#closed = true;
this.#frameBuffer = Buffer.alloc(0);
this.#frameLength = 0;
// A cancellation may race acquisition. Join it before inspecting ownership.
await this.#starting?.catch(() => undefined);
const process = this.#spawned?.process;
process?.stdout?.off("data", this.#read);
process?.stdout?.off("error", this.#error);
process?.stderr?.off("data", this.#countStderr);
process?.off("error", this.#error);
process?.off("exit", this.#exit);
process?.off("close", this.#outputClosed);
const input = process?.stdin;
// A job termination can complete an in-flight pipe write asynchronously.
// Observe its errors until the owned Writable has finished closing.
const inputClosed =
input === undefined || input === null
? undefined
: finished(input, { readable: false, cleanup: true }).catch(
() => undefined,
);
const gracefulInputCloseMs = this.settings.gracefulInputCloseMs ?? 0;
if (input !== undefined && input !== null && gracefulInputCloseMs > 0) {
input.end();
await this.#supervisor?.waitForExit(gracefulInputCloseMs);
}
const stopped = await this.#supervisor?.stop();
input?.destroy();
await inputClosed;
input?.off("error", this.#error);
this.onclose?.();
if (stopped?.status === "incomplete")
throw new ProviderCleanupError(this.settings.provider, [this.runId], {
reason: stopped.reason,
stderr: this.diagnostics()?.stderr ?? "",
});
}
readonly #error = (error: Error): void => {
this.#failure ??= { kind: "process", cause: error };
this.onerror?.(error);
// Reject pending SDK work after joining the same owned teardown on every path.
void this.close().catch((cause: unknown) =>
this.onerror?.(cause instanceof Error ? cause : new Error(String(cause))),
);
};
readonly #exit = (): void => {
this.#failure ??= { kind: "exit" };
if (this.settings.notifyOnExit === true) this.onclose?.();
};
readonly #outputClosed = (): void => {
if (this.settings.notifyOnExit !== true) this.onclose?.();
};
readonly #countStderr = (chunk: Buffer | string): void => {
this.#bytes += Buffer.byteLength(chunk);
if (this.#bytes > this.settings.maxOperationBytes)
this.#fail(
new Error(
`${this.settings.label} operation output exceeds ${this.settings.maxOperationBytes} bytes; no complete result is available`,
),
);
};
readonly #read = (chunk: Buffer | string): void => {
if (this.#closed) return;
const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
this.#bytes += bytes.length;
if (this.#bytes > this.settings.maxOperationBytes) {
this.#fail(
new Error(
`${this.settings.label} operation output exceeds ${this.settings.maxOperationBytes} bytes; no complete result is available`,
),
);
return;
}
try {
let offset = 0;
while (offset < bytes.length) {
const newline = bytes.indexOf(10, offset);
const end = newline === -1 ? bytes.length : newline;
const fragment = bytes.subarray(offset, end);
const frameBytes = this.#frameLength + fragment.length;
if (frameBytes > this.settings.maxFrameBytes)
throw new Error(
`${this.settings.label} MCP frame exceeds ${this.settings.maxFrameBytes} bytes`,
);
this.#appendFrame(fragment, frameBytes);
if (newline === -1) return;
const line = new TextDecoder("utf-8", {
fatal: true,
// Preserve a leading BOM so the JSON-RPC parser can reject it.
ignoreBOM: true,
}).decode(this.#frameBuffer.subarray(0, this.#frameLength));
this.#frameBuffer = Buffer.alloc(0);
this.#frameLength = 0;
this.onmessage?.(deserializeMessage(line));
offset = newline + 1;
}
} catch (cause) {
this.#fail(cause instanceof Error ? cause : new Error(String(cause)));
}
};
#appendFrame(fragment: Buffer, frameBytes: number): void {
if (frameBytes > this.#frameBuffer.length) {
const capacity = Math.min(
this.settings.maxFrameBytes,
Math.max(frameBytes, this.#frameBuffer.length * 2, 1),
);
const grown = Buffer.allocUnsafe(capacity);
this.#frameBuffer.copy(grown, 0, 0, this.#frameLength);
this.#frameBuffer = grown;
}
fragment.copy(this.#frameBuffer, this.#frameLength);
this.#frameLength = frameBytes;
}
#fail(error: Error): void {
this.#failure ??= { kind: "protocol", cause: error };
this.#error(error);
}
}
@@ -5,6 +5,7 @@ import {
type JSONRPCMessage,
} from "@modelcontextprotocol/client";
import { expect, it } from "vitest";
import { OwnedMcpStdioTransport } from "../../../src/process/OwnedMcpStdioTransport.js";
import type { ProviderProcessHandle } from "../../../src/process/ProviderProcess.js";
import {
JadxMcpTransport,
@@ -125,6 +126,47 @@ it("decodes split UTF-8 and newline bytes, then routes multiple complete frames"
await transport.close();
});
it("delivers buffered protocol output before reporting process close", async () => {
const process = new FixtureProcess();
const transport = new OwnedMcpStdioTransport(
{ command: "fixture", arguments: [] },
{
provider: "fixture",
label: "Fixture",
maxFrameBytes: 1024,
maxOperationBytes: 4096,
},
fixtureLauncher(process),
);
const events: string[] = [];
transport.onmessage = () => events.push("message");
transport.onclose = () => events.push("close");
await transport.start();
process.emit("exit", 0, null);
expect(transport.observedFailure()).toEqual({ kind: "exit" });
expect(events).toEqual([]);
process.stdout.write(serializeMessage({ jsonrpc: "2.0", id: 1, result: {} }));
process.emit("close", 0, null);
expect(events).toEqual(["message", "close"]);
await transport.close();
});
it("reports JADX exit even when a descendant holds the protocol pipes", async () => {
const process = new FixtureProcess();
const transport = new JadxMcpTransport(
{ command: "fixture", arguments: [] },
fixtureLauncher(process),
);
let closed = false;
transport.onclose = () => {
closed = true;
};
await transport.start();
process.emit("exit", 0, null);
expect(closed).toBe(true);
await transport.close();
});
it.each([
["invalid UTF-8 byte", Buffer.from([0xff])],
["truncated UTF-8 sequence", Buffer.from([0xe2, 0x82])],
@@ -221,3 +263,46 @@ it("retries an incomplete owned stop while joining concurrent close callers", as
await transport.close();
await transport.close();
});
it("joins late acquisition before releasing the protocol process", async () => {
const process = new FixtureProcess();
let deliver:
| ((launch: Awaited<ReturnType<JadxLauncher>>) => void)
| undefined;
let acquired = false;
let cleaned = false;
const transport = new JadxMcpTransport(
{ command: "fixture", arguments: [] },
async ({ runId }) =>
new Promise((resolve) => {
acquired = true;
deliver = resolve;
// The owned process is returned only when the test releases acquisition.
void runId;
}),
);
const startup = transport.start();
expect(acquired).toBe(true);
let finished = false;
const closing = transport.close().then(() => {
finished = true;
});
await new Promise<void>((resolve) => setImmediate(resolve));
expect(finished).toBe(false);
deliver?.({
process,
ownership: { runId: transport.runId, leaderPid: 1, processGroupId: 1 },
cleanup: async () => {
cleaned = true;
process.emit("exit", 0, null);
process.emit("close", 0, null);
process.stdout.end();
process.stderr.end();
process.stdin.end();
return { cleaned: true, signaled: false };
},
});
await expect(startup).rejects.toThrow("closed during startup");
await closing;
expect(cleaned).toBe(true);
});
@@ -0,0 +1,123 @@
import { spawn } from "node:child_process";
import {
chmod,
readFile,
readdir,
realpath,
rm,
symlink,
writeFile,
} from "node:fs/promises";
import { basename, join } from "node:path";
import { expect, it } from "vitest";
import { createCanonicalTemporaryDirectory } from "../../../scripts/lib/canonical-temporary-directory.mjs";
import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
it("uses the canonical spelling for a verifier directory allocated through a symlink", async () => {
const root = await createTestTempDirectory("rea-pascal-verifier-path-");
const alias = join(root, "alias");
await symlink(root, alias, process.platform === "win32" ? "junction" : "dir");
const allocated = await createCanonicalTemporaryDirectory(
"rea-pascal-calls-",
alias,
);
try {
expect(allocated.path).toBe(await realpath(allocated.allocatedPath));
expect(allocated.path).not.toBe(
join(alias, basename(allocated.allocatedPath)),
);
} finally {
await rm(allocated.allocatedPath, { recursive: true, force: true });
}
});
it.skipIf(process.platform !== "darwin")(
"forwards verifier SIGTERM to the owned compiler group and removes its temporary directory",
async () => {
const root = await createTestTempDirectory("rea-pascal-verifier-cancel-");
const fakeCompiler = join(root, "fake-fpc");
const pidFile = join(root, "compiler-pids.txt");
await writeFile(
fakeCompiler,
'#!/usr/bin/env node\nimport { spawn } from "node:child_process";\nimport { writeFileSync } from "node:fs";\nconst child = spawn(process.execPath, ["-e", "setInterval(() => undefined, 1000)"], { stdio: "ignore" });\nwriteFileSync(process.env.REA_VERIFY_TEST_PID_FILE, `${process.pid} ${child.pid}\\n`);\nawait new Promise(() => undefined);\n',
);
await chmod(fakeCompiler, 0o700);
const verifier = spawn(
process.execPath,
["scripts/verify-native-pascal-calls.mjs"],
{
cwd: process.cwd(),
env: {
...process.env,
TMPDIR: root,
REA_VERIFY_FPC_COMMAND: fakeCompiler,
REA_VERIFY_TEST_PID_FILE: pidFile,
},
stdio: "ignore",
},
);
try {
const deadline = Date.now() + 10_000;
let pids: number[] | undefined;
while (Date.now() < deadline) {
try {
pids = (await readFile(pidFile, "utf8"))
.trim()
.split(/\s+/u)
.map(Number);
break;
} catch (cause: unknown) {
if ((cause as NodeJS.ErrnoException).code !== "ENOENT") throw cause;
await new Promise((resolve) => setTimeout(resolve, 20));
}
}
expect(pids, "the fake compiler started").toHaveLength(2);
verifier.kill("SIGTERM");
const result = await new Promise<{
code: number | null;
signal: NodeJS.Signals | null;
}>((resolve, reject) => {
verifier.once("error", reject);
verifier.once("close", (code, signal) => resolve({ code, signal }));
});
expect(result).toEqual({ code: 143, signal: null });
for (const pid of pids ?? [])
expect(() => process.kill(pid, 0)).toThrow();
expect(
(await readdir(root)).filter((name) =>
name.startsWith("rea-pascal-calls-"),
),
).toEqual([]);
} finally {
if (verifier.exitCode === null && verifier.signalCode === null) {
verifier.kill("SIGTERM");
await new Promise((resolve) => {
const timer = setTimeout(() => {
verifier.kill("SIGKILL");
resolve(undefined);
}, 1_000);
verifier.once("close", () => {
clearTimeout(timer);
resolve(undefined);
});
});
}
try {
const recorded = (await readFile(pidFile, "utf8"))
.trim()
.split(/\s+/u)
.map(Number);
for (const pid of recorded) {
try {
process.kill(pid, "SIGKILL");
} catch {
// The owned compiler group already exited.
}
}
} catch (cause: unknown) {
if ((cause as NodeJS.ErrnoException).code !== "ENOENT") throw cause;
}
await rm(root, { recursive: true, force: true });
}
},
);
@@ -67,7 +67,16 @@ const tracer: NativeCallTracer & { calls: number } = {
breakpoints: request.input.breakpoints.map((_, index) => ({
index,
location_count: 1,
locations: [],
locations: [
{
load_address: "0x1000",
file_address: "0x1000",
module: "Tool",
module_path: request.executable,
symbol: "open",
source_location: { file_path: null, line: 23, column: null },
},
],
})),
events: [],
resource_limit_reached: false,
@@ -133,6 +142,12 @@ it("routes observe_native_calls through MCP with schema-checked input and output
.parse(called.structuredContent).normalized_result,
);
expect(result.process).toMatchObject({ pid: 99, outcome: "exited" });
expect(result.breakpoints[0]?.locations[0]?.source_location).toEqual({
file_path: null,
line: 23,
column: null,
});
expect(result.target.environment).toEqual(
JSON.parse(
'{"__proto__":"selected literal value","constructor":"ordinary value"}',
@@ -124,3 +124,15 @@ it.each(["duplicate-first", "duplicate-last"])(
expect(result.error?.message).toMatch(/Duplicate MCP tool/u);
},
);
it("preserves cancellation received during final transport cleanup", async () => {
const result = await probe("cancel-finalization");
expect(result.accepted).toBe(false);
expect(result.error?.message).toBe("cancelled during final cleanup");
});
it("retains a work failure when cancellation arrives during cleanup", async () => {
const result = await probe("failure-cancel-finalization");
expect(result.accepted).toBe(false);
expect(result.error?.message).toBe("work failed before cancellation");
});
@@ -0,0 +1,66 @@
import { EventEmitter } from "node:events";
import { access, mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { PassThrough } from "node:stream";
import { expect, it } from "vitest";
import { createPackageVerifierCommands } from "../../../scripts/lib/verify-package-commands.mjs";
import { OwnedMcpStdioTransport } from "../../../src/process/OwnedMcpStdioTransport.js";
it("retains the workspace until an owned protocol cleanup can be retried", async () => {
const path = await mkdtemp(join(tmpdir(), "rea-verifier-transport-"));
const process = Object.assign(new EventEmitter(), {
pid: 1,
stdout: new PassThrough(),
stderr: new PassThrough(),
stdin: new PassThrough(),
exitCode: null,
signalCode: null,
kill: () => true,
});
let refuse = true;
const transport = new OwnedMcpStdioTransport(
{ command: "fixture", arguments: [] },
{
provider: "fixture",
label: "Fixture",
maxFrameBytes: 1024,
maxOperationBytes: 4096,
},
async ({ runId }) => ({
process,
ownership: { runId, leaderPid: 1, processGroupId: 1 },
cleanup: async () => {
if (refuse)
return { cleaned: false, reason: "injected cleanup failure" };
process.emit("close", 0, null);
process.stdout.end();
process.stderr.end();
process.stdin.end();
return { cleaned: true, signaled: false };
},
}),
);
const commands = await createPackageVerifierCommands({
workspace: {
path,
close: () => rm(path, { recursive: true, force: true }),
},
});
commands.adopt(transport, "protocol child");
try {
await transport.start();
// The group can still contain work after the leader's exit.
process.emit("exit", 0, null);
await expect(commands.close()).rejects.toMatchObject({
path,
cleanupOwner: commands,
});
await expect(access(path)).resolves.toBeUndefined();
refuse = false;
await commands.close();
await expect(access(path)).rejects.toMatchObject({ code: "ENOENT" });
} finally {
await rm(path, { recursive: true, force: true });
}
});
@@ -55,6 +55,26 @@ const observation = (
};
describe("LLDB retained observations", () => {
it("preserves partial source coordinates while accepting historical omission", () => {
const attributed = {
...event,
source_location: {
file_path: "/missing/fixture.pas",
line: 23,
column: null,
},
};
const journal =
[event, attributed]
.map((entry) => JSON.stringify({ kind: "event", event: entry }))
.join("\n") + "\n";
const partial = observation("cancelled", journal);
expect(partial.events).toEqual([event, attributed]);
expect(partial.limitations).toContain(
"Source paths and lines are LLDB line-table attribution; REA did not read or verify the referenced source bytes.",
);
});
it.each([
["cancellation", "cancelled"],
["bridge failure", "tracer-failure"],
@@ -35,6 +35,11 @@ const LOCATION = {
module: "Fixture",
module_path: "/tmp/Fixture",
symbol: "-[Greeter greet:times:]",
source_location: {
file_path: "/not-acquired/é space/fixture.pas",
line: 23,
column: null,
},
};
const run = (
@@ -84,7 +89,7 @@ const run = (
{ name: "x0", value: "0x6000" },
{ name: "x1", value: "0x7000" },
],
backtrace: [],
backtrace: [{ ...LOCATION, source_location: null }],
},
],
other_stops: [],
@@ -207,7 +212,12 @@ describe("observe_native_calls projection: projects completed observations", ()
expect(result.events[0]).toMatchObject({
receiver_class: "Greeter",
selector: "greet:times:",
source_location: LOCATION.source_location,
backtrace: [{ ...LOCATION, source_location: null }],
});
expect(result.breakpoints[0]?.locations[0]?.source_location).toEqual(
LOCATION.source_location,
);
expect(result.coverage).toEqual({
status: "complete",
event_limit_reached: false,
@@ -581,3 +591,24 @@ it("retains completed call facts when cancellation arrives with the tracer resul
reason: "cancelled",
});
});
it("grows repeated source-location JSON by the location object, not by a token estimate", () => {
const path = "/not-acquired/é space/fixture.pas";
const located = {
...LOCATION,
source_location: { file_path: path, line: 23, column: null },
};
const absent = { ...located, source_location: null };
const copies = 3;
const withPath = Buffer.byteLength(
JSON.stringify(Array.from({ length: copies }, () => located)),
);
const withoutPath = Buffer.byteLength(
JSON.stringify(Array.from({ length: copies }, () => absent)),
);
const perCopy =
Buffer.byteLength(JSON.stringify(located.source_location)) -
Buffer.byteLength("null");
expect(withPath - withoutPath).toBe(copies * perCopy);
expect(perCopy).toBeGreaterThan(Buffer.byteLength(path));
});
+16
View File
@@ -0,0 +1,16 @@
program ReaPascalCalls;
{$mode objfpc}
{$inline off}
procedure ReaPascalCall(Value: LongInt); cdecl; public name 'rea_pascal_call';
begin
WriteLn('pascal ', Value);
end;
var
Index: LongInt;
begin
for Index := 0 to 2 do { REA_VERIFY_CALLER_LOOP }
ReaPascalCall(Index); { REA_VERIFY_CALLER_CALL }
end.
+39 -20
View File
@@ -19,12 +19,24 @@ assert.ok(
"duplicate-last",
"override",
"request-options",
"cancel-finalization",
"failure-cancel-finalization",
].includes(mode),
);
assert.equal(typeof target, "string");
// Isolate the fault to this child client. The server, resource admission and
// advertised schema are real; alter only the reply before tool validation.
const controller = new AbortController();
if (mode === "cancel-finalization" || mode === "failure-cancel-finalization") {
const close = Client.prototype.close;
Client.prototype.close = async function () {
const settled = close.call(this);
controller.abort(new Error("cancelled during final cleanup"));
return settled;
};
}
const request = Client.prototype.request;
let catalogRequests = 0;
let corruptedReplies = 0;
@@ -160,27 +172,34 @@ try {
if (mode === "concurrent" || mode === "reconnect")
await standaloneConnections();
else
await artifact.withBundleResourcesMcp(target, async (client) => {
if (mode === "reuse") await Promise.all([status(client), status(client)]);
if (mode === "override")
await artifact.artifactMcpCall(
client,
{ name: "binary_session", arguments: {} },
{
toolDefinition: {
name: "binary_session",
inputSchema: { type: "object" },
outputSchema: { type: "object" },
await artifact.withBundleResourcesMcp(
target,
async (client) => {
if (mode === "failure-cancel-finalization")
throw new Error("work failed before cancellation");
if (mode === "reuse")
await Promise.all([status(client), status(client)]);
if (mode === "override")
await artifact.artifactMcpCall(
client,
{ name: "binary_session", arguments: {} },
{
toolDefinition: {
name: "binary_session",
inputSchema: { type: "object" },
outputSchema: { type: "object" },
},
},
},
);
if (mode === "request-options")
await artifact.artifactMcpCall(
client,
{ name: "binary_session", arguments: {} },
{ timeout: 12_345 },
);
});
);
if (mode === "request-options")
await artifact.artifactMcpCall(
client,
{ name: "binary_session", arguments: {} },
{ timeout: 12_345 },
);
},
{ signal: controller.signal },
);
} catch (cause) {
error = {
name: cause instanceof Error ? cause.name : "unknown",