experiment(re): compare fact granularity through real MCP

Add a non-shipping MCP facade over the existing admitted analysis port, real-provider comparison runner and offline token accounting. Keep the full catalog, source Evidence and per-input failures. Separate verification runs from equivalent-work timings. Kuna zlib comparisons pass; broader comparisons remain in progress.

Co-authored-by: gensokyo-dev <342580529+gensokyo-dev@users.noreply.github.com>
Co-authored-by: Kaoru0822 <208281322+kaoru0822-kitauji@users.noreply.github.com>
This commit is contained in:
N0zoM1z0
2026-10-15 16:35:23 +00:00
parent ce4fe65170
commit fa8d6a7a23
4 changed files with 673 additions and 0 deletions
+359
View File
@@ -0,0 +1,359 @@
import assert from "node:assert/strict";
import { mkdir, writeFile } from "node:fs/promises";
import { dirname, resolve } from "node:path";
import { createHash } from "node:crypto";
import { readFile } from "node:fs/promises";
import { Client } from "@modelcontextprotocol/client";
import { StdioClientTransport } from "@modelcontextprotocol/client/stdio";
import Ajv2020 from "ajv/dist/2020.js";
import { requireMcpOperationResult } from "../../scripts/lib/mcp-verifier-results.mjs";
import {
createVerifierRun,
completeVerifierRun,
assertVerifierCleanupComplete,
} from "../../scripts/lib/verifier-run.mjs";
const [provider, variant, binary, pattern, prefix, verification] =
process.argv.slice(2);
assert.ok(["ghidra", "kuna"].includes(provider));
assert.ok(["baseline", "fine", "batch", "workflow"].includes(variant));
assert.ok(binary && pattern && prefix);
await mkdir(dirname(prefix), { recursive: true });
const verifier = createVerifierRun();
const client = new Client({ name: "rea-atomic-experiment", version: "0.1.0" });
const transport = new StdioClientTransport({
command: process.execPath,
args: [
resolve(
variant === "baseline"
? "dist/main.js"
: "experiments/atomic-re-tools/server.mjs",
),
],
env: {
...process.env,
REA_LOG_LEVEL: "error",
REA_MCP_OUTPUT_SCHEMA_PROFILE:
process.env.REA_MCP_OUTPUT_SCHEMA_PROFILE ?? "full",
REA_KUNA_CACHE_DIR: resolve(`${prefix}.kuna-cache`),
REA_KUNA_JOBS: "1",
GHIDRA_HEADLESS_MAXMEM: "1G",
REA_GHIDRA_STARTUP_TIMEOUT_MS: "120000",
},
stderr: "pipe",
});
let stderr = "";
transport.stderr?.on("data", (part) => {
stderr = (stderr + part).slice(-32_768);
});
const wire = [],
calls = [],
checks = [],
failures = [];
let phase = "lifecycle";
const call = async (name, arguments_ = {}, expectError = false) => {
const started = performance.now();
const response = await client.callTool(
{ name, arguments: arguments_ },
{ timeout: 180000 },
);
wire.push({ request: { name, arguments: arguments_ }, response, phase });
calls.push({ name, phase, elapsed_ms: performance.now() - started });
if (expectError) {
assert.equal(response.isError, true);
return JSON.parse(
response.content.find((part) => part.type === "text").text,
);
}
assert.notEqual(
response.isError,
true,
JSON.stringify(response).slice(0, 2000),
);
return name.startsWith("experiment_")
? JSON.parse(response.content.find((part) => part.type === "text").text)
: requireMcpOperationResult(response, name);
};
const contextInput = (context) => ({ ref: context.ref, lease: context.lease });
const stringLimit = 3,
referenceLimit = 4;
const query = { pattern, mode: "literal", case_sensitive: true };
const identity = (row) => ({
kind: "string",
address: row.address,
encoding: row.string?.encoding ?? null,
});
const readBatch = async (name, common, { key, items, individually }) => {
if (!individually) return call(name, { ...common, [key]: items });
const results = [];
for (const item of items)
results.push(...(await call(name, { ...common, [key]: [item] })).results);
return { results };
};
const readExperimentalTask = async (supportsOwners) => {
const input = { query, limit: stringLimit, fields: ["value"] };
if (variant === "workflow") {
const traced = await call("experiment_trace_string_uses", {
...input,
reference_limit: referenceLimit,
});
return {
found: traced.strings,
refs: traced.references,
owners: traced.functions,
};
}
const found = await call("experiment_find_strings", input);
const context = contextInput(found.context);
const refs = await readBatch(
"experiment_incoming_refs",
{ context, limit: referenceLimit },
{
key: "targets",
items: found.items.map((row) => row.id),
individually: variant === "fine",
},
);
const locations = [
...new Map(
refs.results
.flatMap((row) => row.items ?? [])
.map((row) => [row.from.address, row.from]),
).values(),
];
const owners = await readBatch(
"experiment_resolve_functions",
{ context },
{
key: "locations",
items: locations,
individually: variant === "fine" && supportsOwners,
},
);
return { found, refs, owners };
};
let catalog,
firstSourceId,
firstContext,
outcomes = [];
try {
await client.connect(transport);
catalog = await client.listTools();
const ajv = new Ajv2020({ strict: false, validateFormats: false });
for (const tool of catalog.tools.filter((tool) =>
tool.name.startsWith("experiment_"),
)) {
assert.equal(
ajv.validateSchema(tool.inputSchema),
true,
JSON.stringify(ajv.errors),
);
ajv.compile(tool.inputSchema);
}
await call("open_binary", { path: resolve(binary), provider_id: provider });
const status = await call("binary_session");
const supportsOwners = status.capabilities.some(
(row) => row.operation === "resolve_containing_procedure" && row.available,
);
for (phase of ["cold", "warm"]) {
const started = performance.now();
let strings, references, functions;
if (variant === "baseline") {
const all = await call("search_strings", query);
strings = all
.slice(0, stringLimit)
.map((row) => ({ id: identity(row), value: row.value }));
references = [];
for (const string of strings)
references.push({
target: string.id,
items: (await call("xrefs", { address: string.id.address })).slice(
0,
referenceLimit,
),
});
const locations = [...new Set(references.flatMap((row) => row.items))];
functions = [];
for (const address of locations)
functions.push(
supportsOwners
? {
address,
resolution: await call("resolve_containing_procedure", {
address,
}),
}
: { address, status: "unknown" },
);
} else {
const { found, refs, owners } =
await readExperimentalTask(supportsOwners);
assert.equal(found.status, "ok");
assert.ok(refs.results.every((row) => row.status === "ok"));
firstSourceId ??= found.evidence_id;
firstContext ??= found.context;
strings = found.items.map((row) => ({ id: row.id, value: row.value }));
references = refs.results.map((row) => ({
target: row.target,
items: row.items.map((edge) => edge.from.address),
}));
functions = owners.results.map((row) =>
row.status === "ok"
? { address: row.location.address, resolution: row.resolution }
: { address: row.location.address, status: "unknown" },
);
}
outcomes.push({
phase,
elapsed_ms: performance.now() - started,
strings,
references,
functions,
});
}
assert.deepEqual(outcomes[0].strings, outcomes[1].strings);
assert.deepEqual(outcomes[0].references, outcomes[1].references);
assert.deepEqual(outcomes[0].functions, outcomes[1].functions);
assert.ok(outcomes[0].strings.length > 0);
phase = "verification";
if (variant !== "baseline" && verification === "--verify") {
const raw = await call("experiment_read_evidence", {
evidence_id: firstSourceId,
});
assert.equal(raw.evidence.operation, "search_strings");
assert.equal(
raw.evidence.analysis_profile.digest,
firstContext.analysis_profile_digest,
);
assert.equal(
raw.evidence.subject.digest.sha256,
createHash("sha256")
.update(await readFile(binary))
.digest("hex"),
);
assert.deepEqual(
raw.evidence.normalized_result
.slice(0, stringLimit)
.map((row) => ({ id: identity(row), value: row.value })),
outcomes[0].strings,
);
const duplicate = await call("experiment_incoming_refs", {
context: contextInput(firstContext),
targets: [outcomes[0].strings[0].id, outcomes[0].strings[0].id],
limit: 1,
});
assert.deepEqual(duplicate.results[0], duplicate.results[1]);
const nextPage = await call("experiment_find_strings", {
query,
offset: stringLimit,
limit: stringLimit,
});
assert.deepEqual(
nextPage.items.map((row) => row.id),
raw.evidence.normalized_result
.slice(stringLimit, stringLimit * 2)
.map(identity),
);
const negative = await call("experiment_find_strings", {
query: { pattern: "\u0000REA-atomic-negative\u0000" },
});
assert.equal(negative.items.length, 0);
assert.equal(negative.coverage.status, "unknown");
const stale = await call(
"experiment_incoming_refs",
{
context: {
ref: "different-artifact-or-profile",
lease: firstContext.lease,
},
targets: [outcomes[0].strings[0].id],
},
true,
);
assert.equal(stale.status, "error");
await call("close_binary");
assert.equal(
(
await call(
"experiment_read_evidence",
{ evidence_id: firstSourceId },
true,
)
).error.code,
"not_found",
);
await call("open_binary", { path: resolve(binary), provider_id: provider });
const expired = await call(
"experiment_incoming_refs",
{
context: contextInput(firstContext),
targets: [outcomes[0].strings[0].id],
},
true,
);
assert.equal(expired.status, "error");
checks.push(
"retained-evidence",
"paging",
"duplicate-targets",
"negative-unknown",
"foreign-context",
"expired-lease",
);
}
} catch (error) {
failures.push(String(error.stack ?? error));
}
phase = "lifecycle";
try {
await call("close_binary");
} catch (error) {
failures.push(String(error));
}
await client.close().catch((error) => failures.push(String(error)));
await transport.close().catch((error) => failures.push(String(error)));
const lifecycle = await completeVerifierRun(verifier);
try {
assertVerifierCleanupComplete(lifecycle.process_lineage);
} catch (error) {
failures.push(String(error));
}
const report = {
provider,
variant,
binary: resolve(binary),
pattern,
node_version: process.versions.node,
target_sha256: createHash("sha256")
.update(await readFile(binary))
.digest("hex"),
string_limit: stringLimit,
reference_limit: referenceLimit,
outcomes,
calls,
checks,
failures,
verification_only: verification === "--verify",
lifecycle,
};
await writeFile(`${prefix}.wire.json`, JSON.stringify(wire));
await writeFile(`${prefix}.catalog.json`, JSON.stringify(catalog));
await writeFile(`${prefix}.report.json`, JSON.stringify(report, null, 2));
await writeFile(`${prefix}.stderr.txt`, stderr);
console.log(
JSON.stringify({
provider,
variant,
calls: calls.length,
outcomes: outcomes.map(({ strings, references, functions, ...rest }) => ({
...rest,
strings: strings.length,
references: references.flatMap((row) => row.items).length,
functions: functions.length,
})),
failures,
}),
);
assert.deepEqual(failures, []);
+40
View File
@@ -0,0 +1,40 @@
"""Run comparable real-provider tasks serially using REA's existing scope owner."""
import argparse
import json
from pathlib import Path
import subprocess
import sys
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--provider", choices=["kuna", "ghidra"], required=True)
parser.add_argument("--binary", required=True)
parser.add_argument("--pattern", required=True)
parser.add_argument("--output", required=True)
parser.add_argument("--variants", nargs="+", choices=["baseline", "fine", "batch", "workflow"],
default=["baseline", "fine", "batch", "workflow"])
options = parser.parse_args()
root = Path(options.output).resolve()
root.mkdir(parents=True, exist_ok=True)
completed = []
for variant in options.variants:
prefix = root / (options.provider + "-" + variant)
command = [sys.executable, "experiments/analysis-efficiency/resource-run.py", "--observe-payload-exit", "--",
"node", "experiments/atomic-re-tools/benchmark.mjs", options.provider, variant,
options.binary, options.pattern, str(prefix)]
with Path(str(prefix) + ".log").open("w") as output:
result = subprocess.run(command, stdout=output, stderr=subprocess.STDOUT, check=False)
if result.returncode:
raise SystemExit(f"{variant} failed ({result.returncode}); retained {prefix}.log")
report = json.loads(Path(str(prefix) + ".report.json").read_text())
if report["failures"] or report["verification_only"]:
raise SystemExit("Not a successful comparable performance sample: " + str(prefix))
completed.append(str(prefix))
(root / (options.provider + "-completed.json")).write_text(json.dumps(completed, indent=2) + "\n")
print(json.dumps({"completed": variant, "provider": options.provider,
"query_ms": [round(row["elapsed_ms"], 2) for row in report["outcomes"]]}), flush=True)
if __name__ == "__main__":
main()
+200
View File
@@ -0,0 +1,200 @@
// Experimental tools are added to the complete production catalog, never shipped.
import { serveStdio } from "@modelcontextprotocol/server/stdio";
import { z } from "zod";
import { run } from "../../dist/main.js";
import { createServer } from "../../dist/server/createServer.js";
import { createEvidence } from "../../dist/domain/evidence.js";
import { projectAnalysisError } from "../../dist/domain/analysisErrorProjection.js";
import { toolContract } from "../../dist/contracts/toolContracts.js";
import { expandProviderCapabilities } from "../../dist/contracts/toolOutputSchemaPrimitives.js";
import { FactQueries, contextFor } from "./facts.mjs";
const contextSchema = z.strictObject({ ref: z.string(), lease: z.string() });
const location = z.strictObject({
kind: z.literal("location"),
address: z.string().min(1),
});
const stringId = z.strictObject({
kind: z.literal("string"),
address: z.string().min(1),
encoding: z.string().nullable(),
});
const paging = {
offset: z.number().int().nonnegative().default(0),
limit: z.number().int().positive().default(20),
};
const stringsInput = z.strictObject({
query: toolContract("search_strings").inputSchema,
...paging,
fields: z.array(z.enum(["value", "string"])).default(["value"]),
});
const contracts = [
[
"experiment_find_strings",
"Find string facts with scoped identities, selected fields, paging and retained source Evidence. Experimental; provider discovery coverage remains unknown.",
stringsInput,
"findStrings",
],
[
"experiment_incoming_refs",
"Read incoming reference facts for a batch of string identities from this session. Preserve per-target failures and unknown reference kinds. Reuse the context ref and lease returned by find_strings.",
z.strictObject({
context: contextSchema,
targets: z.array(stringId),
...paging,
}),
"incomingReferences",
],
[
"experiment_resolve_functions",
"Resolve a batch of instruction locations to provider-analyzed containing functions. Unsupported or failed resolution stays unknown; never guess the nearest symbol.",
z.strictObject({ context: contextSchema, locations: z.array(location) }),
"resolveFunctions",
],
[
"experiment_trace_string_uses",
"Compose the same string, incoming reference and containing-function facts in one call. Page limits and partial outcomes remain explicit.",
stringsInput.extend({
reference_limit: z.number().int().positive().default(20),
include_functions: z.boolean().default(true),
}),
"traceStringUses",
],
];
const encode = (value) => ({
content: [{ type: "text", text: JSON.stringify(value) }],
structuredContent: value,
...(value.status === "error" ? { isError: true } : {}),
});
const createFacts = (session, analysis, signal) => {
const status = session.status();
const context = contextFor(
session.activeTarget().sha256,
session.analysisProfile("search_strings"),
status.analysis_run.run_id,
);
const capabilities = expandProviderCapabilities(status.capabilities);
const facts = new FactQueries({
supports: (operation) =>
capabilities.some(
(capability) =>
capability.operation === operation && capability.available,
),
context: () => context,
observe: async (operation, parameters) => {
const outcome = await analysis.execute(
operation,
toolContract(operation).inputSchema.parse(parameters),
{ signal: signal },
);
if (!outcome.ok)
return {
ok: false,
error: projectAnalysisError(outcome.error),
};
const execution = outcome.value;
const evidence = createEvidence(
execution.subject ?? session.activeTarget(),
execution.provider,
{
operation,
parameters,
result: execution.result,
analysisProfile: execution.analysisProfile,
rawResult: execution.rawResult,
limitations: execution.limitations,
relationshipCoverage: execution.relationshipCoverage,
locations: execution.locations,
},
);
const recorded = session.recordEvidence(evidence);
return recorded.ok
? { ok: true, value: evidence }
: {
ok: false,
error: projectAnalysisError(recorded.error),
};
},
});
return facts;
};
await run({
env: process.env,
serve: serveStdio,
writeStderr: (value) => process.stderr.write(value),
setExitCode: (value) => {
process.exitCode = value;
},
registerShutdown: (handler) => {
const registrations = [
[process, "SIGINT"],
[process, "SIGTERM"],
[process.stdin, "end"],
[process.stdin, "close"],
];
for (const [emitter, event] of registrations) emitter.once(event, handler);
return () => {
for (const [emitter, event] of registrations) emitter.off(event, handler);
};
},
createServer(source, options) {
const server = createServer(source, options);
const session = source.session;
for (const [name, description, inputSchema, method] of contracts) {
server.registerTool(
name,
{ description, inputSchema, annotations: { readOnlyHint: true } },
async (input, request) => {
const admitted = await session.withAdmittedAnalysis(
name,
request.mcpReq.signal,
async (analysis) => {
const facts = createFacts(
session,
analysis,
request.mcpReq.signal,
);
return facts[method](inputSchema.parse(input));
},
);
return encode(
admitted.ok
? admitted.value
: {
status: "error",
error: projectAnalysisError(admitted.error),
},
);
},
);
}
server.registerTool(
"experiment_read_evidence",
{
description:
"Read complete retained source Evidence explicitly by evidence_id while the existing session record owner retains it. Closing the binary expires these records.",
inputSchema: z.strictObject({ evidence_id: z.string().min(1) }),
annotations: { readOnlyHint: true },
},
(input) => {
const evidence = session.evidenceById(input.evidence_id);
return encode(
evidence === undefined
? {
status: "error",
error: {
code: "not_found",
message: "Evidence is not retained in this server.",
},
}
: { status: "ok", evidence },
);
},
);
return server;
},
});
+74
View File
@@ -0,0 +1,74 @@
"""Compare completed equivalent tasks; token counts are offline estimates only."""
import argparse
import importlib.metadata
import json
from pathlib import Path
import tiktoken
def main():
parser = argparse.ArgumentParser()
parser.add_argument("roots", nargs="+")
options = parser.parse_args()
encoding = tiktoken.get_encoding("o200k_base")
def tokens(value):
return len(encoding.encode(json.dumps(value, ensure_ascii=False, separators=(",", ":")), disallowed_special=()))
results = []
equivalence = {}
for root in map(Path, options.roots):
for report_path in sorted(root.glob("*.report.json")):
report = json.loads(report_path.read_text())
if report["failures"] or report["verification_only"]:
continue
prefix = str(report_path).removesuffix(".report.json")
catalog = json.loads(Path(prefix + ".catalog.json").read_text())
wire = json.loads(Path(prefix + ".wire.json").read_text())
resources = [json.loads(line) for line in Path(prefix + ".log").read_text().splitlines()
if line.startswith('{"resource_scope":')][-1]
assert resources["payload_resource_observation_verified"], prefix
counters = resources["payload_resource_observation"]
assert counters["payload_exit_code"] == 0 and counters["payload_cleanup_verified"], prefix
assert counters["resource_events"]["memory.events"]["oom_kill"] == 0, prefix
key = (report["provider"], report["target_sha256"], report["pattern"])
outcome = [{name: row[name] for name in ["strings", "references", "functions"]} for row in report["outcomes"]]
if key in equivalence:
assert outcome == equivalence[key], "Equivalent-task answers differ: " + prefix
else:
equivalence[key] = outcome
phase_traffic = {phase: [{"request": row["request"], "response": row["response"]} for row in wire if row["phase"] == phase]
for phase in ["cold", "warm", "lifecycle"]}
used = {row["request"]["name"] for row in wire}
input_only_catalog = {**catalog, "tools": [{key: value for key, value in tool.items() if key != "outputSchema"} for tool in catalog["tools"]]}
task_traffic = [row for row in wire if row["phase"] in ["cold", "warm", "lifecycle"]]
rows = {phase: {"calls": len(traffic), "estimated_tokens": tokens(traffic)} for phase, traffic in phase_traffic.items()}
results.append({
"provider": report["provider"], "variant": report["variant"], "target_sha256": report["target_sha256"],
"pattern": report["pattern"], "catalog_tools": len(catalog["tools"]), "catalog_estimated_tokens": tokens(catalog),
"observed_output_schema_count": sum("outputSchema" in tool for tool in catalog["tools"]),
"input_only_catalog_estimated_tokens": tokens(input_only_catalog),
"phases": rows,
"query_elapsed_ms": {row["phase"]: round(row["elapsed_ms"], 2) for row in report["outcomes"]},
"session_elapsed_seconds_including_two_tasks": resources["elapsed_seconds"],
"session_cpu_seconds_including_two_tasks": round(counters["cpu_stat"]["usage_usec"] / 1e6, 3),
"peak_process_family_cgroup_mib": round(counters["memory_peak_bytes"] / 1024 ** 2, 1),
"whole_session_estimated_tokens": tokens(catalog) + tokens([{"request": row["request"], "response": row["response"]} for row in task_traffic]),
"whole_session_with_input_only_catalog_estimated_tokens": tokens(input_only_catalog) + tokens([{"request": row["request"], "response": row["response"]} for row in task_traffic]),
"hypothetical_selected_declarations_estimated_tokens": tokens([tool for tool in catalog["tools"] if tool["name"] in used]),
"answer": {"strings": len(outcome[0]["strings"]), "reference_sites": sum(len(row["items"]) for row in outcome[0]["references"]), "locations": len(outcome[0]["functions"])},
"cleanup_verified": True, "oom_kills": 0,
})
summary = {
"token_method": {"encoding": "o200k_base", "tiktoken_version": importlib.metadata.version("tiktoken"),
"estimated_not_billed": True, "scope": "Captured MCP tool catalog and request/response payloads, including text and structured copies; excludes JSON-RPC framing, model reasoning, host prompts and model tool selection."},
"memory_method": "Kernel memory.peak for the complete owned process-family cgroup, including charged file cache; not summed RSS. Payload-exit counters verified while the observer remains alive.",
"workload": "One cold and one warm selected navigation task per fresh session, plus startup, discovery/status, close and cleanup. Additional --verify runs excluded.",
"equivalence": "All included variants agree on selected strings, reference sites and per-location function resolutions within each provider/target/query. This is not cross-provider truth or a live-agent reasoning evaluation.",
"results": results,
}
print(json.dumps(summary, indent=2))
if __name__ == "__main__":
main()