fix(native): bound retained observations and preserve failure evidence

This commit is contained in:
morluto 2026-10-09 01:16:33 +08:00
parent d6cc9ca3a6
commit 7c323870c2
15 changed files with 1045 additions and 36 deletions

View File

@ -30,6 +30,8 @@ MAX_REPORTED_LOCATIONS = 64
MAX_OTHER_STOPS = 256
MAX_TRACE_BYTES = 8 * 1024 * 1024
MAX_TRACE_FRAMES = 65_536
MAX_OBSERVATION_JOURNAL_BYTES = MAX_TRACE_BYTES
MAX_BREAKPOINT_LOCATION_BYTES = 8 * 1024 * 1024
MAX_TARGET_OUTPUT_BYTES = 1024 * 1024
OUTPUT_DRAIN_BYTES_PER_PASS = 1024 * 1024
OUTPUT_EOF_WAIT_SECONDS = 1.0
@ -67,6 +69,24 @@ def _fits_trace_budget(value, remaining_bytes):
return _estimated_json_size(value) <= remaining_bytes
class _ObservationJournal:
"""Flush a bounded JSONL prefix so observations survive tracer failure."""
def __init__(self, handle, max_bytes=MAX_OBSERVATION_JOURNAL_BYTES):
self.handle = handle
self.max_bytes = max_bytes
self.bytes = 0
def append(self, record):
encoded = (json.dumps(record) + "\n").encode("utf-8")
if self.bytes + len(encoded) > self.max_bytes:
return False
self.handle.write(encoded)
self.handle.flush()
self.bytes += len(encoded)
return True
def _set_launch_flags(launch):
launch.SetLaunchFlags(
launch.GetLaunchFlags()
@ -412,13 +432,24 @@ def _handle_stop(context, events, other_stops):
return "resource-limit"
record["backtrace"].append(location)
record_bytes += addition
journal = context.get("observation_journal")
if journal is not None and not journal.append(
{"kind": "event", "event": record}
):
return "resource-limit"
events.append(record)
context["retained_frames"] += event_frame_count
context["retained_trace_bytes"] += record_bytes
if len(events) >= context["config"]["max_events"]:
return "event-limit"
elif reason in (lldb.eStopReasonSignal, lldb.eStopReasonException):
other_stops.append(thread.GetStopDescription(256))
stop_description = thread.GetStopDescription(256)
journal = context.get("observation_journal")
if journal is not None and not journal.append(
{"kind": "other-stop", "reason": stop_description}
):
return "resource-limit"
other_stops.append(stop_description)
if len(other_stops) >= MAX_OTHER_STOPS:
return "stop-limit"
return None
@ -426,6 +457,8 @@ def _handle_stop(context, events, other_stops):
def _breakpoint_report(target, created, config):
report = []
retained_location_bytes = 0
locations_truncated = False
for index, breakpoints in enumerate(created):
locations = []
count = 0
@ -437,9 +470,21 @@ def _breakpoint_report(target, created, config):
continue
count += 1
if len(locations) < MAX_REPORTED_LOCATIONS:
remaining_bytes = (
MAX_BREAKPOINT_LOCATION_BYTES - retained_location_bytes
)
if not _fits_trace_budget(location, remaining_bytes):
locations_truncated = True
continue
location_bytes = len(json.dumps(location).encode("utf-8"))
addition = location_bytes + (1 if locations else 0)
if addition > remaining_bytes:
locations_truncated = True
continue
locations.append(location)
retained_location_bytes += addition
report.append({"index": index, "location_count": count, "locations": locations})
return report
return report, locations_truncated
def _launch_info(target, config, listener):
@ -518,9 +563,15 @@ def trace(debugger, config):
raise
stdout.start()
stderr.start()
observation_journal = None
started = time.monotonic()
process = None
try:
observation_path = config.get("observation_path")
if observation_path:
observation_journal = _ObservationJournal(
open(observation_path, "wb")
)
process = target.Launch(_launch_info(target, config, listener), error)
if not error.Success() or not process.IsValid():
stdout.close()
@ -596,6 +647,7 @@ def trace(debugger, config):
"breakpoint_index": breakpoint_index,
"retained_frames": 0,
"retained_trace_bytes": 0,
"observation_journal": observation_journal,
}
events = []
other_stops = []
@ -641,6 +693,9 @@ def trace(debugger, config):
and module_uuid_before is not None
and module_uuid_after == module_uuid_before
)
breakpoint_report, breakpoint_locations_truncated = _breakpoint_report(
target, created, config
)
return {
"status": "traced",
"version": lldb.SBDebugger.GetVersionString().splitlines()[0],
@ -651,7 +706,8 @@ def trace(debugger, config):
"killed": outcome != "exited",
"terminated": process.GetState() == lldb.eStateExited,
"elapsed_ms": round((time.monotonic() - started) * 1000, 3),
"breakpoints": _breakpoint_report(target, created, config),
"breakpoints": breakpoint_report,
"breakpoint_locations_truncated": breakpoint_locations_truncated,
"events": events,
"resource_limit_reached": outcome == "resource-limit",
"target_output": {
@ -677,6 +733,11 @@ def trace(debugger, config):
}
finally:
if observation_journal is not None:
try:
observation_journal.handle.close()
except Exception:
pass
if process is not None:
try:
if process.IsValid() and process.GetState() != lldb.eStateExited:

View File

@ -1,6 +1,8 @@
"""Source-owned tests for the LLDB bridge's bounded retention helpers."""
from io import BytesIO
import importlib.util
import json
import os
import sys
import tempfile
@ -20,6 +22,50 @@ spec.loader.exec_module(tracer)
class NativeTraceBudgetTests(unittest.TestCase):
def test_journal_keeps_flushed_observations_when_later_execution_fails(self):
stream = BytesIO()
journal = tracer._ObservationJournal(stream)
event = {"sequence": 0, "symbol": "first"}
journal.append({"kind": "event", "event": event})
with self.assertRaisesRegex(RuntimeError, "later failure"):
raise RuntimeError("later failure")
stream.write(b'{"kind":"event"') # Simulate termination during a later row.
rows = stream.getvalue().splitlines()
self.assertEqual(
rows[0],
b'{"kind": "event", "event": {"sequence": 0, "symbol": "first"}}',
)
self.assertEqual(rows[1], b'{"kind":"event"')
def test_journal_checks_exact_byte_budget_before_writing(self):
record = {"kind": "other-stop", "reason": "signal"}
encoded_size = len((tracer.json.dumps(record) + "\n").encode("utf-8"))
stream = BytesIO()
journal = tracer._ObservationJournal(stream, max_bytes=encoded_size - 1)
self.assertFalse(journal.append(record))
self.assertEqual(stream.getvalue(), b"")
self.assertEqual(journal.bytes, 0)
def test_journal_appends_each_row_once_and_flushes_it(self):
class CountingStream(BytesIO):
writes = 0
flushes = 0
def write(self, value):
self.writes += 1
return super().write(value)
def flush(self):
self.flushes += 1
return super().flush()
stream = CountingStream()
journal = tracer._ObservationJournal(stream)
journal.append({"kind": "event", "event": {"sequence": 0}})
journal.append({"kind": "other-stop", "reason": "signal"})
self.assertEqual((stream.writes, stream.flushes), (2, 2))
self.assertEqual(journal.bytes, len(stream.getvalue()))
def test_aggregate_trace_budget_admits_exact_boundary_and_rejects_next_record(self):
self.assertTrue(
tracer._can_retain_trace(
@ -146,7 +192,7 @@ class NativeTraceBudgetTests(unittest.TestCase):
def GetLocationAtIndex(self, index):
return Location()
report = tracer._breakpoint_report(
report, locations_truncated = tracer._breakpoint_report(
object(),
[[Breakpoint()]],
{
@ -161,6 +207,81 @@ class NativeTraceBudgetTests(unittest.TestCase):
)
self.assertEqual(report[0]["location_count"], 0)
self.assertEqual(report[0]["locations"], [])
self.assertFalse(locations_truncated)
def test_breakpoint_report_bounds_aggregate_location_json(self):
tracer.lldb.LLDB_INVALID_ADDRESS = -1
symbol_name = (
"-[VeryLongButPlausibleObjectiveCController "
"performTransactionWithRequestIdentifier:accountContext:"
"securityPolicy:completionHandler:]"
)
module_path = (
"/Applications/Enterprise Product.app/Contents/Frameworks/"
"EnterpriseSubsystem.framework/Versions/A/EnterpriseSubsystem"
)
class Symbol:
def IsValid(self):
return True
def GetName(self):
return symbol_name
class FileSpec:
def GetFilename(self):
return "EnterpriseSubsystem"
def __str__(self):
return module_path
class Module:
def IsValid(self):
return True
def GetFileSpec(self):
return FileSpec()
class Address:
def GetModule(self):
return Module()
def GetSymbol(self):
return Symbol()
def GetFileAddress(self):
return 0x123456789
def GetLoadAddress(self, target):
return 0x7123456789
class Location:
def GetAddress(self):
return Address()
class Breakpoint:
def GetNumLocations(self):
return 64
def GetLocationAtIndex(self, index):
return Location()
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)
if __name__ == "__main__":

View File

@ -78,6 +78,10 @@ Provider calls receive the request cancellation signal. Artifact traversal,
hashing, version comparisons, Hopper requests, and process capture
check the same signal. Cancellation is distinct from timeout. A cleanup failure
uses `cleanup_incomplete` and lists only the owned resource kinds that remain.
Native call tracing and process capture retain available observations in
`details.partial_observation` on failure, including when cleanup succeeds.
The observation reports its partial coverage; cleanup details describe host
state separately from the execution failure.
Derived comparisons and reconstruction verification yield before computation
and before publication, so cancellation cannot race with successful Evidence.

View File

@ -217,6 +217,18 @@ The bridge limits retained event JSON to 8 MiB and caller frames to 65,536
across the run. It checks admission before retaining another complete event;
`resource-limit` reports which observations could not be completed without
silently changing the requested event or frame settings.
Resolved breakpoint locations have a separate aggregate 8 MiB metadata budget.
If their details exceed it, `location_count` retains the full observed match
count, coverage is partial, and `breakpoint_locations_truncated` is true. This
post-run metadata limit does not imply that the target was killed.
Accepted entries and signal stops are also flushed to a bounded 8 MiB journal.
If cancellation, timeout, capture failure, or a later bridge error prevents a
full result, the error retains this evidence in `details.partial_observation`
with partial coverage. Available output prefixes remain inline; their byte
counts are observed lower bounds when final drain counters are unavailable.
The original failure classification and cleanup uncertainty remain separate.
An interrupted journal row is ignored with an explicit limitation.
Only entries are observed: return values, floating-point and stack arguments,
and inlined or `objc_direct` calls are not. The target runs with the current

View File

@ -49,6 +49,7 @@ export interface AnalysisCleanupObservation {
export interface AnalysisErrorOptions extends ErrorOptions {
readonly capturedOutput?: AnalysisCapturedOutput;
readonly cleanup?: AnalysisCleanupObservation;
readonly partialObservation?: AnalysisPartialObservation;
}
/** Base class for expected analysis, provider, and session failures. */
@ -66,16 +67,22 @@ export abstract class AnalysisError extends Error {
this.cleanup = options?.cleanup;
this.cleanupIncomplete = options?.cleanup !== undefined;
this.cleanupResources = options?.cleanup?.resources ?? [];
this.partialObservation = options?.partialObservation;
}
abstract readonly _tag: AnalysisErrorTag;
readonly userMessage: string | undefined = undefined;
readonly userCategory: "cancelled" | undefined = undefined;
readonly executionFailure: string | undefined = undefined;
readonly partialObservation: PartialProcessCaptureObservation | undefined =
undefined;
readonly partialObservation: AnalysisPartialObservation | undefined;
readonly cleanupReport: ProcessCaptureCleanupReport | undefined = undefined;
}
import type {
PartialProcessCaptureObservation,
ProcessCaptureCleanupReport,
} from "./process/processCapture.js";
import type { NativeCallPartialObservation } from "./native/nativeCallPartialObservation.js";
/** Provider-neutral evidence collected before a typed analysis failure. */
export type AnalysisPartialObservation =
| PartialProcessCaptureObservation
| NativeCallPartialObservation;

View File

@ -23,6 +23,34 @@ import { UnknownRegistryError } from "./unknownRegistryError.js";
import { projectAnalysisError } from "./analysisErrorProjection.js";
import { ProviderSelectionError } from "./providerSelectionError.js";
import { analysisErrorProjectionSchema } from "../contracts/errorSchemas.js";
import { nativeCallPartialObservationSchema } from "./native/nativeCallPartialObservation.js";
const nativeCallPartialObservation = nativeCallPartialObservationSchema.parse({
kind: "native-call-observation",
target: {
path: "/tmp/selected-app",
sha256: "a".repeat(64),
architecture: "arm64",
arguments: [],
environment: {},
working_directory: null,
},
process: {
pid: 4242,
stdout: {
text: "partial stdout",
bytes: 13,
truncated: false,
complete: false,
},
stderr: null,
other_stops: [],
},
debugger: { version: null },
events: [],
coverage: { status: "partial", reason: "cancelled" },
limitations: ["Observation ended before the requested window completed."],
});
const retainedOutput = {
stdout: "selected output\u0000",
@ -179,6 +207,50 @@ it("retains the original failure tag and code when cleanup is incomplete", () =>
});
});
it("projects native partial observations through healthy cancellation failures", () => {
const projected = projectAnalysisError(
new AnalysisCancelledError("observe_native_calls", {
partialObservation: nativeCallPartialObservation,
}),
);
expect(projected).toMatchObject({
code: "cancelled",
details: {
operation: "observe_native_calls",
cleanup: "complete",
partial_observation: nativeCallPartialObservation,
},
});
expect(analysisErrorProjectionSchema.safeParse(projected).success).toBe(true);
});
it("retains native partial observations and cleanup uncertainty on provider failures", () => {
const projected = projectAnalysisError(
new ProviderAdapterError("native-lldb", "observe_native_calls", {
partialObservation: nativeCallPartialObservation,
cleanup: {
reason: "target process termination could not be verified",
resources: ["native-target:4242"],
},
}),
);
expect(projected).toMatchObject({
code: "cleanup_incomplete",
details: {
provider_id: "native-lldb",
operation: "observe_native_calls",
cleanup: "incomplete",
cleanup_reason: "target process termination could not be verified",
resources: ["native-target:4242"],
execution_failure: "execution_failure",
partial_observation: nativeCallPartialObservation,
},
});
expect(analysisErrorProjectionSchema.safeParse(projected).success).toBe(true);
});
describe("analysis error projection: provider failures", () => {
it("projects the primary browser failure alongside incomplete cleanup", () => {
const projected = projectAnalysisError(

View File

@ -61,6 +61,11 @@ export const projectAnalysisError = (
...(error.capturedOutput === undefined
? {}
: { captured_output: { ...error.capturedOutput } }),
...(error.partialObservation === undefined
? {}
: {
partial_observation: jsonValueSchema.parse(error.partialObservation),
}),
};
return {
code,
@ -445,13 +450,6 @@ const lifecycleErrorDetails = (
...(error.executionFailure === undefined
? {}
: { execution_failure: error.executionFailure }),
...(error.partialObservation === undefined
? {}
: {
partial_observation: jsonValueSchema.parse(
error.partialObservation,
),
}),
};
if (error._tag === "ProcessCaptureError") {
const details = {
@ -464,13 +462,6 @@ const lifecycleErrorDetails = (
...(error.executionFailure === undefined
? {}
: { execution_failure: error.executionFailure }),
...(error.partialObservation === undefined
? {}
: {
partial_observation: jsonValueSchema.parse(
error.partialObservation,
),
}),
};
return Object.keys(details).length === 0 ? undefined : details;
}

View File

@ -205,6 +205,8 @@ export const nativeCallObservationResultSchema = z.strictObject({
status: z.enum(["complete", "partial"]),
event_limit_reached: z.boolean(),
resource_limit_reached: z.boolean(),
/** Aggregate breakpoint-location metadata exhausted its retained-byte budget. */
breakpoint_locations_truncated: z.boolean().optional(),
/** Requests that matched no code in any image loaded while observing. */
unresolved_breakpoints: z.array(z.number().int().nonnegative()),
}),

View File

@ -0,0 +1,46 @@
import { z } from "zod";
import { nativeCallEventSchema } from "./nativeCallObservation.js";
const capturedOutputSchema = z.strictObject({
text: z.string(),
/** Bytes observed so far; the final stream size may be unknown. */
bytes: z.number().int().nonnegative(),
truncated: z.boolean(),
complete: z.boolean(),
});
/** Native call evidence retained when observation fails before a full result. */
export const nativeCallPartialObservationSchema = z.strictObject({
kind: z.literal("native-call-observation"),
target: z.strictObject({
path: z.string(),
sha256: z.string().regex(/^[0-9a-f]{64}$/u),
architecture: z.string(),
arguments: z.array(z.string()),
environment: z.record(z.string(), z.string()),
working_directory: z.string().nullable(),
}),
process: z.strictObject({
pid: z.number().int().positive().nullable(),
stdout: capturedOutputSchema.nullable(),
stderr: capturedOutputSchema.nullable(),
other_stops: z.array(z.string()),
}),
debugger: z.strictObject({ version: z.string().nullable() }),
events: z.array(nativeCallEventSchema),
coverage: z.strictObject({
status: z.literal("partial"),
reason: z.enum([
"cancelled",
"timeout",
"tracer-failure",
"capture-failure",
"cleanup-failure",
]),
}),
limitations: z.array(z.string()),
});
export type NativeCallPartialObservation = z.infer<
typeof nativeCallPartialObservationSchema
>;

View File

@ -15,6 +15,7 @@ import type {
AnalysisCleanupObservation,
AnalysisError,
} from "../domain/analysisErrorBase.js";
import type { NativeCallPartialObservation } from "../domain/native/nativeCallPartialObservation.js";
import {
nativeCallEventSchema,
nativeCodeLocationSchema,
@ -40,6 +41,12 @@ import {
} from "../process/ProviderDeadline.js";
import { PrivateRuntimeRoot } from "../process/PrivateRuntimeRoot.js";
import { resolveXcrunTool, type ResolvedTool } from "./CommandRunner.js";
import {
readLldbObservationJournal,
readBoundedPrefix,
readPartialCapture,
projectLldbPartialObservation,
} from "./LldbRetainedObservations.js";
const OPERATION = "observe_native_calls";
const PROVIDER = "native-macos";
@ -82,6 +89,7 @@ const tracedSchema = z.strictObject({
stable: z.boolean(),
}),
resource_limit_reached: z.boolean(),
breakpoint_locations_truncated: z.boolean().optional(),
target_output: z.strictObject({
stdout_bytes: z.number().int().nonnegative(),
stderr_bytes: z.number().int().nonnegative(),
@ -193,6 +201,7 @@ export class LldbCallTracer implements NativeCallTracer {
stderr: join(directory, "stderr"),
stdoutCapture: join(directory, "stdout.capture"),
stderrCapture: join(directory, "stderr.capture"),
observations: join(directory, "observations.jsonl"),
};
const { input } = request;
await writeFile(
@ -221,6 +230,7 @@ export class LldbCallTracer implements NativeCallTracer {
result_path: paths.result,
pid_path: paths.pid,
identity_ack_path: paths.identityAck,
observation_path: paths.observations,
}),
{ mode: 0o600 },
);
@ -271,26 +281,56 @@ export class LldbCallTracer implements NativeCallTracer {
: {
reason: cleanupDetails.join("; "),
resources: [
`native-target:${String((await readTargetPid(paths.pid)) ?? "unknown")}`,
...(exited.cleanupFailure === undefined
? []
: ["lldb-process-group"]),
...(terminated
? []
: [
`native-target:${String((await readTargetPid(paths.pid)) ?? "unknown")}`,
]),
],
};
const partialReason =
exited.kind === "cancelled"
? "cancelled"
: exited.kind === "timeout"
? "timeout"
: exited.cleanupFailure !== undefined
? "cleanup-failure"
: "tracer-failure";
const getPartialObservation = (
reason: NativeCallPartialObservation["coverage"]["reason"] = partialReason,
): Promise<NativeCallPartialObservation> =>
partialNativeObservation({
request,
pidPath: paths.pid,
stdoutPath: paths.stdoutCapture,
stderrPath: paths.stderrCapture,
observationPath: paths.observations,
version: output?.status === "traced" ? output.version : null,
reason,
});
if (exited.kind === "cancelled")
return err(
new AnalysisCancelledError(
OPERATION,
lifecycleCleanup === undefined
? undefined
: { cleanup: lifecycleCleanup },
),
new AnalysisCancelledError(OPERATION, {
...(lifecycleCleanup === undefined
? {}
: { cleanup: lifecycleCleanup }),
partialObservation: await getPartialObservation(),
}),
);
if (exited.kind === "timeout")
return err(
new AnalysisTimeoutError(
OPERATION,
input.duration_ms + DEADLINE_GRACE_MS,
lifecycleCleanup === undefined
? undefined
: { cleanup: lifecycleCleanup },
{
...(lifecycleCleanup === undefined
? {}
: { cleanup: lifecycleCleanup }),
partialObservation: await getPartialObservation(),
},
),
);
if (exited.cleanupFailure !== undefined)
@ -307,6 +347,7 @@ export class LldbCallTracer implements NativeCallTracer {
? {}
: { target_cleanup: lifecycleCleanup }),
},
partialObservation: await getPartialObservation(),
}),
);
if (output === undefined)
@ -320,13 +361,16 @@ export class LldbCallTracer implements NativeCallTracer {
? {}
: { target_cleanup: lifecycleCleanup }),
},
partialObservation: await getPartialObservation(),
...(lifecycleCleanup === undefined
? {}
: { cleanup: lifecycleCleanup }),
}),
);
if (output.status !== "traced") {
return err(tracerFailure(output, lifecycleCleanup));
return err(
tracerFailure(output, lifecycleCleanup, await getPartialObservation()),
);
}
const stdout = await capturedOutput(
paths.stdoutCapture,
@ -340,9 +384,11 @@ export class LldbCallTracer implements NativeCallTracer {
output.target_output.stderr_truncated,
output.target_output.stderr_complete,
);
if (!stdout.ok || !stderr.ok)
if (!stdout.ok || !stderr.ok) {
const partial = await getPartialObservation("capture-failure");
return err(
new ProviderAdapterError(PROVIDER, OPERATION, {
partialObservation: partial,
diagnostics: {
reason: "LLDB target output capture could not be read",
...(stdout.ok ? {} : { stdout_capture_error: stdout.error }),
@ -351,8 +397,12 @@ export class LldbCallTracer implements NativeCallTracer {
? {}
: { target_cleanup: lifecycleCleanup }),
},
...(lifecycleCleanup === undefined
? {}
: { cleanup: lifecycleCleanup }),
}),
);
}
return ok({
debugger: { ...lldb, version: output.version },
run: output,
@ -378,8 +428,12 @@ const tracerFailure = (
readonly error: string | null;
},
cleanup?: AnalysisCleanupObservation,
partialObservation?: NativeCallPartialObservation,
): AnalysisError => {
const options = cleanup === undefined ? undefined : { cleanup };
const options = {
...(cleanup === undefined ? {} : { cleanup }),
...(partialObservation === undefined ? {} : { partialObservation }),
};
const message = output.error ?? "no LLDB error text";
if (output.status === "target-integrity-error")
return new EvidenceIntegrityError(
@ -422,6 +476,44 @@ const tracerFailure = (
);
};
const partialNativeObservation = async (options: {
readonly request: Parameters<NativeCallTracer["trace"]>[0];
readonly pidPath: string;
readonly stdoutPath: string;
readonly stderrPath: string;
readonly observationPath: string;
readonly version: string | null;
readonly reason: NativeCallPartialObservation["coverage"]["reason"];
}): Promise<NativeCallPartialObservation> => {
const [journal, pid, stdout, stderr] = await Promise.all([
readLldbObservationJournal(options.observationPath),
readTargetPid(options.pidPath),
readPartialCapture(options.stdoutPath),
readPartialCapture(options.stderrPath),
]);
return projectLldbPartialObservation({
target: {
path: options.request.executable,
sha256: options.request.expectedSha256,
architecture: options.request.architecture,
arguments: options.request.input.arguments,
environment: options.request.input.environment,
working_directory: options.request.input.working_directory ?? null,
},
pid,
stdout: stdout?.capture ?? null,
stderr: stderr?.capture ?? null,
version: options.version,
journal: journal.parsed,
journalLimitations: journal.limitations,
limitations: [
...(stdout?.limitations ?? []),
...(stderr?.limitations ?? []),
],
reason: options.reason,
});
};
const readTracerOutput = async (
path: string,
): Promise<z.infer<typeof tracerOutputSchema> | undefined> => {
@ -447,8 +539,12 @@ const capturedOutput = async (
try {
handle = await open(path, "r");
const { size } = await handle.stat();
const buffer = Buffer.alloc(Math.min(size, MAX_OUTPUT_BYTES));
const { bytesRead } = await handle.read(buffer, 0, buffer.length, 0);
const captureHandle = handle;
const { buffer, bytesRead } = await readBoundedPrefix(
Math.min(size, MAX_OUTPUT_BYTES),
async (target, offset, length, position) =>
(await captureHandle.read(target, offset, length, position)).bytesRead,
);
const captured = {
text: buffer.subarray(0, bytesRead).toString("utf8"),
bytes: observedBytes,

View File

@ -0,0 +1,237 @@
import { open } from "node:fs/promises";
import { nativeCallEventSchema } from "../domain/native/nativeCallObservation.js";
import type { NativeCallPartialObservation } from "../domain/native/nativeCallPartialObservation.js";
/** Maximum journal size enforced by the LLDB bridge. */
export const LLDB_OBSERVATION_JOURNAL_MAX_BYTES = 8 * 1024 * 1024;
const MAX_CAPTURE_BYTES = 1024 * 1024;
export interface ParsedLldbJournal {
readonly events: ReturnType<typeof nativeCallEventSchema.parse>[];
readonly otherStops: string[];
readonly limitations: string[];
}
export interface PartialCaptureRead {
readonly capture: NonNullable<
NativeCallPartialObservation["process"]["stdout"]
>;
readonly limitations: readonly string[];
}
/** Join safely retained journal and process evidence into its failure contract. */
export const projectLldbPartialObservation = (input: {
readonly target: NativeCallPartialObservation["target"];
readonly pid: number | undefined;
readonly stdout: NativeCallPartialObservation["process"]["stdout"];
readonly stderr: NativeCallPartialObservation["process"]["stderr"];
readonly version: string | null;
readonly journal: ParsedLldbJournal;
readonly journalLimitations?: readonly string[];
readonly limitations?: readonly string[];
readonly reason: NativeCallPartialObservation["coverage"]["reason"];
}): NativeCallPartialObservation => ({
kind: "native-call-observation",
target: input.target,
process: {
pid: input.pid ?? null,
stdout: input.stdout,
stderr: input.stderr,
other_stops: input.journal.otherStops,
},
debugger: { version: input.version },
events: input.journal.events,
coverage: { status: "partial", reason: input.reason },
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.",
"Captured output byte counts are observed lower bounds because LLDB did not report final stream counters.",
...input.journal.limitations,
...(input.journalLimitations ?? []),
...(input.limitations ?? []),
...(input.pid === undefined ? ["The target PID was unavailable."] : []),
...(input.stdout === null
? ["The target stdout capture was unavailable."]
: []),
...(input.stderr === null
? ["The target stderr capture was unavailable."]
: []),
],
});
/** Parse completed JSONL records while retaining valid records around bad ones. */
export const parseLldbObservationJournal = (
bytes: Uint8Array,
): ParsedLldbJournal => {
const completeText = Buffer.from(bytes).toString("utf8");
const newline = completeText.lastIndexOf("\n");
const hasTornTail = newline !== completeText.length - 1;
const events: ParsedLldbJournal["events"] = [];
const otherStops: string[] = [];
const limitations: string[] = [];
let malformedCount = 0;
let firstMalformedRow: number | undefined;
let start = 0;
let rowNumber = 0;
while (start <= newline && newline >= 0) {
const end = completeText.indexOf("\n", start);
if (end < 0 || end > newline) break;
const line = completeText.slice(start, end);
start = end + 1;
if (line.length === 0) continue;
rowNumber += 1;
const markMalformed = () => {
malformedCount += 1;
firstMalformedRow ??= rowNumber;
};
let row: unknown;
try {
row = JSON.parse(line);
} catch {
markMalformed();
continue;
}
if (
typeof row !== "object" ||
row === null ||
!("kind" in row) ||
typeof row.kind !== "string"
) {
markMalformed();
continue;
}
if (row.kind === "event" && "event" in row) {
const event = nativeCallEventSchema.safeParse(row.event);
if (event.success) events.push(event.data);
else markMalformed();
} else if (
row.kind === "other-stop" &&
"reason" in row &&
typeof row.reason === "string"
) {
otherStops.push(row.reason);
} else {
markMalformed();
}
}
if (malformedCount > 0)
limitations.push(
`${malformedCount} observation journal row(s) were malformed or invalid and ignored (first at row ${String(firstMalformedRow)}).`,
);
if (hasTornTail)
limitations.push(
"The final observation journal record was incomplete and was ignored.",
);
return { events, otherStops, limitations };
};
/** Keep reading a bounded prefix across short reads until EOF or the bound. */
export const readBoundedPrefix = async (
requestedBytes: number,
readChunk: (
buffer: Buffer,
offset: number,
length: number,
position: number,
) => Promise<number>,
): Promise<{ readonly buffer: Buffer; readonly bytesRead: number }> => {
const buffer = Buffer.alloc(requestedBytes);
let bytesRead = 0;
while (bytesRead < requestedBytes) {
const count = await readChunk(
buffer,
bytesRead,
requestedBytes - bytesRead,
bytesRead,
);
if (count === 0) break;
if (count > requestedBytes - bytesRead)
throw new Error("The bounded read returned more bytes than requested");
bytesRead += count;
}
return { buffer, bytesRead };
};
const readPrefix = (
handle: Awaited<ReturnType<typeof open>>,
requestedBytes: number,
) =>
readBoundedPrefix(
requestedBytes,
async (buffer, offset, length, position) =>
(await handle.read(buffer, offset, length, position)).bytesRead,
);
/** Read at most the bridge's journal bound, checking size before allocation. */
export const readLldbObservationJournal = async (
path: string,
): Promise<{
readonly parsed: ParsedLldbJournal;
readonly limitations: string[];
}> => {
let handle: Awaited<ReturnType<typeof open>> | undefined;
try {
handle = await open(path, "r");
const { size } = await handle.stat();
const length = Math.min(size, LLDB_OBSERVATION_JOURNAL_MAX_BYTES);
const { buffer, bytesRead } = await readPrefix(handle, length);
const parsed = parseLldbObservationJournal(buffer.subarray(0, bytesRead));
const limitations = [
...(size > LLDB_OBSERVATION_JOURNAL_MAX_BYTES
? [
"The observation journal exceeded 8 MiB; only its bounded prefix was read.",
]
: []),
...(bytesRead < length
? [
"The observation journal became shorter while being read; the available prefix was parsed.",
]
: []),
];
return {
parsed,
limitations,
};
} catch (cause: unknown) {
return {
parsed: { events: [], otherStops: [], limitations: [] },
limitations: [
`The observation journal could not be read: ${cause instanceof Error ? cause.message : String(cause)}`,
],
};
} finally {
await handle?.close().catch(() => undefined);
}
};
/** Retain an available output prefix without claiming an unreported total. */
export const readPartialCapture = async (
path: string,
): Promise<PartialCaptureRead | null> => {
let handle: Awaited<ReturnType<typeof open>> | undefined;
try {
handle = await open(path, "r");
const { size } = await handle.stat();
const length = Math.min(size, MAX_CAPTURE_BYTES);
const { buffer, bytesRead } = await readPrefix(handle, length);
return {
capture: {
text: buffer.subarray(0, bytesRead).toString("utf8"),
bytes: Math.max(size, bytesRead),
truncated: true,
complete: false,
},
limitations:
bytesRead < length
? [
"A target output capture became shorter while being read; its retained prefix may be incomplete.",
]
: [],
};
} catch {
return null;
} finally {
await handle?.close().catch(() => undefined);
}
};

View File

@ -105,9 +105,11 @@ export const projectNativeCalls = (
.map(({ index }) => index);
const eventLimit = run.outcome === "event-limit";
const resourceLimit = run.outcome === "resource-limit";
const breakpointLocationLimit = run.breakpoint_locations_truncated === true;
const partial =
eventLimit ||
resourceLimit ||
breakpointLocationLimit ||
run.outcome === "duration-elapsed" ||
run.outcome === "stop-limit" ||
unresolved.length > 0;
@ -155,6 +157,9 @@ export const projectNativeCalls = (
status: partial ? "partial" : "complete",
event_limit_reached: eventLimit,
resource_limit_reached: resourceLimit,
...(breakpointLocationLimit
? { breakpoint_locations_truncated: true }
: {}),
unresolved_breakpoints: unresolved,
},
limitations: [
@ -197,6 +202,11 @@ export const projectNativeCalls = (
"Breakpoints with many matching locations list only the first 64; location_count gives the total.",
]
: []),
...(breakpointLocationLimit
? [
"Resolved breakpoint locations exceeded the aggregate 8 MiB metadata budget; omitted locations are counted in location_count. This metadata limit does not change the process outcome.",
]
: []),
...(trace.stdout.truncated || trace.stderr.truncated
? ["Target output beyond 1 MiB per stream is counted but not kept."]
: []),

View File

@ -1,3 +1,4 @@
import { createHash } from "node:crypto";
import { writeFile } from "node:fs/promises";
import { join } from "node:path";
@ -6,7 +7,10 @@ import { expect, it } from "vitest";
import { z } from "zod";
import { nativeCallObservationResultSchema } from "../../../src/domain/native/nativeCallObservation.js";
import { ok } from "../../../src/domain/result.js";
import { AnalysisCancelledError } from "../../../src/domain/analysisErrorCore.js";
import { nativeCallPartialObservationSchema } from "../../../src/domain/native/nativeCallPartialObservation.js";
import { err, ok } from "../../../src/domain/result.js";
import { ProviderAdapterError } from "../../../src/domain/providerAdapterError.js";
import type { NativeCallTracer } from "../../../src/native/LldbCallTracer.js";
import { NativeMacOSProvider } from "../../../src/native/NativeMacOSProvider.js";
import { createServer } from "../../../src/server/createServer.js";
@ -130,3 +134,169 @@ it("routes observe_native_calls through MCP with schema-checked input and output
await server.close();
}
});
it.each([
["cancelled", "cancelled", undefined],
[
"tracer failure with cleanup uncertainty",
"cleanup_incomplete",
{
reason: "target process termination could not be verified",
resources: ["native-target:4242"],
},
],
] as const)(
"preserves typed native partial observations over MCP after %s",
async (_label, expectedCode, cleanup) => {
const directory = await createTestTempDirectory(
"rea-native-calls-partial-mcp-",
);
const path = join(directory, "Tool");
const selectedBytes = machoHeader();
const sha256 = createHash("sha256").update(selectedBytes).digest("hex");
await writeFile(path, selectedBytes);
const launchArguments = ["--fixture", "selected argument"];
const launchEnvironment = { REA_NATIVE_FIXTURE: "selected value" };
const workingDirectory = directory;
const partialObservation = nativeCallPartialObservationSchema.parse({
kind: "native-call-observation",
target: {
path,
sha256,
architecture: "arm64",
arguments: launchArguments,
environment: launchEnvironment,
working_directory: workingDirectory,
},
process: {
pid: 4242,
stdout: {
text: "partial target output",
bytes: 21,
truncated: true,
complete: false,
},
stderr: {
text: "debugger diagnostic",
bytes: 19,
truncated: false,
complete: false,
},
other_stops: ["signal SIGSTOP"],
},
debugger: { version: "lldb-fixture" },
events: [
{
sequence: 0,
elapsed_ms: 4,
thread_id: 3,
breakpoint_index: 0,
load_address: "0x1000",
file_address: "0x1000",
module: "Tool",
module_path: path,
symbol: "open",
receiver_class: null,
selector: null,
registers: [{ name: "x0", value: "0x2" }],
backtrace: [],
},
],
coverage: {
status: "partial",
reason: cleanup === undefined ? "cancelled" : "cleanup-failure",
},
limitations: ["Observation ended before the requested window completed."],
});
const tracer: NativeCallTracer = {
trace: () =>
Promise.resolve(
err(
cleanup === undefined
? new AnalysisCancelledError("observe_native_calls", {
partialObservation,
})
: new ProviderAdapterError(
"native-macos",
"observe_native_calls",
{
cleanup,
partialObservation,
},
),
),
),
};
const session = createTestBinarySession(
new NativeMacOSProvider(new NativeFixtureRunner(), "darwin", tracer),
);
const server = createServer(session, session);
const client = new Client({
name: "native-calls-partial-mcp-test",
version: "1",
});
const [clientTransport, serverTransport] =
InMemoryTransport.createLinkedPair();
try {
await server.connect(serverTransport);
await client.connect(clientTransport);
const opened = await client.callTool({
name: "open_binary",
arguments: { path },
});
expect(opened.isError, JSON.stringify(opened)).not.toBe(true);
const called = await client.callTool({
name: "observe_native_calls",
arguments: {
breakpoints: [{ kind: "function", name: "open" }],
arguments: launchArguments,
environment: launchEnvironment,
working_directory: workingDirectory,
duration_ms: 1000,
},
});
expect(called.isError).toBe(true);
expect(called.structuredContent).toMatchObject({
error: {
code: expectedCode,
details: {
...(cleanup === undefined
? { cleanup: "complete" }
: {
execution_failure: "execution_failure",
cleanup: "incomplete",
cleanup_reason: cleanup.reason,
}),
partial_observation: {
kind: "native-call-observation",
target: {
path,
sha256,
architecture: "arm64",
arguments: launchArguments,
environment: launchEnvironment,
working_directory: workingDirectory,
},
debugger: { version: "lldb-fixture" },
process: {
pid: 4242,
stdout: {
text: "partial target output",
bytes: 21,
truncated: true,
complete: false,
},
},
events: [{ sequence: 0, symbol: "open" }],
},
},
},
});
} finally {
await client.close();
await server.close();
}
},
);

View File

@ -0,0 +1,133 @@
import { describe, expect, it } from "vitest";
import { nativeCallPartialObservationSchema } from "../../../../src/domain/native/nativeCallPartialObservation.js";
import {
LLDB_OBSERVATION_JOURNAL_MAX_BYTES,
parseLldbObservationJournal,
projectLldbPartialObservation,
readBoundedPrefix,
} from "../../../../src/native/LldbRetainedObservations.js";
const event = {
sequence: 0,
elapsed_ms: 1,
thread_id: 2,
breakpoint_index: 0,
load_address: "0x1000",
file_address: "0x1000",
module: "fixture",
module_path: "/tmp/fixture",
symbol: "main",
receiver_class: null,
selector: null,
registers: [],
backtrace: [],
};
const observation = (
reason: "cancelled" | "tracer-failure" | "cleanup-failure",
journalText: string,
) => {
const parsed = parseLldbObservationJournal(Buffer.from(journalText));
return nativeCallPartialObservationSchema.parse(
projectLldbPartialObservation({
target: {
path: "/tmp/fixture",
sha256: "a".repeat(64),
architecture: "arm64",
arguments: ["--selected"],
environment: { SELECTED_VALUE: "retained" },
working_directory: "/tmp",
},
pid: 42,
stdout: { text: "prefix", bytes: 6, truncated: true, complete: false },
stderr: null,
version: null,
journal: parsed,
reason,
}),
);
};
describe("LLDB retained observations", () => {
it.each([
["cancellation", "cancelled"],
["bridge failure", "tracer-failure"],
["cleanup failure", "cleanup-failure"],
] as const)("retains admitted events after %s", (_case, reason) => {
const journal = `${JSON.stringify({ kind: "event", event })}\n`;
const partial = observation(reason, journal);
expect(partial.events).toEqual([event]);
expect(partial.coverage).toEqual({ status: "partial", reason });
expect(partial.process.stdout).toMatchObject({
text: "prefix",
bytes: 6,
complete: false,
truncated: true,
});
expect(partial.process.stderr).toBeNull();
expect(partial.limitations).toContain(
"The target stderr capture was unavailable.",
);
});
it("keeps valid rows around malformed data and ignores a torn final row", () => {
const journal = [
JSON.stringify({ kind: "event", event }),
"{bad-json}",
JSON.stringify({ kind: "other-stop", reason: "signal SIGTERM" }),
'{"kind":"event","event":',
].join("\n");
const partial = observation("tracer-failure", journal);
expect(partial.events).toEqual([event]);
expect(partial.process.other_stops).toEqual(["signal SIGTERM"]);
expect(partial.limitations).toContain(
"1 observation journal row(s) were malformed or invalid and ignored (first at row 2).",
);
expect(partial.limitations).toContain(
"The final observation journal record was incomplete and was ignored.",
);
});
it("bounds malformed-row diagnostics for an 8 MiB journal", () => {
const valid = JSON.stringify({ kind: "event", event });
const malformed = '{"kind":"unknown"}\n';
const malformedRows = Math.floor(
(LLDB_OBSERVATION_JOURNAL_MAX_BYTES - valid.length * 2 - 2) /
malformed.length,
);
const journal = `${valid}\n${malformed.repeat(malformedRows)}${valid}\n`;
expect(Buffer.byteLength(journal)).toBeLessThanOrEqual(
LLDB_OBSERVATION_JOURNAL_MAX_BYTES,
);
const partial = observation("tracer-failure", journal);
expect(partial.events).toEqual([event, event]);
expect(partial.limitations.length).toBeLessThanOrEqual(6);
expect(
partial.limitations.find((limitation) =>
limitation.includes("malformed"),
),
).toMatch(
/^\d+ observation journal row\(s\) were malformed or invalid and ignored \(first at row 2\)\.$/u,
);
});
it("continues a bounded prefix read after short reads", async () => {
const source = Buffer.from("retained prefix");
let position = 0;
const result = await readBoundedPrefix(
source.length,
async (buffer, offset, length, requestedPosition) => {
expect(requestedPosition).toBe(position);
const count = Math.min(2, length, source.length - position);
source.copy(buffer, offset, position, position + count);
position += count;
return count;
},
);
expect(result.bytesRead).toBe(source.length);
expect(result.buffer.toString("utf8", 0, result.bytesRead)).toBe(
"retained prefix",
);
});
});

View File

@ -212,6 +212,53 @@ describe("observe_native_calls projection: projects completed observations", ()
});
describe("observe_native_calls projection: reports partial observation coverage", () => {
it("retains normal exit and location counts when breakpoint metadata is bounded", async () => {
const observed = await observe(
new FixtureTracer(
ok({
debugger: {
path: "/usr/bin/lldb",
sha256: "a".repeat(64),
version: null,
},
run: run({
breakpoint_locations_truncated: true,
breakpoints: [
{ index: 0, location_count: 64, locations: [LOCATION] },
{ index: 1, location_count: 64, locations: [] },
],
}),
stdout: { text: "", bytes: 0, truncated: false, complete: true },
stderr: { text: "", bytes: 0, truncated: false, complete: true },
terminated: true,
}),
),
{ breakpoints: BREAKPOINTS },
);
if (!observed.ok) throw observed.error;
const result = nativeCallObservationResultSchema.parse(
observed.value.result,
);
expect(result.process).toMatchObject({
outcome: "exited",
exit_status: 0,
terminated: true,
});
expect(
result.breakpoints.map(({ location_count }) => location_count),
).toEqual([64, 64]);
expect(result.coverage).toMatchObject({
status: "partial",
breakpoint_locations_truncated: true,
resource_limit_reached: false,
unresolved_breakpoints: [],
});
expect(result.limitations).toContain(
"Resolved breakpoint locations exceeded the aggregate 8 MiB metadata budget; omitted locations are counted in location_count. This metadata limit does not change the process outcome.",
);
expect(result.limitations.join(" ")).not.toContain("process was killed");
});
it("marks event limits, unresolved breakpoints and unconfirmed exits", async () => {
const tracer = new FixtureTracer(
ok({